This is an automated email from the ASF dual-hosted git repository. leaves12138 pushed a commit to branch safe_snapshot in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 51249c6f43218c7b5b6e71d515aab98f133a637e Author: 仟弋 <[email protected]> AuthorDate: Wed Sep 23 19:40:37 2026 +0800 Fix minus --- .../flink/globalindex/SortedIndexTopoBuilder.java | 31 +++++++++++----------- 1 file changed, 15 insertions(+), 16 deletions(-) diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/globalindex/SortedIndexTopoBuilder.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/globalindex/SortedIndexTopoBuilder.java index 3dd8ffde4c..7b77a32b49 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/globalindex/SortedIndexTopoBuilder.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/globalindex/SortedIndexTopoBuilder.java @@ -118,22 +118,6 @@ public class SortedIndexTopoBuilder { PartitionPredicate partitionPredicate, Options userOptions) throws Exception { - Long lastSafeSnapshotId = table.snapshotManager().latestSnapshotId(); - table = - table.copy( - - new HashMap<String, String>() { - { - put( - CoreOptions.COMMIT_LAST_SAFE_SNAPSHOT.key(), - Long.toString( - lastSafeSnapshotId == null ? 0 : lastSafeSnapshotId)); - put(CoreOptions.COMMIT_STRICT_MODE_ENABLED.key(), "false"); - } - - }); - - Optional<DataStream<Committable>> written = buildIndexStream( env, @@ -166,6 +150,21 @@ public class SortedIndexTopoBuilder { PartitionPredicate partitionPredicate, Options userOptions) throws Exception { + Long lastSafeSnapshotId = table.snapshotManager().latestSnapshotId(); + table = + table.copy( + new HashMap<String, String>() { + { + put( + CoreOptions.COMMIT_LAST_SAFE_SNAPSHOT.key(), + Long.toString( + lastSafeSnapshotId == null + ? 0 + : lastSafeSnapshotId)); + put(CoreOptions.COMMIT_STRICT_MODE_ENABLED.key(), "false"); + } + }); + List<DataStream<Committable>> allStreams = new ArrayList<>(); for (String indexColumn : indexColumns) { SortedGlobalIndexScanner indexScanner =
