(paimon) branch master updated: [flink] Avoid overlapping sorted index key ranges (#9189)

[hope]

This is an automated email from the ASF dual-hosted git repository.

XiaoHongbo-Hope pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git


The following commit(s) were added to refs/heads/master by this push:
     new 83f7409c60 [flink] Avoid overlapping sorted index key ranges (#9189)
83f7409c60 is described below

commit 83f7409c609e1c6d91d17e28b004f6cb14805909
Author: YeJunHao <41894543+leaves12...@users.noreply.github.com>
AuthorDate: Wed Aug 12 17:58:34 2026 +0800

    [flink] Avoid overlapping sorted index key ranges (#9189)
    
    ## What changed
    
    - Add `_ROW_ID` as the final sort key for sorted global index builds.
    - Add a unit test that locks the sort-key order to build task ID, index
    key, and row ID.
    
    ## Why
    
    Range shuffle groups duplicate sampled boundaries and randomly assigns
    matching records among their range indexes. When the sort key contains
    only the build task ID and index key, a hot index key can therefore be
    spread across writer partitions in an order that produces overlapping
    output key ranges within the same row-range group.
    
    Using `_ROW_ID` as a tie-breaker makes the full shuffle key unique while
    keeping equal index keys adjacent. The actual index key is unchanged,
    and neighboring output files may still meet at the same index-key
    boundary without crossing each other.
    
    ## Impact
    
    New sorted global index builds preserve ordered file key ranges,
    preventing downstream BTree compaction from receiving overlapping inputs
    caused by duplicate range-shuffle boundaries.
    
    ## Validation
    
    - `mvn -pl paimon-flink/paimon-flink-common -Pflink1
    -DwildcardSuites=none -Dtest=SortedIndexTopoBuilderTest test` (8 tests
    passed)
    - `mvn -pl paimon-flink/paimon-flink-common -Pflink1 -DskipTests
    spotless:check`
    - `mvn -pl paimon-flink/paimon-flink-1.20 -am -Pflink1 -Pfast-build
    -DskipTests install -T 16`
    - `git diff --check`
---
 .../paimon/flink/globalindex/SortedIndexTopoBuilder.java       | 10 +++++++---
 .../paimon/flink/globalindex/SortedIndexTopoBuilderTest.java   |  7 +++++++
 2 files changed, 14 insertions(+), 3 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 775ad0722e..7e3154bdbc 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
@@ -184,9 +184,7 @@ public class SortedIndexTopoBuilder {
             // 4. Build one topology for all contiguous row ranges
             CoreOptions coreOptions = table.coreOptions();
             ReadBuilder readBuilder = 
table.newReadBuilder().withReadType(dataReadType);
-            List<String> sortColumns = new ArrayList<>();
-            sortColumns.add(buildTaskIdField);
-            sortColumns.add(indexColumn);
+            List<String> sortColumns = createSortColumns(buildTaskIdField, 
indexColumn);
             int partitionFieldSize = table.partitionKeys().size();
             BinaryRowSerializer binaryRowSerializer = new 
BinaryRowSerializer(partitionFieldSize);
             List<SortedBuildTask> buildTasks = new ArrayList<>();
@@ -373,6 +371,12 @@ public class SortedIndexTopoBuilder {
         return (int) Math.min(parallelism, maxParallelism);
     }
 
+    static List<String> createSortColumns(String buildTaskIdField, String 
indexColumn) {
+        // Range shuffle may spread duplicate boundary keys randomly. ROW_ID 
makes the full sort key
+        // unique while keeping equal index keys adjacent, so output file key 
ranges stay ordered.
+        return Arrays.asList(buildTaskIdField, indexColumn, 
SpecialFields.ROW_ID.name());
+    }
+
     private static String buildTaskIdFieldName(RowType readType) {
         String fieldName = BUILD_TASK_ID_FIELD;
         while (readType.containsField(fieldName)) {
diff --git 
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/globalindex/SortedIndexTopoBuilderTest.java
 
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/globalindex/SortedIndexTopoBuilderTest.java
index 6a33b2288c..4fc485bfd9 100644
--- 
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/globalindex/SortedIndexTopoBuilderTest.java
+++ 
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/globalindex/SortedIndexTopoBuilderTest.java
@@ -24,6 +24,7 @@ import 
org.apache.paimon.globalindex.sorted.SortedGlobalIndexScanner;
 import org.apache.paimon.globalindex.sorted.SortedGlobalIndexWriter;
 import org.apache.paimon.options.Options;
 import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.SpecialFields;
 import org.apache.paimon.types.DataTypes;
 import org.apache.paimon.utils.Range;
 
@@ -169,4 +170,10 @@ public class SortedIndexTopoBuilderTest {
 
         assertThat(SortedIndexTopoBuilder.calculateParallelism(tasks, 1000L, 
16)).isEqualTo(1);
     }
+
+    @Test
+    public void testSortColumnsUseRowIdAsTieBreaker() {
+        assertThat(SortedIndexTopoBuilder.createSortColumns("task-id", 
"index-key"))
+                .containsExactly("task-id", "index-key", 
SpecialFields.ROW_ID.name());
+    }
 }