stateSchemaFilePath -> stateSchemaFilePaths

ericm-db · Dec 2, 2024 · f161db3 · f161db3
1 parent fe80545
commit f161db3
Show file tree

Hide file tree

Showing 4 changed files with 7 additions and 7 deletions.
diff --git a/...la/org/apache/spark/sql/execution/datasources/v2/state/metadata/StateMetadataSource.scala b/...la/org/apache/spark/sql/execution/datasources/v2/state/metadata/StateMetadataSource.scala
@@ -270,7 +270,7 @@ class StateMetadataPartitionReader(
               operatorStateMetadata.version,
               v2.operatorPropertiesJson,
               -1, // numColsPrefixKey is not available in OperatorStateMetadataV2
-              Some(stateStoreMetadata.stateSchemaFilePath(stateStoreMetadata.stateSchemaId))
+              Some(stateStoreMetadata.stateSchemaFilePaths(stateStoreMetadata.stateSchemaId))
             )
           }
         }

diff --git a/...core/src/main/scala/org/apache/spark/sql/execution/streaming/TransformWithStateExec.scala b/...core/src/main/scala/org/apache/spark/sql/execution/streaming/TransformWithStateExec.scala
@@ -475,7 +475,7 @@ case class TransformWithStateExec(
         metadata match {
           case v2: OperatorStateMetadataV2 =>
             val ssInfo = v2.stateStoreInfo.head
-            val stateSchemaFilePath = ssInfo.stateSchemaFilePath(ssInfo.stateSchemaId)
+            val stateSchemaFilePath = ssInfo.stateSchemaFilePaths(ssInfo.stateSchemaId)
             Some(new Path(stateSchemaFilePath))
           case _ => None
         }
@@ -498,7 +498,7 @@ case class TransformWithStateExec(
       val (oldSchemaId, oldSchemaPaths): (Int, Map[Int, String]) = oldMetadata match {
         case Some(v2: OperatorStateMetadataV2) =>
           val ssInfo = v2.stateStoreInfo.head
-          (ssInfo.stateSchemaId, ssInfo.stateSchemaFilePath)
+          (ssInfo.stateSchemaId, ssInfo.stateSchemaFilePaths)
         case _ => (-1, Map.empty)
       }
       List(oldSchemaPaths + (oldSchemaId + 1 -> validationResult.schemaPath))
@@ -507,7 +507,7 @@ case class TransformWithStateExec(
         case Some(v2: OperatorStateMetadataV2) =>
           // If schema hasn't evolved, keep existing mappings
           val ssInfo = v2.stateStoreInfo.head
-          List(ssInfo.stateSchemaFilePath)
+          List(ssInfo.stateSchemaFilePaths)
         case _ =>
           // If no previous metadata and no evolution, start with schema ID 0
           List(Map(0 -> validationResult.schemaPath))

diff --git a/...src/main/scala/org/apache/spark/sql/execution/streaming/state/OperatorStateMetadata.scala b/...src/main/scala/org/apache/spark/sql/execution/streaming/state/OperatorStateMetadata.scala
@@ -52,7 +52,7 @@ case class StateStoreMetadataV2(
     numColsPrefixKey: Int,
     numPartitions: Int,
     stateSchemaId: Int,
-    stateSchemaFilePath: Map[Int, String])
+    stateSchemaFilePaths: Map[Int, String])
   extends StateStoreMetadata with Serializable
 
 object StateStoreMetadataV2 {
@@ -470,7 +470,7 @@ class OperatorStateMetadataV2FileManager(
     val earliestBatchToKeep = latestMetadata match {
       case Some(OperatorStateMetadataV2(_, stateStoreInfo, _)) =>
         val ssInfo = stateStoreInfo.head
-        val schemaFilePath = ssInfo.stateSchemaFilePath.minBy(_._1)._2
+        val schemaFilePath = ssInfo.stateSchemaFilePaths.minBy(_._1)._2
         new Path(schemaFilePath).getName.split("_").head.toLong
       case _ => 0
     }

diff --git a/...est/scala/org/apache/spark/sql/execution/streaming/state/OperatorStateMetadataSuite.scala b/...est/scala/org/apache/spark/sql/execution/streaming/state/OperatorStateMetadataSuite.scala
@@ -62,7 +62,7 @@ class OperatorStateMetadataSuite extends StreamTest with SharedSparkSession {
       assert(operatorMetadataV2.operatorPropertiesJson.nonEmpty)
       val stateStoreInfo = operatorMetadataV2.stateStoreInfo.head
       val expectedStateStoreInfo = expectedMetadataV2.stateStoreInfo.head
-      assert(stateStoreInfo.stateSchemaFilePath.nonEmpty)
+      assert(stateStoreInfo.stateSchemaFilePaths.nonEmpty)
       assert(stateStoreInfo.storeName == expectedStateStoreInfo.storeName)
       assert(stateStoreInfo.numPartitions == expectedStateStoreInfo.numPartitions)
     }