rebase master

liming30 · Nov 14, 2023 · d5eabea · d5eabea
1 parent 05f46ac
commit d5eabea
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 2 deletions.
diff --git a/...nk-common/src/main/java/org/apache/paimon/flink/source/ContinuousFileSplitEnumerator.java b/...nk-common/src/main/java/org/apache/paimon/flink/source/ContinuousFileSplitEnumerator.java
@@ -71,7 +71,7 @@ public class ContinuousFileSplitEnumerator
 
     protected final SplitAssigner splitAssigner;
 
-    private final ConsumerProgressCalculator consumerProgressCalculator;
+    protected final ConsumerProgressCalculator consumerProgressCalculator;
 
     @Nullable protected Long nextSnapshotId;
 

diff --git a/...c/main/java/org/apache/paimon/flink/source/assigners/DynamicPartitionPruningAssigner.java b/...c/main/java/org/apache/paimon/flink/source/assigners/DynamicPartitionPruningAssigner.java
@@ -34,6 +34,7 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
+import java.util.Optional;
 import java.util.stream.Collectors;
 
 /** Assigner to perform dynamic partition pruning by given {@link DynamicFilteringData}. */
@@ -99,10 +100,15 @@ public static SplitAssigner createDynamicPartitionPruningAssignerIfNeeded(
                 dynamicFilteringData.isFiltering());
         return dynamicFilteringData.isFiltering()
                 ? new DynamicPartitionPruningAssigner(
-                        oriAssigner, partitionRowProjection, dynamicFilteringData)
+                oriAssigner, partitionRowProjection, dynamicFilteringData)
                 : oriAssigner;
     }
 
+    @Override
+    public Optional<Long> getNextSnapshotId(int subtask) {
+        return innerAssigner.getNextSnapshotId(subtask);
+    }
+
     private boolean filter(FileStoreSourceSplit sourceSplit) {
         DataSplit dataSplit = (DataSplit) sourceSplit.split();
         BinaryRow partition = dataSplit.partition();