You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by ju...@apache.org on 2026/03/19 03:32:59 UTC
(paimon) branch master updated: [core] refactor flink create global index (#7445)
This is an automated email from the ASF dual-hosted git repository.
junhao 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 635170d08e [core] refactor flink create global index (#7445)
635170d08e is described below
commit 635170d08e479bc81964fda4ec4965739e94c2ad
Author: jerry <li...@alibaba-inc.com>
AuthorDate: Thu Mar 19 11:32:53 2026 +0800
[core] refactor flink create global index (#7445)
---
.../shortcodes/generated/core_configuration.html | 12 +
.../main/java/org/apache/paimon/CoreOptions.java | 18 +
.../globalindex/GlobalIndexBuilderUtils.java | 17 +
.../globalindex/GlobalIndexBuilderUtilsTest.java | 81 +++
.../globalindex/GenericGlobalIndexBuilder.java | 82 +++
.../flink/globalindex/GenericIndexTopoBuilder.java | 554 +++++++++++++++++++++
.../procedure/CreateGlobalIndexProcedure.java | 256 +---------
.../flink/LuminaVectorGlobalIndexITCase.java | 140 ++++++
.../globalindex/GenericIndexTopoBuilderTest.java | 256 ++++++++++
.../index/LuminaVectorGlobalIndexWriter.java | 89 ++--
.../globalindex/DefaultGlobalIndexBuilder.java | 33 +-
.../globalindex/DefaultGlobalIndexTopoBuilder.java | 18 +-
12 files changed, 1283 insertions(+), 273 deletions(-)
diff --git a/docs/layouts/shortcodes/generated/core_configuration.html b/docs/layouts/shortcodes/generated/core_configuration.html
index b36a45a606..5a654a97c9 100644
--- a/docs/layouts/shortcodes/generated/core_configuration.html
+++ b/docs/layouts/shortcodes/generated/core_configuration.html
@@ -638,6 +638,18 @@ under the License.
<td>Integer</td>
<td>For streaming write, full compaction will be constantly triggered after delta commits. For batch write, full compaction will be triggered with each commit as long as this value is greater than 0.</td>
</tr>
+ <tr>
+ <td><h5>global-index.build.max-parallelism</h5></td>
+ <td style="word-wrap: break-word;">4096</td>
+ <td>Integer</td>
+ <td>The max parallelism of Flink/Spark for building global index.</td>
+ </tr>
+ <tr>
+ <td><h5>global-index.build.max-shard</h5></td>
+ <td style="word-wrap: break-word;">32</td>
+ <td>Integer</td>
+ <td>The max number of shards for building global index. If the number of shards calculated by 'global-index.row-count-per-shard' exceeds this value, 'global-index.row-count-per-shard' will be recalculated as ceil(total-row-count / max-shard) to guarantee the shard count does not exceed max-shard.</td>
+ </tr>
<tr>
<td><h5>global-index.column-update-action</h5></td>
<td style="word-wrap: break-word;">THROW_ERROR</td>
diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
index 07f5fc620b..f93b85f218 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -2281,6 +2281,24 @@ public class CoreOptions implements Serializable {
.defaultValue(100000L)
.withDescription("Row count per shard for global index.");
+ public static final ConfigOption<Integer> GLOBAL_INDEX_BUILD_MAX_SHARD =
+ key("global-index.build.max-shard")
+ .intType()
+ .defaultValue(32)
+ .withDescription(
+ "The max number of shards for building global index. "
+ + "If the number of shards calculated by 'global-index.row-count-per-shard' "
+ + "exceeds this value, 'global-index.row-count-per-shard' will be "
+ + "recalculated as ceil(total-row-count / max-shard) to guarantee "
+ + "the shard count does not exceed max-shard.");
+
+ public static final ConfigOption<Integer> GLOBAL_INDEX_BUILD_MAX_PARALLELISM =
+ key("global-index.build.max-parallelism")
+ .intType()
+ .defaultValue(4096)
+ .withDescription(
+ "The max parallelism of Flink/Spark for building global index.");
+
public static final ConfigOption<Boolean> GLOBAL_INDEX_ENABLED =
key("global-index.enabled")
.booleanType()
diff --git a/paimon-core/src/main/java/org/apache/paimon/globalindex/GlobalIndexBuilderUtils.java b/paimon-core/src/main/java/org/apache/paimon/globalindex/GlobalIndexBuilderUtils.java
index 085423efa8..19ffc3bb40 100644
--- a/paimon-core/src/main/java/org/apache/paimon/globalindex/GlobalIndexBuilderUtils.java
+++ b/paimon-core/src/main/java/org/apache/paimon/globalindex/GlobalIndexBuilderUtils.java
@@ -71,6 +71,23 @@ public class GlobalIndexBuilderUtils {
return results;
}
+ /**
+ * Adjust rowsPerShard if the estimated shard count exceeds maxShard. Returns the adjusted
+ * rowsPerShard value.
+ */
+ public static long adjustRowsPerShard(long rowsPerShard, long totalRowCount, int maxShard) {
+ long estimatedShards = ceilDivision(totalRowCount, rowsPerShard);
+ if (estimatedShards > maxShard) {
+ return ceilDivision(totalRowCount, maxShard);
+ }
+ return rowsPerShard;
+ }
+
+ /** Integer ceiling division: returns ceil(a / b) for positive a and b. */
+ private static long ceilDivision(long a, long b) {
+ return (a + b - 1) / b;
+ }
+
public static GlobalIndexWriter createIndexWriter(
FileStoreTable table, String indexType, DataField indexField, Options options)
throws IOException {
diff --git a/paimon-core/src/test/java/org/apache/paimon/globalindex/GlobalIndexBuilderUtilsTest.java b/paimon-core/src/test/java/org/apache/paimon/globalindex/GlobalIndexBuilderUtilsTest.java
new file mode 100644
index 0000000000..612f41e0e9
--- /dev/null
+++ b/paimon-core/src/test/java/org/apache/paimon/globalindex/GlobalIndexBuilderUtilsTest.java
@@ -0,0 +1,81 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.globalindex;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link GlobalIndexBuilderUtils#adjustRowsPerShard}. */
+public class GlobalIndexBuilderUtilsTest {
+
+ @Test
+ void testAdjustRowsPerShardNoAdjustmentNeeded() {
+ // 1000 rows, 100 per shard = 10 shards, maxShard = 20 -> no adjustment
+ long result = GlobalIndexBuilderUtils.adjustRowsPerShard(100, 1000, 20);
+ assertThat(result).isEqualTo(100);
+ }
+
+ @Test
+ void testAdjustRowsPerShardExactMatch() {
+ // 1000 rows, 100 per shard = 10 shards, maxShard = 10 -> no adjustment
+ long result = GlobalIndexBuilderUtils.adjustRowsPerShard(100, 1000, 10);
+ assertThat(result).isEqualTo(100);
+ }
+
+ @Test
+ void testAdjustRowsPerShardExceedsMaxShard() {
+ // 1000 rows, 100 per shard = 10 shards, maxShard = 3 -> adjust to ceil(1000/3) = 334
+ long result = GlobalIndexBuilderUtils.adjustRowsPerShard(100, 1000, 3);
+ assertThat(result).isEqualTo(334);
+ // Verify: ceil(1000/334) = 3 shards
+ assertThat((1000 + result - 1) / result).isEqualTo(3);
+ }
+
+ @Test
+ void testAdjustRowsPerShardMaxShardOne() {
+ // 1000 rows, 100 per shard = 10 shards, maxShard = 1 -> all in one shard
+ long result = GlobalIndexBuilderUtils.adjustRowsPerShard(100, 1000, 1);
+ assertThat(result).isEqualTo(1000);
+ assertThat((1000 + result - 1) / result).isEqualTo(1);
+ }
+
+ @Test
+ void testAdjustRowsPerShardEvenDivision() {
+ // 1000 rows, 100 per shard = 10 shards, maxShard = 5 -> adjust to 200
+ long result = GlobalIndexBuilderUtils.adjustRowsPerShard(100, 1000, 5);
+ assertThat(result).isEqualTo(200);
+ assertThat((1000 + result - 1) / result).isEqualTo(5);
+ }
+
+ @Test
+ void testAdjustRowsPerShardLargeRowCount() {
+ // 10M rows, 100K per shard = 100 shards, maxShard = 10 -> adjust to 1M
+ long result = GlobalIndexBuilderUtils.adjustRowsPerShard(100000, 10000000, 10);
+ assertThat(result).isEqualTo(1000000);
+ assertThat((10000000 + result - 1) / result).isEqualTo(10);
+ }
+
+ @Test
+ void testAdjustRowsPerShardTotalRowsLessThanRowsPerShard() {
+ // 50 rows, 100 per shard = 1 shard, maxShard = 3 -> no adjustment
+ long result = GlobalIndexBuilderUtils.adjustRowsPerShard(100, 50, 3);
+ assertThat(result).isEqualTo(100);
+ }
+}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/globalindex/GenericGlobalIndexBuilder.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/globalindex/GenericGlobalIndexBuilder.java
new file mode 100644
index 0000000000..48ea935f5b
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/globalindex/GenericGlobalIndexBuilder.java
@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.globalindex;
+
+import org.apache.paimon.manifest.IndexManifestEntry;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.partition.PartitionPredicate;
+import org.apache.paimon.table.FileStoreTable;
+
+import javax.annotation.Nullable;
+
+import java.io.Serializable;
+import java.util.Collections;
+import java.util.List;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** Builder for generic (non-btree) global index. */
+public class GenericGlobalIndexBuilder implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ protected final FileStoreTable table;
+
+ @Nullable protected PartitionPredicate partitionPredicate;
+
+ public GenericGlobalIndexBuilder(FileStoreTable table) {
+ this.table = table;
+ }
+
+ public GenericGlobalIndexBuilder withPartitionPredicate(PartitionPredicate partitionPredicate) {
+ this.partitionPredicate = partitionPredicate;
+ return this;
+ }
+
+ public FileStoreTable table() {
+ return table;
+ }
+
+ /**
+ * Scans manifest entries to determine which files need to be indexed.
+ *
+ * @return manifest entries to build index from
+ */
+ public List<ManifestEntry> scan() {
+ checkArgument(
+ table.coreOptions().bucket() == -1,
+ "Generic global index only supports unaware-bucket tables (bucket = -1), "
+ + "but table '%s' has bucket = %d.",
+ table.name(),
+ table.coreOptions().bucket());
+ checkArgument(
+ !table.coreOptions().deletionVectorsEnabled(),
+ "Generic global index does not support tables with deletion vectors enabled. "
+ + "Table '%s' has 'deletion-vectors.enabled' = true, which may cause "
+ + "deleted rows to be indexed.",
+ table.name());
+
+ return table.store().newScan().withPartitionFilter(partitionPredicate).plan().files();
+ }
+
+ /** Returns old index file entries that should be deleted after new indexes are built. */
+ public List<IndexManifestEntry> deletedIndexEntries() {
+ return Collections.emptyList();
+ }
+}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/globalindex/GenericIndexTopoBuilder.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/globalindex/GenericIndexTopoBuilder.java
new file mode 100644
index 0000000000..44239b3876
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/globalindex/GenericIndexTopoBuilder.java
@@ -0,0 +1,554 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.globalindex;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.flink.sink.Committable;
+import org.apache.paimon.flink.sink.CommittableTypeInfo;
+import org.apache.paimon.flink.sink.CommitterOperatorFactory;
+import org.apache.paimon.flink.sink.NoopCommittableStateManager;
+import org.apache.paimon.flink.sink.StoreCommitter;
+import org.apache.paimon.flink.utils.BoundedOneInputOperator;
+import org.apache.paimon.flink.utils.JavaTypeInfo;
+import org.apache.paimon.globalindex.GlobalIndexBuilderUtils;
+import org.apache.paimon.globalindex.GlobalIndexSingletonWriter;
+import org.apache.paimon.globalindex.ResultEntry;
+import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.manifest.IndexManifestEntry;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.partition.PartitionPredicate;
+import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.SpecialFields;
+import org.apache.paimon.table.sink.BatchWriteBuilder;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.CommitMessageImpl;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.ReadBuilder;
+import org.apache.paimon.table.source.TableRead;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.CloseableIterator;
+import org.apache.paimon.utils.Range;
+
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperatorFactory;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+
+import static org.apache.paimon.globalindex.GlobalIndexBuilderUtils.createIndexWriter;
+import static org.apache.paimon.globalindex.GlobalIndexBuilderUtils.toIndexFileMetas;
+import static org.apache.paimon.io.CompactIncrement.emptyIncrement;
+import static org.apache.paimon.io.DataIncrement.deleteIndexIncrement;
+import static org.apache.paimon.io.DataIncrement.indexIncrement;
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/**
+ * Builds a Flink topology for creating generic (non-btree) global indexes with parallelism. Each
+ * shard becomes an independent Flink task. Files are assigned to shards at file level, so each
+ * shard only contains the files whose row ID ranges overlap with its shard range. When a file spans
+ * multiple shard boundaries, it is included in each overlapping shard and rows outside the shard's
+ * range are filtered during reading.
+ */
+public class GenericIndexTopoBuilder {
+
+ private static final Logger LOG = LoggerFactory.getLogger(GenericIndexTopoBuilder.class);
+
+ public static void buildIndexAndExecute(
+ StreamExecutionEnvironment env,
+ FileStoreTable table,
+ String indexColumn,
+ String indexType,
+ PartitionPredicate partitionPredicate,
+ Options userOptions)
+ throws Exception {
+ boolean hasIndexToBuild =
+ buildIndex(
+ env,
+ () -> new GenericGlobalIndexBuilder(table),
+ table,
+ indexColumn,
+ indexType,
+ partitionPredicate,
+ userOptions);
+ if (hasIndexToBuild) {
+ env.execute("Create " + indexType + " global index for table: " + table.name());
+ } else {
+ LOG.info("No index to build, nothing to do.");
+ }
+ }
+
+ /**
+ * Builds a generic global index topology using a {@link GenericGlobalIndexBuilder} supplier.
+ *
+ * @return {@code true} if a Flink topology was built and is ready to execute, {@code false} if
+ * there was nothing to index
+ */
+ public static boolean buildIndex(
+ StreamExecutionEnvironment env,
+ Supplier<GenericGlobalIndexBuilder> indexBuilderSupplier,
+ FileStoreTable table,
+ String indexColumn,
+ String indexType,
+ PartitionPredicate partitionPredicate,
+ Options userOptions)
+ throws Exception {
+ GenericGlobalIndexBuilder indexBuilder = indexBuilderSupplier.get();
+ if (partitionPredicate != null) {
+ indexBuilder.withPartitionPredicate(partitionPredicate);
+ }
+
+ List<ManifestEntry> entries = indexBuilder.scan();
+ List<IndexManifestEntry> deletedIndexEntries = indexBuilder.deletedIndexEntries();
+ long totalRowCount = entries.stream().mapToLong(e -> e.file().rowCount()).sum();
+ LOG.info(
+ "Scanned {} files ({} rows) across {} partitions for {} index on column '{}'.",
+ entries.size(),
+ totalRowCount,
+ entries.stream().map(ManifestEntry::partition).distinct().count(),
+ indexType,
+ indexColumn);
+
+ RowType rowType = table.rowType();
+ DataField indexField = rowType.getField(indexColumn);
+ // Project indexColumn + _ROW_ID so we can read the actual row ID from data
+ List<String> readColumns = new ArrayList<>();
+ readColumns.add(indexColumn);
+ readColumns.add(SpecialFields.ROW_ID.name());
+ RowType projectedRowType = SpecialFields.rowTypeWithRowId(rowType).project(readColumns);
+
+ Options mergedOptions = new Options(table.options(), userOptions.toMap());
+
+ long rowsPerShard = mergedOptions.get(CoreOptions.GLOBAL_INDEX_ROW_COUNT_PER_SHARD);
+ checkArgument(
+ rowsPerShard > 0,
+ "Option 'global-index.row-count-per-shard' must be greater than 0.");
+
+ int maxShard = mergedOptions.get(CoreOptions.GLOBAL_INDEX_BUILD_MAX_SHARD);
+ checkArgument(
+ maxShard > 0, "Option 'global-index.build.max-shard' must be greater than 0.");
+ rowsPerShard =
+ GlobalIndexBuilderUtils.adjustRowsPerShard(rowsPerShard, totalRowCount, maxShard);
+
+ // Compute shard tasks at file level from the provided entries
+ List<ShardTask> shardTasks = computeShardTasks(table, entries, rowsPerShard);
+ if (shardTasks.isEmpty()) {
+ LOG.info("No shard tasks generated, nothing to index.");
+ return false;
+ }
+ LOG.info("Generated {} shard tasks with rowsPerShard={}.", shardTasks.size(), rowsPerShard);
+
+ int maxParallelism =
+ mergedOptions
+ .getOptional(CoreOptions.GLOBAL_INDEX_BUILD_MAX_PARALLELISM)
+ .orElse(CoreOptions.GLOBAL_INDEX_BUILD_MAX_PARALLELISM.defaultValue());
+ checkArgument(
+ maxParallelism > 0,
+ "Option 'global-index.build.max-parallelism' must be greater than 0.");
+ int parallelism = Math.min(shardTasks.size(), maxParallelism);
+
+ // Build Flink topology
+ ReadBuilder readBuilder = table.newReadBuilder().withReadType(projectedRowType);
+
+ DataStream<ShardTask> source =
+ env.fromData(
+ new JavaTypeInfo<>(ShardTask.class),
+ shardTasks.toArray(new ShardTask[0]))
+ .name("Generic Index Source")
+ .setParallelism(1);
+
+ DataStream<Committable> built =
+ source.transform(
+ "build-" + indexType + "-index",
+ new CommittableTypeInfo(),
+ new BuildIndexOperator(
+ readBuilder,
+ table,
+ indexType,
+ indexField,
+ projectedRowType,
+ mergedOptions))
+ .setParallelism(parallelism);
+
+ if (!deletedIndexEntries.isEmpty()) {
+ List<Committable> deleteCommittables = createDeleteCommittables(deletedIndexEntries);
+ DataStream<Committable> deletes =
+ env.fromData(
+ new CommittableTypeInfo(),
+ deleteCommittables.toArray(new Committable[0]))
+ .name("Index Delete Source")
+ .setParallelism(1);
+ built = built.union(deletes);
+ }
+
+ commit(table, indexType, built);
+ return true;
+ }
+
+ /**
+ * Compute shard tasks at file level from the given manifest entries. Each shard only contains
+ * the files whose row ID ranges overlap with its shard range. A file spanning multiple shard
+ * boundaries is included in each overlapping shard.
+ */
+ static List<ShardTask> computeShardTasks(
+ FileStoreTable table, List<ManifestEntry> entries, long rowsPerShard) {
+ // Group by partition (bucket is always 0 for unaware-bucket tables)
+ Map<BinaryRow, List<ManifestEntry>> entriesByPartition =
+ entries.stream().collect(Collectors.groupingBy(ManifestEntry::partition));
+
+ List<ShardTask> tasks = new ArrayList<>();
+
+ for (Map.Entry<BinaryRow, List<ManifestEntry>> partitionEntry :
+ entriesByPartition.entrySet()) {
+ BinaryRow partition = partitionEntry.getKey();
+ String partBucketPath = table.store().pathFactory().bucketPath(partition, 0).toString();
+
+ // Assign files to shards by row ID range
+ Map<Long, List<DataFileMeta>> filesByShard = new LinkedHashMap<>();
+ for (ManifestEntry entry : partitionEntry.getValue()) {
+ DataFileMeta file = entry.file();
+ if (file.firstRowId() == null) {
+ LOG.warn(
+ "Skipping file '{}' in partition {} because it has no row ID. "
+ + "This file will NOT be indexed. "
+ + "Ensure row tracking is enabled for the table.",
+ file.fileName(),
+ partition);
+ continue;
+ }
+ Range fileRange = file.nonNullRowIdRange();
+ long startShardId = fileRange.from / rowsPerShard;
+ long endShardId = fileRange.to / rowsPerShard;
+
+ for (long shardId = startShardId; shardId <= endShardId; shardId++) {
+ long shardStartRowId = shardId * rowsPerShard;
+ filesByShard.computeIfAbsent(shardStartRowId, k -> new ArrayList<>()).add(file);
+ }
+ }
+
+ // Create ShardTask for each shard group
+ for (Map.Entry<Long, List<DataFileMeta>> shardEntry : filesByShard.entrySet()) {
+ long shardStart = shardEntry.getKey();
+ long shardEnd = shardStart + rowsPerShard - 1;
+ List<DataFileMeta> shardFiles = shardEntry.getValue();
+ if (shardFiles.isEmpty()) {
+ continue;
+ }
+
+ shardFiles.sort(Comparator.comparingLong(DataFileMeta::nonNullFirstRowId));
+
+ // Group contiguous files; gaps produce separate tasks
+ List<DataFileMeta> currentGroup = new ArrayList<>();
+ long currentGroupEnd = -1;
+
+ for (DataFileMeta file : shardFiles) {
+ long fileStart = file.nonNullFirstRowId();
+ long fileEnd = fileStart + file.rowCount() - 1;
+
+ if (currentGroup.isEmpty()) {
+ currentGroup.add(file);
+ currentGroupEnd = fileEnd;
+ } else if (fileStart <= currentGroupEnd + 1) {
+ currentGroup.add(file);
+ currentGroupEnd = Math.max(currentGroupEnd, fileEnd);
+ } else {
+ tasks.add(
+ createShardTask(
+ currentGroup,
+ shardStart,
+ shardEnd,
+ partition,
+ partBucketPath));
+ currentGroup = new ArrayList<>();
+ currentGroup.add(file);
+ currentGroupEnd = fileEnd;
+ }
+ }
+ if (!currentGroup.isEmpty()) {
+ tasks.add(
+ createShardTask(
+ currentGroup, shardStart, shardEnd, partition, partBucketPath));
+ }
+ }
+ }
+ return tasks;
+ }
+
+ private static ShardTask createShardTask(
+ List<DataFileMeta> files,
+ long shardStart,
+ long shardEnd,
+ BinaryRow partition,
+ String bucketPath) {
+ long groupMinRowId = files.get(0).nonNullFirstRowId();
+ long groupMaxRowId =
+ files.stream().mapToLong(f -> f.nonNullRowIdRange().to).max().getAsLong();
+
+ // Clamp to shard boundaries
+ long rangeFrom = Math.max(groupMinRowId, shardStart);
+ long rangeTo = Math.min(groupMaxRowId, shardEnd);
+
+ DataSplit dataSplit =
+ DataSplit.builder()
+ .withPartition(partition)
+ .withBucket(0)
+ .withDataFiles(files)
+ .withBucketPath(bucketPath)
+ .rawConvertible(false)
+ .build();
+
+ return new ShardTask(dataSplit, new Range(rangeFrom, rangeTo));
+ }
+
+ private static List<Committable> createDeleteCommittables(
+ List<IndexManifestEntry> deletedEntries) {
+ List<Committable> committables = new ArrayList<>();
+ for (IndexManifestEntry entry : deletedEntries) {
+ CommitMessage msg =
+ new CommitMessageImpl(
+ entry.partition(),
+ entry.bucket(),
+ null,
+ deleteIndexIncrement(Collections.singletonList(entry.indexFile())),
+ emptyIncrement());
+ committables.add(new Committable(BatchWriteBuilder.COMMIT_IDENTIFIER, msg));
+ }
+ return committables;
+ }
+
+ private static void commit(
+ FileStoreTable table, String indexType, DataStream<Committable> written) {
+ OneInputStreamOperatorFactory<Committable, Committable> committerOperator =
+ new CommitterOperatorFactory<>(
+ false,
+ true,
+ "GenericIndexCommitter-" + indexType + "-" + UUID.randomUUID(),
+ context ->
+ new StoreCommitter(
+ table, table.newCommit(context.commitUser()), context),
+ new NoopCommittableStateManager());
+
+ written.transform("COMMIT OPERATOR", new CommittableTypeInfo(), committerOperator)
+ .setParallelism(1)
+ .setMaxParallelism(1);
+ }
+
+ /** Serializable descriptor for one shard's work. Each shard has its own DataSplit and Range. */
+ static class ShardTask implements Serializable {
+ private static final long serialVersionUID = 1L;
+
+ final DataSplit split;
+ final Range shardRange;
+
+ ShardTask(DataSplit split, Range shardRange) {
+ this.split = split;
+ this.shardRange = shardRange;
+ }
+ }
+
+ /**
+ * Operator that receives a {@link ShardTask}, reads data from its split, builds the index, and
+ * emits a {@link Committable}. Each shard's split contains only the files relevant to this
+ * shard, so no redundant I/O occurs.
+ */
+ private static class BuildIndexOperator
+ extends BoundedOneInputOperator<ShardTask, Committable> {
+
+ private static final long serialVersionUID = 1L;
+
+ private final ReadBuilder readBuilder;
+ private final FileStoreTable table;
+ private final String indexType;
+ private final DataField indexField;
+ private final RowType projectedRowType;
+ private final Options mergedOptions;
+
+ private transient TableRead tableRead;
+ private transient InternalRow.FieldGetter indexFieldGetter;
+ private transient int rowIdFieldIndex;
+
+ BuildIndexOperator(
+ ReadBuilder readBuilder,
+ FileStoreTable table,
+ String indexType,
+ DataField indexField,
+ RowType projectedRowType,
+ Options mergedOptions) {
+ this.readBuilder = readBuilder;
+ this.table = table;
+ this.indexType = indexType;
+ this.indexField = indexField;
+ this.projectedRowType = projectedRowType;
+ this.mergedOptions = mergedOptions;
+ }
+
+ @Override
+ public void open() throws Exception {
+ super.open();
+ this.tableRead = readBuilder.newRead();
+ this.indexFieldGetter =
+ InternalRow.createFieldGetter(
+ indexField.type(), projectedRowType.getFieldIndex(indexField.name()));
+ this.rowIdFieldIndex = projectedRowType.getFieldIndex(SpecialFields.ROW_ID.name());
+ }
+
+ @Override
+ public void processElement(StreamRecord<ShardTask> element) throws Exception {
+ ShardTask task = element.getValue();
+ BinaryRow partition = task.split.partition();
+
+ LOG.info(
+ "Building {} index for partition={}, shardRange=[{}, {}], files={}.",
+ indexType,
+ partition,
+ task.shardRange.from,
+ task.shardRange.to,
+ task.split.dataFiles().size());
+ long startTime = System.currentTimeMillis();
+
+ GlobalIndexSingletonWriter indexWriter =
+ (GlobalIndexSingletonWriter)
+ createIndexWriter(table, indexType, indexField, mergedOptions);
+
+ try {
+ long rowsWritten = 0;
+ long lastRowId = Long.MIN_VALUE;
+ try (RecordReader<InternalRow> reader = tableRead.createReader(task.split);
+ CloseableIterator<InternalRow> iter = reader.toCloseableIterator()) {
+ while (iter.hasNext()) {
+ InternalRow row = iter.next();
+ long currentRowId = row.getLong(rowIdFieldIndex);
+
+ if (currentRowId < lastRowId) {
+ throw new IllegalStateException(
+ String.format(
+ "Row IDs are not monotonically increasing: "
+ + "previous=%d, current=%d in shard [%d, %d]. "
+ + "This may indicate corrupted data files.",
+ lastRowId,
+ currentRowId,
+ task.shardRange.from,
+ task.shardRange.to));
+ }
+ lastRowId = currentRowId;
+
+ if (currentRowId > task.shardRange.to) {
+ break;
+ }
+ // Only write rows within this shard's range
+ if (currentRowId >= task.shardRange.from) {
+ indexWriter.write(indexFieldGetter.getFieldOrNull(row));
+ rowsWritten++;
+ }
+ }
+ }
+
+ List<ResultEntry> resultEntries = indexWriter.finish();
+ long elapsed = System.currentTimeMillis() - startTime;
+ LOG.info(
+ "Finished shard [{}, {}]: wrote {} rows, "
+ + "produced {} result entries in {} ms.",
+ task.shardRange.from,
+ task.shardRange.to,
+ rowsWritten,
+ resultEntries.size(),
+ elapsed);
+
+ if (rowsWritten == 0) {
+ LOG.warn(
+ "Shard [{}, {}] produced 0 rows, skipping index flush.",
+ task.shardRange.from,
+ task.shardRange.to);
+ return;
+ }
+
+ CommitMessage commitMessage =
+ flushIndex(
+ table,
+ partition,
+ task.shardRange,
+ indexField,
+ indexType,
+ resultEntries);
+ output.collect(
+ new StreamRecord<>(
+ new Committable(
+ BatchWriteBuilder.COMMIT_IDENTIFIER, commitMessage)));
+ } catch (Exception e) {
+ throw e;
+ } finally {
+ closeWriterQuietly(indexWriter);
+ }
+ }
+
+ @Override
+ public void endInput() {
+ // All work is done in processElement
+ }
+ }
+
+ private static CommitMessage flushIndex(
+ FileStoreTable table,
+ BinaryRow partition,
+ Range rowRange,
+ DataField indexField,
+ String indexType,
+ List<ResultEntry> resultEntries)
+ throws IOException {
+ List<IndexFileMeta> indexFileMetas =
+ toIndexFileMetas(
+ table.fileIO(),
+ table.store().pathFactory().globalIndexFileFactory(),
+ table.coreOptions(),
+ rowRange,
+ indexField.id(),
+ indexType,
+ resultEntries);
+ return new CommitMessageImpl(
+ partition, 0, null, indexIncrement(indexFileMetas), emptyIncrement());
+ }
+
+ private static void closeWriterQuietly(GlobalIndexSingletonWriter writer) {
+ if (writer instanceof Closeable) {
+ try {
+ ((Closeable) writer).close();
+ } catch (IOException ignored) {
+ }
+ }
+ }
+}
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 664b69be4f..5f48555670 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
@@ -18,49 +18,23 @@
package org.apache.paimon.flink.procedure;
-import org.apache.paimon.CoreOptions;
-import org.apache.paimon.data.BinaryRow;
-import org.apache.paimon.data.InternalRow;
import org.apache.paimon.flink.btree.BTreeIndexTopoBuilder;
-import org.apache.paimon.globalindex.GlobalIndexSingletonWriter;
-import org.apache.paimon.globalindex.ResultEntry;
-import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.flink.globalindex.GenericIndexTopoBuilder;
import org.apache.paimon.options.Options;
import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.predicate.Predicate;
-import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.table.FileStoreTable;
-import org.apache.paimon.table.sink.CommitMessage;
-import org.apache.paimon.table.sink.CommitMessageImpl;
-import org.apache.paimon.table.sink.TableCommitImpl;
-import org.apache.paimon.table.source.DataSplit;
-import org.apache.paimon.table.source.ReadBuilder;
-import org.apache.paimon.table.source.snapshot.SnapshotReader;
-import org.apache.paimon.types.DataField;
import org.apache.paimon.types.RowType;
-import org.apache.paimon.utils.CloseableIterator;
import org.apache.paimon.utils.ParameterUtils;
-import org.apache.paimon.utils.Range;
import org.apache.flink.table.annotation.ArgumentHint;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.ProcedureHint;
import org.apache.flink.table.procedure.ProcedureContext;
-import java.io.Closeable;
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Collections;
import java.util.List;
import java.util.Map;
-import java.util.UUID;
-import static org.apache.paimon.globalindex.GlobalIndexBuilderUtils.createIndexWriter;
-import static org.apache.paimon.globalindex.GlobalIndexBuilderUtils.toIndexFileMetas;
-import static org.apache.paimon.globalindex.btree.BTreeGlobalIndexBuilder.groupSplitsByRange;
-import static org.apache.paimon.globalindex.btree.BTreeGlobalIndexBuilder.splitByContiguousRowRange;
-import static org.apache.paimon.io.CompactIncrement.emptyIncrement;
-import static org.apache.paimon.io.DataIncrement.indexIncrement;
import static org.apache.paimon.utils.ParameterUtils.getPartitions;
import static org.apache.paimon.utils.Preconditions.checkArgument;
@@ -118,212 +92,36 @@ public class CreateGlobalIndexProcedure extends ProcedureBase {
// Build global index based on index type
indexType = indexType.toLowerCase().trim();
- if ("btree".equals(indexType)) {
- BTreeIndexTopoBuilder.buildIndexAndExecute(
- procedureContext.getExecutionEnvironment(),
- table,
- indexColumn,
- partitionPredicate,
- userOptions);
- return new String[] {
- "BTree global index created successfully for table: " + table.name()
- };
- } else {
- buildGenericIndex(table, indexColumn, indexType, partitionPredicate, userOptions);
- return new String[] {
- indexType + " global index created successfully for table: " + table.name()
- };
- }
- }
-
- /**
- * Builds a global index using the generic SPI-based GlobalIndexer mechanism. This supports any
- * index type registered via {@link org.apache.paimon.globalindex.GlobalIndexerFactory} (e.g.
- * lumina-vector-ann, bitmap).
- */
- private void buildGenericIndex(
- FileStoreTable table,
- String indexColumn,
- String indexType,
- PartitionPredicate partitionPredicate,
- Options userOptions)
- throws Exception {
- RowType rowType = table.rowType();
- DataField indexField = rowType.getField(indexColumn);
- RowType projectedRowType = rowType.project(Collections.singletonList(indexColumn));
-
- // Merge table options with user options (user options take precedence)
- Options mergedOptions = new Options(table.options(), userOptions.toMap());
-
- long rowsPerShard =
- mergedOptions
- .getOptional(CoreOptions.GLOBAL_INDEX_ROW_COUNT_PER_SHARD)
- .orElse(CoreOptions.GLOBAL_INDEX_ROW_COUNT_PER_SHARD.defaultValue());
-
- // Use snapshot reader to get proper DataSplits (preserves bucket paths)
- SnapshotReader snapshotReader = table.newSnapshotReader();
- if (partitionPredicate != null) {
- snapshotReader = snapshotReader.withPartitionFilter(partitionPredicate);
- }
- List<DataSplit> splits = splitByContiguousRowRange(snapshotReader.read().dataSplits());
- if (splits.isEmpty()) {
- return;
- }
-
- // Group splits by partition and contiguous row range
- Map<BinaryRow, Map<Range, List<DataSplit>>> partitionRangeSplits =
- groupSplitsByRange(splits);
-
- ReadBuilder readBuilder = table.newReadBuilder().withReadType(projectedRowType);
- InternalRow.FieldGetter getter =
- InternalRow.createFieldGetter(
- indexField.type(), projectedRowType.getFieldIndex(indexField.name()));
-
- List<CommitMessage> allCommitMessages = new ArrayList<>();
-
- for (Map.Entry<BinaryRow, Map<Range, List<DataSplit>>> partEntry :
- partitionRangeSplits.entrySet()) {
- BinaryRow partition = partEntry.getKey();
- for (Map.Entry<Range, List<DataSplit>> rangeEntry : partEntry.getValue().entrySet()) {
- Range rowRange = rangeEntry.getKey();
- List<DataSplit> rangeSplits = rangeEntry.getValue();
-
- buildForRange(
- table,
- partition,
- rowRange,
- rangeSplits,
- readBuilder,
- getter,
- indexField,
- indexType,
- mergedOptions,
- rowsPerShard,
- allCommitMessages);
- }
- }
-
- // Commit all index files
- if (!allCommitMessages.isEmpty()) {
- try (TableCommitImpl commit =
- table.newCommit("global-index-create-" + UUID.randomUUID())) {
- commit.commit(allCommitMessages);
- }
- }
- }
-
- /**
- * Builds index files for a single partition + contiguous row range, sharding into multiple
- * index files when the row count exceeds {@code rowsPerShard}. Each shard receives its own
- * sub-range so the reader creates one reader per shard file and merges results via {@link
- * org.apache.paimon.globalindex.UnionGlobalIndexReader}.
- */
- private void buildForRange(
- FileStoreTable table,
- BinaryRow partition,
- Range rowRange,
- List<DataSplit> rangeSplits,
- ReadBuilder readBuilder,
- InternalRow.FieldGetter getter,
- DataField indexField,
- String indexType,
- Options mergedOptions,
- long rowsPerShard,
- List<CommitMessage> allCommitMessages)
- throws Exception {
- long totalWritten = 0;
- long shardCount = 0;
- GlobalIndexSingletonWriter currentWriter = null;
-
try {
- for (DataSplit split : rangeSplits) {
- try (RecordReader<InternalRow> reader = readBuilder.newRead().createReader(split);
- CloseableIterator<InternalRow> iter = reader.toCloseableIterator()) {
- while (iter.hasNext()) {
- InternalRow row = iter.next();
-
- // Shard: flush current writer when threshold reached
- if (currentWriter != null && shardCount >= rowsPerShard) {
- Range shardRange =
- new Range(
- rowRange.from + totalWritten - shardCount,
- rowRange.from + totalWritten - 1);
- allCommitMessages.add(
- flushIndex(
- table,
- partition,
- shardRange,
- indexField,
- indexType,
- currentWriter.finish()));
- currentWriter = null;
- shardCount = 0;
- }
-
- if (currentWriter == null) {
- currentWriter =
- (GlobalIndexSingletonWriter)
- createIndexWriter(
- table, indexType, indexField, mergedOptions);
- }
-
- currentWriter.write(getter.getFieldOrNull(row));
- shardCount++;
- totalWritten++;
- }
- }
- }
-
- // Flush remaining data
- if (currentWriter != null && shardCount > 0) {
- Range shardRange =
- new Range(
- rowRange.from + totalWritten - shardCount,
- rowRange.from + totalWritten - 1);
- allCommitMessages.add(
- flushIndex(
- table,
- partition,
- shardRange,
- indexField,
- indexType,
- currentWriter.finish()));
- currentWriter = null;
- }
- } finally {
- // Clean up writer on exception to prevent resource leaks
- closeWriterQuietly(currentWriter);
- }
- }
-
- private CommitMessage flushIndex(
- FileStoreTable table,
- BinaryRow partition,
- Range rowRange,
- DataField indexField,
- String indexType,
- List<ResultEntry> resultEntries)
- throws IOException {
- List<IndexFileMeta> indexFileMetas =
- toIndexFileMetas(
- table.fileIO(),
- table.store().pathFactory().globalIndexFileFactory(),
- table.coreOptions(),
- rowRange,
- indexField.id(),
+ if ("btree".equals(indexType)) {
+ BTreeIndexTopoBuilder.buildIndexAndExecute(
+ procedureContext.getExecutionEnvironment(),
+ table,
+ indexColumn,
+ partitionPredicate,
+ userOptions);
+ return new String[] {
+ "BTree global index created successfully for table: " + table.name()
+ };
+ } else {
+ GenericIndexTopoBuilder.buildIndexAndExecute(
+ procedureContext.getExecutionEnvironment(),
+ table,
+ indexColumn,
indexType,
- resultEntries);
- return new CommitMessageImpl(
- partition, 0, null, indexIncrement(indexFileMetas), emptyIncrement());
- }
-
- private static void closeWriterQuietly(GlobalIndexSingletonWriter writer) {
- if (writer instanceof Closeable) {
- try {
- ((Closeable) writer).close();
- } catch (IOException ignored) {
+ partitionPredicate,
+ userOptions);
}
+ } catch (Exception e) {
+ throw new RuntimeException(
+ String.format(
+ "Failed to create %s index for column '%s' on table '%s'.",
+ indexType, indexColumn, table.name()),
+ e);
}
+ return new String[] {
+ indexType + " global index created successfully for table: " + table.name()
+ };
}
private PartitionPredicate parsePartitionPredicate(FileStoreTable table, String partitions) {
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/LuminaVectorGlobalIndexITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/LuminaVectorGlobalIndexITCase.java
index 8e58fe87cf..aae3ca05f3 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/LuminaVectorGlobalIndexITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/LuminaVectorGlobalIndexITCase.java
@@ -19,10 +19,15 @@
package org.apache.paimon.flink;
import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.flink.globalindex.GenericGlobalIndexBuilder;
+import org.apache.paimon.flink.globalindex.GenericIndexTopoBuilder;
import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.manifest.IndexManifestEntry;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.options.Options;
import org.apache.paimon.table.FileStoreTable;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.junit.jupiter.api.Assumptions;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
@@ -359,6 +364,141 @@ public class LuminaVectorGlobalIndexITCase extends CatalogITCaseBase {
}
}
+ @Test
+ public void testCustomBuilderSupplier() throws Exception {
+ sql(
+ "CREATE TABLE T_CUSTOM (id INT, v ARRAY<FLOAT>) WITH ("
+ + "'bucket' = '-1', "
+ + "'row-tracking.enabled' = 'true', "
+ + "'data-evolution.enabled' = 'true', "
+ + "'lumina.index.dimension' = '3', "
+ + "'lumina.distance.metric' = 'l2'"
+ + ")");
+
+ // Two separate inserts create two data files with different firstRowId ranges
+ sql("INSERT INTO T_CUSTOM VALUES " + vectorValues(0, 50, 3));
+ sql("INSERT INTO T_CUSTOM VALUES " + vectorValues(50, 50, 3));
+
+ FileStoreTable table = paimonTable("T_CUSTOM");
+ Options mergedOptions = new Options(table.options());
+
+ // Use a custom builder that only indexes the first batch (firstRowId < 50)
+ StreamExecutionEnvironment env = streamExecutionEnvironmentBuilder().batchMode().build();
+ boolean hasIndex =
+ GenericIndexTopoBuilder.buildIndex(
+ env,
+ () ->
+ new GenericGlobalIndexBuilder(table) {
+ @Override
+ public List<ManifestEntry> scan() {
+ return super.scan().stream()
+ .filter(
+ e ->
+ e.file().firstRowId() != null
+ && e.file().firstRowId()
+ < 50)
+ .collect(Collectors.toList());
+ }
+ },
+ table,
+ "v",
+ INDEX_TYPE,
+ null,
+ mergedOptions);
+ assertThat(hasIndex).isTrue();
+ env.execute("test-partial-index");
+
+ List<IndexFileMeta> vectorEntries = getVectorIndexFiles(table);
+ assertThat(vectorEntries).isNotEmpty();
+ // Only the first batch (50 rows) should be indexed
+ long totalRowCount = vectorEntries.stream().mapToLong(IndexFileMeta::rowCount).sum();
+ assertThat(totalRowCount).isEqualTo(50L);
+ }
+
+ @Test
+ public void testDeletedIndexEntriesDefault() throws Catalog.TableNotExistException {
+ sql(
+ "CREATE TABLE T_DEL_DEFAULT (id INT, v ARRAY<FLOAT>) WITH ("
+ + "'bucket' = '-1', "
+ + "'row-tracking.enabled' = 'true', "
+ + "'data-evolution.enabled' = 'true', "
+ + "'lumina.index.dimension' = '3', "
+ + "'lumina.distance.metric' = 'l2'"
+ + ")");
+
+ FileStoreTable table = paimonTable("T_DEL_DEFAULT");
+ GenericGlobalIndexBuilder builder = new GenericGlobalIndexBuilder(table);
+ assertThat(builder.deletedIndexEntries()).isEmpty();
+ }
+
+ @Test
+ public void testDeletedIndexEntriesAtomicCommit() throws Exception {
+ sql(
+ "CREATE TABLE T_DEL_ATOMIC (id INT, v ARRAY<FLOAT>) WITH ("
+ + "'bucket' = '-1', "
+ + "'row-tracking.enabled' = 'true', "
+ + "'data-evolution.enabled' = 'true', "
+ + "'lumina.index.dimension' = '3', "
+ + "'lumina.distance.metric' = 'l2'"
+ + ")");
+
+ sql("INSERT INTO T_DEL_ATOMIC VALUES " + vectorValues(0, 100, 3));
+
+ // First build: create initial index
+ sql(
+ "CALL sys.create_global_index("
+ + "`table` => 'default.T_DEL_ATOMIC', "
+ + "index_column => 'v', "
+ + "index_type => '"
+ + INDEX_TYPE
+ + "')");
+
+ FileStoreTable table = paimonTable("T_DEL_ATOMIC");
+ List<IndexManifestEntry> oldEntries =
+ table.store().newIndexFileHandler().scanEntries().stream()
+ .filter(e -> INDEX_TYPE.equals(e.indexFile().indexType()))
+ .collect(Collectors.toList());
+ assertThat(oldEntries).isNotEmpty();
+
+ List<String> oldFileNames =
+ oldEntries.stream().map(e -> e.indexFile().fileName()).collect(Collectors.toList());
+
+ // Second build: use a custom builder that reports old entries as deletedIndexEntries
+ StreamExecutionEnvironment env = streamExecutionEnvironmentBuilder().batchMode().build();
+ boolean hasIndex =
+ GenericIndexTopoBuilder.buildIndex(
+ env,
+ () ->
+ new GenericGlobalIndexBuilder(table) {
+ @Override
+ public List<IndexManifestEntry> deletedIndexEntries() {
+ return oldEntries;
+ }
+ },
+ table,
+ "v",
+ INDEX_TYPE,
+ null,
+ new Options(table.options()));
+ assertThat(hasIndex).isTrue();
+ env.execute("test-rebuild-index");
+
+ // After the atomic commit, old index files should be deleted and new ones created
+ List<IndexManifestEntry> newEntries =
+ table.store().newIndexFileHandler().scanEntries().stream()
+ .filter(e -> INDEX_TYPE.equals(e.indexFile().indexType()))
+ .collect(Collectors.toList());
+
+ assertThat(newEntries).isNotEmpty();
+ long totalRowCount = newEntries.stream().mapToLong(e -> e.indexFile().rowCount()).sum();
+ assertThat(totalRowCount).isEqualTo(100L);
+
+ // New files should be different from old files (old ones were deleted)
+ List<String> newFileNames =
+ newEntries.stream().map(e -> e.indexFile().fileName()).collect(Collectors.toList());
+ assertThat(newFileNames).doesNotContainAnyElementsOf(oldFileNames);
+ }
+
// -- Helpers --
private List<IndexFileMeta> getVectorIndexFiles(FileStoreTable table) {
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/globalindex/GenericIndexTopoBuilderTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/globalindex/GenericIndexTopoBuilderTest.java
new file mode 100644
index 0000000000..8a01a8f7e5
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/globalindex/GenericIndexTopoBuilderTest.java
@@ -0,0 +1,256 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.globalindex;
+
+import org.apache.paimon.FileStore;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.BinaryRowWriter;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.io.PojoDataFileMeta;
+import org.apache.paimon.manifest.FileKind;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.stats.SimpleStats;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.FileStorePathFactory;
+import org.apache.paimon.utils.Range;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.UUID;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/** Tests for shard computation in {@link GenericIndexTopoBuilder}. */
+class GenericIndexTopoBuilderTest {
+
+ private FileStoreTable table;
+
+ @BeforeEach
+ void setUp() {
+ table = mock(FileStoreTable.class);
+ when(table.partitionKeys()).thenReturn(Collections.singletonList("pt"));
+
+ FileStore<?> store = mock(FileStore.class);
+ FileStorePathFactory pathFactory = mock(FileStorePathFactory.class);
+ org.mockito.Mockito.doReturn(store).when(table).store();
+ when(store.pathFactory()).thenReturn(pathFactory);
+ when(pathFactory.bucketPath(any(BinaryRow.class), eq(0)))
+ .thenReturn(new Path("/warehouse/table/bucket-0"));
+ }
+
+ @Test
+ void testSingleFileSingleShard() throws IOException {
+ // One file [0, 99] with rowsPerShard=100 → one shard [0, 99]
+ List<ManifestEntry> entries = new ArrayList<>();
+ entries.add(createEntry(BinaryRow.EMPTY_ROW, 0L, 100));
+
+ List<GenericIndexTopoBuilder.ShardTask> tasks =
+ GenericIndexTopoBuilder.computeShardTasks(table, entries, 100);
+
+ assertThat(tasks).hasSize(1);
+ assertThat(tasks.get(0).shardRange).isEqualTo(new Range(0, 99));
+ assertThat(tasks.get(0).split.dataFiles()).hasSize(1);
+ }
+
+ @Test
+ void testFileSpanningMultipleShards() throws IOException {
+ // One file [0, 249] with rowsPerShard=100 → spans shards 0,1,2
+ List<ManifestEntry> entries = new ArrayList<>();
+ entries.add(createEntry(BinaryRow.EMPTY_ROW, 0L, 250));
+
+ List<GenericIndexTopoBuilder.ShardTask> tasks =
+ GenericIndexTopoBuilder.computeShardTasks(table, entries, 100);
+
+ assertThat(tasks).hasSize(3);
+ // Shard 0: [0, 99], shard 1: [0, 199] clamped to [100, 199], shard 2: [0, 249] clamped
+ assertThat(tasks.get(0).shardRange).isEqualTo(new Range(0, 99));
+ assertThat(tasks.get(1).shardRange).isEqualTo(new Range(100, 199));
+ assertThat(tasks.get(2).shardRange).isEqualTo(new Range(200, 249));
+ // Each shard should contain the same file
+ for (GenericIndexTopoBuilder.ShardTask task : tasks) {
+ assertThat(task.split.dataFiles()).hasSize(1);
+ }
+ }
+
+ @Test
+ void testMultipleContiguousFiles() throws IOException {
+ // Two contiguous files [0, 49] and [50, 99] in one shard
+ List<ManifestEntry> entries = new ArrayList<>();
+ entries.add(createEntry(BinaryRow.EMPTY_ROW, 0L, 50));
+ entries.add(createEntry(BinaryRow.EMPTY_ROW, 50L, 50));
+
+ List<GenericIndexTopoBuilder.ShardTask> tasks =
+ GenericIndexTopoBuilder.computeShardTasks(table, entries, 100);
+
+ assertThat(tasks).hasSize(1);
+ assertThat(tasks.get(0).shardRange).isEqualTo(new Range(0, 99));
+ assertThat(tasks.get(0).split.dataFiles()).hasSize(2);
+ }
+
+ @Test
+ void testFilesWithGapInSameShard() throws IOException {
+ // Two files [0, 29] and [70, 99] with a gap in the middle, same shard
+ List<ManifestEntry> entries = new ArrayList<>();
+ entries.add(createEntry(BinaryRow.EMPTY_ROW, 0L, 30));
+ entries.add(createEntry(BinaryRow.EMPTY_ROW, 70L, 30));
+
+ List<GenericIndexTopoBuilder.ShardTask> tasks =
+ GenericIndexTopoBuilder.computeShardTasks(table, entries, 100);
+
+ // Gap produces two separate tasks within the same shard
+ assertThat(tasks).hasSize(2);
+ assertThat(tasks.get(0).shardRange).isEqualTo(new Range(0, 29));
+ assertThat(tasks.get(0).split.dataFiles()).hasSize(1);
+ assertThat(tasks.get(1).shardRange).isEqualTo(new Range(70, 99));
+ assertThat(tasks.get(1).split.dataFiles()).hasSize(1);
+ }
+
+ @Test
+ void testFileWithNullFirstRowIdSkipped() throws IOException {
+ // One file with null firstRowId + one normal file
+ List<ManifestEntry> entries = new ArrayList<>();
+ entries.add(createEntry(BinaryRow.EMPTY_ROW, null, 100));
+ entries.add(createEntry(BinaryRow.EMPTY_ROW, 0L, 50));
+
+ List<GenericIndexTopoBuilder.ShardTask> tasks =
+ GenericIndexTopoBuilder.computeShardTasks(table, entries, 100);
+
+ // Only the file with valid firstRowId should produce a task
+ assertThat(tasks).hasSize(1);
+ assertThat(tasks.get(0).split.dataFiles()).hasSize(1);
+ assertThat(tasks.get(0).shardRange).isEqualTo(new Range(0, 49));
+ }
+
+ @Test
+ void testMultiplePartitions() throws IOException {
+ BinaryRow partA = createPartition("a");
+ BinaryRow partB = createPartition("b");
+
+ List<ManifestEntry> entries = new ArrayList<>();
+ entries.add(createEntry(partA, 0L, 50));
+ entries.add(createEntry(partB, 100L, 50));
+
+ List<GenericIndexTopoBuilder.ShardTask> tasks =
+ GenericIndexTopoBuilder.computeShardTasks(table, entries, 100);
+
+ assertThat(tasks).hasSize(2);
+ // Each partition should have its own task
+ assertThat(tasks.stream().map(t -> t.split.partition()).distinct().count()).isEqualTo(2);
+ }
+
+ @Test
+ void testEmptyEntries() throws IOException {
+ List<GenericIndexTopoBuilder.ShardTask> tasks =
+ GenericIndexTopoBuilder.computeShardTasks(table, Collections.emptyList(), 100);
+
+ assertThat(tasks).isEmpty();
+ }
+
+ @Test
+ void testAllFilesNullRowId() throws IOException {
+ List<ManifestEntry> entries = new ArrayList<>();
+ entries.add(createEntry(BinaryRow.EMPTY_ROW, null, 100));
+ entries.add(createEntry(BinaryRow.EMPTY_ROW, null, 200));
+
+ List<GenericIndexTopoBuilder.ShardTask> tasks =
+ GenericIndexTopoBuilder.computeShardTasks(table, entries, 100);
+
+ assertThat(tasks).isEmpty();
+ }
+
+ @Test
+ void testMultipleFilesAcrossShardBoundary() throws IOException {
+ // File1 [0, 79] in shard 0, File2 [80, 159] spans shard 0 and shard 1
+ List<ManifestEntry> entries = new ArrayList<>();
+ entries.add(createEntry(BinaryRow.EMPTY_ROW, 0L, 80));
+ entries.add(createEntry(BinaryRow.EMPTY_ROW, 80L, 80));
+
+ List<GenericIndexTopoBuilder.ShardTask> tasks =
+ GenericIndexTopoBuilder.computeShardTasks(table, entries, 100);
+
+ // Shard 0: both files (contiguous), range [0, 99]
+ // Shard 1: only file2, range [100, 159]
+ assertThat(tasks).hasSize(2);
+ assertThat(tasks.get(0).shardRange).isEqualTo(new Range(0, 99));
+ assertThat(tasks.get(0).split.dataFiles()).hasSize(2);
+ assertThat(tasks.get(1).shardRange).isEqualTo(new Range(100, 159));
+ assertThat(tasks.get(1).split.dataFiles()).hasSize(1);
+ }
+
+ @Test
+ void testShardRangeClampedToFileRange() throws IOException {
+ // File [50, 149] with rowsPerShard=100 → shard 0 [50,99], shard 1 [100,149]
+ List<ManifestEntry> entries = new ArrayList<>();
+ entries.add(createEntry(BinaryRow.EMPTY_ROW, 50L, 100));
+
+ List<GenericIndexTopoBuilder.ShardTask> tasks =
+ GenericIndexTopoBuilder.computeShardTasks(table, entries, 100);
+
+ assertThat(tasks).hasSize(2);
+ // Clamped: shard 0 starts at 50 (not 0), shard 1 ends at 149 (not 199)
+ assertThat(tasks.get(0).shardRange).isEqualTo(new Range(50, 99));
+ assertThat(tasks.get(1).shardRange).isEqualTo(new Range(100, 149));
+ }
+
+ // -- Helpers --
+
+ private static ManifestEntry createEntry(BinaryRow partition, Long firstRowId, long rowCount) {
+ PojoDataFileMeta file =
+ new PojoDataFileMeta(
+ "test-file-" + UUID.randomUUID(),
+ 1024L,
+ rowCount,
+ BinaryRow.EMPTY_ROW,
+ BinaryRow.EMPTY_ROW,
+ SimpleStats.EMPTY_STATS,
+ SimpleStats.EMPTY_STATS,
+ 0L,
+ 0L,
+ 0L,
+ 0,
+ Collections.emptyList(),
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ firstRowId,
+ null);
+ return ManifestEntry.create(FileKind.ADD, partition, 0, 1, file);
+ }
+
+ private static BinaryRow createPartition(String value) {
+ BinaryRow row = new BinaryRow(1);
+ BinaryRowWriter writer = new BinaryRowWriter(row);
+ writer.writeString(0, BinaryString.fromString(value));
+ writer.complete();
+ return row;
+ }
+}
diff --git a/paimon-lumina/src/main/java/org/apache/paimon/lumina/index/LuminaVectorGlobalIndexWriter.java b/paimon-lumina/src/main/java/org/apache/paimon/lumina/index/LuminaVectorGlobalIndexWriter.java
index b21b55eed6..e2e257c28c 100644
--- a/paimon-lumina/src/main/java/org/apache/paimon/lumina/index/LuminaVectorGlobalIndexWriter.java
+++ b/paimon-lumina/src/main/java/org/apache/paimon/lumina/index/LuminaVectorGlobalIndexWriter.java
@@ -43,6 +43,7 @@ import java.nio.channels.FileChannel;
import java.util.Collections;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
/**
* Vector global index writer using Lumina. Builds a single index file per shard.
@@ -62,6 +63,14 @@ public class LuminaVectorGlobalIndexWriter implements GlobalIndexSingletonWriter
private static final Logger LOG = LoggerFactory.getLogger(LuminaVectorGlobalIndexWriter.class);
+ /**
+ * Tracks which environment variables have already been set in this JVM. Used to avoid
+ * concurrent reflection-based modification of {@code System.getenv()} when multiple shard tasks
+ * run in parallel on the same TaskManager.
+ */
+ private static final ConcurrentHashMap<String, String> CONFIGURED_ENV_VARS =
+ new ConcurrentHashMap<>();
+
/** I/O buffer size for reading/writing the temp vector file (~8 MB). */
private static final int IO_BUFFER_SIZE = 8 * 1024 * 1024;
@@ -117,26 +126,33 @@ public class LuminaVectorGlobalIndexWriter implements GlobalIndexSingletonWriter
@Override
public void write(Object fieldData) {
- float[] vector;
if (fieldData == null) {
throw new IllegalArgumentException("Field data must not be null");
}
- if (fieldData instanceof float[]) {
- vector = (float[]) fieldData;
- } else if (fieldData instanceof InternalArray) {
- vector = ((InternalArray) fieldData).toFloatArray();
- } else {
- throw new RuntimeException(
- "Unsupported vector type: " + fieldData.getClass().getName());
- }
- checkDimension(vector);
int bytesNeeded = dim * Float.BYTES;
if (writeBuf.remaining() < bytesNeeded) {
flushWriteBuffer();
}
- for (int i = 0; i < dim; i++) {
- writeBuf.putFloat(vector[i]);
+
+ if (fieldData instanceof float[]) {
+ float[] vector = (float[]) fieldData;
+ checkDimension(vector.length);
+ for (int i = 0; i < dim; i++) {
+ writeBuf.putFloat(vector[i]);
+ }
+ } else if (fieldData instanceof InternalArray) {
+ InternalArray array = (InternalArray) fieldData;
+ checkDimension(array.size());
+ for (int i = 0; i < dim; i++) {
+ if (array.isNullAt(i)) {
+ throw new IllegalArgumentException("Vector element at index " + i + " is null");
+ }
+ writeBuf.putFloat(array.getFloat(i));
+ }
+ } else {
+ throw new RuntimeException(
+ "Unsupported vector type: " + fieldData.getClass().getName());
}
count++;
}
@@ -220,31 +236,46 @@ public class LuminaVectorGlobalIndexWriter implements GlobalIndexSingletonWriter
}
/**
- * Best-effort attempt to set an environment variable in the current process. Uses reflection to
- * modify the JVM's process environment map. This is necessary because Lumina's native executor
- * thread pool reads {@code LUMINA_EXECUTOR_THREAD_COUNT} from the OS environment.
+ * Best-effort, thread-safe, idempotent attempt to set an environment variable in the current
+ * process. Uses reflection to modify the JVM's process environment map. This is necessary
+ * because Lumina's native executor thread pool reads {@code LUMINA_EXECUTOR_THREAD_COUNT} from
+ * the OS environment.
+ *
+ * <p>When multiple shard tasks run in parallel on the same TaskManager JVM, this method ensures
+ * the reflection-based map modification happens at most once per key-value pair.
*/
@SuppressWarnings("unchecked")
private static void setEnv(String key, String value) {
- try {
- Map<String, String> env = System.getenv();
- Field field = env.getClass().getDeclaredField("m");
- field.setAccessible(true);
- ((Map<String, String>) field.get(env)).put(key, value);
- } catch (Exception e) {
- LOG.warn(
- "Failed to set environment variable '{}' for Lumina executor. Thread-count tuning may not take effect.",
- key,
- e);
+ // Skip if already set to the same value in this JVM
+ if (value.equals(CONFIGURED_ENV_VARS.get(key))) {
+ return;
+ }
+ synchronized (CONFIGURED_ENV_VARS) {
+ // Double-check after acquiring lock
+ if (value.equals(CONFIGURED_ENV_VARS.get(key))) {
+ return;
+ }
+ try {
+ Map<String, String> env = System.getenv();
+ Field field = env.getClass().getDeclaredField("m");
+ field.setAccessible(true);
+ ((Map<String, String>) field.get(env)).put(key, value);
+ CONFIGURED_ENV_VARS.put(key, value);
+ } catch (Exception e) {
+ LOG.warn(
+ "Failed to set environment variable '{}' for Lumina executor. "
+ + "Thread-count tuning may not take effect.",
+ key,
+ e);
+ }
}
}
- private void checkDimension(float[] vector) {
- if (vector.length != dim) {
+ private void checkDimension(int actualDim) {
+ if (actualDim != dim) {
throw new IllegalArgumentException(
String.format(
- "Vector dimension mismatch: expected %d, but got %d",
- dim, vector.length));
+ "Vector dimension mismatch: expected %d, but got %d", dim, actualDim));
}
}
diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/DefaultGlobalIndexBuilder.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/DefaultGlobalIndexBuilder.java
index 20d8bc5de2..1485d14fac 100644
--- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/DefaultGlobalIndexBuilder.java
+++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/DefaultGlobalIndexBuilder.java
@@ -103,15 +103,28 @@ public class DefaultGlobalIndexBuilder implements Serializable {
(GlobalIndexSingletonWriter)
createIndexWriter(table, indexType, indexField, options);
- InternalRow.FieldGetter getter =
- InternalRow.createFieldGetter(
- indexField.type(), readType.getFieldIndex(indexField.name()));
- rows.forEachRemaining(
- row -> {
- Object indexO = getter.getFieldOrNull(row);
- indexWriter.write(indexO);
- rowCounter.add(1);
- });
- return indexWriter.finish();
+ try {
+ InternalRow.FieldGetter getter =
+ InternalRow.createFieldGetter(
+ indexField.type(), readType.getFieldIndex(indexField.name()));
+ rows.forEachRemaining(
+ row -> {
+ Object indexO = getter.getFieldOrNull(row);
+ indexWriter.write(indexO);
+ rowCounter.add(1);
+ });
+ return indexWriter.finish();
+ } finally {
+ closeWriterQuietly(indexWriter);
+ }
+ }
+
+ private static void closeWriterQuietly(GlobalIndexSingletonWriter writer) {
+ if (writer instanceof java.io.Closeable) {
+ try {
+ ((java.io.Closeable) writer).close();
+ } catch (IOException ignored) {
+ }
+ }
}
}
diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/DefaultGlobalIndexTopoBuilder.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/DefaultGlobalIndexTopoBuilder.java
index 711e65e8c8..04593c55a3 100644
--- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/DefaultGlobalIndexTopoBuilder.java
+++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/DefaultGlobalIndexTopoBuilder.java
@@ -21,6 +21,7 @@ package org.apache.paimon.spark.globalindex;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.fs.Path;
+import org.apache.paimon.globalindex.GlobalIndexBuilderUtils;
import org.apache.paimon.globalindex.IndexedSplit;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.manifest.ManifestEntry;
@@ -55,6 +56,7 @@ import java.util.Map;
import java.util.function.BiFunction;
import java.util.stream.Collectors;
+import static org.apache.paimon.CoreOptions.GLOBAL_INDEX_BUILD_MAX_SHARD;
import static org.apache.paimon.CoreOptions.GLOBAL_INDEX_ROW_COUNT_PER_SHARD;
import static org.apache.paimon.utils.Preconditions.checkArgument;
@@ -86,8 +88,17 @@ public class DefaultGlobalIndexTopoBuilder implements GlobalIndexTopologyBuilder
rowsPerShard > 0,
"Option 'global-index.row-count-per-shard' must be greater than 0.");
+ int maxShard = tableOptions.get(GLOBAL_INDEX_BUILD_MAX_SHARD);
+ checkArgument(
+ maxShard > 0, "Option 'global-index.build.max-shard' must be greater than 0.");
+ List<ManifestEntry> entries =
+ table.store().newScan().withPartitionFilter(partitionPredicate).plan().files();
+ long totalRowCount = entries.stream().mapToLong(e -> e.file().rowCount()).sum();
+ rowsPerShard =
+ GlobalIndexBuilderUtils.adjustRowsPerShard(rowsPerShard, totalRowCount, maxShard);
+
// generate splits for each partition && shard
- Map<BinaryRow, List<IndexedSplit>> splits = split(table, partitionPredicate, rowsPerShard);
+ Map<BinaryRow, List<IndexedSplit>> splits = split(table, entries, rowsPerShard);
JavaSparkContext javaSparkContext = new JavaSparkContext(spark.sparkContext());
List<Pair<byte[], byte[]>> taskList = new ArrayList<>();
@@ -139,11 +150,8 @@ public class DefaultGlobalIndexTopoBuilder implements GlobalIndexTopologyBuilder
}
private static Map<BinaryRow, List<IndexedSplit>> split(
- FileStoreTable table, PartitionPredicate partitions, long rowsPerShard) {
+ FileStoreTable table, List<ManifestEntry> entries, long rowsPerShard) {
FileStorePathFactory pathFactory = table.store().pathFactory();
- // Get all manifest entries from the table scan
- List<ManifestEntry> entries =
- table.store().newScan().withPartitionFilter(partitions).plan().files();
// Group manifest entries by partition
Map<BinaryRow, List<ManifestEntry>> entriesByPartition =