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 cf381b9aeb [flink] Support grouped sorted index definitions in
topology builders (#10469)
cf381b9aeb is described below
commit cf381b9aeb3a6450ccc88b7bd590125955b68087
Author: YeJunHao <[email protected]>
AuthorDate: Fri Oct 9 17:45:49 2026 +0800
[flink] Support grouped sorted index definitions in topology builders
(#10469)
---
.../flink/globalindex/SortedIndexTopoBuilder.java | 52 +++++----------
.../procedure/CreateGlobalIndexProcedure.java | 3 +-
.../paimon/flink/SortedGlobalIndexITCase.java | 75 ++++++++++++++++++++++
.../globalindex/SortedIndexTopoBuilderTest.java | 4 +-
4 files changed, 94 insertions(+), 40 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 f40a224265..11900b0fad 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
@@ -110,11 +110,12 @@ public class SortedIndexTopoBuilder {
return SUPPORTED_INDEX_TYPES.contains(indexType);
}
+ /** Builds independent indexes, each defined by its ordered columns. */
public static boolean buildIndex(
StreamExecutionEnvironment env,
Supplier<SortedGlobalIndexScanner> indexScannerSupplier,
FileStoreTable table,
- List<String> indexColumns,
+ List<List<String>> definitions,
String indexType,
PartitionPredicate partitionPredicate,
Options userOptions)
@@ -124,7 +125,7 @@ public class SortedIndexTopoBuilder {
env,
indexScannerSupplier,
table,
- indexColumns,
+ definitions,
indexType,
partitionPredicate,
userOptions);
@@ -143,29 +144,6 @@ public class SortedIndexTopoBuilder {
* to one shared committer.
*/
public static Optional<DataStream<Committable>> buildIndexStream(
- StreamExecutionEnvironment env,
- Supplier<SortedGlobalIndexScanner> indexScannerSupplier,
- FileStoreTable table,
- List<String> indexColumns,
- String indexType,
- PartitionPredicate partitionPredicate,
- Options userOptions)
- throws Exception {
- List<List<String>> definitions = new ArrayList<>();
- for (String name : indexColumns) {
- definitions.add(Collections.singletonList(name));
- }
- return buildIndexDefinitions(
- env,
- indexScannerSupplier,
- table,
- definitions,
- indexType,
- partitionPredicate,
- userOptions);
- }
-
- private static Optional<DataStream<Committable>> buildIndexDefinitions(
StreamExecutionEnvironment env,
Supplier<SortedGlobalIndexScanner> indexScannerSupplier,
FileStoreTable table,
@@ -176,6 +154,9 @@ public class SortedIndexTopoBuilder {
throws Exception {
List<DataStream<Committable>> allStreams = new ArrayList<>();
for (List<String> indexColumns : definitions) {
+ if (indexColumns.isEmpty()) {
+ throw new IllegalArgumentException("At least one index column
is required.");
+ }
String indexColumn = indexColumns.get(0);
SortedGlobalIndexScanner indexScanner =
indexScannerSupplier.get().withIndexFields(indexColumns);
@@ -322,22 +303,19 @@ public class SortedIndexTopoBuilder {
public static void buildIndexAndExecute(
StreamExecutionEnvironment env,
FileStoreTable table,
- List<String> indexColumns,
+ List<List<String>> definitions,
String indexType,
PartitionPredicate partitionPredicate,
Options userOptions)
throws Exception {
- Optional<DataStream<Committable>> written =
- buildIndexDefinitions(
- env,
- () -> new SortedGlobalIndexScanner(table, indexType,
userOptions),
- table,
- Collections.singletonList(indexColumns),
- indexType,
- partitionPredicate,
- userOptions);
- if (written.isPresent()) {
- commit(table, written.get(),
CoreOptions.createCommitUser(userOptions));
+ if (buildIndex(
+ env,
+ () -> new SortedGlobalIndexScanner(table, indexType,
userOptions),
+ table,
+ definitions,
+ indexType,
+ partitionPredicate,
+ userOptions)) {
env.execute("Create " + indexType + " global index for table: " +
table.name());
}
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CreateGlobalIndexProcedure.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CreateGlobalIndexProcedure.java
index bb38de474f..6c40f09371 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CreateGlobalIndexProcedure.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CreateGlobalIndexProcedure.java
@@ -35,6 +35,7 @@ import org.apache.flink.table.annotation.ProcedureHint;
import org.apache.flink.table.procedure.ProcedureContext;
import java.util.Arrays;
+import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Locale;
@@ -135,7 +136,7 @@ public class CreateGlobalIndexProcedure extends
ProcedureBase {
SortedIndexTopoBuilder.buildIndexAndExecute(
procedureContext.getExecutionEnvironment(),
table,
- indexColumns,
+ Collections.singletonList(indexColumns),
indexType,
partitionPredicate,
userOptions);
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/SortedGlobalIndexITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/SortedGlobalIndexITCase.java
index a73f9b8dbe..05053eccb1 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/SortedGlobalIndexITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/SortedGlobalIndexITCase.java
@@ -22,12 +22,14 @@ import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.flink.globalindex.SortedIndexTopoBuilder;
import org.apache.paimon.globalindex.IndexedSplit;
import org.apache.paimon.globalindex.sorted.SortedGlobalIndexTestUtils;
import org.apache.paimon.index.DataEvolutionIndexSourceMeta;
import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.manifest.IndexManifestEntry;
+import org.apache.paimon.options.Options;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
import org.apache.paimon.reader.RecordReader;
@@ -66,6 +68,7 @@ import java.util.stream.IntStream;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.assertj.core.api.Assertions.entry;
/** Test case for sorted global indexes. */
public class SortedGlobalIndexITCase extends CatalogITCaseBase {
@@ -102,6 +105,78 @@ public class SortedGlobalIndexITCase extends
CatalogITCaseBase {
assertThat(sql("SELECT * FROM T WHERE id =
100")).containsOnly(Row.of(100, "name_100"));
}
+ @Test
+ public void testGroupedBTreeIndexes() throws Exception {
+ tEnv.getConfig().set(TableConfigOptions.TABLE_DML_SYNC, true);
+ sql(
+ "CREATE TABLE T_GROUPED (a INT, b INT, c INT, d INT, e INT)
WITH ("
+ + "'global-index.enabled' = 'true', "
+ + "'row-tracking.enabled' = 'true',
'data-evolution.enabled' = 'true', "
+ + "'sorted-index.records-per-file' = '3',
'sorted-index.build.max-parallelism' = '2')");
+ sql(
+ "INSERT INTO T_GROUPED VALUES "
+ + IntStream.range(0, 12)
+ .mapToObj(
+ i ->
+ String.format(
+ "(%d, %d, %d, %d, %s)",
+ i,
+ i % 4,
+ i % 3,
+ i % 5,
+ i == 7 ? "CAST(NULL AS
INT)" : i))
+ .collect(Collectors.joining(",")));
+ // Preserve composite key order and distinguish definitions sharing
their first column.
+ List<List<String>> definitions =
+ Arrays.asList(
+ Collections.singletonList("a"),
+ Arrays.asList("d", "b", "c"),
+ Collections.singletonList("e"),
+ Collections.singletonList("d"));
+ FileStoreTable table = paimonTable("T_GROUPED");
+ Options userOptions = new Options(table.options());
+ long snapshot = table.snapshotManager().latestSnapshot().id();
+ SortedIndexTopoBuilder.buildIndexAndExecute(
+
streamExecutionEnvironmentBuilder().batchMode().parallelism(2).build(),
+ table,
+ definitions,
+ "btree",
+ null,
+ userOptions);
+
assertThat(table.snapshotManager().latestSnapshot().id()).isEqualTo(snapshot +
1);
+ Map<List<Integer>, Long> rowCountsByFields =
+ indexEntries(table, "btree").stream()
+ .collect(
+ Collectors.groupingBy(
+ index ->
+ index.indexFile()
+ .globalIndexMeta()
+ .getIndexedFieldIds(),
+ Collectors.summingLong(
+ index ->
index.indexFile().rowCount())));
+ assertThat(rowCountsByFields)
+ .containsOnly(
+ entry(Collections.singletonList(0), 12L),
+ entry(Arrays.asList(3, 1, 2), 12L),
+ entry(Collections.singletonList(4), 12L),
+ entry(Collections.singletonList(3), 12L));
+ assertThat(sql("SELECT a FROM T_GROUPED WHERE a =
7")).containsExactly(Row.of(7));
+ assertThat(sql("SELECT a FROM T_GROUPED WHERE d = 2 AND b = 3 AND c =
1"))
+ .containsExactly(Row.of(7));
+ assertThat(sql("SELECT a FROM T_GROUPED WHERE e IS
NULL")).containsExactly(Row.of(7));
+
+ Set<String> indexedFiles = fileNames(indexEntries(table, "btree"));
+ SortedIndexTopoBuilder.buildIndexAndExecute(
+
streamExecutionEnvironmentBuilder().batchMode().parallelism(2).build(),
+ table,
+ definitions,
+ "btree",
+ null,
+ userOptions);
+
assertThat(table.snapshotManager().latestSnapshot().id()).isEqualTo(snapshot +
1);
+ assertThat(fileNames(indexEntries(table,
"btree"))).isEqualTo(indexedFiles);
+ }
+
@Test
public void testCompositeBTreeIndex() throws Exception {
tEnv.getConfig().set(TableConfigOptions.TABLE_DML_SYNC, true);
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 f5abf37b70..afcc6ec829 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
@@ -127,7 +127,7 @@ public class SortedIndexTopoBuilderTest {
env,
() -> indexScanner,
mock(FileStoreTable.class),
- Collections.singletonList("id"),
+
Collections.singletonList(Collections.singletonList("id")),
"btree",
null,
new Options()))
@@ -149,7 +149,7 @@ public class SortedIndexTopoBuilderTest {
env,
() -> indexScanner,
mock(FileStoreTable.class),
- Collections.singletonList("id"),
+
Collections.singletonList(Collections.singletonList("id")),
"btree",
null,
new Options()))
