solve conflict

eason-yuchen-liu · Jul 2, 2024 · 261c671 · 261c671
1 parent 39d0b17
commit 261c671
Showing 1 changed file with 8 additions and 6 deletions.
diff --git a/...main/scala/org/apache/spark/sql/execution/datasources/v2/state/StatePartitionReader.scala b/...main/scala/org/apache/spark/sql/execution/datasources/v2/state/StatePartitionReader.scala
@@ -42,10 +42,10 @@ class StatePartitionReaderFactory(
     val stateStoreInputPartition = partition.asInstanceOf[StateStoreInputPartition]
     if (stateStoreInputPartition.sourceOptions.readChangeFeed) {
       new StateStoreChangeDataPartitionReader(storeConf, hadoopConf,
-        stateStoreInputPartition, schema)
+        partition.asInstanceOf[StateStoreInputPartition], schema, stateStoreMetadata)
     } else {
       new StatePartitionReader(storeConf, hadoopConf,
-        stateStoreInputPartition, schema)
+        partition.asInstanceOf[StateStoreInputPartition], schema, stateStoreMetadata)
     }
   }
 }
@@ -146,10 +146,12 @@ class StatePartitionReader(
 }
 
 class StateStoreChangeDataPartitionReader(
-  storeConf: StateStoreConf,
-  hadoopConf: SerializableConfiguration,
-  partition: StateStoreInputPartition,
-  schema: StructType) extends StatePartitionReader(storeConf, hadoopConf, partition, schema) {
+    storeConf: StateStoreConf,
+    hadoopConf: SerializableConfiguration,
+    partition: StateStoreInputPartition,
+    schema: StructType,
+    stateStoreMetadata: Array[StateMetadataTableEntry])
+  extends StatePartitionReader(storeConf, hadoopConf, partition, schema, stateStoreMetadata) {
 
   private lazy val changeDataReader: StateStoreChangeDataReader = {
     if (!provider.isInstanceOf[SupportsFineGrainedReplay]) {