address more comments from Anish

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

@@ -209,10 +209,12 @@ object StateSourceOptions extends DataSourceOptions {
       throw StateDataSourceErrors.invalidOptionValueIsNegative(SNAPSHOT_PARTITION_ID)
     }
 
+    // both snapshotPartitionId and snapshotStartBatchId are required at the same time, because
+    // each partition may have different checkpoint status
     if (snapshotPartitionId.isDefined && snapshotStartBatchId.isEmpty) {
-      throw StateDataSourceErrors.requiredOptionUnspecified(SNAPSHOT_START_BATCH_ID.toString)
+      throw StateDataSourceErrors.requiredOptionUnspecified(SNAPSHOT_START_BATCH_ID)
     } else if (snapshotPartitionId.isEmpty && snapshotStartBatchId.isDefined) {
-      throw StateDataSourceErrors.requiredOptionUnspecified(SNAPSHOT_PARTITION_ID.toString)
+      throw StateDataSourceErrors.requiredOptionUnspecified(SNAPSHOT_PARTITION_ID)
     }
 
     StateSourceOptions(

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

@@ -26,7 +26,7 @@ import org.apache.spark.sql.SparkSession
 import org.apache.spark.sql.connector.read.{Batch, InputPartition, PartitionReaderFactory, Scan, ScanBuilder}
 import org.apache.spark.sql.execution.datasources.v2.state.StateSourceOptions.JoinSideValues
 import org.apache.spark.sql.execution.streaming.StreamingSymmetricHashJoinHelper.{LeftSide, RightSide}
-import org.apache.spark.sql.execution.streaming.state.{StateStoreConf, StateStoreSnapshotPartitionNotFound}
+import org.apache.spark.sql.execution.streaming.state.{StateStoreConf, StateStoreErrors}
 import org.apache.spark.sql.types.StructType
 import org.apache.spark.util.SerializableConfiguration
 
@@ -82,14 +82,14 @@ class StateScan(
         s"No continuous partitions in state: ${partitionNums.mkString("Array(", ", ", ")")}")
 
       sourceOptions.snapshotPartitionId match {
-        case None => partitionNums.map {
-          pn => new StateStoreInputPartition(pn, queryId, sourceOptions)
+        case None => partitionNums.map { pn =>
+          new StateStoreInputPartition(pn, queryId, sourceOptions)
         }.toArray
         case Some(snapshotPartitionId) =>
           if (partitionNums.contains(snapshotPartitionId)) {
             Array(new StateStoreInputPartition(snapshotPartitionId, queryId, sourceOptions))
           } else {
-            throw new StateStoreSnapshotPartitionNotFound(
+            throw StateStoreErrors.stateStoreSnapshotPartitionNotFound(
               snapshotPartitionId, sourceOptions.operatorId,
               sourceOptions.stateCheckpointLocation.toString)
           }

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/HDFSBackedStateStoreProvider.scala

@@ -613,7 +613,8 @@ private[sql] class HDFSBackedStateStoreProvider extends StateStoreProvider with
           readSnapshotFile(startVersion)
       }
       if (startVersionMap.isEmpty) {
-        throw new StateStoreSnapshotFileNotFound(snapshotFile(startVersion).toString, toString())
+        throw StateStoreErrors.stateStoreSnapshotFileNotFound(
+          snapshotFile(startVersion).toString, toString())
       }
       synchronized { putStateIntoStateCacheMap(startVersion, startVersionMap.get) }
 
@@ -772,7 +773,7 @@ private[sql] class HDFSBackedStateStoreProvider extends StateStoreProvider with
   }
 
   /**
-  * try to read the snapshot file of the given version.
+   * try to read the snapshot file of the given version.
    * If the snapshot file is not available, return [[None]].
    *
    * @param version the version of the snapshot file

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateStoreErrors.scala

@@ -149,6 +149,17 @@ object StateStoreErrors {
       newValueSchema: String): StateStoreValueSchemaNotCompatible = {
     new StateStoreValueSchemaNotCompatible(storedValueSchema, newValueSchema)
   }
+
+  def stateStoreSnapshotFileNotFound(fileToRead: String, clazz: String):
+  StateStoreSnapshotFileNotFound = {
+    new StateStoreSnapshotFileNotFound(fileToRead, clazz)
+  }
+
+  def stateStoreSnapshotPartitionNotFound(
+    snapshotPartitionId: Long, operatorId: Int, checkpointLocation: String):
+  StateStoreSnapshotPartitionNotFound = {
+    new StateStoreSnapshotPartitionNotFound(snapshotPartitionId, operatorId, checkpointLocation)
+  }
 }
 
 class StateStoreMultipleColumnFamiliesNotSupportedException(stateStoreProvider: String)

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/SymmetricHashJoinStateManager.scala

@@ -494,8 +494,7 @@ class SymmetricHashJoinStateManager(
           useMultipleValuesPerKey = false)
         if (snapshotStartVersion.isDefined) {
           stateStoreProvider.getStore(snapshotStartVersion.get, stateInfo.get.storeVersion)
-        }
-        else {
+        } else {
           stateStoreProvider.getStore(stateInfo.get.storeVersion)
         }
       }

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

@@ -196,7 +196,7 @@ class StateDataSourceNegativeTestSuite extends StateDataSourceTestBase {
     }
   }
 
-  test("ERROR: snapshotStartBatchId specified to negative") {
+  test("ERROR: snapshotStartBatchId specified as a negative value") {
     withTempDir { tempDir =>
       val exc = intercept[StateDataSourceInvalidOptionValueIsNegative] {
         spark.read.format("statestore")
@@ -210,7 +210,7 @@ class StateDataSourceNegativeTestSuite extends StateDataSourceTestBase {
     }
   }
 
-  test("ERROR: snapshotPartitionId specified to negative") {
+  test("ERROR: snapshotPartitionId specified as a negative value") {
     withTempDir { tempDir =>
       val exc = intercept[StateDataSourceInvalidOptionValueIsNegative] {
         spark.read.format("statestore")
@@ -257,7 +257,6 @@ class StateDataSourceNegativeTestSuite extends StateDataSourceTestBase {
       val exc = intercept[StateDataSourceInvalidOptionValue] {
         spark.read.format("statestore")
           // trick to bypass getting the last committed batch before validating operator ID
-          .option(StateSourceOptions.BATCH_ID, 0)
           .option(StateSourceOptions.SNAPSHOT_START_BATCH_ID, startBatchId)
           .option(StateSourceOptions.BATCH_ID, endBatchId)
           .load(tempDir.getAbsolutePath)