[flink] Fix replace logic for reused row

yunfengzhou-hub · Dec 17, 2024 · b4e68f6 · b4e68f6
1 parent 1c10be1
commit b4e68f6
Showing 1 changed file with 3 additions and 1 deletion.
diff --git a/...imon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/ReadOperator.java b/...imon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/ReadOperator.java
@@ -91,7 +91,6 @@ public void open() throws Exception {
         this.read = readBuilder.newRead().withIOManager(ioManager);
         this.reuseRow = new FlinkRowData(null);
         if (projectionRowData != null) {
-            projectionRowData.replaceRow(this.reuseRow);
             this.reuseRecord = new StreamRecord<>(projectionRowData);
         } else {
             this.reuseRecord = new StreamRecord<>(reuseRow);
@@ -126,6 +125,9 @@ public void processElement(StreamRecord<Split> record) throws Exception {
                 }
 
                 reuseRow.replace(iterator.next());
+                if (projectionRowData != null) {
+                    projectionRowData.replaceRow(this.reuseRow);
+                }
                 output.collect(reuseRecord);
             }
         }