reflect comments from Jungtaek

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/state/StateDataSource.scala

@@ -123,9 +123,9 @@ case class StateSourceOptions(
 
   override def toString: String = {
     s"StateSourceOptions(checkpointLocation=$resolvedCpLocation, batchId=$batchId, " +
-      s"operatorId=$operatorId, storeName=$storeName, joinSide=$joinSide" +
-      snapshotStartBatchId.map(", snapshotStartBatchId=" + _).getOrElse("") +
-      snapshotPartitionId.map(", snapshotPartitionId=" + _).getOrElse("") + ")"
+      s"operatorId=$operatorId, storeName=$storeName, joinSide=$joinSide, " +
+      s"snapshotStartBatchId=${snapshotStartBatchId.getOrElse("None")}, " +
+      s"snapshotPartitionId=${snapshotPartitionId.getOrElse("None")})"
   }
 }
 

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/state/StatePartitionReader.scala

@@ -95,6 +95,7 @@ class StatePartitionReader(
   private lazy val store: ReadStateStore = {
     partition.sourceOptions.snapshotStartBatchId match {
       case None => provider.getReadStore(partition.sourceOptions.batchId + 1)
+
       case Some(snapshotStartBatchId) => provider.getReadStore(
         snapshotStartBatchId + 1,
         partition.sourceOptions.batchId + 1)

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/state/StateScanBuilder.scala

@@ -85,6 +85,7 @@ class StateScan(
         case None => partitionNums.map { pn =>
           new StateStoreInputPartition(pn, queryId, sourceOptions)
         }.toArray
+
         case Some(snapshotPartitionId) =>
           if (partitionNums.contains(snapshotPartitionId)) {
             Array(new StateStoreInputPartition(snapshotPartitionId, queryId, sourceOptions))

sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/state/StateDataSourceTestBase.scala

@@ -383,7 +383,7 @@ trait StateDataSourceTestBase extends StreamTest with StateStoreMetricsTest {
     }
   }
 
-  protected def getStreamStreamJoinQuery(inputStream: MemoryStream[(Int, Long)]): DataFrame = {
+  private def getStreamStreamJoinQuery(inputStream: MemoryStream[(Int, Long)]): DataFrame = {
     val df = inputStream.toDS()
       .select(col("_1").as("value"), timestamp_seconds($"_2").as("timestamp"))