Github user zsxwing commented on a diff in the pull request:
https://github.com/apache/spark/pull/19611#discussion_r147784280
--- Diff:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/HDFSBackedStateStoreProvider.scala
---
@@ -297,17 +297,41 @@ private[state] class HDFSBackedStateStoreProvider
extends StateStoreProvider wit
/** Load the required version of the map data from the backing files */
private def loadMap(version: Long): MapType = {
- if (version <= 0) return new MapType
- synchronized { loadedMaps.get(version) }.getOrElse {
- val mapFromFile = readSnapshotFile(version).getOrElse {
- val prevMap = loadMap(version - 1)
- val newMap = new MapType(prevMap)
- updateFromDeltaFile(version, newMap)
- newMap
+
+ // Shortcut if the map for this version is already there to avoid a
redundant put.
+ val currentVersionMap =
+ synchronized { loadedMaps.get(version)
}.orElse(readSnapshotFile(version))
+ if (currentVersionMap.isDefined) {
+ return currentVersionMap.get
+ }
+
--- End diff --
nit: extra empty line
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
