Reflect more comments from Anish

common/utils/src/main/resources/error/error-conditions.json

@@ -229,7 +229,7 @@
           "Error reading snapshot file <fileToRead> of <clazz>: key size cannot be <keySize>."
         ]
       },
-      "CANNOT_READ_SNAPSHOT_FILE_NOT_EXISTS" : {
+      "CANNOT_READ_MISSING_SNAPSHOT_FILE" : {
         "message" : [
           "Error reading snapshot file <fileToRead> of <clazz>: <fileToRead> does not exist."
         ]

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

@@ -263,7 +263,7 @@ private[sql] class HDFSBackedStateStoreProvider extends StateStoreProvider with
 
   /**
    * Get the state store of endVersion for reading by applying delta files on the snapshot of
-   * startVersion. If startVersion does not exist, an error will be thrown.
+   * startVersion. If snapshot for startVersion does not exist, an error will be thrown.
    *
    * @param startVersion checkpoint version of the snapshot to start with
    * @param endVersion   checkpoint version to end with
@@ -287,7 +287,7 @@ private[sql] class HDFSBackedStateStoreProvider extends StateStoreProvider with
 
   /**
    * Get the state store of endVersion for reading by applying delta files on the snapshot of
-   * startVersion. If startVersion does not exist, an error will be thrown.
+   * startVersion. If snapshot for startVersion does not exist, an error will be thrown.
    *
    * @param startVersion checkpoint version of the snapshot to start with
    * @param endVersion   checkpoint version to end with
@@ -317,6 +317,12 @@ private[sql] class HDFSBackedStateStoreProvider extends StateStoreProvider with
     }
   }
 
+  /**
+   * Consturct the state at endVersion from snapshot from startVersion.
+   * Returns a new [[HDFSBackedStateStoreMap]]
+   * @param startVersion checkpoint version of the snapshot to start with
+   * @param endVersion   checkpoint version to end with
+   */
   private def getLoadedMapForStore(startVersion: Long, endVersion: Long):
     HDFSBackedStateStoreMap = synchronized {
     try {
@@ -598,9 +604,8 @@ private[sql] class HDFSBackedStateStoreProvider extends StateStoreProvider with
   }
 
   private def loadMap(startVersion: Long, endVersion: Long): HDFSBackedStateStoreMap = {
-
     val (result, elapsedMs) = Utils.timeTakenMs {
-      val startVersionMap = Option(loadedMaps.get(startVersion)) match {
+      val startVersionMap = synchronized { Option(loadedMaps.get(startVersion)) } match {
         case Some(value) =>
           loadedMapCacheHitCount.increment()
           Option(value)
@@ -773,8 +778,7 @@ private[sql] class HDFSBackedStateStoreProvider extends StateStoreProvider with
   }
 
   /**
-   * try to read the snapshot file of the given version.
-   * If the snapshot file is not available, return [[None]].
+   * Try to read the snapshot file. 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/RocksDB.scala

@@ -227,14 +227,14 @@ class RocksDB(
   }
 
   /**
-   * Load from the start checkpoint version and apply all the changelog records to reach the
-   * end version. Note that this will copy all the necessary file from DFS to local disk as needed,
+   * Load from the start snapshot version and apply all the changelog records to reach the
+   * end version. Note that this will copy all the necessary files from DFS to local disk as needed,
    * and possibly restart the native RocksDB instance.
    *
-   * @param startVersion
-   * @param endVersion
-   * @param readOnly
-   * @return
+   * @param startVersion version of the snapshot to start with
+   * @param endVersion end version
+   * @param readOnly whether the RocksDB instance is read-only
+   * @return A RocksDB instance loaded with the state endVersion replayed from startVersion
    */
   def load(startVersion: Long, endVersion: Long, readOnly: Boolean): RocksDB = {
     assert(startVersion >= 0 && endVersion >= startVersion)

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

@@ -151,13 +151,13 @@ object StateStoreErrors {
   }
 
   def stateStoreSnapshotFileNotFound(fileToRead: String, clazz: String):
-  StateStoreSnapshotFileNotFound = {
+    StateStoreSnapshotFileNotFound = {
     new StateStoreSnapshotFileNotFound(fileToRead, clazz)
   }
 
   def stateStoreSnapshotPartitionNotFound(
-    snapshotPartitionId: Long, operatorId: Int, checkpointLocation: String):
-  StateStoreSnapshotPartitionNotFound = {
+      snapshotPartitionId: Long, operatorId: Int, checkpointLocation: String):
+    StateStoreSnapshotPartitionNotFound = {
     new StateStoreSnapshotPartitionNotFound(snapshotPartitionId, operatorId, checkpointLocation)
   }
 }
@@ -268,7 +268,7 @@ class StateStoreValueSchemaNotCompatible(
 
 class StateStoreSnapshotFileNotFound(fileToRead: String, clazz: String)
   extends SparkUnsupportedOperationException(
-    errorClass = "CANNOT_LOAD_STATE_STORE.CANNOT_READ_SNAPSHOT_FILE_NOT_EXISTS",
+    errorClass = "CANNOT_LOAD_STATE_STORE.CANNOT_READ_MISSING_SNAPSHOT_FILE",
     messageParameters = Map(
       "fileToRead" -> fileToRead,
       "clazz" -> clazz))