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 ec79aacada [index] Reuse incremental global index planning (#8407)
ec79aacada is described below
commit ec79aacada6c7f885d2fa65cb96114d955f6ee09
Author: Jingsong Lee <[email protected]>
AuthorDate: Wed Jul 1 15:21:13 2026 +0800
[index] Reuse incremental global index planning (#8407)
This PR makes global index build incremental by default across
Java/Flink/Spark/Python and extracts shared planning logic for reusable
row-range/shard handling.
---
.../globalindex/GlobalIndexBuilderUtils.java | 275 +++++++++++++++-
.../sorted/SortedGlobalIndexBuilder.java | 31 +-
.../globalindex/GlobalIndexBuilderUtilsTest.java | 64 ++++
.../globalindex/GenericGlobalIndexBuilder.java | 12 +-
.../flink/globalindex/GenericIndexTopoBuilder.java | 353 +++++++++------------
.../flink/globalindex/SortedIndexTopoBuilder.java | 2 +-
.../globalindex/GenericIndexTopoBuilderTest.java | 183 ++++++-----
.../globalindex/SortedIndexTopoBuilderTest.java | 4 +-
paimon-python/pypaimon/globalindex/build_plan.py | 331 +++++++++++++++++++
.../pypaimon/globalindex/create_global_index.py | 283 +++--------------
.../pypaimon/tests/global_index_build_test.py | 160 +++++++++-
.../globalindex/DefaultGlobalIndexTopoBuilder.java | 223 ++++---------
.../globalindex/sorted/SortedIndexTopoBuilder.java | 3 +-
.../procedure/CreateGlobalIndexProcedureTest.java | 28 ++
14 files changed, 1225 insertions(+), 727 deletions(-)
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 39f7fb2b0e..0537e50114 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
@@ -19,15 +19,21 @@
package org.apache.paimon.globalindex;
import org.apache.paimon.CoreOptions;
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.Path;
import org.apache.paimon.index.GlobalIndexMeta;
import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.index.IndexPathFactory;
+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.schema.SchemaManager;
import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.source.DataSplit;
import org.apache.paimon.types.DataField;
import org.apache.paimon.utils.Range;
@@ -38,9 +44,16 @@ import javax.annotation.Nullable;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Comparator;
import java.util.HashMap;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import java.util.function.BiFunction;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
/** Utils for global index build. */
public class GlobalIndexBuilderUtils {
@@ -78,12 +91,7 @@ public class GlobalIndexBuilderUtils {
// The first column is the primary index column and is stored as
indexFieldId; the
// remaining columns (if any) go into extraFieldIds.
int indexFieldId = fields.get(0).id();
- int[] extraFieldIds =
- fields.size() > 1
- ? fields.subList(1, fields.size()).stream()
- .mapToInt(DataField::id)
- .toArray()
- : null;
+ int[] extraFieldIds = extraFieldIds(fields);
return toIndexFileMetas(
fileIO,
indexPathFactory,
@@ -95,6 +103,244 @@ public class GlobalIndexBuilderUtils {
entries);
}
+ public static List<Range> unindexedRowRanges(
+ FileStoreTable table,
+ @Nullable Snapshot snapshot,
+ String indexType,
+ List<DataField> fields,
+ @Nullable PartitionPredicate partitionPredicate) {
+ if (snapshot == null || snapshot.nextRowId() == null ||
snapshot.nextRowId() <= 0) {
+ return Collections.emptyList();
+ }
+
+ Range dataRange = new Range(0, snapshot.nextRowId() - 1);
+ List<Range> indexedRanges =
+ indexedRowRanges(table, snapshot, indexType, fields,
partitionPredicate);
+ return Range.sortAndMergeOverlap(dataRange.exclude(indexedRanges),
true);
+ }
+
+ public static List<Range> indexedRowRanges(
+ FileStoreTable table,
+ @Nullable Snapshot snapshot,
+ String indexType,
+ List<DataField> fields,
+ @Nullable PartitionPredicate partitionPredicate) {
+ if (snapshot == null || fields.isEmpty()) {
+ return Collections.emptyList();
+ }
+
+ int indexFieldId = fields.get(0).id();
+ int[] extraFieldIds = extraFieldIds(fields);
+ List<Range> ranges = new ArrayList<>();
+ for (IndexManifestEntry entry :
+ table.store().newIndexFileHandler().scan(snapshot, indexType))
{
+ if (partitionPredicate != null &&
!partitionPredicate.test(entry.partition())) {
+ continue;
+ }
+ GlobalIndexMeta meta = entry.indexFile().globalIndexMeta();
+ if (meta == null) {
+ continue;
+ }
+ if (meta.indexFieldId() != indexFieldId) {
+ continue;
+ }
+ if (!sameExtraFieldIds(meta.extraFieldIds(), extraFieldIds)) {
+ continue;
+ }
+ ranges.add(meta.rowRange());
+ }
+ return Range.sortAndMergeOverlap(ranges, true);
+ }
+
+ @Nullable
+ public static List<Range> rowRangesAfter(long maxIndexedRowId) {
+ if (maxIndexedRowId < 0) {
+ return null;
+ }
+ if (maxIndexedRowId == Long.MAX_VALUE) {
+ return Collections.emptyList();
+ }
+ return Collections.singletonList(new Range(maxIndexedRowId + 1,
Long.MAX_VALUE));
+ }
+
+ public static List<IndexedSplit> createShardIndexedSplits(
+ FileStoreTable table, List<ManifestEntry> entries, long
rowsPerShard) {
+ return createShardIndexedSplits(table, entries, rowsPerShard, null);
+ }
+
+ public static List<IndexedSplit> createShardIndexedSplits(
+ FileStoreTable table,
+ List<ManifestEntry> entries,
+ long rowsPerShard,
+ @Nullable List<Range> rowRangesToBuild) {
+ return createShardIndexedSplits(
+ entries,
+ rowsPerShard,
+ (partition, bucket) ->
+ table.store().pathFactory().bucketPath(partition,
bucket).toString(),
+ rowRangesToBuild);
+ }
+
+ public static List<IndexedSplit> createShardIndexedSplits(
+ List<ManifestEntry> entries,
+ long rowsPerShard,
+ BiFunction<BinaryRow, Integer, String> bucketPathFactory,
+ @Nullable List<Range> rowRangesToBuild) {
+ checkArgument(
+ rowsPerShard > 0,
+ "Option 'global-index.row-count-per-shard' must be greater
than 0.");
+ if (rowRangesToBuild != null) {
+ rowRangesToBuild = Range.sortAndMergeOverlap(rowRangesToBuild,
true);
+ if (rowRangesToBuild.isEmpty()) {
+ return Collections.emptyList();
+ }
+ }
+
+ Map<BinaryRow, Map<Integer, List<ManifestEntry>>>
entriesByPartitionAndBucket =
+ new LinkedHashMap<>();
+ for (ManifestEntry entry : entries) {
+ entriesByPartitionAndBucket
+ .computeIfAbsent(entry.partition(), key -> new
LinkedHashMap<>())
+ .computeIfAbsent(entry.bucket(), key -> new ArrayList<>())
+ .add(entry);
+ }
+
+ List<IndexedSplit> result = new ArrayList<>();
+ for (Map.Entry<BinaryRow, Map<Integer, List<ManifestEntry>>>
partitionEntry :
+ entriesByPartitionAndBucket.entrySet()) {
+ BinaryRow partition = partitionEntry.getKey();
+ for (Map.Entry<Integer, List<ManifestEntry>> bucketEntry :
+ partitionEntry.getValue().entrySet()) {
+ addShardIndexedSplits(
+ result,
+ partition,
+ bucketEntry.getKey(),
+ bucketEntry.getValue(),
+ rowsPerShard,
+ bucketPathFactory,
+ rowRangesToBuild);
+ }
+ }
+ return result;
+ }
+
+ private static void addShardIndexedSplits(
+ List<IndexedSplit> result,
+ BinaryRow partition,
+ int bucket,
+ List<ManifestEntry> entries,
+ long rowsPerShard,
+ BiFunction<BinaryRow, Integer, String> bucketPathFactory,
+ @Nullable List<Range> rowRangesToBuild) {
+ Map<Long, List<DataFileMeta>> filesByShard = new LinkedHashMap<>();
+ for (ManifestEntry entry : entries) {
+ DataFileMeta file = entry.file();
+ if (file.firstRowId() == null) {
+ LOG.warn(
+ "Skipping file '{}' in partition {} bucket {} because
it has no row ID.",
+ file.fileName(),
+ partition,
+ bucket);
+ 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, key -> new
ArrayList<>()).add(file);
+ }
+ }
+
+ 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));
+ List<DataFileMeta> currentGroup = new ArrayList<>();
+ long currentGroupEnd = -1;
+
+ for (DataFileMeta file : shardFiles) {
+ long fileStart = file.nonNullFirstRowId();
+ long fileEnd = file.nonNullRowIdRange().to;
+ if (currentGroup.isEmpty()) {
+ currentGroup.add(file);
+ currentGroupEnd = fileEnd;
+ } else if (fileStart <= currentGroupEnd + 1) {
+ currentGroup.add(file);
+ currentGroupEnd = Math.max(currentGroupEnd, fileEnd);
+ } else {
+ addIndexedSplitForFileGroup(
+ result,
+ currentGroup,
+ shardStart,
+ shardEnd,
+ partition,
+ bucket,
+ entries.get(0).totalBuckets(),
+ bucketPathFactory.apply(partition, bucket),
+ rowRangesToBuild);
+ currentGroup = new ArrayList<>();
+ currentGroup.add(file);
+ currentGroupEnd = fileEnd;
+ }
+ }
+ if (!currentGroup.isEmpty()) {
+ addIndexedSplitForFileGroup(
+ result,
+ currentGroup,
+ shardStart,
+ shardEnd,
+ partition,
+ bucket,
+ entries.get(0).totalBuckets(),
+ bucketPathFactory.apply(partition, bucket),
+ rowRangesToBuild);
+ }
+ }
+ }
+
+ private static void addIndexedSplitForFileGroup(
+ List<IndexedSplit> result,
+ List<DataFileMeta> files,
+ long shardStart,
+ long shardEnd,
+ BinaryRow partition,
+ int bucket,
+ int totalBuckets,
+ String bucketPath,
+ @Nullable List<Range> rowRangesToBuild) {
+ long groupMinRowId = files.get(0).nonNullFirstRowId();
+ long groupMaxRowId =
+ files.stream().mapToLong(file ->
file.nonNullRowIdRange().to).max().getAsLong();
+ Range groupRange =
+ new Range(Math.max(groupMinRowId, shardStart),
Math.min(groupMaxRowId, shardEnd));
+ List<Range> taskRanges =
+ rowRangesToBuild == null
+ ? Collections.singletonList(groupRange)
+ : Range.and(Collections.singletonList(groupRange),
rowRangesToBuild);
+ if (taskRanges.isEmpty()) {
+ return;
+ }
+
+ DataSplit dataSplit =
+ DataSplit.builder()
+ .withPartition(partition)
+ .withBucket(bucket)
+ .withTotalBuckets(totalBuckets)
+ .withDataFiles(files)
+ .withBucketPath(bucketPath)
+ .rawConvertible(false)
+ .build();
+ for (Range taskRange : taskRanges) {
+ result.add(new IndexedSplit(dataSplit,
Collections.singletonList(taskRange), null));
+ }
+ }
+
private static List<IndexFileMeta> toIndexFileMetas(
FileIO fileIO,
IndexPathFactory indexPathFactory,
@@ -219,4 +465,21 @@ public class GlobalIndexBuilderUtils {
IndexPathFactory indexPathFactory =
table.store().pathFactory().globalIndexFileFactory();
return new GlobalIndexFileReadWrite(table.fileIO(), indexPathFactory);
}
+
+ @Nullable
+ private static int[] extraFieldIds(List<DataField> fields) {
+ return fields.size() > 1
+ ? fields.subList(1,
fields.size()).stream().mapToInt(DataField::id).toArray()
+ : null;
+ }
+
+ private static boolean sameExtraFieldIds(@Nullable int[] left, @Nullable
int[] right) {
+ if (left == null || left.length == 0) {
+ return right == null || right.length == 0;
+ }
+ if (right == null || right.length == 0) {
+ return false;
+ }
+ return Arrays.equals(left, right);
+ }
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexBuilder.java
b/paimon-core/src/main/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexBuilder.java
index e8ba93ffc8..6ee00b3bb8 100644
---
a/paimon-core/src/main/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexBuilder.java
+++
b/paimon-core/src/main/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexBuilder.java
@@ -34,7 +34,6 @@ import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.io.CompactIncrement;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.io.DataIncrement;
-import org.apache.paimon.manifest.IndexManifestEntry;
import org.apache.paimon.options.Options;
import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.reader.RecordReader;
@@ -76,6 +75,7 @@ import java.util.stream.IntStream;
import static java.util.Collections.singletonList;
import static org.apache.paimon.format.blob.BlobFileFormat.isBlobFile;
import static
org.apache.paimon.globalindex.GlobalIndexBuilderUtils.createIndexWriter;
+import static
org.apache.paimon.globalindex.GlobalIndexBuilderUtils.indexedRowRanges;
import static
org.apache.paimon.globalindex.GlobalIndexBuilderUtils.toIndexFileMetas;
import static org.apache.paimon.types.VectorType.isVectorStoreFile;
import static org.apache.paimon.utils.Preconditions.checkArgument;
@@ -174,7 +174,13 @@ public class SortedGlobalIndexBuilder implements
Serializable {
Preconditions.checkArgument(indexField != null, "indexField must be
set before scan.");
Range dataRange = new Range(0, snapshot.nextRowId() - 1);
- List<Range> indexedRanges = indexedRowRanges(snapshot);
+ List<Range> indexedRanges =
+ indexedRowRanges(
+ table,
+ snapshot,
+ indexType,
+ Collections.singletonList(indexField),
+ partitionPredicate);
List<Range> nonIndexedRanges = dataRange.exclude(indexedRanges);
if (nonIndexedRanges.isEmpty()) {
return Optional.empty();
@@ -193,27 +199,6 @@ public class SortedGlobalIndexBuilder implements
Serializable {
&&
!isVectorStoreFile(entry.file().fileName()));
}
- private List<Range> indexedRowRanges(Snapshot snapshot) {
- List<Range> ranges = new ArrayList<>();
- for (IndexManifestEntry entry :
- table.store().newIndexFileHandler().scan(snapshot, indexType))
{
- if (partitionPredicate != null &&
!partitionPredicate.test(entry.partition())) {
- continue;
- }
- if (entry.indexFile().globalIndexMeta() == null) {
- continue;
- }
- if (entry.indexFile().globalIndexMeta().indexFieldId() !=
indexField.id()) {
- continue;
- }
- ranges.add(
- new Range(
-
entry.indexFile().globalIndexMeta().rowRangeStart(),
-
entry.indexFile().globalIndexMeta().rowRangeEnd()));
- }
- return Range.sortAndMergeOverlap(ranges, true);
- }
-
@VisibleForTesting
public List<CommitMessage> build(DataSplit split, IOManager ioManager)
throws IOException {
BinaryRow partition = split.partition();
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
index 67852ae925..11c8ddb64c 100644
---
a/paimon-core/src/test/java/org/apache/paimon/globalindex/GlobalIndexBuilderUtilsTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/globalindex/GlobalIndexBuilderUtilsTest.java
@@ -19,12 +19,17 @@
package org.apache.paimon.globalindex;
import org.apache.paimon.CoreOptions;
+import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.index.IndexPathFactory;
+import org.apache.paimon.io.PojoDataFileMeta;
+import org.apache.paimon.manifest.FileKind;
+import org.apache.paimon.manifest.ManifestEntry;
import org.apache.paimon.options.Options;
+import org.apache.paimon.stats.SimpleStats;
import org.apache.paimon.types.ArrayType;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.FloatType;
@@ -138,10 +143,69 @@ class GlobalIndexBuilderUtilsTest {
assertThat(metas.get(0).globalIndexMeta().extraFieldIds()).isEqualTo(new int[]
{2, 3});
}
+ @Test
+ void testCreateShardIndexedSplitsUsesUnindexedRanges() {
+ List<ManifestEntry> entries = Arrays.asList(createEntry(0L, 100),
createEntry(100L, 100));
+
+ List<IndexedSplit> splits =
+ GlobalIndexBuilderUtils.createShardIndexedSplits(
+ entries,
+ 100,
+ (partition, bucket) -> "/bucket-" + bucket,
+ Collections.singletonList(new Range(0, 99)));
+
+ assertThat(splits).hasSize(1);
+ assertThat(splits.get(0).rowRanges()).containsExactly(new Range(0,
99));
+
assertThat(splits.get(0).dataSplit().dataFiles()).containsExactly(entries.get(0).file());
+ }
+
+ @Test
+ void testCreateShardIndexedSplitsCanSplitOneShardIntoMultipleRanges() {
+ List<ManifestEntry> entries =
Collections.singletonList(createEntry(0L, 100));
+
+ List<IndexedSplit> splits =
+ GlobalIndexBuilderUtils.createShardIndexedSplits(
+ entries,
+ 100,
+ (partition, bucket) -> "/bucket-" + bucket,
+ Arrays.asList(new Range(0, 9), new Range(90, 99)));
+
+ assertThat(splits).hasSize(2);
+ assertThat(splits.get(0).rowRanges()).containsExactly(new Range(0, 9));
+ assertThat(splits.get(1).rowRanges()).containsExactly(new Range(90,
99));
+
assertThat(splits.get(0).dataSplit()).isEqualTo(splits.get(1).dataSplit());
+ }
+
private List<ResultEntry> createDummyResultEntries() throws IOException {
String fileName = "test-index-" + UUID.randomUUID();
Path filePath = indexPathFactory.toPath(fileName);
fileIO.newOutputStream(filePath, false).close();
return Collections.singletonList(new ResultEntry(fileName, 100, null));
}
+
+ private ManifestEntry createEntry(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, BinaryRow.EMPTY_ROW, 0, 1,
file);
+ }
}
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
index 415c1590a5..b20c513cd0 100644
---
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
@@ -40,6 +40,7 @@ public class GenericGlobalIndexBuilder implements
Serializable {
protected final FileStoreTable table;
@Nullable protected PartitionPredicate partitionPredicate;
+ @Nullable private Snapshot scanSnapshot;
public GenericGlobalIndexBuilder(FileStoreTable table) {
this.table = table;
@@ -73,19 +74,24 @@ public class GenericGlobalIndexBuilder implements
Serializable {
+ "deleted rows to be indexed.",
table.name());
- Snapshot snapshot = table.snapshotManager().latestSnapshot();
- if (snapshot == null) {
+ scanSnapshot = table.snapshotManager().latestSnapshot();
+ if (scanSnapshot == null) {
return Collections.emptyList();
}
return table.store()
.newScan()
- .withSnapshot(snapshot)
+ .withSnapshot(scanSnapshot)
.withPartitionFilter(partitionPredicate)
.plan()
.files();
}
+ @Nullable
+ public Snapshot scanSnapshot() {
+ return scanSnapshot;
+ }
+
/** 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
index 39e6db0cf8..34461d7f75 100644
---
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
@@ -19,6 +19,7 @@
package org.apache.paimon.flink.globalindex;
import org.apache.paimon.CoreOptions;
+import org.apache.paimon.Snapshot;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.flink.sink.Committable;
@@ -32,9 +33,9 @@ import
org.apache.paimon.flink.utils.StreamExecutionEnvironmentUtils;
import org.apache.paimon.globalindex.GlobalIndexMultiColumnWriter;
import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter;
import org.apache.paimon.globalindex.GlobalIndexWriter;
+import org.apache.paimon.globalindex.IndexedSplit;
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;
@@ -45,7 +46,6 @@ 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;
@@ -61,23 +61,24 @@ import
org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import javax.annotation.Nullable;
+
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.createShardIndexedSplits;
import static
org.apache.paimon.globalindex.GlobalIndexBuilderUtils.filterEntriesBefore;
import static
org.apache.paimon.globalindex.GlobalIndexBuilderUtils.findMinNonIndexableRowId;
+import static
org.apache.paimon.globalindex.GlobalIndexBuilderUtils.rowRangesAfter;
import static
org.apache.paimon.globalindex.GlobalIndexBuilderUtils.toIndexFileMetas;
+import static
org.apache.paimon.globalindex.GlobalIndexBuilderUtils.unindexedRowRanges;
import static org.apache.paimon.io.CompactIncrement.emptyIncrement;
import static org.apache.paimon.io.DataIncrement.deleteIndexIncrement;
import static org.apache.paimon.io.DataIncrement.indexIncrement;
@@ -103,7 +104,7 @@ public class GenericIndexTopoBuilder {
PartitionPredicate partitionPredicate,
Options userOptions)
throws Exception {
- buildIndexAndExecute(
+ buildIndexAndExecuteInternal(
env,
table,
indexColumn,
@@ -111,7 +112,8 @@ public class GenericIndexTopoBuilder {
indexType,
partitionPredicate,
userOptions,
- NO_MAX_INDEXED_ROW_ID);
+ NO_MAX_INDEXED_ROW_ID,
+ true);
}
public static void buildIndexAndExecute(
@@ -123,7 +125,7 @@ public class GenericIndexTopoBuilder {
Options userOptions,
long maxIndexedRowId)
throws Exception {
- buildIndexAndExecute(
+ buildIndexAndExecuteInternal(
env,
table,
indexColumn,
@@ -131,7 +133,8 @@ public class GenericIndexTopoBuilder {
indexType,
partitionPredicate,
userOptions,
- maxIndexedRowId);
+ maxIndexedRowId,
+ false);
}
public static void buildIndexAndExecute(
@@ -143,7 +146,7 @@ public class GenericIndexTopoBuilder {
PartitionPredicate partitionPredicate,
Options userOptions)
throws Exception {
- buildIndexAndExecute(
+ buildIndexAndExecuteInternal(
env,
table,
indexColumn,
@@ -151,7 +154,8 @@ public class GenericIndexTopoBuilder {
indexType,
partitionPredicate,
userOptions,
- NO_MAX_INDEXED_ROW_ID);
+ NO_MAX_INDEXED_ROW_ID,
+ true);
}
public static void buildIndexAndExecute(
@@ -164,8 +168,31 @@ public class GenericIndexTopoBuilder {
Options userOptions,
long maxIndexedRowId)
throws Exception {
+ buildIndexAndExecuteInternal(
+ env,
+ table,
+ indexColumn,
+ extraColumns,
+ indexType,
+ partitionPredicate,
+ userOptions,
+ maxIndexedRowId,
+ false);
+ }
+
+ private static void buildIndexAndExecuteInternal(
+ StreamExecutionEnvironment env,
+ FileStoreTable table,
+ String indexColumn,
+ List<String> extraColumns,
+ String indexType,
+ PartitionPredicate partitionPredicate,
+ Options userOptions,
+ long maxIndexedRowId,
+ boolean autoIncremental)
+ throws Exception {
boolean hasIndexToBuild =
- buildIndex(
+ buildIndexInternal(
env,
() -> new GenericGlobalIndexBuilder(table),
table,
@@ -174,7 +201,8 @@ public class GenericIndexTopoBuilder {
indexType,
partitionPredicate,
userOptions,
- maxIndexedRowId);
+ maxIndexedRowId,
+ autoIncremental);
if (hasIndexToBuild) {
env.execute("Create " + indexType + " global index for table: " +
table.name());
} else {
@@ -191,7 +219,7 @@ public class GenericIndexTopoBuilder {
PartitionPredicate partitionPredicate,
Options userOptions)
throws Exception {
- return buildIndex(
+ return buildIndexInternal(
env,
indexBuilderSupplier,
table,
@@ -200,7 +228,8 @@ public class GenericIndexTopoBuilder {
indexType,
partitionPredicate,
userOptions,
- NO_MAX_INDEXED_ROW_ID);
+ NO_MAX_INDEXED_ROW_ID,
+ true);
}
public static boolean buildIndex(
@@ -213,7 +242,7 @@ public class GenericIndexTopoBuilder {
Options userOptions,
long maxIndexedRowId)
throws Exception {
- return buildIndex(
+ return buildIndexInternal(
env,
indexBuilderSupplier,
table,
@@ -222,7 +251,8 @@ public class GenericIndexTopoBuilder {
indexType,
partitionPredicate,
userOptions,
- maxIndexedRowId);
+ maxIndexedRowId,
+ false);
}
/**
@@ -242,6 +272,31 @@ public class GenericIndexTopoBuilder {
Options userOptions,
long maxIndexedRowId)
throws Exception {
+ return buildIndexInternal(
+ env,
+ indexBuilderSupplier,
+ table,
+ indexColumn,
+ extraColumns,
+ indexType,
+ partitionPredicate,
+ userOptions,
+ maxIndexedRowId,
+ false);
+ }
+
+ private static boolean buildIndexInternal(
+ StreamExecutionEnvironment env,
+ Supplier<GenericGlobalIndexBuilder> indexBuilderSupplier,
+ FileStoreTable table,
+ String indexColumn,
+ List<String> extraColumns,
+ String indexType,
+ PartitionPredicate partitionPredicate,
+ Options userOptions,
+ long maxIndexedRowId,
+ boolean autoIncremental)
+ throws Exception {
GenericGlobalIndexBuilder indexBuilder = indexBuilderSupplier.get();
if (partitionPredicate != null) {
indexBuilder.withPartitionPredicate(partitionPredicate);
@@ -259,16 +314,16 @@ public class GenericIndexTopoBuilder {
userOptions,
entries,
deletedIndexEntries,
- maxIndexedRowId);
+ partitionPredicate,
+ indexBuilder.scanSnapshot(),
+ maxIndexedRowId,
+ autoIncremental);
}
/**
* Builds the Flink topology for global index creation from pre-scanned
entries. Supports both
- * full builds ({@code maxIndexedRowId = NO_MAX_INDEXED_ROW_ID}) and
incremental builds where
- * rows up to {@code maxIndexedRowId} are skipped.
+ * full builds and incremental builds where already indexed row ranges are
skipped.
*
- * @param maxIndexedRowId the maximum row ID already indexed; use {@link
#NO_MAX_INDEXED_ROW_ID}
- * for a full build
* @return {@code true} if a Flink topology was built, {@code false} if
nothing to index
*/
private static boolean buildTopology(
@@ -280,7 +335,10 @@ public class GenericIndexTopoBuilder {
Options userOptions,
List<ManifestEntry> entries,
List<IndexManifestEntry> deletedIndexEntries,
- long maxIndexedRowId)
+ PartitionPredicate partitionPredicate,
+ @Nullable Snapshot scanSnapshot,
+ long maxIndexedRowId,
+ boolean autoIncremental)
throws Exception {
// The primary column followed by the extra columns, in index order.
List<String> indexColumns = new ArrayList<>(1 + extraColumns.size());
@@ -289,14 +347,12 @@ public class GenericIndexTopoBuilder {
long totalRowCount = entries.stream().mapToLong(e ->
e.file().rowCount()).sum();
LOG.info(
- "Scanned {} files ({} rows) across {} partitions for {} index
on columns '{}'"
- + (maxIndexedRowId >= 0 ? ", maxIndexedRowId={}." :
"."),
+ "Scanned {} files ({} rows) across {} partitions for {} index
on columns '{}'.",
entries.size(),
totalRowCount,
entries.stream().map(ManifestEntry::partition).distinct().count(),
indexType,
- indexColumns,
- maxIndexedRowId);
+ indexColumns);
long minNonIndexableRowId =
findMinNonIndexableRowId(table.schemaManager(), entries,
indexColumns);
@@ -306,6 +362,9 @@ public class GenericIndexTopoBuilder {
DataField indexField = rowType.getField(indexColumn);
List<DataField> extraFields =
extraColumns.stream().map(rowType::getField).collect(Collectors.toList());
+ List<DataField> indexedFields = new ArrayList<>(1 +
extraFields.size());
+ indexedFields.add(indexField);
+ indexedFields.addAll(extraFields);
// Project indexColumns + _ROW_ID so we can read the actual row ID
from data
List<String> readColumns = new ArrayList<>(indexColumns);
readColumns.add(SpecialFields.ROW_ID.name());
@@ -318,9 +377,33 @@ public class GenericIndexTopoBuilder {
rowsPerShard > 0,
"Option 'global-index.row-count-per-shard' must be greater
than 0.");
+ List<Range> rowRangesToBuild = null;
+ if (deletedIndexEntries.isEmpty()) {
+ if (autoIncremental) {
+ Snapshot snapshot =
+ scanSnapshot == null
+ ? table.snapshotManager().latestSnapshot()
+ : scanSnapshot;
+ rowRangesToBuild =
+ unindexedRowRanges(
+ table, snapshot, indexType, indexedFields,
partitionPredicate);
+ LOG.info("Automatically selected unindexed row ranges: {}.",
rowRangesToBuild);
+ } else if (maxIndexedRowId != NO_MAX_INDEXED_ROW_ID) {
+ rowRangesToBuild = rowRangesAfter(maxIndexedRowId);
+ LOG.info(
+ "Selected row ranges after maxIndexedRowId={}: {}.",
+ maxIndexedRowId,
+ rowRangesToBuild);
+ }
+ if (rowRangesToBuild != null && rowRangesToBuild.isEmpty()) {
+ LOG.info("No unindexed row ranges found, nothing to index.");
+ return false;
+ }
+ }
+
// Compute shard tasks at file level from the provided entries
- List<ShardTask> shardTasks =
- computeShardTasks(table, entries, rowsPerShard,
maxIndexedRowId);
+ List<IndexedSplit> shardTasks =
+ computeShardTasks(table, entries, rowsPerShard,
rowRangesToBuild);
if (shardTasks.isEmpty()) {
LOG.info("No shard tasks generated, nothing to index.");
return false;
@@ -339,11 +422,11 @@ public class GenericIndexTopoBuilder {
// Build Flink topology
ReadBuilder readBuilder =
table.newReadBuilder().withReadType(projectedRowType);
- DataStream<ShardTask> source =
+ DataStream<IndexedSplit> source =
StreamExecutionEnvironmentUtils.fromData(
env,
- new JavaTypeInfo<>(ShardTask.class),
- shardTasks.toArray(new ShardTask[0]))
+ new JavaTypeInfo<>(IndexedSplit.class),
+ shardTasks.toArray(new IndexedSplit[0]))
.name("Generic Index Source")
.setParallelism(1);
@@ -377,152 +460,25 @@ public class GenericIndexTopoBuilder {
return true;
}
- /**
- * Compute shard tasks for a full build (no rows to skip).
- *
- * @see #computeShardTasks(FileStoreTable, List, long, long)
- */
- static List<ShardTask> computeShardTasks(
+ static List<IndexedSplit> computeShardTasks(
FileStoreTable table, List<ManifestEntry> entries, long
rowsPerShard) {
- return computeShardTasks(table, entries, rowsPerShard,
NO_MAX_INDEXED_ROW_ID);
+ return createShardIndexedSplits(table, entries, rowsPerShard);
}
- /**
- * 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.
- *
- * <p>When {@code maxIndexedRowId >= 0}, each shard's effective start is
advanced past {@code
- * maxIndexedRowId}, skipping fully-indexed shards entirely. This enables
incremental index
- * building where only new (un-indexed) rows are processed.
- *
- * @param maxIndexedRowId the maximum row ID already indexed; use {@link
#NO_MAX_INDEXED_ROW_ID}
- * for a full build
- */
- static List<ShardTask> computeShardTasks(
+ static List<IndexedSplit> computeShardTasks(
FileStoreTable table,
List<ManifestEntry> entries,
long rowsPerShard,
long maxIndexedRowId) {
- // 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;
- }
-
- // For incremental builds, advance past already-indexed rows
- long effectiveStart =
- maxIndexedRowId >= 0
- ? Math.max(shardStart, maxIndexedRowId + 1)
- : shardStart;
- if (effectiveStart > shardEnd) {
- continue; // entire shard already indexed
- }
-
-
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,
- effectiveStart,
- shardEnd,
- partition,
- partBucketPath));
- currentGroup = new ArrayList<>();
- currentGroup.add(file);
- currentGroupEnd = fileEnd;
- }
- }
- if (!currentGroup.isEmpty()) {
- tasks.add(
- createShardTask(
- currentGroup,
- effectiveStart,
- shardEnd,
- partition,
- partBucketPath));
- }
- }
- }
- return tasks;
+ return computeShardTasks(table, entries, rowsPerShard,
rowRangesAfter(maxIndexedRowId));
}
- private static ShardTask createShardTask(
- List<DataFileMeta> files,
- long effectiveStart,
- 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 effective boundaries
- long rangeFrom = Math.max(groupMinRowId, effectiveStart);
- 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));
+ static List<IndexedSplit> computeShardTasks(
+ FileStoreTable table,
+ List<ManifestEntry> entries,
+ long rowsPerShard,
+ @Nullable List<Range> rowRangesToBuild) {
+ return createShardIndexedSplits(table, entries, rowsPerShard,
rowRangesToBuild);
}
private static List<Committable> createDeleteCommittables(
@@ -558,26 +514,13 @@ public class GenericIndexTopoBuilder {
.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
+ * Operator that receives an {@link IndexedSplit}, 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> {
+ extends BoundedOneInputOperator<IndexedSplit, Committable> {
private static final long serialVersionUID = 1L;
@@ -641,17 +584,21 @@ public class GenericIndexTopoBuilder {
}
@Override
- public void processElement(StreamRecord<ShardTask> element) throws
Exception {
- ShardTask task = element.getValue();
- BinaryRow partition = task.split.partition();
+ public void processElement(StreamRecord<IndexedSplit> element) throws
Exception {
+ IndexedSplit task = element.getValue();
+ checkArgument(
+ task.rowRanges().size() == 1,
+ "Each generic global index task should contain exactly one
row range.");
+ Range shardRange = task.rowRanges().get(0);
+ BinaryRow partition = task.dataSplit().partition();
LOG.info(
"Building {} index for partition={}, shardRange=[{}, {}],
files={}.",
indexType,
partition,
- task.shardRange.from,
- task.shardRange.to,
- task.split.dataFiles().size());
+ shardRange.from,
+ shardRange.to,
+ task.dataSplit().dataFiles().size());
long startTime = System.currentTimeMillis();
GlobalIndexWriter indexWriter =
@@ -660,7 +607,7 @@ public class GenericIndexTopoBuilder {
try {
long rowsSeen = 0;
long lastRowId = Long.MIN_VALUE;
- try (RecordReader<InternalRow> reader =
tableRead.createReader(task.split);
+ try (RecordReader<InternalRow> reader =
tableRead.createReader(task.dataSplit());
CloseableIterator<InternalRow> iter =
reader.toCloseableIterator()) {
while (iter.hasNext()) {
InternalRow row = iter.next();
@@ -674,17 +621,17 @@ public class GenericIndexTopoBuilder {
+ "This may indicate
corrupted data files.",
lastRowId,
currentRowId,
- task.shardRange.from,
- task.shardRange.to));
+ shardRange.from,
+ shardRange.to));
}
lastRowId = currentRowId;
- if (currentRowId > task.shardRange.to) {
+ if (currentRowId > shardRange.to) {
break;
}
// Only write rows within this shard's range
- if (currentRowId >= task.shardRange.from) {
- long rowId = currentRowId - task.shardRange.from;
+ if (currentRowId >= shardRange.from) {
+ long rowId = currentRowId - shardRange.from;
if (multiColumn) {
((GlobalIndexMultiColumnWriter) indexWriter)
.write(rowId,
writerProjection.replaceRow(row));
@@ -709,8 +656,8 @@ public class GenericIndexTopoBuilder {
LOG.info(
"Finished shard [{}, {}]: saw {} rows, "
+ "produced {} result entries in {} ms.",
- task.shardRange.from,
- task.shardRange.to,
+ shardRange.from,
+ shardRange.to,
rowsSeen,
resultEntries.size(),
elapsed);
@@ -719,8 +666,8 @@ public class GenericIndexTopoBuilder {
LOG.warn(
"Shard [{}, {}] produced no index (all null or
empty), "
+ "skipping index flush.",
- task.shardRange.from,
- task.shardRange.to);
+ shardRange.from,
+ shardRange.to);
return;
}
@@ -728,7 +675,7 @@ public class GenericIndexTopoBuilder {
flushIndex(
table,
partition,
- task.shardRange,
+ shardRange,
indexedFields,
indexType,
resultEntries);
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 4a17bb68f1..b68611eb31 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
@@ -111,7 +111,7 @@ public class SortedIndexTopoBuilder {
}
Optional<Pair<RowRangeIndex, List<DataSplit>>> indexRangeAndSplits
=
- indexBuilder.scan();
+ indexBuilder.incrementalScan();
if (!indexRangeAndSplits.isPresent()) {
continue;
}
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
index c69b59ad6e..916d6c2578 100644
---
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
@@ -24,6 +24,7 @@ import org.apache.paimon.data.BinaryRowWriter;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.fs.Path;
import org.apache.paimon.globalindex.GlobalIndexBuilderUtils;
+import org.apache.paimon.globalindex.IndexedSplit;
import org.apache.paimon.io.PojoDataFileMeta;
import org.apache.paimon.manifest.FileKind;
import org.apache.paimon.manifest.ManifestEntry;
@@ -74,12 +75,11 @@ class GenericIndexTopoBuilderTest {
List<ManifestEntry> entries = new ArrayList<>();
entries.add(createEntry(BinaryRow.EMPTY_ROW, 0L, 100));
- List<GenericIndexTopoBuilder.ShardTask> tasks =
- GenericIndexTopoBuilder.computeShardTasks(table, entries, 100);
+ List<IndexedSplit> 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);
+ assertThat(tasks.get(0).rowRanges().get(0)).isEqualTo(new Range(0,
99));
+ assertThat(tasks.get(0).dataSplit().dataFiles()).hasSize(1);
}
@Test
@@ -88,17 +88,16 @@ class GenericIndexTopoBuilderTest {
List<ManifestEntry> entries = new ArrayList<>();
entries.add(createEntry(BinaryRow.EMPTY_ROW, 0L, 250));
- List<GenericIndexTopoBuilder.ShardTask> tasks =
- GenericIndexTopoBuilder.computeShardTasks(table, entries, 100);
+ List<IndexedSplit> 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));
+ assertThat(tasks.get(0).rowRanges().get(0)).isEqualTo(new Range(0,
99));
+ assertThat(tasks.get(1).rowRanges().get(0)).isEqualTo(new Range(100,
199));
+ assertThat(tasks.get(2).rowRanges().get(0)).isEqualTo(new Range(200,
249));
// Each shard should contain the same file
- for (GenericIndexTopoBuilder.ShardTask task : tasks) {
- assertThat(task.split.dataFiles()).hasSize(1);
+ for (IndexedSplit task : tasks) {
+ assertThat(task.dataSplit().dataFiles()).hasSize(1);
}
}
@@ -109,12 +108,11 @@ class GenericIndexTopoBuilderTest {
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);
+ List<IndexedSplit> 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);
+ assertThat(tasks.get(0).rowRanges().get(0)).isEqualTo(new Range(0,
99));
+ assertThat(tasks.get(0).dataSplit().dataFiles()).hasSize(2);
}
@Test
@@ -124,15 +122,14 @@ class GenericIndexTopoBuilderTest {
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);
+ List<IndexedSplit> 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);
+ assertThat(tasks.get(0).rowRanges().get(0)).isEqualTo(new Range(0,
29));
+ assertThat(tasks.get(0).dataSplit().dataFiles()).hasSize(1);
+ assertThat(tasks.get(1).rowRanges().get(0)).isEqualTo(new Range(70,
99));
+ assertThat(tasks.get(1).dataSplit().dataFiles()).hasSize(1);
}
@Test
@@ -142,13 +139,12 @@ class GenericIndexTopoBuilderTest {
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);
+ List<IndexedSplit> 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));
+ assertThat(tasks.get(0).dataSplit().dataFiles()).hasSize(1);
+ assertThat(tasks.get(0).rowRanges().get(0)).isEqualTo(new Range(0,
49));
}
@Test
@@ -160,17 +156,17 @@ class GenericIndexTopoBuilderTest {
entries.add(createEntry(partA, 0L, 50));
entries.add(createEntry(partB, 100L, 50));
- List<GenericIndexTopoBuilder.ShardTask> tasks =
- GenericIndexTopoBuilder.computeShardTasks(table, entries, 100);
+ List<IndexedSplit> 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);
+ assertThat(tasks.stream().map(t ->
t.dataSplit().partition()).distinct().count())
+ .isEqualTo(2);
}
@Test
void testEmptyEntries() throws IOException {
- List<GenericIndexTopoBuilder.ShardTask> tasks =
+ List<IndexedSplit> tasks =
GenericIndexTopoBuilder.computeShardTasks(table,
Collections.emptyList(), 100);
assertThat(tasks).isEmpty();
@@ -182,8 +178,7 @@ class GenericIndexTopoBuilderTest {
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);
+ List<IndexedSplit> tasks =
GenericIndexTopoBuilder.computeShardTasks(table, entries, 100);
assertThat(tasks).isEmpty();
}
@@ -195,16 +190,15 @@ class GenericIndexTopoBuilderTest {
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);
+ List<IndexedSplit> 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);
+ assertThat(tasks.get(0).rowRanges().get(0)).isEqualTo(new Range(0,
99));
+ assertThat(tasks.get(0).dataSplit().dataFiles()).hasSize(2);
+ assertThat(tasks.get(1).rowRanges().get(0)).isEqualTo(new Range(100,
159));
+ assertThat(tasks.get(1).dataSplit().dataFiles()).hasSize(1);
}
@Test
@@ -213,13 +207,12 @@ class GenericIndexTopoBuilderTest {
List<ManifestEntry> entries = new ArrayList<>();
entries.add(createEntry(BinaryRow.EMPTY_ROW, 50L, 100));
- List<GenericIndexTopoBuilder.ShardTask> tasks =
- GenericIndexTopoBuilder.computeShardTasks(table, entries, 100);
+ List<IndexedSplit> 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));
+ assertThat(tasks.get(0).rowRanges().get(0)).isEqualTo(new Range(50,
99));
+ assertThat(tasks.get(1).rowRanges().get(0)).isEqualTo(new Range(100,
149));
}
// ========== Incremental build scenarios (maxIndexedRowId) ==========
@@ -232,12 +225,12 @@ class GenericIndexTopoBuilderTest {
entries.add(createEntry(BinaryRow.EMPTY_ROW, 0L, 100));
entries.add(createEntry(BinaryRow.EMPTY_ROW, 100L, 100));
- List<GenericIndexTopoBuilder.ShardTask> tasks =
+ List<IndexedSplit> tasks =
GenericIndexTopoBuilder.computeShardTasks(table, entries, 200,
-1);
assertThat(tasks).hasSize(1);
- assertThat(tasks.get(0).shardRange).isEqualTo(new Range(0, 199));
- assertThat(tasks.get(0).split.dataFiles()).hasSize(2);
+ assertThat(tasks.get(0).rowRanges().get(0)).isEqualTo(new Range(0,
199));
+ assertThat(tasks.get(0).dataSplit().dataFiles()).hasSize(2);
}
@Test
@@ -246,11 +239,41 @@ class GenericIndexTopoBuilderTest {
List<ManifestEntry> entries = new ArrayList<>();
entries.add(createEntry(BinaryRow.EMPTY_ROW, 200L, 200));
- List<GenericIndexTopoBuilder.ShardTask> tasks =
+ List<IndexedSplit> tasks =
GenericIndexTopoBuilder.computeShardTasks(table, entries, 200,
199);
assertThat(tasks).hasSize(1);
- assertThat(tasks.get(0).shardRange).isEqualTo(new Range(200, 399));
+ assertThat(tasks.get(0).rowRanges().get(0)).isEqualTo(new Range(200,
399));
+ }
+
+ @Test
+ void testIncrementalUsesUnindexedRangesInsteadOfMaxRowId() {
+ // Existing index may cover a later range while an earlier
partition/range
+ // is still unindexed. Building from ranges must not skip the earlier
gap.
+ List<ManifestEntry> entries = new ArrayList<>();
+ entries.add(createEntry(BinaryRow.EMPTY_ROW, 0L, 100));
+ entries.add(createEntry(BinaryRow.EMPTY_ROW, 100L, 100));
+ List<Range> unindexedRanges = Collections.singletonList(new Range(0,
99));
+
+ List<IndexedSplit> tasks =
+ GenericIndexTopoBuilder.computeShardTasks(table, entries, 100,
unindexedRanges);
+
+ assertThat(tasks).hasSize(1);
+ assertThat(tasks.get(0).rowRanges().get(0)).isEqualTo(new Range(0,
99));
+ }
+
+ @Test
+ void testIncrementalRangeCanSplitOneShardIntoMultipleTasks() {
+ List<ManifestEntry> entries = new ArrayList<>();
+ entries.add(createEntry(BinaryRow.EMPTY_ROW, 0L, 100));
+ List<Range> unindexedRanges = Arrays.asList(new Range(0, 9), new
Range(90, 99));
+
+ List<IndexedSplit> tasks =
+ GenericIndexTopoBuilder.computeShardTasks(table, entries, 100,
unindexedRanges);
+
+ assertThat(tasks).hasSize(2);
+ assertThat(tasks.get(0).rowRanges().get(0)).isEqualTo(new Range(0, 9));
+ assertThat(tasks.get(1).rowRanges().get(0)).isEqualTo(new Range(90,
99));
}
@Test
@@ -259,7 +282,7 @@ class GenericIndexTopoBuilderTest {
List<ManifestEntry> entries = new ArrayList<>();
entries.add(createEntry(BinaryRow.EMPTY_ROW, 0L, 400));
- List<GenericIndexTopoBuilder.ShardTask> tasks =
+ List<IndexedSplit> tasks =
GenericIndexTopoBuilder.computeShardTasks(table, entries, 200,
399);
assertThat(tasks).isEmpty();
@@ -273,17 +296,17 @@ class GenericIndexTopoBuilderTest {
List<ManifestEntry> entries = new ArrayList<>();
entries.add(createEntry(BinaryRow.EMPTY_ROW, 100L, 200)); // D[100,299]
- List<GenericIndexTopoBuilder.ShardTask> tasks =
+ List<IndexedSplit> tasks =
GenericIndexTopoBuilder.computeShardTasks(table, entries, 200,
199);
assertThat(tasks).hasSize(1);
- assertThat(tasks.get(0).shardRange).isEqualTo(new Range(200, 299));
+ assertThat(tasks.get(0).rowRanges().get(0)).isEqualTo(new Range(200,
299));
}
@Test
void testIncrementalCompactOnlyIndexedFiles() {
// Compact two indexed files → empty entries → no tasks.
- List<GenericIndexTopoBuilder.ShardTask> tasks =
+ List<IndexedSplit> tasks =
GenericIndexTopoBuilder.computeShardTasks(table,
Collections.emptyList(), 200, 199);
assertThat(tasks).isEmpty();
@@ -297,11 +320,11 @@ class GenericIndexTopoBuilderTest {
List<ManifestEntry> entries = new ArrayList<>();
entries.add(createEntry(BinaryRow.EMPTY_ROW, 200L, 400)); // D[200,599]
- List<GenericIndexTopoBuilder.ShardTask> tasks =
+ List<IndexedSplit> tasks =
GenericIndexTopoBuilder.computeShardTasks(table, entries, 200,
399);
assertThat(tasks).hasSize(1);
- assertThat(tasks.get(0).shardRange).isEqualTo(new Range(400, 599));
+ assertThat(tasks.get(0).rowRanges().get(0)).isEqualTo(new Range(400,
599));
}
@Test
@@ -312,11 +335,11 @@ class GenericIndexTopoBuilderTest {
List<ManifestEntry> entries = new ArrayList<>();
entries.add(createEntry(BinaryRow.EMPTY_ROW, 0L, 400));
- List<GenericIndexTopoBuilder.ShardTask> tasks =
+ List<IndexedSplit> tasks =
GenericIndexTopoBuilder.computeShardTasks(table, entries, 200,
199);
assertThat(tasks).hasSize(1);
- assertThat(tasks.get(0).shardRange).isEqualTo(new Range(200, 399));
+ assertThat(tasks.get(0).rowRanges().get(0)).isEqualTo(new Range(200,
399));
}
@Test
@@ -325,12 +348,12 @@ class GenericIndexTopoBuilderTest {
List<ManifestEntry> entries = new ArrayList<>();
entries.add(createEntry(BinaryRow.EMPTY_ROW, 0L, 250));
- List<GenericIndexTopoBuilder.ShardTask> tasks =
+ List<IndexedSplit> tasks =
GenericIndexTopoBuilder.computeShardTasks(table, entries, 200,
-1);
assertThat(tasks).hasSize(2);
- assertThat(tasks.get(0).shardRange).isEqualTo(new Range(0, 199));
- assertThat(tasks.get(1).shardRange).isEqualTo(new Range(200, 249));
+ assertThat(tasks.get(0).rowRanges().get(0)).isEqualTo(new Range(0,
199));
+ assertThat(tasks.get(1).rowRanges().get(0)).isEqualTo(new Range(200,
249));
}
@Test
@@ -342,12 +365,12 @@ class GenericIndexTopoBuilderTest {
List<ManifestEntry> entries = new ArrayList<>();
entries.add(createEntry(BinaryRow.EMPTY_ROW, 0L, 600));
- List<GenericIndexTopoBuilder.ShardTask> tasks =
+ List<IndexedSplit> tasks =
GenericIndexTopoBuilder.computeShardTasks(table, entries, 200,
199);
assertThat(tasks).hasSize(2);
- assertThat(tasks.get(0).shardRange).isEqualTo(new Range(200, 399));
- assertThat(tasks.get(1).shardRange).isEqualTo(new Range(400, 599));
+ assertThat(tasks.get(0).rowRanges().get(0)).isEqualTo(new Range(200,
399));
+ assertThat(tasks.get(1).rowRanges().get(0)).isEqualTo(new Range(400,
599));
}
@Test
@@ -356,11 +379,11 @@ class GenericIndexTopoBuilderTest {
List<ManifestEntry> entries = new ArrayList<>();
entries.add(createEntry(BinaryRow.EMPTY_ROW, 200L, 200));
- List<GenericIndexTopoBuilder.ShardTask> tasks =
+ List<IndexedSplit> tasks =
GenericIndexTopoBuilder.computeShardTasks(table, entries, 200,
199);
assertThat(tasks).hasSize(1);
- assertThat(tasks.get(0).shardRange).isEqualTo(new Range(200, 399));
+ assertThat(tasks.get(0).rowRanges().get(0)).isEqualTo(new Range(200,
399));
}
@Test
@@ -371,12 +394,12 @@ class GenericIndexTopoBuilderTest {
List<ManifestEntry> entries = new ArrayList<>();
entries.add(createEntry(BinaryRow.EMPTY_ROW, 0L, 350));
- List<GenericIndexTopoBuilder.ShardTask> tasks =
+ List<IndexedSplit> tasks =
GenericIndexTopoBuilder.computeShardTasks(table, entries, 200,
149);
assertThat(tasks).hasSize(2);
- assertThat(tasks.get(0).shardRange).isEqualTo(new Range(150, 199));
- assertThat(tasks.get(1).shardRange).isEqualTo(new Range(200, 349));
+ assertThat(tasks.get(0).rowRanges().get(0)).isEqualTo(new Range(150,
199));
+ assertThat(tasks.get(1).rowRanges().get(0)).isEqualTo(new Range(200,
349));
}
@Test
@@ -386,12 +409,12 @@ class GenericIndexTopoBuilderTest {
List<ManifestEntry> entries = new ArrayList<>();
entries.add(createEntry(BinaryRow.EMPTY_ROW, 0L, 600));
- List<GenericIndexTopoBuilder.ShardTask> tasks =
+ List<IndexedSplit> tasks =
GenericIndexTopoBuilder.computeShardTasks(table, entries, 200,
199);
assertThat(tasks).hasSize(2);
- assertThat(tasks.get(0).shardRange).isEqualTo(new Range(200, 399));
- assertThat(tasks.get(1).shardRange).isEqualTo(new Range(400, 599));
+ assertThat(tasks.get(0).rowRanges().get(0)).isEqualTo(new Range(200,
399));
+ assertThat(tasks.get(1).rowRanges().get(0)).isEqualTo(new Range(400,
599));
}
@Test
@@ -400,12 +423,12 @@ class GenericIndexTopoBuilderTest {
entries.add(createEntry(BinaryRow.EMPTY_ROW, null, 100));
entries.add(createEntry(BinaryRow.EMPTY_ROW, 200L, 100));
- List<GenericIndexTopoBuilder.ShardTask> tasks =
+ List<IndexedSplit> tasks =
GenericIndexTopoBuilder.computeShardTasks(table, entries, 200,
199);
assertThat(tasks).hasSize(1);
- assertThat(tasks.get(0).shardRange).isEqualTo(new Range(200, 299));
- assertThat(tasks.get(0).split.dataFiles()).hasSize(1);
+ assertThat(tasks.get(0).rowRanges().get(0)).isEqualTo(new Range(200,
299));
+ assertThat(tasks.get(0).dataSplit().dataFiles()).hasSize(1);
}
@Test
@@ -415,12 +438,12 @@ class GenericIndexTopoBuilderTest {
entries.add(createEntry(BinaryRow.EMPTY_ROW, 200L, 100));
entries.add(createEntry(BinaryRow.EMPTY_ROW, 300L, 100));
- List<GenericIndexTopoBuilder.ShardTask> tasks =
+ List<IndexedSplit> tasks =
GenericIndexTopoBuilder.computeShardTasks(table, entries, 400,
199);
assertThat(tasks).hasSize(1);
- assertThat(tasks.get(0).shardRange).isEqualTo(new Range(200, 399));
- assertThat(tasks.get(0).split.dataFiles()).hasSize(2);
+ assertThat(tasks.get(0).rowRanges().get(0)).isEqualTo(new Range(200,
399));
+ assertThat(tasks.get(0).dataSplit().dataFiles()).hasSize(2);
}
@Test
@@ -430,12 +453,12 @@ class GenericIndexTopoBuilderTest {
entries.add(createEntry(BinaryRow.EMPTY_ROW, 0L, 50));
entries.add(createEntry(BinaryRow.EMPTY_ROW, 150L, 50));
- List<GenericIndexTopoBuilder.ShardTask> tasks =
+ List<IndexedSplit> tasks =
GenericIndexTopoBuilder.computeShardTasks(table, entries, 200,
-1);
assertThat(tasks).hasSize(2);
- assertThat(tasks.get(0).shardRange).isEqualTo(new Range(0, 49));
- assertThat(tasks.get(1).shardRange).isEqualTo(new Range(150, 199));
+ assertThat(tasks.get(0).rowRanges().get(0)).isEqualTo(new Range(0,
49));
+ assertThat(tasks.get(1).rowRanges().get(0)).isEqualTo(new Range(150,
199));
}
@Test
@@ -446,12 +469,12 @@ class GenericIndexTopoBuilderTest {
List<ManifestEntry> entries = new ArrayList<>();
entries.add(createEntry(BinaryRow.EMPTY_ROW, 300L, 200));
- List<GenericIndexTopoBuilder.ShardTask> tasks =
+ List<IndexedSplit> tasks =
GenericIndexTopoBuilder.computeShardTasks(table, entries, 200,
250);
assertThat(tasks).hasSize(2);
- assertThat(tasks.get(0).shardRange).isEqualTo(new Range(300, 399));
- assertThat(tasks.get(1).shardRange).isEqualTo(new Range(400, 499));
+ assertThat(tasks.get(0).rowRanges().get(0)).isEqualTo(new Range(300,
399));
+ assertThat(tasks.get(1).rowRanges().get(0)).isEqualTo(new Range(400,
499));
}
@Test
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 89a15bef31..21941e35e6 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
@@ -34,6 +34,7 @@ import java.util.Optional;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoInteractions;
import static org.mockito.Mockito.when;
@@ -44,7 +45,7 @@ public class SortedIndexTopoBuilderTest {
public void testBuildIndexReturnsFalseWhenNoBuildTask() throws Exception {
SortedGlobalIndexBuilder indexBuilder =
mock(SortedGlobalIndexBuilder.class);
when(indexBuilder.withIndexField("id")).thenReturn(indexBuilder);
- when(indexBuilder.scan()).thenReturn(Optional.empty());
+ when(indexBuilder.incrementalScan()).thenReturn(Optional.empty());
StreamExecutionEnvironment env =
mock(StreamExecutionEnvironment.class);
assertThat(
@@ -56,6 +57,7 @@ public class SortedIndexTopoBuilderTest {
null,
new Options()))
.isFalse();
+ verify(indexBuilder).incrementalScan();
verifyNoInteractions(env);
}
diff --git a/paimon-python/pypaimon/globalindex/build_plan.py
b/paimon-python/pypaimon/globalindex/build_plan.py
new file mode 100644
index 0000000000..74893d16b2
--- /dev/null
+++ b/paimon-python/pypaimon/globalindex/build_plan.py
@@ -0,0 +1,331 @@
+# 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.
+
+"""Reusable global index build planning helpers."""
+
+from typing import List, Optional, Sequence
+
+from pypaimon.globalindex.indexed_split import IndexedSplit
+from pypaimon.read.split import DataSplit
+from pypaimon.utils.range import Range
+from pypaimon.utils.range_helper import RangeHelper
+
+
+def calc_row_range(split) -> Range:
+ ranges = []
+ for file in split.files:
+ row_range = file.row_id_range()
+ if row_range is None:
+ raise ValueError(
+ "Cannot build global index because file '%s' has no row id
range."
+ % file.file_name
+ )
+ ranges.append(row_range)
+ if not ranges:
+ raise ValueError("Cannot build global index for an empty split.")
+ merged = Range.sort_and_merge_overlap(ranges, True, True)
+ return Range(merged[0].from_, merged[-1].to)
+
+
+def unindexed_row_ranges(
+ table,
+ snapshot,
+ partition_filter,
+ index_field_id: int,
+ index_type: str,
+) -> List[Range]:
+ next_row_id = getattr(snapshot, "next_row_id", None)
+ if snapshot is None or next_row_id is None or next_row_id <= 0:
+ return []
+
+ indexed_ranges = indexed_row_ranges(
+ table, snapshot, partition_filter, index_field_id, index_type)
+ return Range.sort_and_merge_overlap(
+ Range(0, next_row_id - 1).exclude(indexed_ranges), True)
+
+
+def indexed_row_ranges(
+ table,
+ snapshot,
+ partition_filter,
+ index_field_id: int,
+ index_type: str,
+) -> List[Range]:
+ from pypaimon.index.index_file_handler import IndexFileHandler
+
+ ranges = []
+ for entry in IndexFileHandler(table).scan(snapshot):
+ if getattr(entry, "kind", 0) != 0:
+ continue
+ if (partition_filter is not None
+ and not partition_filter.test(entry.partition)):
+ continue
+
+ index_file = entry.index_file
+ meta = index_file.global_index_meta
+ if (
+ meta is None
+ or index_file.index_type != index_type
+ or meta.index_field_id != index_field_id
+ or meta.extra_field_ids
+ ):
+ continue
+ ranges.append(Range(meta.row_range_start, meta.row_range_end))
+ return Range.sort_and_merge_overlap(ranges, True)
+
+
+def split_by_contiguous_row_range(splits):
+ result = []
+ for split in splits:
+ result.extend(split_one_by_contiguous_row_range(split))
+ return result
+
+
+def split_by_contiguous_unindexed_row_range(splits, unindexed_ranges):
+ result = []
+ for split in split_by_contiguous_row_range(splits):
+ row_range = calc_row_range(split)
+ for index_range in Range.and_([row_range], unindexed_ranges):
+ index_split = indexed_split_for_row_range(split, index_range)
+ if index_split is not None:
+ result.append((index_split, index_range))
+ return result
+
+
+def indexed_split_for_row_range(split, row_range):
+ files = [
+ file for file in split.files
+ if file.row_id_range() is not None
+ and file.row_id_range().overlaps(row_range)
+ ]
+ if not files:
+ return None
+ return IndexedSplit(copy_split_with_files(split, files), [row_range])
+
+
+def filter_non_indexable_splits(table, splits, index_columns: Sequence[str]):
+ boundary = find_min_non_indexable_row_id(
+ table.schema_manager,
+ [file for split in splits for file in split.files],
+ index_columns,
+ )
+ if boundary is None:
+ return splits
+
+ result = []
+ for split in splits:
+ files = [
+ file for file in split.files
+ if file.row_id_range() is not None
+ and file.row_id_range().from_ < boundary
+ ]
+ if files:
+ result.append(copy_split_with_files(split, files))
+ return result
+
+
+def find_min_non_indexable_row_id(schema_manager, files, index_columns):
+ schema_contains_columns = {}
+ index_column_set = set(index_columns)
+ boundary = None
+ for file in files:
+ row_range = file.row_id_range()
+ if row_range is None:
+ continue
+
+ schema_id = file.schema_id
+ if schema_id not in schema_contains_columns:
+ schema = schema_manager.get_schema(schema_id)
+ schema_field_names = {field.name for field in schema.fields}
+ schema_contains_columns[schema_id] = (
+ index_column_set.issubset(schema_field_names)
+ )
+ if not schema_contains_columns[schema_id]:
+ if boundary is None or row_range.from_ < boundary:
+ boundary = row_range.from_
+ return boundary
+
+
+def split_by_global_index_shard(
+ splits,
+ rows_per_shard: int,
+ row_ranges_to_build: Optional[List[Range]] = None,
+):
+ if rows_per_shard <= 0:
+ raise ValueError(
+ "Option 'global-index.row-count-per-shard' must be greater than 0."
+ )
+ if row_ranges_to_build is not None:
+ row_ranges_to_build = Range.sort_and_merge_overlap(
+ row_ranges_to_build, True)
+ if not row_ranges_to_build:
+ return []
+
+ groups = {}
+ for split in splits:
+ key = (partition_key(split.partition), split.bucket)
+ if key not in groups:
+ groups[key] = {
+ "partition": split.partition,
+ "bucket": split.bucket,
+ "files": [],
+ }
+ for file in split.files:
+ if file.row_id_range() is None:
+ continue
+ groups[key]["files"].append(file)
+
+ result = []
+ for group in groups.values():
+ files_by_shard = {}
+ for file in group["files"]:
+ file_range = file.row_id_range()
+ start_shard = file_range.from_ // rows_per_shard
+ end_shard = file_range.to // rows_per_shard
+ for shard_id in range(start_shard, end_shard + 1):
+ shard_start = shard_id * rows_per_shard
+ files_by_shard.setdefault(shard_start, []).append(file)
+
+ for shard_start in sorted(files_by_shard):
+ shard_end = shard_start + rows_per_shard - 1
+ shard_files = sorted(
+ files_by_shard[shard_start],
+ key=lambda file: file.row_id_range().from_,
+ )
+ current_group = []
+ current_group_end = None
+ for file in shard_files:
+ file_range = file.row_id_range()
+ if not current_group:
+ current_group.append(file)
+ current_group_end = file_range.to
+ elif file_range.from_ <= current_group_end + 1:
+ current_group.append(file)
+ current_group_end = max(current_group_end, file_range.to)
+ else:
+ append_shard_split(
+ result,
+ current_group,
+ shard_start,
+ shard_end,
+ group["partition"],
+ group["bucket"],
+ row_ranges_to_build,
+ )
+ current_group = [file]
+ current_group_end = file_range.to
+
+ if current_group:
+ append_shard_split(
+ result,
+ current_group,
+ shard_start,
+ shard_end,
+ group["partition"],
+ group["bucket"],
+ row_ranges_to_build,
+ )
+
+ return result
+
+
+def partition_key(partition):
+ values = getattr(partition, "values", None)
+ return tuple(values) if values is not None else partition
+
+
+def append_shard_split(
+ result,
+ files,
+ shard_start,
+ shard_end,
+ partition,
+ bucket,
+ row_ranges_to_build,
+):
+ group_start = min(file.row_id_range().from_ for file in files)
+ group_end = max(file.row_id_range().to for file in files)
+ row_range = Range(max(group_start, shard_start), min(group_end, shard_end))
+ task_ranges = (
+ [row_range]
+ if row_ranges_to_build is None
+ else Range.and_([row_range], row_ranges_to_build)
+ )
+ if not task_ranges:
+ return
+
+ data_split = DataSplit(
+ files=list(files),
+ partition=partition,
+ bucket=bucket,
+ raw_convertible=False,
+ )
+ for task_range in task_ranges:
+ result.append((IndexedSplit(data_split, [task_range]), task_range))
+
+
+def split_one_by_contiguous_row_range(split):
+ for file in split.files:
+ if file.row_id_range() is None:
+ raise ValueError(
+ "Cannot build global index because file '%s' has no row id
range."
+ % file.file_name
+ )
+
+ range_helper = RangeHelper(lambda file: file.row_id_range())
+ ranges = range_helper.merge_overlapping_ranges(split.files)
+ if not ranges:
+ return []
+
+ result = []
+ current_segment = []
+ current_max_row_id = None
+ for range_files in ranges:
+ min_row_id = min(file.row_id_range().from_ for file in range_files)
+ max_row_id = max(file.row_id_range().to for file in range_files)
+ if (
+ not current_segment
+ or current_max_row_id is None
+ or current_max_row_id >= min_row_id - 1
+ ):
+ current_segment.extend(range_files)
+ current_max_row_id = max_row_id
+ else:
+ result.append(copy_split_with_files(split, current_segment))
+ current_segment = list(range_files)
+ current_max_row_id = max_row_id
+
+ if current_segment:
+ result.append(copy_split_with_files(split, current_segment))
+ return result
+
+
+def copy_split_with_files(split, files):
+ data_deletion_files = None
+ if getattr(split, "data_deletion_files", None) is not None:
+ index_by_file = {id(file): i for i, file in enumerate(split.files)}
+ data_deletion_files = [
+ split.data_deletion_files[index_by_file[id(file)]]
+ for file in files
+ ]
+ return DataSplit(
+ files=list(files),
+ partition=split.partition,
+ bucket=split.bucket,
+ raw_convertible=getattr(split, "raw_convertible", False),
+ data_deletion_files=data_deletion_files,
+ )
diff --git a/paimon-python/pypaimon/globalindex/create_global_index.py
b/paimon-python/pypaimon/globalindex/create_global_index.py
index 6fe5ae82e3..3cf1547edf 100644
--- a/paimon-python/pypaimon/globalindex/create_global_index.py
+++ b/paimon-python/pypaimon/globalindex/create_global_index.py
@@ -33,6 +33,12 @@ from pypaimon.globalindex.bitmap.bitmap_index_writer import (
BITMAP_IDENTIFIER,
BitmapIndexWriter,
)
+from pypaimon.globalindex.build_plan import (
+ filter_non_indexable_splits as _filter_non_indexable_splits,
+ split_by_contiguous_unindexed_row_range as
_split_by_contiguous_unindexed_row_range,
+ split_by_global_index_shard as _split_by_global_index_shard,
+ unindexed_row_ranges as _unindexed_row_ranges,
+)
from pypaimon.globalindex.global_index_meta import GlobalIndexMeta
from pypaimon.globalindex.key_serializer import create_serializer
from pypaimon.globalindex.tantivy.tantivy_full_text_global_index_reader import
(
@@ -49,11 +55,9 @@ from pypaimon.globalindex.vindex.vindex_vector_index_writer
import (
)
from pypaimon.index.index_file_meta import IndexFileMeta
from pypaimon.manifest.index_manifest_entry import IndexManifestEntry
-from pypaimon.read.split import DataSplit
from pypaimon.table.row.generic_row import GenericRow
from pypaimon.table.special_fields import SpecialFields
from pypaimon.utils.range import Range
-from pypaimon.utils.range_helper import RangeHelper
from pypaimon.write.commit_message import CommitMessage
@@ -156,11 +160,23 @@ class GlobalIndexBuilder:
read_builder = read_builder.with_partition_filter(partition_filter)
scan = read_builder.new_scan()
- splits = scan.plan().splits()
+ plan = scan.plan()
+ splits = plan.splits()
if not splits:
return []
index_field = self._table.field_dict[self._index_columns[0]]
+ snapshot = self._snapshot_for_plan(plan)
+ unindexed_ranges = _unindexed_row_ranges(
+ self._table,
+ snapshot,
+ partition_filter,
+ index_field.id,
+ self._index_type,
+ )
+ if not unindexed_ranges:
+ return []
+
if self._index_type in _GENERIC_INDEX_IDENTIFIERS:
splits = _filter_non_indexable_splits(
self._table, splits, self._index_columns)
@@ -180,12 +196,19 @@ class GlobalIndexBuilder:
if self._index_type in _SORTED_INDEX_IDENTIFIERS:
return self._build_sorted_index(
- splits, index_field, table_read, index_path)
+ splits, unindexed_ranges, index_field, table_read, index_path)
return self._build_generic_index(
- splits, index_field, table_read, index_path)
+ splits, unindexed_ranges, index_field, table_read, index_path)
+
+ def _snapshot_for_plan(self, plan):
+ snapshot_id = getattr(plan, "snapshot_id", None)
+ snapshot_manager = self._table.snapshot_manager()
+ if snapshot_id is not None:
+ return snapshot_manager.get_snapshot_by_id(snapshot_id)
+ return snapshot_manager.get_latest_snapshot()
def _build_sorted_index(
- self, splits, index_field, table_read, index_path: str
+ self, splits, unindexed_ranges, index_field, table_read, index_path:
str
) -> List[CommitMessage]:
key_serializer = create_serializer(index_field.type)
configured_records_per_range = (
@@ -198,8 +221,9 @@ class GlobalIndexBuilder:
)
messages = []
- for split in _split_by_contiguous_row_range(splits):
- row_range = _calc_row_range(split)
+ for split, row_range in _split_by_contiguous_unindexed_row_range(
+ splits, unindexed_ranges
+ ):
table = table_read.to_arrow([split])
if table is None or table.num_rows == 0:
continue
@@ -208,6 +232,7 @@ class GlobalIndexBuilder:
self._index_columns[0],
SpecialFields.ROW_ID.name,
key_serializer,
+ row_range,
)
if not rows:
continue
@@ -259,7 +284,7 @@ class GlobalIndexBuilder:
raise ValueError("Unsupported sorted global index type: %s" %
self._index_type)
def _build_generic_index(
- self, splits, index_field, table_read, index_path: str
+ self, splits, unindexed_ranges, index_field, table_read, index_path:
str
) -> List[CommitMessage]:
rows_per_shard = self._core_options.global_index_row_count_per_shard()
if rows_per_shard <= 0:
@@ -268,10 +293,10 @@ class GlobalIndexBuilder:
)
messages = []
- for split, row_range in _split_by_global_index_shard(
- splits, rows_per_shard
+ for index_split, index_range in _split_by_global_index_shard(
+ splits, rows_per_shard, unindexed_ranges
):
- table = table_read.to_arrow([split])
+ table = table_read.to_arrow([index_split])
if table is None or table.num_rows == 0:
continue
@@ -281,14 +306,14 @@ class GlobalIndexBuilder:
table,
self._index_columns[0],
SpecialFields.ROW_ID.name,
- row_range,
+ index_range,
):
- writer.write(value, row_id - row_range.from_)
+ writer.write(value, row_id - index_range.from_)
index_adds = _to_index_manifest_entries(
self._table,
- split.partition,
- row_range,
+ index_split.partition,
+ index_range,
index_field.id,
self._index_type,
writer.finish(),
@@ -298,7 +323,7 @@ class GlobalIndexBuilder:
if index_adds:
messages.append(
CommitMessage(
- partition=tuple(split.partition.values),
+ partition=tuple(index_split.partition.values),
bucket=0,
new_files=[],
index_adds=index_adds,
@@ -385,229 +410,12 @@ def _merged_options(table, options: Optional[Dict[str,
object]]) -> Options:
return Options(merged)
-def _calc_row_range(split) -> Range:
- ranges = []
- for file in split.files:
- row_range = file.row_id_range()
- if row_range is None:
- raise ValueError(
- "Cannot build global index because file '%s' has no row id
range."
- % file.file_name
- )
- ranges.append(row_range)
- if not ranges:
- raise ValueError("Cannot build global index for an empty split.")
- merged = Range.sort_and_merge_overlap(ranges, True, True)
- return Range(merged[0].from_, merged[-1].to)
-
-
-def _split_by_contiguous_row_range(splits):
- result = []
- for split in splits:
- result.extend(_split_one_by_contiguous_row_range(split))
- return result
-
-
-def _filter_non_indexable_splits(table, splits, index_columns):
- boundary = _find_min_non_indexable_row_id(
- table.schema_manager,
- [file for split in splits for file in split.files],
- index_columns,
- )
- if boundary is None:
- return splits
-
- result = []
- for split in splits:
- files = [
- file for file in split.files
- if file.row_id_range() is not None
- and file.row_id_range().from_ < boundary
- ]
- if files:
- result.append(_copy_split_with_files(split, files))
- return result
-
-
-def _find_min_non_indexable_row_id(schema_manager, files, index_columns):
- schema_contains_columns = {}
- index_column_set = set(index_columns)
- boundary = None
- for file in files:
- row_range = file.row_id_range()
- if row_range is None:
- continue
-
- schema_id = file.schema_id
- if schema_id not in schema_contains_columns:
- schema = schema_manager.get_schema(schema_id)
- schema_field_names = {field.name for field in schema.fields}
- schema_contains_columns[schema_id] = (
- index_column_set.issubset(schema_field_names)
- )
- if not schema_contains_columns[schema_id]:
- if boundary is None or row_range.from_ < boundary:
- boundary = row_range.from_
- return boundary
-
-
-def _split_by_global_index_shard(splits, rows_per_shard):
- if rows_per_shard <= 0:
- raise ValueError(
- "Option 'global-index.row-count-per-shard' must be greater than 0."
- )
-
- groups = {}
- for split in splits:
- key = (_partition_key(split.partition), split.bucket)
- if key not in groups:
- groups[key] = {
- "partition": split.partition,
- "bucket": split.bucket,
- "files": [],
- }
- for file in split.files:
- if file.row_id_range() is None:
- continue
- groups[key]["files"].append(file)
-
- result = []
- for group in groups.values():
- files_by_shard = {}
- for file in group["files"]:
- file_range = file.row_id_range()
- start_shard = file_range.from_ // rows_per_shard
- end_shard = file_range.to // rows_per_shard
- for shard_id in range(start_shard, end_shard + 1):
- shard_start = shard_id * rows_per_shard
- files_by_shard.setdefault(shard_start, []).append(file)
-
- for shard_start in sorted(files_by_shard):
- shard_end = shard_start + rows_per_shard - 1
- shard_files = sorted(
- files_by_shard[shard_start],
- key=lambda file: file.row_id_range().from_,
- )
- current_group = []
- current_group_end = None
- for file in shard_files:
- file_range = file.row_id_range()
- if not current_group:
- current_group.append(file)
- current_group_end = file_range.to
- elif file_range.from_ <= current_group_end + 1:
- current_group.append(file)
- current_group_end = max(current_group_end, file_range.to)
- else:
- _append_shard_split(
- result,
- current_group,
- shard_start,
- shard_end,
- group["partition"],
- group["bucket"],
- )
- current_group = [file]
- current_group_end = file_range.to
-
- if current_group:
- _append_shard_split(
- result,
- current_group,
- shard_start,
- shard_end,
- group["partition"],
- group["bucket"],
- )
-
- return result
-
-
-def _partition_key(partition):
- values = getattr(partition, "values", None)
- return tuple(values) if values is not None else partition
-
-
-def _append_shard_split(
- result,
- files,
- shard_start,
- shard_end,
- partition,
- bucket,
-):
- group_start = min(file.row_id_range().from_ for file in files)
- group_end = max(file.row_id_range().to for file in files)
- row_range = Range(max(group_start, shard_start), min(group_end, shard_end))
- result.append((
- DataSplit(
- files=list(files),
- partition=partition,
- bucket=bucket,
- raw_convertible=False,
- ),
- row_range,
- ))
-
-
-def _split_one_by_contiguous_row_range(split):
- for file in split.files:
- if file.row_id_range() is None:
- raise ValueError(
- "Cannot build global index because file '%s' has no row id
range."
- % file.file_name
- )
-
- range_helper = RangeHelper(lambda file: file.row_id_range())
- ranges = range_helper.merge_overlapping_ranges(split.files)
- if not ranges:
- return []
-
- result = []
- current_segment = []
- current_max_row_id = None
- for range_files in ranges:
- min_row_id = min(file.row_id_range().from_ for file in range_files)
- max_row_id = max(file.row_id_range().to for file in range_files)
- if (
- not current_segment
- or current_max_row_id is None
- or current_max_row_id >= min_row_id - 1
- ):
- current_segment.extend(range_files)
- current_max_row_id = max_row_id
- else:
- result.append(_copy_split_with_files(split, current_segment))
- current_segment = list(range_files)
- current_max_row_id = max_row_id
-
- if current_segment:
- result.append(_copy_split_with_files(split, current_segment))
- return result
-
-
-def _copy_split_with_files(split, files):
- data_deletion_files = None
- if getattr(split, "data_deletion_files", None) is not None:
- index_by_file = {id(file): i for i, file in enumerate(split.files)}
- data_deletion_files = [
- split.data_deletion_files[index_by_file[id(file)]]
- for file in files
- ]
- return DataSplit(
- files=list(files),
- partition=split.partition,
- bucket=split.bucket,
- raw_convertible=getattr(split, "raw_convertible", False),
- data_deletion_files=data_deletion_files,
- )
-
-
def _extract_sorted_rows(
table: pa.Table,
index_column: str,
row_id_column: str,
key_serializer,
+ row_range: Optional[Range] = None,
):
keys = table.column(index_column).to_pylist()
row_ids = table.column(row_id_column).to_pylist()
@@ -615,7 +423,10 @@ def _extract_sorted_rows(
for key, row_id in zip(keys, row_ids):
if row_id is None:
raise ValueError("Cannot build global index because _ROW_ID is
null.")
- rows.append((key, int(row_id)))
+ row_id = int(row_id)
+ if row_range is not None and not row_range.contains(row_id):
+ continue
+ rows.append((key, row_id))
comparator = key_serializer.create_comparator()
diff --git a/paimon-python/pypaimon/tests/global_index_build_test.py
b/paimon-python/pypaimon/tests/global_index_build_test.py
index 9210d82091..ee73162319 100644
--- a/paimon-python/pypaimon/tests/global_index_build_test.py
+++ b/paimon-python/pypaimon/tests/global_index_build_test.py
@@ -25,10 +25,10 @@ import types
import pyarrow as pa
-from pypaimon.globalindex.create_global_index import (
- _filter_non_indexable_splits,
- _split_by_global_index_shard,
- _split_one_by_contiguous_row_range,
+from pypaimon.globalindex.build_plan import (
+ filter_non_indexable_splits as _filter_non_indexable_splits,
+ split_by_global_index_shard as _split_by_global_index_shard,
+ split_one_by_contiguous_row_range as _split_one_by_contiguous_row_range,
)
from pypaimon.globalindex.key_serializer import create_serializer
from pypaimon.globalindex.tantivy.tantivy_full_text_global_index_reader import
(
@@ -536,7 +536,7 @@ class GlobalIndexBuildTest(
self.assertTrue(external_path.startswith(external_root + '/'))
self.assertTrue(table.file_io.exists(external_path))
- def test_create_global_index_rejects_overlapping_existing_range(self):
+ def test_create_global_index_skips_existing_ranges(self):
table = self._create_table()
self._write_arrow(table, pa.table(
{
@@ -553,12 +553,47 @@ class GlobalIndexBuildTest(
snapshot = table.snapshot_manager().get_latest_snapshot()
self.assertEqual(2, len(IndexFileHandler(table).scan(snapshot)))
- with self.assertRaisesRegex(RuntimeError, 'overlapping row range'):
- table.create_global_index('id', options=options)
+ self.assertEqual(0, table.create_global_index('id', options=options))
latest_snapshot = table.snapshot_manager().get_latest_snapshot()
self.assertEqual(2, len(IndexFileHandler(table).scan(latest_snapshot)))
+ self._write_arrow(table, pa.table(
+ {
+ 'id': [5, 4],
+ 'name': ['e', 'd'],
+ 'age': [50, 40],
+ 'city': ['v', 'u'],
+ },
+ schema=self.pa_schema,
+ ))
+
+ self.assertEqual(1, table.create_global_index('id', options=options))
+ latest_snapshot = table.snapshot_manager().get_latest_snapshot()
+ entries = sorted(
+ IndexFileHandler(table).scan(latest_snapshot),
+ key=lambda entry: (
+ entry.index_file.global_index_meta.row_range_start,
+ entry.index_file.file_name,
+ ),
+ )
+ self.assertEqual(
+ [(0, 3), (0, 3), (4, 5)],
+ [
+ (
+ entry.index_file.global_index_meta.row_range_start,
+ entry.index_file.global_index_meta.row_range_end,
+ )
+ for entry in entries
+ ],
+ )
+ self.assertEqual(0, table.create_global_index('id', options=options))
+ self.assertEqual(
+ 3,
+ len(IndexFileHandler(table).scan(
+ table.snapshot_manager().get_latest_snapshot())),
+ )
+
def test_create_btree_global_index_for_java_scalar_types(self):
schema = pa.schema([
('flag', pa.bool_()),
@@ -707,6 +742,98 @@ class GlobalIndexBuildTest(
],
)
+ def test_create_vindex_global_index_skips_existing_ranges(self):
+ schema = pa.schema([
+ ('id', pa.int32()),
+ ('embedding', pa.list_(pa.float32())),
+ ])
+ table = self._create_table(pa_schema=schema,
options=self.table_options)
+ self._write_arrow(table, pa.table(
+ {
+ 'id': [1, 2, 3],
+ 'embedding': pa.array(
+ [[1.0, 0.0], [0.0, 1.0], [0.5, 0.5]],
+ type=pa.list_(pa.float32()),
+ ),
+ },
+ schema=schema,
+ ))
+
+ old_module = sys.modules.get("paimon_vindex")
+ sys.modules["paimon_vindex"] = types.SimpleNamespace(
+ VectorIndexWriter=_FakeVectorIndexWriter)
+ _FakeVectorIndexWriter.instances = []
+ options = {
+ 'global-index.row-count-per-shard': '2',
+ 'ivf-flat.dimension': '2',
+ }
+ try:
+ self.assertEqual(
+ 2,
+ table.create_global_index(
+ 'embedding',
+ index_type='ivf-flat',
+ options=options,
+ ),
+ )
+ self.assertEqual(
+ 0,
+ table.create_global_index(
+ 'embedding',
+ index_type='ivf-flat',
+ options=options,
+ ),
+ )
+
+ self._write_arrow(table, pa.table(
+ {
+ 'id': [4, 5],
+ 'embedding': pa.array(
+ [[0.2, 0.8], [0.9, 0.1]],
+ type=pa.list_(pa.float32()),
+ ),
+ },
+ schema=schema,
+ ))
+
+ self.assertEqual(
+ 2,
+ table.create_global_index(
+ 'embedding',
+ index_type='ivf-flat',
+ options=options,
+ ),
+ )
+ self.assertEqual(
+ 0,
+ table.create_global_index(
+ 'embedding',
+ index_type='ivf-flat',
+ options=options,
+ ),
+ )
+ finally:
+ if old_module is None:
+ sys.modules.pop("paimon_vindex", None)
+ else:
+ sys.modules["paimon_vindex"] = old_module
+
+ snapshot = table.snapshot_manager().get_latest_snapshot()
+ entries = sorted(
+ IndexFileHandler(table).scan(snapshot),
+ key=lambda entry:
entry.index_file.global_index_meta.row_range_start,
+ )
+ self.assertEqual(
+ [(0, 1), (2, 2), (3, 3), (4, 4)],
+ [
+ (
+ entry.index_file.global_index_meta.row_range_start,
+ entry.index_file.global_index_meta.row_range_end,
+ )
+ for entry in entries
+ ],
+ )
+
def test_create_tantivy_fulltext_global_index_from_python(self):
schema = pa.schema([
('id', pa.int32()),
@@ -973,6 +1100,25 @@ class GlobalIndexBuildTest(
],
)
+ def test_split_by_global_index_shard_uses_unindexed_ranges(self):
+ split = _FakeSplit([
+ _FakeFile('a', 0, 100),
+ ])
+
+ shards = _split_by_global_index_shard(
+ [split], 100, [Range(0, 9), Range(90, 99)])
+
+ self.assertEqual(
+ [
+ (['a'], 0, 9),
+ (['a'], 90, 99),
+ ],
+ [
+ ([file.file_name for file in shard.files], row_range.from_,
row_range.to)
+ for shard, row_range in shards
+ ],
+ )
+
def test_split_by_global_index_shard_skips_files_without_row_ids(self):
split = _FakeSplit([
_FakeFile('no-row-id', None, 2),
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 b2112149a1..c8dc0bc011 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
@@ -18,12 +18,12 @@
package org.apache.paimon.spark.globalindex;
+import org.apache.paimon.Snapshot;
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;
import org.apache.paimon.options.Options;
import org.apache.paimon.partition.PartitionPredicate;
@@ -32,12 +32,10 @@ import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.sink.CommitMessage;
import org.apache.paimon.table.sink.CommitMessageSerializer;
-import org.apache.paimon.table.source.DataSplit;
import org.apache.paimon.table.source.ReadBuilder;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.CloseableIterator;
-import org.apache.paimon.utils.FileStorePathFactory;
import org.apache.paimon.utils.InstantiationUtil;
import org.apache.paimon.utils.Pair;
import org.apache.paimon.utils.Range;
@@ -46,12 +44,11 @@ import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation;
+import javax.annotation.Nullable;
+
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
-import java.util.Comparator;
-import java.util.HashMap;
-import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.function.BiFunction;
@@ -107,8 +104,17 @@ public class DefaultGlobalIndexTopoBuilder implements
GlobalIndexTopologyBuilder
rowsPerShard > 0,
"Option 'global-index.row-count-per-shard' must be greater
than 0.");
+ Snapshot snapshot = table.snapshotManager().latestSnapshot();
+ if (snapshot == null) {
+ return Collections.emptyList();
+ }
List<ManifestEntry> entries =
-
table.store().newScan().withPartitionFilter(partitionPredicate).plan().files();
+ table.store()
+ .newScan()
+ .withSnapshot(snapshot)
+ .withPartitionFilter(partitionPredicate)
+ .plan()
+ .files();
List<DataField> indexFields = new ArrayList<>();
indexFields.add(indexField);
indexFields.addAll(extraFields);
@@ -119,33 +125,36 @@ public class DefaultGlobalIndexTopoBuilder implements
GlobalIndexTopologyBuilder
GlobalIndexBuilderUtils.findMinNonIndexableRowId(
schemaManager, entries, indexColumns);
entries = GlobalIndexBuilderUtils.filterEntriesBefore(entries,
boundaryRowId);
+ List<Range> rowRangesToBuild =
+ GlobalIndexBuilderUtils.unindexedRowRanges(
+ table, snapshot, indexType, indexFields,
partitionPredicate);
+ if (rowRangesToBuild.isEmpty()) {
+ return Collections.emptyList();
+ }
// generate splits for each partition && shard
- Map<BinaryRow, List<IndexedSplit>> splits = split(table, entries,
rowsPerShard);
+ List<IndexedSplit> splits =
+ GlobalIndexBuilderUtils.createShardIndexedSplits(
+ table, entries, rowsPerShard, rowRangesToBuild);
JavaSparkContext javaSparkContext = new
JavaSparkContext(spark.sparkContext());
List<Pair<byte[], byte[]>> taskList = new ArrayList<>();
- for (Map.Entry<BinaryRow, List<IndexedSplit>> entry :
splits.entrySet()) {
- BinaryRow partition = entry.getKey();
- List<IndexedSplit> partitions = entry.getValue();
-
- for (IndexedSplit indexedSplit : partitions) {
- checkArgument(
- indexedSplit.rowRanges().size() == 1,
- "Each IndexedSplit should contain exactly one row
range.");
- DefaultGlobalIndexBuilder builder =
- new DefaultGlobalIndexBuilder(
- table,
- partition,
- readType,
- indexField,
- extraFields,
- indexType,
- indexedSplit.rowRanges().get(0),
- options);
- byte[] builderBytes =
InstantiationUtil.serializeObject(builder);
- byte[] splitBytes =
InstantiationUtil.serializeObject(indexedSplit);
- taskList.add(Pair.of(builderBytes, splitBytes));
- }
+ for (IndexedSplit indexedSplit : splits) {
+ checkArgument(
+ indexedSplit.rowRanges().size() == 1,
+ "Each IndexedSplit should contain exactly one row range.");
+ DefaultGlobalIndexBuilder builder =
+ new DefaultGlobalIndexBuilder(
+ table,
+ indexedSplit.dataSplit().partition(),
+ readType,
+ indexField,
+ extraFields,
+ indexType,
+ indexedSplit.rowRanges().get(0),
+ options);
+ byte[] builderBytes = InstantiationUtil.serializeObject(builder);
+ byte[] splitBytes =
InstantiationUtil.serializeObject(indexedSplit);
+ taskList.add(Pair.of(builderBytes, splitBytes));
}
if (taskList.isEmpty()) {
@@ -176,21 +185,9 @@ public class DefaultGlobalIndexTopoBuilder implements
GlobalIndexTopologyBuilder
}
}
- private static Map<BinaryRow, List<IndexedSplit>> split(
- FileStoreTable table, List<ManifestEntry> entries, long
rowsPerShard) {
- FileStorePathFactory pathFactory = table.store().pathFactory();
-
- // Group manifest entries by partition
- Map<BinaryRow, List<ManifestEntry>> entriesByPartition =
-
entries.stream().collect(Collectors.groupingBy(ManifestEntry::partition));
-
- return groupFilesIntoShardsByPartition(
- entriesByPartition, rowsPerShard, pathFactory::bucketPath);
- }
-
/**
- * Groups files into shards by partition. This method is extracted from
split() to make it more
- * testable.
+ * Groups files into shards by partition. This method delegates to the
generic global index
+ * build planner and keeps the previous test surface stable.
*
* @param entriesByPartition manifest entries grouped by partition
* @param rowsPerShard number of rows per shard
@@ -201,130 +198,24 @@ public class DefaultGlobalIndexTopoBuilder implements
GlobalIndexTopologyBuilder
Map<BinaryRow, List<ManifestEntry>> entriesByPartition,
long rowsPerShard,
BiFunction<BinaryRow, Integer, Path> pathFactory) {
- Map<BinaryRow, List<IndexedSplit>> result = new HashMap<>();
-
- for (Map.Entry<BinaryRow, List<ManifestEntry>> partitionEntry :
- entriesByPartition.entrySet()) {
- BinaryRow partition = partitionEntry.getKey();
- List<ManifestEntry> partitionEntries = partitionEntry.getValue();
-
- // Group files into shards - a file may belong to multiple shards
- Map<Long, List<DataFileMeta>> filesByShard = new LinkedHashMap<>();
-
- for (ManifestEntry entry : partitionEntries) {
- DataFileMeta file = entry.file();
- Long firstRowId = file.firstRowId();
- if (firstRowId == null) {
- continue; // Skip files without row tracking
- }
-
- // Calculate the row ID range this file covers
- Range fileRange = file.nonNullRowIdRange();
-
- // Calculate which shards this file overlaps with
- long startShardId = fileRange.from / rowsPerShard;
- long endShardId = fileRange.to / rowsPerShard;
-
- // Add this file to all shards it overlaps with
- for (long shardId = startShardId; shardId <= endShardId;
shardId++) {
- long shardStartRowId = shardId * rowsPerShard;
- filesByShard.computeIfAbsent(shardStartRowId, k -> new
ArrayList<>()).add(file);
- }
- }
-
- // Create DataSplit for each shard with exact ranges
- List<IndexedSplit> shardSplits = new ArrayList<>();
- 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;
- }
-
- // Sort files by firstRowId to ensure sequential order
-
shardFiles.sort(Comparator.comparingLong(DataFileMeta::nonNullFirstRowId));
-
- // Group contiguous files and create separate DataSplits for
each group
- 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()) {
- // Start a new group
- currentGroup.add(file);
- currentGroupEnd = fileEnd;
- } else if (fileStart <= currentGroupEnd + 1) {
- // File is contiguous with current group (adjacent or
overlapping)
- currentGroup.add(file);
- currentGroupEnd = Math.max(currentGroupEnd, fileEnd);
- } else {
- // Gap detected, finalize current group and start a
new one
- createDataSplitForGroup(
- currentGroup,
- shardStart,
- shardEnd,
- partition,
- pathFactory,
- shardSplits);
- currentGroup = new ArrayList<>();
- currentGroup.add(file);
- currentGroupEnd = fileEnd;
- }
- }
-
- // Don't forget to process the last group
- if (!currentGroup.isEmpty()) {
- createDataSplitForGroup(
- currentGroup,
- shardStart,
- shardEnd,
- partition,
- pathFactory,
- shardSplits);
- }
- }
-
- if (!shardSplits.isEmpty()) {
- result.put(partition, shardSplits);
- }
- }
-
- return result;
+ return groupFilesIntoShardsByPartition(entriesByPartition,
rowsPerShard, pathFactory, null);
}
- private static void createDataSplitForGroup(
- List<DataFileMeta> files,
- long shardStart,
- long shardEnd,
- BinaryRow partition,
+ public static Map<BinaryRow, List<IndexedSplit>>
groupFilesIntoShardsByPartition(
+ Map<BinaryRow, List<ManifestEntry>> entriesByPartition,
+ long rowsPerShard,
BiFunction<BinaryRow, Integer, Path> pathFactory,
- List<IndexedSplit> shardSplits) {
- // Calculate the actual row range covered by the files
- long groupMinRowId = files.get(0).nonNullFirstRowId();
- long groupMaxRowId =
- files.stream().mapToLong(f ->
f.nonNullRowIdRange().to).max().getAsLong();
-
- // Clamp to shard boundaries
- // Range.from >= shardStart, Range.to <= shardEnd
- long rangeFrom = Math.max(groupMinRowId, shardStart);
- long rangeTo = Math.min(groupMaxRowId, shardEnd);
-
- Range range = new Range(rangeFrom, rangeTo);
-
- DataSplit dataSplit =
- DataSplit.builder()
- .withPartition(partition)
- .withBucket(0)
- .withDataFiles(files)
- .withBucketPath(pathFactory.apply(partition,
0).toString())
- .rawConvertible(false)
- .build();
-
- shardSplits.add(new IndexedSplit(dataSplit,
Collections.singletonList(range), null));
+ @Nullable List<Range> rowRangesToBuild) {
+ List<ManifestEntry> entries =
+ entriesByPartition.values().stream()
+ .flatMap(List::stream)
+ .collect(Collectors.toList());
+ return GlobalIndexBuilderUtils.createShardIndexedSplits(
+ entries,
+ rowsPerShard,
+ (partition, bucket) -> pathFactory.apply(partition,
bucket).toString(),
+ rowRangesToBuild)
+ .stream()
+ .collect(Collectors.groupingBy(split ->
split.dataSplit().partition()));
}
}
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/sorted/SortedIndexTopoBuilder.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/sorted/SortedIndexTopoBuilder.java
index 2174418009..9a23d1944a 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/sorted/SortedIndexTopoBuilder.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/sorted/SortedIndexTopoBuilder.java
@@ -91,7 +91,8 @@ public class SortedIndexTopoBuilder implements
GlobalIndexTopologyBuilder {
indexBuilder =
indexBuilder.withPartitionPredicate(partitionPredicate);
}
- Optional<Pair<RowRangeIndex, List<DataSplit>>> indexRangeAndSplits =
indexBuilder.scan();
+ Optional<Pair<RowRangeIndex, List<DataSplit>>> indexRangeAndSplits =
+ indexBuilder.incrementalScan();
if (!indexRangeAndSplits.isPresent()) {
return Collections.emptyList();
}
diff --git
a/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/procedure/CreateGlobalIndexProcedureTest.java
b/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/procedure/CreateGlobalIndexProcedureTest.java
index 991f5d8627..f40c751fe2 100644
---
a/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/procedure/CreateGlobalIndexProcedureTest.java
+++
b/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/procedure/CreateGlobalIndexProcedureTest.java
@@ -231,6 +231,34 @@ public class CreateGlobalIndexProcedureTest {
assertThat(shardToSplit.get(shard2).dataFiles()).contains(file3);
}
+ @Test
+ void testGroupFilesIntoShardsByPartitionUsesUnindexedRanges() {
+ BinaryRow partition = createPartition(0);
+ DataFileMeta indexedLaterFile = createDataFileMeta(1000L, 100L);
+ DataFileMeta unindexedEarlierFile = createDataFileMeta(0L, 100L);
+
+ Map<BinaryRow, List<ManifestEntry>> entriesByPartition = new
HashMap<>();
+ entriesByPartition.put(
+ partition,
+ Arrays.asList(
+ createManifestEntry(partition, indexedLaterFile),
+ createManifestEntry(partition, unindexedEarlierFile)));
+
+ Map<BinaryRow, List<IndexedSplit>> result =
+ DefaultGlobalIndexTopoBuilder.groupFilesIntoShardsByPartition(
+ entriesByPartition,
+ 1000L,
+ pathFactory,
+ Collections.singletonList(new Range(0L, 99L)));
+
+ assertThat(result).hasSize(1);
+ List<IndexedSplit> shardSplits = result.get(partition);
+ assertThat(shardSplits).hasSize(1);
+ assertThat(shardSplits.get(0).rowRanges()).containsExactly(new
Range(0L, 99L));
+ assertThat(shardSplits.get(0).dataSplit().dataFiles())
+ .containsExactly(unindexedEarlierFile);
+ }
+
@Test
void testGroupFilesIntoShardsByPartitionMultiplePartitions() {
// Create two partitions
