(paimon) branch master updated: [flink] Expose sorted index build stream (#8998)

[lzljs3620320]

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

JingsongLi 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 4a468bb370 [flink] Expose sorted index build stream (#8998)
4a468bb370 is described below

commit 4a468bb370094b1974bee35f276726051abccadc
Author: YeJunHao <41894543+leaves12...@users.noreply.github.com>
AuthorDate: Mon Aug 3 21:14:32 2026 +0800

    [flink] Expose sorted index build stream (#8998)
---
 .../flink/globalindex/SortedIndexTopoBuilder.java  | 46 +++++++++++++++++-----
 .../globalindex/SortedIndexTopoBuilderTest.java    | 20 ++++++++++
 2 files changed, 57 insertions(+), 9 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 d9f63cf64a..a361687f38 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
@@ -101,6 +101,36 @@ public class SortedIndexTopoBuilder {
             PartitionPredicate partitionPredicate,
             Options userOptions)
             throws Exception {
+        Optional<DataStream<Committable>> written =
+                buildIndexStream(
+                        env,
+                        indexBuilderSupplier,
+                        table,
+                        indexColumns,
+                        partitionPredicate,
+                        userOptions);
+        if (!written.isPresent()) {
+            return false;
+        }
+
+        commit(table, written.get(), 
CoreOptions.createCommitUser(userOptions));
+        return true;
+    }
+
+    /**
+     * Builds sorted indexes and returns their committables without attaching 
a committer.
+     *
+     * <p>This allows callers which combine multiple maintenance topologies to 
send all committables
+     * to one shared committer.
+     */
+    public static Optional<DataStream<Committable>> buildIndexStream(
+            StreamExecutionEnvironment env,
+            Supplier<SortedGlobalIndexBuilder> indexBuilderSupplier,
+            FileStoreTable table,
+            List<String> indexColumns,
+            PartitionPredicate partitionPredicate,
+            Options userOptions)
+            throws Exception {
         List<DataStream<Committable>> allStreams = new ArrayList<>();
         for (String indexColumn : indexColumns) {
             SortedGlobalIndexBuilder indexBuilder =
@@ -175,7 +205,7 @@ public class SortedIndexTopoBuilder {
             }
 
             if (buildTasks.isEmpty()) {
-                return false;
+                return Optional.empty();
             }
 
             DataStream<Committable> commitMessages =
@@ -198,15 +228,13 @@ public class SortedIndexTopoBuilder {
 
             allStreams.add(commitMessages);
         }
-        if (!allStreams.isEmpty()) {
-            @SuppressWarnings("unchecked")
-            DataStream<Committable>[] rest =
-                    allStreams.subList(1, allStreams.size()).toArray(new 
DataStream[0]);
-            commit(table, allStreams.get(0).union(rest), 
CoreOptions.createCommitUser(userOptions));
-            return true;
+        if (allStreams.isEmpty()) {
+            return Optional.empty();
         }
-
-        return false;
+        @SuppressWarnings("unchecked")
+        DataStream<Committable>[] rest =
+                allStreams.subList(1, allStreams.size()).toArray(new 
DataStream[0]);
+        return Optional.of(allStreams.get(0).union(rest));
     }
 
     public static void buildIndexAndExecute(
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 5c47ce05ba..19b80e5bd5 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
@@ -117,6 +117,26 @@ public class SortedIndexTopoBuilderTest {
         verifyNoInteractions(env);
     }
 
+    @Test
+    public void testBuildIndexStreamReturnsEmptyWhenNoBuildTask() throws 
Exception {
+        SortedGlobalIndexBuilder indexBuilder = 
mock(SortedGlobalIndexBuilder.class);
+        when(indexBuilder.withIndexField("id")).thenReturn(indexBuilder);
+        when(indexBuilder.incrementalScan()).thenReturn(Optional.empty());
+        StreamExecutionEnvironment env = 
mock(StreamExecutionEnvironment.class);
+
+        assertThat(
+                        SortedIndexTopoBuilder.buildIndexStream(
+                                env,
+                                () -> indexBuilder,
+                                mock(FileStoreTable.class),
+                                Collections.singletonList("id"),
+                                null,
+                                new Options()))
+                .isEmpty();
+        verify(indexBuilder).incrementalScan();
+        verifyNoInteractions(env);
+    }
+
     @Test
     public void testCalculateParallelismByTotalRowsInsteadOfRangeCount() {
         List<SortedBuildTask> tasks = new ArrayList<>();