You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2026/08/06 04:15:09 UTC
(paimon) branch master updated: [core][flink][spark] Refactor global index building and support incremental refresh (#8818)
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 875699e843 [core][flink][spark] Refactor global index building and support incremental refresh (#8818)
875699e843 is described below
commit 875699e843256772f32f648266606abc444c630e
Author: YeJunHao <41...@users.noreply.github.com>
AuthorDate: Thu Aug 6 12:14:59 2026 +0800
[core][flink][spark] Refactor global index building and support incremental refresh (#8818)
---
docs/docs/multimodal-table/global-index.mdx | 2 +-
docs/generated/core_configuration.html | 2 +-
.../main/java/org/apache/paimon/CoreOptions.java | 2 +-
.../DataEvolutionRowIdReassigner.java | 3 +-
.../DataEvolutionGlobalIndexRefreshPlanner.java | 230 ++++++++++
.../globalindex/GlobalIndexBuilderUtils.java | 314 ++++++++-----
.../org/apache/paimon/globalindex/ScanResult.java | 70 +++
.../generic/GenericGlobalIndexScanner.java | 170 +++++++
.../sorted/SortedGlobalIndexBuilder.java | 469 -------------------
.../sorted/SortedGlobalIndexScanner.java | 191 ++++++++
.../sorted/SortedGlobalIndexWriter.java | 145 ++++++
.../paimon/index/DataEvolutionIndexSourceMeta.java | 101 +++++
.../paimon/manifest/IndexManifestFileHandler.java | 15 +-
.../operation/DataEvolutionFileStoreScan.java | 24 +-
.../apache/paimon/utils/DataEvolutionUtils.java | 22 +
.../test/java/org/apache/paimon/JavaPyE2ETest.java | 106 ++---
.../DataEvolutionRowIdReassignerTest.java | 75 +++-
...DataEvolutionGlobalIndexRefreshPlannerTest.java | 494 +++++++++++++++++++++
.../globalindex/GlobalIndexBuilderUtilsTest.java | 139 +++++-
...enericGlobalIndexScannerDeletionVectorTest.java | 10 +-
.../generic/GenericGlobalIndexScannerTest.java | 111 +++++
.../sorted/SortedGlobalIndexBuilderSplitTest.java | 131 ------
...Test.java => SortedGlobalIndexScannerTest.java} | 164 +++----
.../sorted/SortedGlobalIndexTestUtils.java | 96 ++++
.../sorted/SortedGlobalIndexWriterTest.java | 128 ++++++
.../index/DataEvolutionIndexSourceMetaTest.java | 100 +++++
.../manifest/IndexManifestFileHandlerTest.java | 69 +++
.../metastore/VisibilityWaitCallbackTest.java | 34 +-
.../paimon/operation/FileStoreCommitTest.java | 43 ++
.../paimon/table/BtreeGlobalIndexTableTest.java | 30 +-
.../table/source/FullTextSearchBuilderTest.java | 9 +-
.../table/source/VectorSearchBuilderTest.java | 3 +-
.../paimon/utils/DataEvolutionUtilsTest.java | 83 +++-
.../globalindex/GenericGlobalIndexBuilder.java | 92 ----
.../flink/globalindex/GenericIndexTopoBuilder.java | 248 ++---------
.../flink/globalindex/SortedIndexTopoBuilder.java | 102 +++--
.../flink/LuminaVectorGlobalIndexITCase.java | 158 ++++---
.../paimon/flink/SortedGlobalIndexITCase.java | 243 ++++++++++
.../globalindex/GenericIndexTopoBuilderTest.java | 344 +-------------
.../globalindex/SortedIndexTopoBuilderTest.java | 31 +-
.../procedure/FullTextSearchProcedureITCase.java | 91 ++++
.../procedure/VectorSearchProcedureITCase.java | 3 +-
.../globalindex/DefaultGlobalIndexBuilder.java | 32 +-
.../globalindex/DefaultGlobalIndexTopoBuilder.java | 81 ++--
.../globalindex/sorted/SortedIndexTopoBuilder.java | 63 ++-
.../procedure/CreateGlobalIndexProcedureTest.scala | 104 +++++
.../paimon/spark/sql/FullTextSearchTest.scala | 7 +
.../paimon/spark/sql/LuminaVectorIndexTest.scala | 135 ++++++
48 files changed, 3588 insertions(+), 1731 deletions(-)
diff --git a/docs/docs/multimodal-table/global-index.mdx b/docs/docs/multimodal-table/global-index.mdx
index 26ca42cc14..63295276f1 100644
--- a/docs/docs/multimodal-table/global-index.mdx
+++ b/docs/docs/multimodal-table/global-index.mdx
@@ -399,7 +399,7 @@ These table options affect global index build and read behavior:
| `vector-index.search-mode` | `fast` | Search mode for vector queries. |
| `full-text-index.search-mode` | `fast` | Search mode for full-text queries. |
| `global-index.external-path` | Not set | Root directory for global index files. If not set, files are stored under the table index directory. |
-| `global-index.column-update-action` | `THROW_ERROR` | Action for updates to indexed columns. `THROW_ERROR` rejects the update, `DROP_PARTITION_INDEX` drops affected partition indexes, and `IGNORE` keeps existing index files unchanged. Use `IGNORE` only when a stale index is acceptable; for example, a vector updated from `NULL` to a value remains invisible until the index is rebuilt. |
+| `global-index.column-update-action` | `THROW_ERROR` | Action for updates to indexed columns. `THROW_ERROR` rejects the update, `DROP_PARTITION_INDEX` drops affected partition indexes, and `IGNORE` keeps existing index files unchanged during the update and causes the next incremental index build to scan active data manifests and rebuild affected row ranges. Index files created before source metadata was introduced are not refreshed automatically. |
| `sorted-index.records-per-range` | `10000000` | Expected number of records per sorted global index file for BTree and Bitmap builds. |
| `sorted-index.build.max-parallelism` | `4096` | Maximum Flink or Spark parallelism for building sorted global indexes. |
| `global-index.row-count-per-shard` | `100000` | Target row count per shard for non-sorted global index builds such as vector and full-text indexes. |
diff --git a/docs/generated/core_configuration.html b/docs/generated/core_configuration.html
index bb5fba90d3..0c0c1a0a8d 100644
--- a/docs/generated/core_configuration.html
+++ b/docs/generated/core_configuration.html
@@ -798,7 +798,7 @@ under the License.
<td><h5>global-index.column-update-action</h5></td>
<td style="word-wrap: break-word;">THROW_ERROR</td>
<td><p>Enum</p></td>
- <td>Defines the action to take when an update modifies columns that are covered by a global index. IGNORE leaves existing index files unchanged and may make the index stale.<br /><br />Possible values:<ul><li>"THROW_ERROR"</li><li>"DROP_PARTITION_INDEX"</li><li>"IGNORE"</li></ul></td>
+ <td>Defines the action to take when an update modifies columns that are covered by a global index. IGNORE leaves existing index files unchanged during the update and enables a later incremental index build to refresh affected row ranges.<br /><br />Possible values:<ul><li>"THROW_ERROR"</li><li>"DROP_PARTITION_INDEX"</li><li>"IGNORE"</li></ul></td>
</tr>
<tr>
<td><h5>global-index.enabled</h5></td>
diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
index 2454b84c80..96fdbf731b 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -2607,7 +2607,7 @@ public class CoreOptions implements Serializable {
.defaultValue(GlobalIndexColumnUpdateAction.THROW_ERROR)
.withDescription(
"Defines the action to take when an update modifies columns that are covered by a global index. "
- + "IGNORE leaves existing index files unchanged and may make the index stale.");
+ + "IGNORE leaves existing index files unchanged during the update and enables a later incremental index build to refresh affected row ranges.");
public static final ConfigOption<MemorySize> LOOKUP_MERGE_BUFFER_SIZE =
key("lookup.merge-buffer-size")
diff --git a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassigner.java b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassigner.java
index 3a731143f8..05d173ed0a 100644
--- a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassigner.java
+++ b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassigner.java
@@ -670,7 +670,8 @@ public class DataEvolutionRowIdReassigner {
rewrittenRange.to,
globalIndex.indexFieldId(),
globalIndex.extraFieldIds(),
- globalIndex.indexMeta());
+ globalIndex.indexMeta(),
+ globalIndex.sourceMeta());
IndexFileMeta newIndexFile =
new IndexFileMeta(
indexFile.indexType(),
diff --git a/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexRefreshPlanner.java b/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexRefreshPlanner.java
new file mode 100644
index 0000000000..44ed39685d
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexRefreshPlanner.java
@@ -0,0 +1,230 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.globalindex;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.index.DataEvolutionIndexSourceMeta;
+import org.apache.paimon.index.GlobalIndexMeta;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.manifest.FileKind;
+import org.apache.paimon.manifest.IndexManifestEntry;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.utils.Pair;
+import org.apache.paimon.utils.Range;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.NavigableMap;
+import java.util.Set;
+import java.util.TreeMap;
+
+import static org.apache.paimon.utils.DataEvolutionUtils.fileFieldIds;
+
+/** Plans existing global index files which need refresh after data-evolution updates. */
+public final class DataEvolutionGlobalIndexRefreshPlanner {
+
+ private DataEvolutionGlobalIndexRefreshPlanner() {}
+
+ public static List<IndexManifestEntry> findIndexesToRefresh(
+ SchemaManager schemaManager,
+ List<ManifestEntry> dataEntries,
+ List<IndexManifestEntry> indexEntries,
+ List<DataField> indexedFields) {
+ Set<Integer> indexedFieldIds = new HashSet<>();
+ for (DataField field : indexedFields) {
+ indexedFieldIds.add(field.id());
+ }
+
+ Map<Pair<BinaryRow, Integer>, RefreshGroup> groups = new HashMap<>();
+ for (int i = 0; i < indexEntries.size(); i++) {
+ IndexManifestEntry indexEntry = indexEntries.get(i);
+ GlobalIndexMeta indexMeta = indexEntry.indexFile().globalIndexMeta();
+ if (indexEntry.kind() != FileKind.ADD
+ || indexMeta == null
+ || !matchesFields(indexMeta, indexedFields)) {
+ continue;
+ }
+
+ byte[] sourceMeta = indexMeta.sourceMeta();
+ if (!DataEvolutionIndexSourceMeta.isDataEvolutionMeta(sourceMeta)) {
+ // Legacy indexes have no trustworthy scan baseline and require an explicit rebuild.
+ continue;
+ }
+ long scanSnapshotId =
+ DataEvolutionIndexSourceMeta.deserialize(sourceMeta).scanSnapshotId();
+ groups.computeIfAbsent(
+ Pair.of(indexEntry.partition(), indexEntry.bucket()),
+ key -> new RefreshGroup())
+ .addIndex(i, indexMeta.rowRange(), scanSnapshotId);
+ }
+
+ Map<Pair<Long, List<String>>, Set<Integer>> fileFieldIdsCache = new HashMap<>();
+ for (ManifestEntry dataEntry : dataEntries) {
+ DataFileMeta file = dataEntry.file();
+ if (dataEntry.kind() != FileKind.ADD || file.firstRowId() == null) {
+ continue;
+ }
+
+ RefreshGroup group = groups.get(Pair.of(dataEntry.partition(), dataEntry.bucket()));
+ if (group == null || !group.mayContainUpdate(file)) {
+ continue;
+ }
+
+ Set<Integer> physicalFieldIds =
+ fileFieldIdsCache.computeIfAbsent(
+ Pair.of(file.schemaId(), file.writeCols()),
+ key -> fileFieldIds(schemaManager::schema, file));
+ if (!disjoint(indexedFieldIds, physicalFieldIds)) {
+ group.addDataFile(file);
+ }
+ }
+
+ boolean[] indexesToRefresh = new boolean[indexEntries.size()];
+ for (RefreshGroup group : groups.values()) {
+ group.markIndexesToRefresh(indexesToRefresh);
+ }
+
+ List<IndexManifestEntry> result = new ArrayList<>();
+ for (int i = 0; i < indexEntries.size(); i++) {
+ if (indexesToRefresh[i]) {
+ result.add(indexEntries.get(i));
+ }
+ }
+ return result;
+ }
+
+ private static final class RefreshGroup {
+
+ private final List<IndexQuery> indexes = new ArrayList<>();
+ private final List<DataFileMeta> dataFiles = new ArrayList<>();
+ private final MergedRanges indexedRanges = new MergedRanges();
+ private long minScanSnapshotId = Long.MAX_VALUE;
+
+ private void addIndex(int ordinal, Range rowRange, long scanSnapshotId) {
+ indexes.add(new IndexQuery(ordinal, rowRange, scanSnapshotId));
+ indexedRanges.add(rowRange);
+ minScanSnapshotId = Math.min(minScanSnapshotId, scanSnapshotId);
+ }
+
+ private boolean mayContainUpdate(DataFileMeta file) {
+ return file.maxSequenceNumber() > minScanSnapshotId
+ && indexedRanges.intersects(file.nonNullRowIdRange());
+ }
+
+ private void addDataFile(DataFileMeta file) {
+ dataFiles.add(file);
+ }
+
+ private void markIndexesToRefresh(boolean[] result) {
+ // As scan watermarks decrease, eligible data files only grow.
+ dataFiles.sort(Comparator.comparingLong(DataFileMeta::maxSequenceNumber).reversed());
+ indexes.sort((left, right) -> Long.compare(right.scanSnapshotId, left.scanSnapshotId));
+
+ MergedRanges updatedRanges = new MergedRanges();
+ int nextFile = 0;
+ for (IndexQuery index : indexes) {
+ while (nextFile < dataFiles.size()
+ && dataFiles.get(nextFile).maxSequenceNumber() > index.scanSnapshotId) {
+ updatedRanges.add(dataFiles.get(nextFile).nonNullRowIdRange());
+ nextFile++;
+ }
+ if (updatedRanges.intersects(index.rowRange)) {
+ result[index.ordinal] = true;
+ }
+ }
+ }
+ }
+
+ private static final class IndexQuery {
+
+ private final int ordinal;
+ private final Range rowRange;
+ private final long scanSnapshotId;
+
+ private IndexQuery(int ordinal, Range rowRange, long scanSnapshotId) {
+ this.ordinal = ordinal;
+ this.rowRange = rowRange;
+ this.scanSnapshotId = scanSnapshotId;
+ }
+ }
+
+ /** Dynamically merged inclusive ranges supporting logarithmic intersection checks. */
+ private static final class MergedRanges {
+
+ private final NavigableMap<Long, Long> ranges = new TreeMap<>();
+
+ private void add(Range range) {
+ long from = range.from;
+ long to = range.to;
+
+ Map.Entry<Long, Long> floor = ranges.floorEntry(from);
+ if (floor != null && floor.getValue() >= from) {
+ from = floor.getKey();
+ to = Math.max(to, floor.getValue());
+ ranges.remove(floor.getKey());
+ }
+
+ Map.Entry<Long, Long> next = ranges.ceilingEntry(from);
+ while (next != null && next.getKey() <= to) {
+ to = Math.max(to, next.getValue());
+ ranges.remove(next.getKey());
+ next = ranges.ceilingEntry(from);
+ }
+ ranges.put(from, to);
+ }
+
+ private boolean intersects(Range range) {
+ Map.Entry<Long, Long> floor = ranges.floorEntry(range.to);
+ return floor != null && floor.getValue() >= range.from;
+ }
+ }
+
+ private static boolean matchesFields(GlobalIndexMeta meta, List<DataField> fields) {
+ if (fields.isEmpty() || meta.indexFieldId() != fields.get(0).id()) {
+ return false;
+ }
+ int[] expectedExtraFields =
+ fields.size() == 1
+ ? null
+ : fields.subList(1, fields.size()).stream()
+ .mapToInt(DataField::id)
+ .toArray();
+ int[] actualExtraFields = meta.extraFieldIds();
+ if (actualExtraFields == null || actualExtraFields.length == 0) {
+ return expectedExtraFields == null || expectedExtraFields.length == 0;
+ }
+ return expectedExtraFields != null && Arrays.equals(actualExtraFields, expectedExtraFields);
+ }
+
+ private static boolean disjoint(Set<Integer> left, Set<Integer> right) {
+ for (Integer value : left) {
+ if (right.contains(value)) {
+ return false;
+ }
+ }
+ return true;
+ }
+}
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 0537e50114..5e838e204b 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
@@ -29,13 +29,17 @@ 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.operation.FileStoreScan;
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.table.source.Split;
import org.apache.paimon.types.DataField;
+import org.apache.paimon.utils.Pair;
import org.apache.paimon.utils.Range;
+import org.apache.paimon.utils.RangeHelper;
+import org.apache.paimon.utils.RowRangeIndex;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -52,6 +56,7 @@ import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.function.BiFunction;
+import java.util.function.Supplier;
import static org.apache.paimon.utils.Preconditions.checkArgument;
@@ -70,7 +75,15 @@ public class GlobalIndexBuilderUtils {
List<ResultEntry> entries)
throws IOException {
return toIndexFileMetas(
- fileIO, indexPathFactory, options, range, indexFieldId, null, indexType, entries);
+ fileIO,
+ indexPathFactory,
+ options,
+ range,
+ indexFieldId,
+ null,
+ indexType,
+ entries,
+ null);
}
/**
@@ -86,52 +99,52 @@ public class GlobalIndexBuilderUtils {
Range range,
List<DataField> fields,
String indexType,
- List<ResultEntry> entries)
+ List<ResultEntry> entries,
+ @Nullable byte[] sourceMeta)
throws IOException {
- // 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 = extraFieldIds(fields);
return toIndexFileMetas(
fileIO,
indexPathFactory,
options,
range,
- indexFieldId,
- extraFieldIds,
+ fields.get(0).id(),
+ extraFieldIds(fields),
indexType,
- entries);
+ entries,
+ sourceMeta);
}
public static List<Range> unindexedRowRanges(
- FileStoreTable table,
- @Nullable Snapshot snapshot,
- String indexType,
- List<DataField> fields,
- @Nullable PartitionPredicate partitionPredicate) {
+ @Nullable Snapshot snapshot, List<IndexManifestEntry> currentIndexes) {
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);
+ List<Range> indexedRanges = new ArrayList<>(currentIndexes.size());
+ for (IndexManifestEntry entry : currentIndexes) {
+ GlobalIndexMeta meta = entry.indexFile().globalIndexMeta();
+ if (meta != null) {
+ indexedRanges.add(meta.rowRange());
+ }
+ }
+ indexedRanges = Range.sortAndMergeOverlap(indexedRanges, true);
return Range.sortAndMergeOverlap(dataRange.exclude(indexedRanges), true);
}
- public static List<Range> indexedRowRanges(
+ public static List<IndexManifestEntry> currentIndexEntries(
FileStoreTable table,
- @Nullable Snapshot snapshot,
+ Snapshot snapshot,
String indexType,
List<DataField> fields,
@Nullable PartitionPredicate partitionPredicate) {
- if (snapshot == null || fields.isEmpty()) {
+ if (fields.isEmpty()) {
return Collections.emptyList();
}
int indexFieldId = fields.get(0).id();
int[] extraFieldIds = extraFieldIds(fields);
- List<Range> ranges = new ArrayList<>();
+ List<IndexManifestEntry> entries = new ArrayList<>();
for (IndexManifestEntry entry :
table.store().newIndexFileHandler().scan(snapshot, indexType)) {
if (partitionPredicate != null && !partitionPredicate.test(entry.partition())) {
@@ -147,20 +160,183 @@ public class GlobalIndexBuilderUtils {
if (!sameExtraFieldIds(meta.extraFieldIds(), extraFieldIds)) {
continue;
}
- ranges.add(meta.rowRange());
+ entries.add(entry);
}
- return Range.sortAndMergeOverlap(ranges, true);
+ return entries;
+ }
+
+ public static List<Pair<Range, Split>> splitByRowRangeIndex(
+ RowRangeIndex rowRangeIndex, DataSplit dataSplit) {
+ if (rowRangeIndex == null) {
+ Range range = calcRowRange(dataSplit);
+ return range == null
+ ? Collections.emptyList()
+ : Collections.singletonList(Pair.of(range, dataSplit));
+ }
+
+ List<Pair<Range, Split>> result = new ArrayList<>();
+ for (Split split :
+ DataEvolutionBatchScan.wrapToIndexSplits(
+ Collections.singletonList(dataSplit), rowRangeIndex, null)
+ .splits()) {
+ IndexedSplit indexedSplit = (IndexedSplit) split;
+ for (Range rowRange : indexedSplit.rowRanges()) {
+ result.add(
+ Pair.of(
+ rowRange,
+ new IndexedSplit(
+ indexedSplit.dataSplit(),
+ Collections.singletonList(rowRange),
+ null)));
+ }
+ }
+ return result;
}
@Nullable
- public static List<Range> rowRangesAfter(long maxIndexedRowId) {
- if (maxIndexedRowId < 0) {
+ public static Range calcRowRange(DataSplit dataSplit) {
+ List<Range> ranges = calcRowRanges(Collections.singletonList(dataSplit));
+ if (ranges.isEmpty()) {
return null;
}
- if (maxIndexedRowId == Long.MAX_VALUE) {
- return Collections.emptyList();
+ return new Range(ranges.get(0).from, ranges.get(ranges.size() - 1).to);
+ }
+
+ public static List<Range> calcRowRanges(List<DataSplit> dataSplits) {
+ List<Range> ranges = new ArrayList<>();
+ for (DataSplit dataSplit : dataSplits) {
+ for (DataFileMeta file : dataSplit.dataFiles()) {
+ ranges.add(file.nonNullRowIdRange());
+ }
+ }
+ return Range.sortAndMergeOverlap(ranges, true);
+ }
+
+ public static List<DataSplit> splitByContiguousRowRange(List<DataSplit> splits) {
+ List<DataSplit> result = new ArrayList<>();
+ for (DataSplit split : splits) {
+ result.addAll(splitByContiguousRowRange(split));
}
- return Collections.singletonList(new Range(maxIndexedRowId + 1, Long.MAX_VALUE));
+ return result;
+ }
+
+ public static Map<BinaryRow, Map<Range, List<Split>>> groupSplitsByRange(
+ RowRangeIndex rowRangeIndex, List<DataSplit> splits) {
+ Map<BinaryRow, List<Pair<Range, Split>>> partitionSplitRanges = new HashMap<>();
+ for (DataSplit split : splits) {
+ for (Pair<Range, Split> keyPair : splitByRowRangeIndex(rowRangeIndex, split)) {
+ Range splitRange = keyPair.getKey();
+ Split splitWithRange = keyPair.getValue();
+ if (splitRange == null) {
+ continue;
+ }
+ BinaryRow partition = split.partition();
+ partitionSplitRanges
+ .computeIfAbsent(partition, p -> new ArrayList<>())
+ .add(Pair.of(splitRange, splitWithRange));
+ }
+ }
+
+ Map<BinaryRow, Map<Range, List<Split>>> result = new HashMap<>();
+ for (Map.Entry<BinaryRow, List<Pair<Range, Split>>> partitionEntry :
+ partitionSplitRanges.entrySet()) {
+ List<Pair<Range, Split>> splitRanges = partitionEntry.getValue();
+ splitRanges.sort(
+ Comparator.comparingLong((Pair<Range, Split> e) -> e.getKey().from)
+ .thenComparingLong(e -> e.getKey().to));
+
+ Map<Range, List<Split>> partitionRanges = new LinkedHashMap<>();
+ Range current = null;
+ List<Split> currentSplits = new ArrayList<>();
+ for (Map.Entry<Range, Split> entry : splitRanges) {
+ Range splitRange = entry.getKey();
+ if (current == null) {
+ current = splitRange;
+ currentSplits.add(entry.getValue());
+ continue;
+ }
+ Range merged = Range.union(current, splitRange);
+ if (merged != null) {
+ current = merged;
+ currentSplits.add(entry.getValue());
+ } else {
+ partitionRanges.put(current, currentSplits);
+ current = splitRange;
+ currentSplits = new ArrayList<>();
+ currentSplits.add(entry.getValue());
+ }
+ }
+ if (current != null) {
+ partitionRanges.put(current, currentSplits);
+ }
+ result.put(partitionEntry.getKey(), partitionRanges);
+ }
+
+ return result;
+ }
+
+ private static List<DataSplit> splitByContiguousRowRange(DataSplit split) {
+ List<DataFileMeta> input = split.dataFiles();
+ RangeHelper<DataFileMeta> rangeHelper = new RangeHelper<>(DataFileMeta::nonNullRowIdRange);
+ List<List<DataFileMeta>> ranges = rangeHelper.mergeOverlappingRanges(input);
+
+ Supplier<DataSplit.Builder> builderSupplier =
+ () ->
+ DataSplit.builder()
+ .withSnapshot(split.snapshotId())
+ .withPartition(split.partition())
+ .withBucket(split.bucket())
+ .withBucketPath(split.bucketPath())
+ .withTotalBuckets(split.totalBuckets())
+ .isStreaming(split.isStreaming())
+ .rawConvertible(split.rawConvertible());
+ return packByContiguousRanges(builderSupplier, ranges);
+ }
+
+ private static List<DataSplit> packByContiguousRanges(
+ Supplier<DataSplit.Builder> builderFactory, List<List<DataFileMeta>> ranges) {
+ if (ranges.isEmpty()) {
+ return new ArrayList<>();
+ }
+
+ List<DataSplit> result = new ArrayList<>();
+ List<DataFileMeta> currentSegment = new ArrayList<>();
+ long currentMaxRowId = Long.MIN_VALUE;
+
+ for (List<DataFileMeta> rangeFiles : ranges) {
+ long minRowId = minRowId(rangeFiles);
+ long maxRowId = maxRowId(rangeFiles);
+ if (currentSegment.isEmpty() || areContiguous(currentMaxRowId, minRowId)) {
+ currentSegment.addAll(rangeFiles);
+ currentMaxRowId = maxRowId;
+ } else {
+ DataSplit.Builder builder = builderFactory.get();
+ builder.withDataFiles(currentSegment);
+ result.add(builder.build());
+ currentSegment = new ArrayList<>(rangeFiles);
+ currentMaxRowId = maxRowId;
+ }
+ }
+
+ DataSplit.Builder builder = builderFactory.get();
+ builder.withDataFiles(currentSegment);
+ result.add(builder.build());
+ return result;
+ }
+
+ private static long minRowId(List<DataFileMeta> files) {
+ return files.stream()
+ .mapToLong(f -> f.nonNullRowIdRange().from)
+ .min()
+ .orElse(Long.MAX_VALUE);
+ }
+
+ private static long maxRowId(List<DataFileMeta> files) {
+ return files.stream().mapToLong(f -> f.nonNullRowIdRange().to).max().orElse(Long.MIN_VALUE);
+ }
+
+ private static boolean areContiguous(long previousMaxRowId, long currentMinRowId) {
+ return previousMaxRowId >= currentMinRowId - 1;
}
public static List<IndexedSplit> createShardIndexedSplits(
@@ -197,13 +373,7 @@ public class GlobalIndexBuilderUtils {
}
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);
- }
+ FileStoreScan.Plan.groupByPartFiles(entries);
List<IndexedSplit> result = new ArrayList<>();
for (Map.Entry<BinaryRow, Map<Integer, List<ManifestEntry>>> partitionEntry :
@@ -349,7 +519,8 @@ public class GlobalIndexBuilderUtils {
int indexFieldId,
@Nullable int[] extraFieldIds,
String indexType,
- List<ResultEntry> entries)
+ List<ResultEntry> entries,
+ @Nullable byte[] sourceMeta)
throws IOException {
List<IndexFileMeta> results = new ArrayList<>();
for (ResultEntry entry : entries) {
@@ -357,7 +528,12 @@ public class GlobalIndexBuilderUtils {
long fileSize = fileIO.getFileSize(indexPathFactory.toPath(fileName));
GlobalIndexMeta globalIndexMeta =
new GlobalIndexMeta(
- range.from, range.to, indexFieldId, extraFieldIds, entry.meta());
+ range.from,
+ range.to,
+ indexFieldId,
+ extraFieldIds,
+ entry.meta(),
+ sourceMeta);
Path externalPathDir = options.globalIndexExternalPath();
String externalPathString = null;
@@ -397,70 +573,6 @@ public class GlobalIndexBuilderUtils {
return globalIndexer.createWriter(createGlobalIndexFileReadWrite(table));
}
- /**
- * Find the minimum firstRowId among files whose schema does not contain all index columns.
- * Files at or beyond this rowId cannot be indexed because the column was added later via ALTER
- * TABLE.
- *
- * @return the boundary rowId, or {@link Long#MAX_VALUE} if all files contain the columns
- */
- public static long findMinNonIndexableRowId(
- SchemaManager schemaManager, List<ManifestEntry> entries, List<String> indexColumns) {
- Map<Long, Boolean> schemaContainsColumns = new HashMap<>();
- long minRowId = Long.MAX_VALUE;
- long minSchemaId = -1;
- for (ManifestEntry entry : entries) {
- long sid = entry.file().schemaId();
- boolean contains =
- schemaContainsColumns.computeIfAbsent(
- sid,
- id -> schemaManager.schema(id).fieldNames().containsAll(indexColumns));
- if (!contains && entry.file().firstRowId() != null) {
- long rowId = entry.file().nonNullFirstRowId();
- if (rowId < minRowId) {
- minRowId = rowId;
- minSchemaId = sid;
- }
- }
- }
- if (minRowId != Long.MAX_VALUE) {
- List<String> schemaFields = schemaManager.schema(minSchemaId).fieldNames();
- List<String> missingColumns = new ArrayList<>();
- for (String col : indexColumns) {
- if (!schemaFields.contains(col)) {
- missingColumns.add(col);
- }
- }
- LOG.info(
- "Found non-indexable files: schemaId={} missing columns {}, boundaryRowId={}.",
- minSchemaId,
- missingColumns,
- minRowId);
- }
- return minRowId;
- }
-
- /** Keep only entries whose firstRowId is strictly less than the given boundary. */
- public static List<ManifestEntry> filterEntriesBefore(
- List<ManifestEntry> entries, long boundaryRowId) {
- if (boundaryRowId == Long.MAX_VALUE) {
- return entries;
- }
- List<ManifestEntry> result = new ArrayList<>();
- for (ManifestEntry entry : entries) {
- if (entry.file().firstRowId() != null
- && entry.file().nonNullFirstRowId() < boundaryRowId) {
- result.add(entry);
- }
- }
- LOG.info(
- "Filtered {} files to {} indexable files (boundaryRowId={}).",
- entries.size(),
- result.size(),
- boundaryRowId);
- return result;
- }
-
private static GlobalIndexFileReadWrite createGlobalIndexFileReadWrite(FileStoreTable table) {
IndexPathFactory indexPathFactory = table.store().pathFactory().globalIndexFileFactory();
return new GlobalIndexFileReadWrite(table.fileIO(), indexPathFactory);
diff --git a/paimon-core/src/main/java/org/apache/paimon/globalindex/ScanResult.java b/paimon-core/src/main/java/org/apache/paimon/globalindex/ScanResult.java
new file mode 100644
index 0000000000..619f24b9f2
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/globalindex/ScanResult.java
@@ -0,0 +1,70 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.globalindex;
+
+import org.apache.paimon.manifest.IndexManifestEntry;
+import org.apache.paimon.utils.RowRangeIndex;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** Result of scanning data files for a global index build. */
+public final class ScanResult<T> {
+
+ private final long scanSnapshotId;
+ private final RowRangeIndex rowRangeIndex;
+ private final List<T> entries;
+ private final List<IndexManifestEntry> deletedIndexEntries;
+
+ public ScanResult(
+ long scanSnapshotId,
+ RowRangeIndex rowRangeIndex,
+ List<T> entries,
+ List<IndexManifestEntry> deletedIndexEntries) {
+ checkArgument(scanSnapshotId > 0, "Scan snapshot id must be positive.");
+ this.scanSnapshotId = scanSnapshotId;
+ this.rowRangeIndex = rowRangeIndex;
+ this.entries = Collections.unmodifiableList(new ArrayList<>(entries));
+ this.deletedIndexEntries =
+ Collections.unmodifiableList(new ArrayList<>(deletedIndexEntries));
+ }
+
+ public long scanSnapshotId() {
+ return scanSnapshotId;
+ }
+
+ public RowRangeIndex rowRangeIndex() {
+ return rowRangeIndex;
+ }
+
+ public List<T> entries() {
+ return entries;
+ }
+
+ public List<IndexManifestEntry> deletedIndexEntries() {
+ return deletedIndexEntries;
+ }
+
+ public ScanResult<T> withEntries(List<T> entries) {
+ return new ScanResult<>(scanSnapshotId, rowRangeIndex, entries, deletedIndexEntries);
+ }
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/globalindex/generic/GenericGlobalIndexScanner.java b/paimon-core/src/main/java/org/apache/paimon/globalindex/generic/GenericGlobalIndexScanner.java
new file mode 100644
index 0000000000..619467daad
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/globalindex/generic/GenericGlobalIndexScanner.java
@@ -0,0 +1,170 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.globalindex.generic;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.globalindex.DataEvolutionGlobalIndexRefreshPlanner;
+import org.apache.paimon.globalindex.ScanResult;
+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.table.FileStoreTable;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.utils.Range;
+import org.apache.paimon.utils.RowRangeIndex;
+
+import javax.annotation.Nullable;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+
+import static org.apache.paimon.globalindex.GlobalIndexBuilderUtils.currentIndexEntries;
+import static org.apache.paimon.globalindex.GlobalIndexBuilderUtils.unindexedRowRanges;
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** Scanner for generic (non-btree) global index. */
+public class GenericGlobalIndexScanner implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ protected final FileStoreTable table;
+
+ @Nullable protected PartitionPredicate partitionPredicate;
+ @Nullable private String indexType;
+ private List<DataField> indexFields = Collections.emptyList();
+ private Options options = new Options();
+
+ public GenericGlobalIndexScanner(FileStoreTable table) {
+ this.table = table;
+ }
+
+ public GenericGlobalIndexScanner withPartitionPredicate(PartitionPredicate partitionPredicate) {
+ this.partitionPredicate = partitionPredicate;
+ return this;
+ }
+
+ public GenericGlobalIndexScanner withIndex(
+ String indexType, List<String> indexColumns, Options options) {
+ checkArgument(!indexColumns.isEmpty(), "Index columns must not be empty.");
+ List<DataField> indexFields = new ArrayList<>(indexColumns.size());
+ for (String indexColumn : indexColumns) {
+ checkArgument(
+ table.rowType().containsField(indexColumn),
+ "Column '%s' does not exist in table '%s'.",
+ indexColumn,
+ table.fullName());
+ indexFields.add(table.rowType().getField(indexColumn));
+ }
+ this.indexType = indexType;
+ this.indexFields = Collections.unmodifiableList(indexFields);
+ this.options = options;
+ return this;
+ }
+
+ public FileStoreTable table() {
+ return table;
+ }
+
+ /**
+ * Scans manifest entries to determine which files need to be indexed.
+ *
+ * @return scan result containing manifest entries to build index from
+ */
+ public Optional<ScanResult<ManifestEntry>> scan() {
+ checkArgument(
+ table.coreOptions().bucket() == -1,
+ "Generic global index only supports unaware-bucket tables (bucket = -1), "
+ + "but table '%s' has bucket = %d.",
+ table.name(),
+ table.coreOptions().bucket());
+ Snapshot scanSnapshot = table.snapshotManager().latestSnapshot();
+ if (scanSnapshot == null) {
+ return Optional.empty();
+ }
+
+ Long nextRowId = scanSnapshot.nextRowId();
+ if (nextRowId == null || nextRowId <= 0) {
+ return Optional.empty();
+ }
+ Range dataRange = new Range(0, nextRowId - 1);
+
+ List<ManifestEntry> entries =
+ table.store()
+ .newScan()
+ .withSnapshot(scanSnapshot)
+ .withPartitionFilter(partitionPredicate)
+ .dropStats()
+ .plan()
+ .files();
+ return Optional.of(
+ new ScanResult<>(
+ scanSnapshot.id(),
+ RowRangeIndex.create(Collections.singletonList(dataRange)),
+ entries,
+ Collections.emptyList()));
+ }
+
+ public Optional<ScanResult<ManifestEntry>> incrementalScan() {
+ Optional<ScanResult<ManifestEntry>> optionalScanResult = scan();
+ if (!optionalScanResult.isPresent()) {
+ return Optional.empty();
+ }
+
+ checkArgument(indexType != null, "Index type must be set before incremental scan.");
+ checkArgument(!indexFields.isEmpty(), "Index fields must be set before incremental scan.");
+
+ ScanResult<ManifestEntry> scanResult = optionalScanResult.get();
+ Snapshot scanSnapshot = table.snapshotManager().snapshot(scanResult.scanSnapshotId());
+ List<IndexManifestEntry> currentIndexes =
+ currentIndexEntries(
+ table, scanSnapshot, indexType, indexFields, partitionPredicate);
+ List<Range> rangesToBuild =
+ new ArrayList<>(unindexedRowRanges(scanSnapshot, currentIndexes));
+ List<IndexManifestEntry> deletedIndexEntries = Collections.emptyList();
+ Options mergedOptions = new Options(table.options(), options.toMap());
+ if (mergedOptions.get(CoreOptions.GLOBAL_INDEX_COLUMN_UPDATE_ACTION)
+ == CoreOptions.GlobalIndexColumnUpdateAction.IGNORE) {
+ deletedIndexEntries =
+ DataEvolutionGlobalIndexRefreshPlanner.findIndexesToRefresh(
+ table.schemaManager(),
+ scanResult.entries(),
+ currentIndexes,
+ indexFields);
+ for (IndexManifestEntry entry : deletedIndexEntries) {
+ rangesToBuild.add(entry.indexFile().globalIndexMeta().rowRange());
+ }
+ }
+
+ rangesToBuild = Range.sortAndMergeOverlap(rangesToBuild, true);
+ if (rangesToBuild.isEmpty()) {
+ return Optional.empty();
+ }
+ return Optional.of(
+ new ScanResult<>(
+ scanResult.scanSnapshotId(),
+ RowRangeIndex.create(rangesToBuild),
+ scanResult.entries(),
+ deletedIndexEntries));
+ }
+}
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
deleted file mode 100644
index f97e80fca3..0000000000
--- a/paimon-core/src/main/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexBuilder.java
+++ /dev/null
@@ -1,469 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.paimon.globalindex.sorted;
-
-import org.apache.paimon.CoreOptions;
-import org.apache.paimon.Snapshot;
-import org.apache.paimon.annotation.VisibleForTesting;
-import org.apache.paimon.data.BinaryRow;
-import org.apache.paimon.data.InternalRow;
-import org.apache.paimon.data.InternalRow.FieldGetter;
-import org.apache.paimon.disk.IOManager;
-import org.apache.paimon.globalindex.DataEvolutionBatchScan;
-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.CompactIncrement;
-import org.apache.paimon.io.DataFileMeta;
-import org.apache.paimon.io.DataIncrement;
-import org.apache.paimon.options.Options;
-import org.apache.paimon.partition.PartitionPredicate;
-import org.apache.paimon.reader.RecordReader;
-import org.apache.paimon.sort.BinaryExternalSortBuffer;
-import org.apache.paimon.table.FileStoreTable;
-import org.apache.paimon.table.SpecialFields;
-import org.apache.paimon.table.Table;
-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.Split;
-import org.apache.paimon.table.source.snapshot.SnapshotReader;
-import org.apache.paimon.types.DataField;
-import org.apache.paimon.types.RowType;
-import org.apache.paimon.utils.CloseableIterator;
-import org.apache.paimon.utils.MutableObjectIteratorAdapter;
-import org.apache.paimon.utils.Pair;
-import org.apache.paimon.utils.Preconditions;
-import org.apache.paimon.utils.Range;
-import org.apache.paimon.utils.RangeHelper;
-import org.apache.paimon.utils.RowRangeIndex;
-
-import javax.annotation.Nullable;
-
-import java.io.IOException;
-import java.io.Serializable;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.Comparator;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.LinkedHashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-import java.util.function.Supplier;
-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;
-
-/** Builder to build sorted global index. */
-public class SortedGlobalIndexBuilder implements Serializable {
-
- private static final long serialVersionUID = 1L;
- private static final double FLOATING = 1.2;
- private final String indexType;
- private final FileStoreTable table;
- private final RowType rowType;
- private final Options options;
- private final long recordsPerRange;
-
- private DataField indexField;
-
- // readRowType is composed by partition fields, indexed field and _ROW_ID field
- private RowType readRowType;
- @Nullable private Snapshot snapshot;
-
- @Nullable private PartitionPredicate partitionPredicate;
-
- public SortedGlobalIndexBuilder(Table table, String indexType) {
- this(table, indexType, ((FileStoreTable) table).coreOptions().toConfiguration());
- }
-
- public SortedGlobalIndexBuilder(Table table, String indexType, Options options) {
- this.indexType = indexType;
- this.table = (FileStoreTable) table;
- this.rowType = this.table.rowType();
- this.options = options;
- this.recordsPerRange =
- (long) (options.get(SortedIndexOptions.SORTED_INDEX_RECORDS_PER_RANGE) * FLOATING);
- }
-
- public SortedGlobalIndexBuilder withIndexField(String indexField) {
- checkArgument(
- rowType.containsField(indexField),
- "Column '%s' does not exist in table '%s'.",
- indexField,
- table.fullName());
- this.indexField = rowType.getField(indexField);
- List<String> readColumns = new ArrayList<>();
- readColumns.add(this.indexField.name());
- readColumns.add(SpecialFields.ROW_ID.name());
-
- this.readRowType = SpecialFields.rowTypeWithRowId(table.rowType()).project(readColumns);
- return this;
- }
-
- public SortedGlobalIndexBuilder withPartitionPredicate(PartitionPredicate partitionPredicate) {
- this.partitionPredicate = partitionPredicate;
- return this;
- }
-
- public SortedGlobalIndexBuilder withSnapshot(Snapshot snapshot) {
- this.snapshot = snapshot;
- return this;
- }
-
- public Optional<Pair<RowRangeIndex, List<DataSplit>>> scan() {
- SnapshotReader snapshotReader = table.newSnapshotReader();
- if (partitionPredicate != null) {
- snapshotReader = snapshotReader.withPartitionFilter(partitionPredicate);
- }
- Snapshot snapshot =
- this.snapshot != null
- ? this.snapshot
- : snapshotReader.snapshotManager().latestSnapshot();
- if (snapshot == null) {
- return Optional.empty();
- }
- snapshotReader = withManifestEntryFilter(snapshotReader.withSnapshot(snapshot));
- Range dataRange = new Range(0, snapshot.nextRowId() - 1);
-
- return Optional.of(
- Pair.of(
- RowRangeIndex.create(Collections.singletonList(dataRange)),
- snapshotReader.read().dataSplits()));
- }
-
- public Optional<Pair<RowRangeIndex, List<DataSplit>>> incrementalScan() {
- SnapshotReader snapshotReader = table.newSnapshotReader();
- if (partitionPredicate != null) {
- snapshotReader = snapshotReader.withPartitionFilter(partitionPredicate);
- }
- Snapshot snapshot =
- this.snapshot != null
- ? this.snapshot
- : snapshotReader.snapshotManager().latestSnapshot();
- if (snapshot == null) {
- return Optional.empty();
- }
- snapshotReader = withManifestEntryFilter(snapshotReader.withSnapshot(snapshot));
-
- Preconditions.checkArgument(indexField != null, "indexField must be set before scan.");
- Range dataRange = new Range(0, snapshot.nextRowId() - 1);
- List<Range> indexedRanges =
- indexedRowRanges(
- table,
- snapshot,
- indexType,
- Collections.singletonList(indexField),
- partitionPredicate);
- List<Range> nonIndexedRanges = dataRange.exclude(indexedRanges);
- if (nonIndexedRanges.isEmpty()) {
- return Optional.empty();
- }
- snapshotReader = snapshotReader.withRowRanges(nonIndexedRanges);
- return Optional.of(
- Pair.of(
- RowRangeIndex.create(nonIndexedRanges),
- snapshotReader.read().dataSplits()));
- }
-
- private SnapshotReader withManifestEntryFilter(SnapshotReader snapshotReader) {
- return snapshotReader.withManifestEntryFilter(
- entry ->
- !isBlobFile(entry.file().fileName())
- && !isVectorStoreFile(entry.file().fileName()));
- }
-
- @VisibleForTesting
- public List<CommitMessage> build(DataSplit split, IOManager ioManager) throws IOException {
- BinaryRow partition = split.partition();
- Range rowRange = calcRowRange(split);
-
- CoreOptions options = new CoreOptions(this.options);
- BinaryExternalSortBuffer buffer =
- BinaryExternalSortBuffer.create(
- ioManager,
- readRowType,
- // sort by <partition, indexed_field>
- IntStream.range(0, readRowType.getFieldCount() - 1).toArray(),
- options.writeBufferSize(),
- options.pageSize(),
- options.localSortMaxNumFileHandles(),
- options.spillCompressOptions(),
- options.writeBufferSpillDiskSize());
-
- List<Split> splitList = Collections.singletonList(split);
- RecordReader<InternalRow> reader =
- table.newReadBuilder().withReadType(readRowType).newRead().createReader(splitList);
- try (CloseableIterator<InternalRow> iterator = reader.toCloseableIterator()) {
- while (iterator.hasNext()) {
- InternalRow row = iterator.next();
- buffer.write(row);
- }
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
-
- Iterator<InternalRow> iterator =
- new MutableObjectIteratorAdapter<>(
- buffer.sortedIterator(), new BinaryRow(readRowType.getFieldCount()));
- List<CommitMessage> result = buildForSinglePartition(rowRange, partition, iterator);
- buffer.clear();
-
- return result;
- }
-
- public long recordsPerRange() {
- return recordsPerRange;
- }
-
- public List<CommitMessage> buildForSinglePartition(
- Range rowRange, BinaryRow partition, Iterator<InternalRow> data) throws IOException {
- FieldGetter indexFieldGetter = InternalRow.createFieldGetter(indexField.type(), 0);
- try (SortedSingleColumnIndexWriter writer =
- new SortedSingleColumnIndexWriter(recordsPerRange, this::createWriter)) {
- while (data.hasNext()) {
- InternalRow row = data.next();
- long localRowId = row.getLong(1) - rowRange.from;
- writer.write(indexFieldGetter.getFieldOrNull(row), localRowId);
- }
-
- List<CommitMessage> commitMessages = new ArrayList<>();
- for (List<ResultEntry> resultEntries : writer.finish()) {
- commitMessages.add(flushIndex(rowRange, resultEntries, partition));
- }
- return commitMessages;
- }
- }
-
- public GlobalIndexSingleColumnWriter createWriter() throws IOException {
- GlobalIndexSingleColumnWriter currentWriter;
- GlobalIndexWriter indexWriter = createIndexWriter(table, indexType, indexField, options);
- if (!(indexWriter instanceof GlobalIndexSingleColumnWriter)) {
- throw new RuntimeException(
- "Unexpected implementation, the index writer of "
- + indexType
- + " should be an instance of GlobalIndexSingleColumnWriter, but found: "
- + indexWriter.getClass().getName());
- }
- currentWriter = (GlobalIndexSingleColumnWriter) indexWriter;
- return currentWriter;
- }
-
- public CommitMessage flushIndex(
- Range rowRange, List<ResultEntry> resultEntries, BinaryRow partition)
- throws IOException {
- List<IndexFileMeta> indexFileMetas =
- toIndexFileMetas(
- table.fileIO(),
- table.store().pathFactory().globalIndexFileFactory(),
- table.coreOptions(),
- rowRange,
- indexField.id(),
- indexType,
- resultEntries);
- DataIncrement dataIncrement = DataIncrement.indexIncrement(indexFileMetas);
- return new CommitMessageImpl(
- partition, 0, null, dataIncrement, CompactIncrement.emptyIncrement());
- }
-
- public static List<Pair<Range, Split>> splitByRowRangeIndex(
- RowRangeIndex rowRangeIndex, DataSplit dataSplit) {
- if (rowRangeIndex == null) {
- Range range = calcRowRange(dataSplit);
- return range == null
- ? Collections.emptyList()
- : Collections.singletonList(Pair.of(range, dataSplit));
- }
-
- List<Pair<Range, Split>> result = new ArrayList<>();
- for (Split split :
- DataEvolutionBatchScan.wrapToIndexSplits(
- Collections.singletonList(dataSplit), rowRangeIndex, null)
- .splits()) {
- IndexedSplit indexedSplit = (IndexedSplit) split;
- for (Range rowRange : indexedSplit.rowRanges()) {
- result.add(
- Pair.of(
- rowRange,
- new IndexedSplit(
- indexedSplit.dataSplit(),
- Collections.singletonList(rowRange),
- null)));
- }
- }
- return result;
- }
-
- public static Range calcRowRange(DataSplit dataSplit) {
- List<Range> ranges = calcRowRanges(singletonList(dataSplit));
- if (ranges.isEmpty()) {
- return null;
- }
- return new Range(ranges.get(0).from, ranges.get(ranges.size() - 1).to);
- }
-
- public static List<Range> calcRowRanges(List<DataSplit> dataSplits) {
- List<Range> ranges = new ArrayList<>();
- for (DataSplit dataSplit : dataSplits) {
- for (DataFileMeta file : dataSplit.dataFiles()) {
- ranges.add(file.nonNullRowIdRange());
- }
- }
- return Range.sortAndMergeOverlap(ranges, true);
- }
-
- public static List<DataSplit> splitByContiguousRowRange(List<DataSplit> splits) {
- List<DataSplit> result = new ArrayList<>();
- for (DataSplit split : splits) {
- result.addAll(splitByContiguousRowRange(split));
- }
- return result;
- }
-
- public static Map<BinaryRow, Map<Range, List<Split>>> groupSplitsByRange(
- RowRangeIndex rowRangeIndex, List<DataSplit> splits) {
- Map<BinaryRow, List<Pair<Range, Split>>> partitionSplitRanges = new HashMap<>();
- for (DataSplit split : splits) {
- for (Pair<Range, Split> keyPair : splitByRowRangeIndex(rowRangeIndex, split)) {
- Range splitRange = keyPair.getKey();
- Split splitWithRange = keyPair.getValue();
- if (splitRange == null) {
- continue;
- }
- BinaryRow partition = split.partition();
- partitionSplitRanges
- .computeIfAbsent(partition, p -> new ArrayList<>())
- .add(Pair.of(splitRange, splitWithRange));
- }
- }
-
- Map<BinaryRow, Map<Range, List<Split>>> result = new HashMap<>();
- for (Map.Entry<BinaryRow, List<Pair<Range, Split>>> partitionEntry :
- partitionSplitRanges.entrySet()) {
- List<Pair<Range, Split>> splitRanges = partitionEntry.getValue();
- splitRanges.sort(
- Comparator.comparingLong((Pair<Range, Split> e) -> e.getKey().from)
- .thenComparingLong(e -> e.getKey().to));
-
- Map<Range, List<Split>> partitionRanges = new LinkedHashMap<>();
- Range current = null;
- List<Split> currentSplits = new ArrayList<>();
- for (Map.Entry<Range, Split> entry : splitRanges) {
- Range splitRange = entry.getKey();
- if (current == null) {
- current = splitRange;
- currentSplits.add(entry.getValue());
- continue;
- }
- Range merged = Range.union(current, splitRange);
- if (merged != null) {
- current = merged;
- currentSplits.add(entry.getValue());
- } else {
- partitionRanges.put(current, currentSplits);
- current = splitRange;
- currentSplits = new ArrayList<>();
- currentSplits.add(entry.getValue());
- }
- }
- if (current != null) {
- partitionRanges.put(current, currentSplits);
- }
- result.put(partitionEntry.getKey(), partitionRanges);
- }
-
- return result;
- }
-
- private static List<DataSplit> splitByContiguousRowRange(DataSplit split) {
- List<DataFileMeta> input = split.dataFiles();
- RangeHelper<DataFileMeta> rangeHelper = new RangeHelper<>(DataFileMeta::nonNullRowIdRange);
- List<List<DataFileMeta>> ranges = rangeHelper.mergeOverlappingRanges(input);
-
- Supplier<DataSplit.Builder> builderSupplier =
- () ->
- DataSplit.builder()
- .withSnapshot(split.snapshotId())
- .withPartition(split.partition())
- .withBucket(split.bucket())
- .withBucketPath(split.bucketPath())
- .withTotalBuckets(split.totalBuckets())
- .isStreaming(split.isStreaming())
- .rawConvertible(split.rawConvertible());
- return packByContiguousRanges(builderSupplier, ranges);
- }
-
- private static List<DataSplit> packByContiguousRanges(
- Supplier<DataSplit.Builder> builderFactory, List<List<DataFileMeta>> ranges) {
- if (ranges.isEmpty()) {
- return new ArrayList<>();
- }
-
- List<DataSplit> result = new ArrayList<>();
- List<DataFileMeta> currentSegment = new ArrayList<>();
- long currentMaxRowId = Long.MIN_VALUE;
-
- for (List<DataFileMeta> rangeFiles : ranges) {
- long minRowId = minRowId(rangeFiles);
- long maxRowId = maxRowId(rangeFiles);
- if (currentSegment.isEmpty() || areContiguous(currentMaxRowId, minRowId)) {
- currentSegment.addAll(rangeFiles);
- currentMaxRowId = maxRowId;
- } else {
- DataSplit.Builder builder = builderFactory.get();
- builder.withDataFiles(currentSegment);
- result.add(builder.build());
- currentSegment = new ArrayList<>(rangeFiles);
- currentMaxRowId = maxRowId;
- }
- }
-
- DataSplit.Builder builder = builderFactory.get();
- builder.withDataFiles(currentSegment);
- result.add(builder.build());
- return result;
- }
-
- private static long minRowId(List<DataFileMeta> files) {
- return files.stream()
- .mapToLong(f -> f.nonNullRowIdRange().from)
- .min()
- .orElse(Long.MAX_VALUE);
- }
-
- private static long maxRowId(List<DataFileMeta> files) {
- return files.stream().mapToLong(f -> f.nonNullRowIdRange().to).max().orElse(Long.MIN_VALUE);
- }
-
- private static boolean areContiguous(long previousMaxRowId, long currentMinRowId) {
- // Contiguous means no gap between adjacent ranges.
- // e.g. previous max == current min (as requested) or previous max + 1 == current min.
- return previousMaxRowId >= currentMinRowId - 1;
- }
-}
diff --git a/paimon-core/src/main/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexScanner.java b/paimon-core/src/main/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexScanner.java
new file mode 100644
index 0000000000..e1417dd90c
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexScanner.java
@@ -0,0 +1,191 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.globalindex.sorted;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.globalindex.DataEvolutionGlobalIndexRefreshPlanner;
+import org.apache.paimon.globalindex.ScanResult;
+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.table.FileStoreTable;
+import org.apache.paimon.table.Table;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.snapshot.SnapshotReader;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.Preconditions;
+import org.apache.paimon.utils.Range;
+import org.apache.paimon.utils.RowRangeIndex;
+
+import javax.annotation.Nullable;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+
+import static org.apache.paimon.format.blob.BlobFileFormat.isBlobFile;
+import static org.apache.paimon.globalindex.GlobalIndexBuilderUtils.currentIndexEntries;
+import static org.apache.paimon.globalindex.GlobalIndexBuilderUtils.unindexedRowRanges;
+import static org.apache.paimon.types.VectorType.isVectorStoreFile;
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** Scanner for sorted global indexes. */
+public class SortedGlobalIndexScanner implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+ private final String indexType;
+ private final FileStoreTable table;
+ private final RowType rowType;
+ private final Options options;
+
+ private DataField indexField;
+
+ @Nullable private Snapshot snapshot;
+
+ @Nullable private PartitionPredicate partitionPredicate;
+
+ public SortedGlobalIndexScanner(Table table, String indexType) {
+ this(table, indexType, ((FileStoreTable) table).coreOptions().toConfiguration());
+ }
+
+ public SortedGlobalIndexScanner(Table table, String indexType, Options options) {
+ this.indexType = indexType;
+ this.table = (FileStoreTable) table;
+ this.rowType = this.table.rowType();
+ this.options = options;
+ }
+
+ public SortedGlobalIndexScanner withIndexField(String indexField) {
+ checkArgument(
+ rowType.containsField(indexField),
+ "Column '%s' does not exist in table '%s'.",
+ indexField,
+ table.fullName());
+ this.indexField = rowType.getField(indexField);
+ return this;
+ }
+
+ public SortedGlobalIndexScanner withPartitionPredicate(PartitionPredicate partitionPredicate) {
+ this.partitionPredicate = partitionPredicate;
+ return this;
+ }
+
+ public SortedGlobalIndexScanner withSnapshot(Snapshot snapshot) {
+ this.snapshot = snapshot;
+ return this;
+ }
+
+ public Optional<ScanResult<DataSplit>> scan() {
+ SnapshotReader snapshotReader = table.newSnapshotReader();
+ if (partitionPredicate != null) {
+ snapshotReader = snapshotReader.withPartitionFilter(partitionPredicate);
+ }
+ Snapshot snapshot =
+ this.snapshot != null
+ ? this.snapshot
+ : snapshotReader.snapshotManager().latestSnapshot();
+ if (snapshot == null) {
+ return Optional.empty();
+ }
+ snapshotReader = withManifestEntryFilter(snapshotReader.withSnapshot(snapshot));
+ Range dataRange = new Range(0, snapshot.nextRowId() - 1);
+
+ return Optional.of(
+ new ScanResult<>(
+ snapshot.id(),
+ RowRangeIndex.create(Collections.singletonList(dataRange)),
+ snapshotReader.read().dataSplits(),
+ Collections.emptyList()));
+ }
+
+ public Optional<ScanResult<DataSplit>> incrementalScan() {
+ SnapshotReader snapshotReader = table.newSnapshotReader();
+ if (partitionPredicate != null) {
+ snapshotReader = snapshotReader.withPartitionFilter(partitionPredicate);
+ }
+ Snapshot snapshot =
+ this.snapshot != null
+ ? this.snapshot
+ : snapshotReader.snapshotManager().latestSnapshot();
+ if (snapshot == null) {
+ return Optional.empty();
+ }
+ snapshotReader = withManifestEntryFilter(snapshotReader.withSnapshot(snapshot));
+
+ Preconditions.checkArgument(indexField != null, "indexField must be set before scan.");
+ List<IndexManifestEntry> currentIndexes =
+ currentIndexEntries(
+ table,
+ snapshot,
+ indexType,
+ Collections.singletonList(indexField),
+ partitionPredicate);
+ List<Range> rangesToBuild = new ArrayList<>(unindexedRowRanges(snapshot, currentIndexes));
+ List<IndexManifestEntry> deletedIndexEntries = Collections.emptyList();
+ if (detectDataFileChange()) {
+ List<ManifestEntry> dataEntries =
+ table.store()
+ .newScan()
+ .withSnapshot(snapshot)
+ .withPartitionFilter(partitionPredicate)
+ .dropStats()
+ .plan()
+ .files();
+ deletedIndexEntries =
+ DataEvolutionGlobalIndexRefreshPlanner.findIndexesToRefresh(
+ table.schemaManager(),
+ dataEntries,
+ currentIndexes,
+ Collections.singletonList(indexField));
+ for (IndexManifestEntry entry : deletedIndexEntries) {
+ rangesToBuild.add(entry.indexFile().globalIndexMeta().rowRange());
+ }
+ }
+
+ rangesToBuild = Range.sortAndMergeOverlap(rangesToBuild, true);
+ if (rangesToBuild.isEmpty()) {
+ return Optional.empty();
+ }
+ snapshotReader = snapshotReader.withRowRanges(rangesToBuild);
+ return Optional.of(
+ new ScanResult<>(
+ snapshot.id(),
+ RowRangeIndex.create(rangesToBuild),
+ snapshotReader.read().dataSplits(),
+ deletedIndexEntries));
+ }
+
+ private boolean detectDataFileChange() {
+ Options mergedOptions = new Options(table.options(), options.toMap());
+ return mergedOptions.get(CoreOptions.GLOBAL_INDEX_COLUMN_UPDATE_ACTION)
+ == CoreOptions.GlobalIndexColumnUpdateAction.IGNORE;
+ }
+
+ private SnapshotReader withManifestEntryFilter(SnapshotReader snapshotReader) {
+ return snapshotReader.withManifestEntryFilter(
+ entry ->
+ !isBlobFile(entry.file().fileName())
+ && !isVectorStoreFile(entry.file().fileName()));
+ }
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexWriter.java b/paimon-core/src/main/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexWriter.java
new file mode 100644
index 0000000000..52e6d12ee8
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexWriter.java
@@ -0,0 +1,145 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.globalindex.sorted;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.data.InternalRow.FieldGetter;
+import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter;
+import org.apache.paimon.globalindex.GlobalIndexWriter;
+import org.apache.paimon.globalindex.ResultEntry;
+import org.apache.paimon.index.DataEvolutionIndexSourceMeta;
+import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.io.CompactIncrement;
+import org.apache.paimon.io.DataIncrement;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.Table;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.CommitMessageImpl;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.Range;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+
+import static org.apache.paimon.globalindex.GlobalIndexBuilderUtils.createIndexWriter;
+import static org.apache.paimon.globalindex.GlobalIndexBuilderUtils.toIndexFileMetas;
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** Writer for sorted global indexes. */
+public class SortedGlobalIndexWriter implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+ private static final double FLOATING = 1.2;
+
+ private final String indexType;
+ private final FileStoreTable table;
+ private final RowType rowType;
+ private final Options options;
+ private final long recordsPerRange;
+
+ private DataField indexField;
+
+ public SortedGlobalIndexWriter(Table table, String indexType) {
+ this(table, indexType, ((FileStoreTable) table).coreOptions().toConfiguration());
+ }
+
+ public SortedGlobalIndexWriter(Table table, String indexType, Options options) {
+ this.indexType = indexType;
+ this.table = (FileStoreTable) table;
+ this.rowType = this.table.rowType();
+ this.options = options;
+ this.recordsPerRange =
+ (long) (options.get(SortedIndexOptions.SORTED_INDEX_RECORDS_PER_RANGE) * FLOATING);
+ }
+
+ public SortedGlobalIndexWriter withIndexField(String indexField) {
+ checkArgument(
+ rowType.containsField(indexField),
+ "Column '%s' does not exist in table '%s'.",
+ indexField,
+ table.fullName());
+ this.indexField = rowType.getField(indexField);
+ return this;
+ }
+
+ public long recordsPerRange() {
+ return recordsPerRange;
+ }
+
+ public List<CommitMessage> buildForSinglePartition(
+ Range rowRange, BinaryRow partition, Iterator<InternalRow> data, long scanSnapshotId)
+ throws IOException {
+ FieldGetter indexFieldGetter = InternalRow.createFieldGetter(indexField.type(), 0);
+ try (SortedSingleColumnIndexWriter writer =
+ new SortedSingleColumnIndexWriter(recordsPerRange, this::createWriter)) {
+ while (data.hasNext()) {
+ InternalRow row = data.next();
+ long localRowId = row.getLong(1) - rowRange.from;
+ writer.write(indexFieldGetter.getFieldOrNull(row), localRowId);
+ }
+
+ List<CommitMessage> commitMessages = new ArrayList<>();
+ for (List<ResultEntry> resultEntries : writer.finish()) {
+ commitMessages.add(flushIndex(rowRange, resultEntries, partition, scanSnapshotId));
+ }
+ return commitMessages;
+ }
+ }
+
+ public GlobalIndexSingleColumnWriter createWriter() throws IOException {
+ GlobalIndexWriter indexWriter = createIndexWriter(table, indexType, indexField, options);
+ if (!(indexWriter instanceof GlobalIndexSingleColumnWriter)) {
+ throw new RuntimeException(
+ "Unexpected implementation, the index writer of "
+ + indexType
+ + " should be an instance of GlobalIndexSingleColumnWriter, but found: "
+ + indexWriter.getClass().getName());
+ }
+ return (GlobalIndexSingleColumnWriter) indexWriter;
+ }
+
+ public CommitMessage flushIndex(
+ Range rowRange,
+ List<ResultEntry> resultEntries,
+ BinaryRow partition,
+ long scanSnapshotId)
+ throws IOException {
+ byte[] sourceMeta = new DataEvolutionIndexSourceMeta(scanSnapshotId).serialize();
+ List<IndexFileMeta> indexFileMetas =
+ toIndexFileMetas(
+ table.fileIO(),
+ table.store().pathFactory().globalIndexFileFactory(),
+ table.coreOptions(),
+ rowRange,
+ Collections.singletonList(indexField),
+ indexType,
+ resultEntries,
+ sourceMeta);
+ DataIncrement dataIncrement = DataIncrement.indexIncrement(indexFileMetas);
+ return new CommitMessageImpl(
+ partition, 0, null, dataIncrement, CompactIncrement.emptyIncrement());
+ }
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/index/DataEvolutionIndexSourceMeta.java b/paimon-core/src/main/java/org/apache/paimon/index/DataEvolutionIndexSourceMeta.java
new file mode 100644
index 0000000000..fb39bc2f8e
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/index/DataEvolutionIndexSourceMeta.java
@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.index;
+
+import org.apache.paimon.io.DataInputDeserializer;
+import org.apache.paimon.io.DataOutputSerializer;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** Data snapshot scanned to build a global index for a data-evolution table. */
+public final class DataEvolutionIndexSourceMeta {
+
+ // "DEIX". The marker distinguishes this metadata from primary-key index source metadata.
+ private static final int MAGIC = 0x44454958;
+ private static final int VERSION = 1;
+
+ private final long scanSnapshotId;
+
+ public DataEvolutionIndexSourceMeta(long scanSnapshotId) {
+ checkArgument(scanSnapshotId > 0, "Scan snapshot id must be positive.");
+ this.scanSnapshotId = scanSnapshotId;
+ }
+
+ public long scanSnapshotId() {
+ return scanSnapshotId;
+ }
+
+ public byte[] serialize() {
+ try {
+ DataOutputSerializer output = new DataOutputSerializer(16);
+ output.writeInt(MAGIC);
+ output.writeInt(VERSION);
+ output.writeLong(scanSnapshotId);
+ return output.getCopyOfBuffer();
+ } catch (IOException e) {
+ throw new RuntimeException(
+ "Failed to serialize data-evolution index source metadata.", e);
+ }
+ }
+
+ public static boolean isDataEvolutionMeta(@Nullable byte[] bytes) {
+ if (bytes == null || bytes.length < Integer.BYTES) {
+ return false;
+ }
+ try {
+ return new DataInputDeserializer(bytes).readInt() == MAGIC;
+ } catch (IOException e) {
+ return false;
+ }
+ }
+
+ public static DataEvolutionIndexSourceMeta deserialize(byte[] bytes) {
+ try {
+ DataInputDeserializer input = new DataInputDeserializer(bytes);
+ int magic = input.readInt();
+ checkArgument(magic == MAGIC, "Not data-evolution index source metadata.");
+ int version = input.readInt();
+ checkArgument(
+ version == VERSION,
+ "Unsupported data-evolution index source version: %s.",
+ version);
+ long scanSnapshotId = input.readLong();
+ checkArgument(
+ input.available() == 0,
+ "Unexpected trailing bytes in data-evolution index source metadata.");
+ return new DataEvolutionIndexSourceMeta(scanSnapshotId);
+ } catch (IOException e) {
+ throw new IllegalArgumentException(
+ "Failed to deserialize data-evolution index source metadata.", e);
+ }
+ }
+
+ public static DataEvolutionIndexSourceMeta fromIndexFile(IndexFileMeta indexFile) {
+ GlobalIndexMeta globalIndexMeta = indexFile.globalIndexMeta();
+ checkArgument(
+ globalIndexMeta != null && isDataEvolutionMeta(globalIndexMeta.sourceMeta()),
+ "Index file %s has no data-evolution source metadata.",
+ indexFile.fileName());
+ return deserialize(globalIndexMeta.sourceMeta());
+ }
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestFileHandler.java b/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestFileHandler.java
index 486f39fef4..16cfd163ed 100644
--- a/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestFileHandler.java
+++ b/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestFileHandler.java
@@ -19,6 +19,7 @@
package org.apache.paimon.manifest;
import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.index.DataEvolutionIndexSourceMeta;
import org.apache.paimon.index.DeletionVectorMeta;
import org.apache.paimon.index.GlobalIndexMeta;
import org.apache.paimon.index.IndexFileMeta;
@@ -220,7 +221,12 @@ public class IndexManifestFileHandler {
.filter(f -> f.kind() == FileKind.ADD)
.collect(Collectors.toList());
for (IndexManifestEntry entry : removed) {
- indexEntries.remove(entry.indexFile().fileName());
+ String fileName = entry.indexFile().fileName();
+ checkState(
+ indexEntries.containsKey(fileName),
+ "Trying to delete global index file %s which does not exist.",
+ fileName);
+ indexEntries.remove(fileName);
}
validateRetainedIndexFiles(indexEntries.values(), added);
for (IndexManifestEntry entry : added) {
@@ -241,7 +247,12 @@ public class IndexManifestFileHandler {
for (IndexManifestEntry added : addedIndexFiles) {
GlobalIndexMeta addedMeta = added.indexFile().globalIndexMeta();
if (addedMeta == null
- || (retainedMeta.sourceMeta() != null && addedMeta.sourceMeta() != null)
+ || (retainedMeta.sourceMeta() != null
+ && addedMeta.sourceMeta() != null
+ && !DataEvolutionIndexSourceMeta.isDataEvolutionMeta(
+ retainedMeta.sourceMeta())
+ && !DataEvolutionIndexSourceMeta.isDataEvolutionMeta(
+ addedMeta.sourceMeta()))
|| retainedMeta.indexFieldId() != addedMeta.indexFieldId()
|| (Arrays.equals(
retainedMeta.extraFieldIds(), addedMeta.extraFieldIds())
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionFileStoreScan.java b/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionFileStoreScan.java
index 66f1aadbee..a19f02320c 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionFileStoreScan.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionFileStoreScan.java
@@ -62,6 +62,7 @@ import java.util.stream.Collectors;
import static org.apache.paimon.format.blob.BlobFileFormat.isBlobFile;
import static org.apache.paimon.manifest.ManifestFileMeta.allContainsRowId;
import static org.apache.paimon.types.VectorType.isVectorStoreFile;
+import static org.apache.paimon.utils.DataEvolutionUtils.fileFieldIds;
import static org.apache.paimon.utils.DataEvolutionUtils.retrieveAnchorFile;
/** {@link FileStoreScan} for data-evolution enabled table. */
@@ -254,24 +255,7 @@ public class DataEvolutionFileStoreScan extends AppendOnlyFileStoreScan {
private Set<Integer> fileFieldIdsForEntry(ManifestEntry entry) {
return fileFieldIdsCache.computeIfAbsent(
Pair.of(entry.file().schemaId(), entry.file().writeCols()),
- pair -> computeFileFieldIds(this::scanTableSchema, entry.file()));
- }
-
- /**
- * Field ids of the columns physically present in {@code file}, resolved through the file's own
- * schema (i.e. the schema the file was written under). Field id, not field name, is the stable
- * identity across schemas — necessary so a renamed column matches an old file written under the
- * pre-rename name.
- */
- @VisibleForTesting
- static Set<Integer> computeFileFieldIds(
- Function<Long, TableSchema> scanTableSchema, DataFileMeta file) {
- Set<Integer> ids = new HashSet<>();
- for (DataField f :
- scanTableSchema.apply(file.schemaId()).project(file.writeCols()).fields()) {
- ids.add(f.id());
- }
- return ids;
+ pair -> fileFieldIds(this::scanTableSchema, entry.file()));
}
/** TODO: Optimize implementation of this method. */
@@ -286,9 +270,7 @@ public class DataEvolutionFileStoreScan extends AppendOnlyFileStoreScan {
entry ->
isBlobFile(entry.file().fileName())
|| isVectorStoreFile(entry.file().fileName()))
- .flatMap(
- entry ->
- computeFileFieldIds(scanTableSchema, entry.file()).stream())
+ .flatMap(entry -> fileFieldIds(scanTableSchema, entry.file()).stream())
.collect(Collectors.toSet());
// exclude blob and vector-store files, useless for predicate eval
metas =
diff --git a/paimon-core/src/main/java/org/apache/paimon/utils/DataEvolutionUtils.java b/paimon-core/src/main/java/org/apache/paimon/utils/DataEvolutionUtils.java
index 57ba93ae19..c1294f3946 100644
--- a/paimon-core/src/main/java/org/apache/paimon/utils/DataEvolutionUtils.java
+++ b/paimon-core/src/main/java/org/apache/paimon/utils/DataEvolutionUtils.java
@@ -19,10 +19,14 @@
package org.apache.paimon.utils;
import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.types.DataField;
import java.util.Collection;
import java.util.Comparator;
+import java.util.HashSet;
import java.util.List;
+import java.util.Set;
import java.util.function.Function;
import java.util.stream.Collectors;
@@ -34,6 +38,24 @@ import static org.apache.paimon.utils.Preconditions.checkState;
/** Util class for data evolution. */
public class DataEvolutionUtils {
+ /**
+ * Table field ids physically present in a file, resolved through the schema used to write it.
+ */
+ public static Set<Integer> fileFieldIds(
+ Function<Long, TableSchema> scanTableSchema, DataFileMeta file) {
+ TableSchema schema = scanTableSchema.apply(file.schemaId());
+ List<String> writeCols = file.writeCols();
+ Set<String> writeColNames = writeCols == null ? null : new HashSet<>(writeCols);
+ Set<Integer> ids = new HashSet<>();
+ for (DataField field : schema.fields()) {
+ // writeCols may also contain physical row-tracking fields outside the table schema.
+ if (writeColNames == null || writeColNames.contains(field.name())) {
+ ids.add(field.id());
+ }
+ }
+ return ids;
+ }
+
/**
* Retrieve the anchor file of a row range group. Always the oldest normal file. Files are
* compared by (max_seq, fileName) pairs.
diff --git a/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java b/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java
index a45990f158..81188e64a9 100644
--- a/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java
@@ -45,7 +45,9 @@ import org.apache.paimon.disk.IOManager;
import org.apache.paimon.fs.FileIOFinder;
import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.local.LocalFileIO;
-import org.apache.paimon.globalindex.sorted.SortedGlobalIndexBuilder;
+import org.apache.paimon.globalindex.ScanResult;
+import org.apache.paimon.globalindex.sorted.SortedGlobalIndexScanner;
+import org.apache.paimon.globalindex.sorted.SortedGlobalIndexTestUtils;
import org.apache.paimon.index.HashBucketAssigner;
import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.io.CompactIncrement;
@@ -75,6 +77,7 @@ import org.apache.paimon.table.sink.CommitMessageImpl;
import org.apache.paimon.table.sink.InnerTableCommit;
import org.apache.paimon.table.sink.StreamTableCommit;
import org.apache.paimon.table.sink.StreamTableWrite;
+import org.apache.paimon.table.source.DataSplit;
import org.apache.paimon.table.source.EndOfScanException;
import org.apache.paimon.table.source.ReadBuilder;
import org.apache.paimon.table.source.Split;
@@ -658,19 +661,8 @@ public class JavaPyE2ETest {
commit.commit(write.prepareCommit());
}
- SortedGlobalIndexBuilder builder =
- new SortedGlobalIndexBuilder(table, "btree").withIndexField("k");
try (BatchTableCommit commit = writeBuilder.newCommit()) {
- commit.commit(
- builder.build(
- builder.scan()
- .map(org.apache.paimon.utils.Pair::getValue)
- .orElseThrow(
- () ->
- new IllegalStateException(
- "Expected scan result when building index."))
- .get(0),
- IOManager.create(warehouse.toString())));
+ commit.commit(buildSortedIndex(table, "btree", "k"));
}
try (BatchTableWrite write = writeBuilder.newWrite();
@@ -735,19 +727,8 @@ public class JavaPyE2ETest {
}
// build index
- SortedGlobalIndexBuilder builder =
- new SortedGlobalIndexBuilder(table, "bitmap").withIndexField("k");
try (BatchTableCommit commit = writeBuilder.newCommit()) {
- commit.commit(
- builder.build(
- builder.scan()
- .map(org.apache.paimon.utils.Pair::getValue)
- .orElseThrow(
- () ->
- new IllegalStateException(
- "Expected scan result when building index."))
- .get(0),
- IOManager.create(warehouse.toString())));
+ commit.commit(buildSortedIndex(table, "bitmap", "k"));
}
// assert index
@@ -838,19 +819,8 @@ public class JavaPyE2ETest {
}
// build index
- SortedGlobalIndexBuilder builder =
- new SortedGlobalIndexBuilder(table, "btree").withIndexField("k");
try (BatchTableCommit commit = writeBuilder.newCommit()) {
- commit.commit(
- builder.build(
- builder.scan()
- .map(org.apache.paimon.utils.Pair::getValue)
- .orElseThrow(
- () ->
- new IllegalStateException(
- "Expected scan result when building index."))
- .get(0),
- IOManager.create(warehouse.toString())));
+ commit.commit(buildSortedIndex(table, "btree", "k"));
}
// assert index
@@ -919,19 +889,8 @@ public class JavaPyE2ETest {
commit.commit(write.prepareCommit());
}
- SortedGlobalIndexBuilder builder =
- new SortedGlobalIndexBuilder(table, indexType).withIndexField("k");
try (BatchTableCommit commit = writeBuilder.newCommit()) {
- commit.commit(
- builder.build(
- builder.scan()
- .map(org.apache.paimon.utils.Pair::getValue)
- .orElseThrow(
- () ->
- new IllegalStateException(
- "Expected scan result when building index."))
- .get(0),
- IOManager.create(warehouse.toString())));
+ commit.commit(buildSortedIndex(table, indexType, "k"));
}
List<IndexManifestEntry> indexEntries =
@@ -999,19 +958,8 @@ public class JavaPyE2ETest {
}
// build index
- SortedGlobalIndexBuilder builder =
- new SortedGlobalIndexBuilder(table, "btree").withIndexField("k");
try (BatchTableCommit commit = writeBuilder.newCommit()) {
- commit.commit(
- builder.build(
- builder.scan()
- .map(org.apache.paimon.utils.Pair::getValue)
- .orElseThrow(
- () ->
- new IllegalStateException(
- "Expected scan result when building index."))
- .get(0),
- IOManager.create(warehouse.toString())));
+ commit.commit(buildSortedIndex(table, "btree", "k"));
}
// assert index
@@ -1078,19 +1026,8 @@ public class JavaPyE2ETest {
}
// build index
- SortedGlobalIndexBuilder builder =
- new SortedGlobalIndexBuilder(table, "btree").withIndexField("k");
try (BatchTableCommit commit = writeBuilder.newCommit()) {
- commit.commit(
- builder.build(
- builder.scan()
- .map(org.apache.paimon.utils.Pair::getValue)
- .orElseThrow(
- () ->
- new IllegalStateException(
- "Expected scan result when building index."))
- .get(0),
- IOManager.create(warehouse.toString())));
+ commit.commit(buildSortedIndex(table, "btree", "k"));
}
// assert index
@@ -2147,6 +2084,29 @@ public class JavaPyE2ETest {
return result;
}
+ private List<CommitMessage> buildSortedIndex(
+ FileStoreTable table, String indexType, String indexFieldName) throws Exception {
+ SortedGlobalIndexScanner scanner =
+ new SortedGlobalIndexScanner(table, indexType).withIndexField(indexFieldName);
+ ScanResult<DataSplit> scanResult =
+ scanner.scan()
+ .orElseThrow(
+ () ->
+ new IllegalStateException(
+ "Expected scan result when building index."));
+ List<CommitMessage> commitMessages = new ArrayList<>();
+ for (DataSplit dataSplit : scanResult.entries()) {
+ commitMessages.addAll(
+ SortedGlobalIndexTestUtils.buildIndex(
+ table,
+ indexType,
+ indexFieldName,
+ dataSplit,
+ scanResult.scanSnapshotId()));
+ }
+ return commitMessages;
+ }
+
private static class E2eDvSpec {
private final Range range;
diff --git a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassignerTest.java b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassignerTest.java
index 96b2a8f230..ee3bcac92b 100644
--- a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassignerTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassignerTest.java
@@ -27,8 +27,12 @@ import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.data.serializer.InternalRowSerializer;
import org.apache.paimon.fs.Path;
+import org.apache.paimon.globalindex.GlobalIndexBuilderUtils;
+import org.apache.paimon.globalindex.ScanResult;
import org.apache.paimon.globalindex.btree.BTreeIndexOptions;
-import org.apache.paimon.globalindex.sorted.SortedGlobalIndexBuilder;
+import org.apache.paimon.globalindex.sorted.SortedGlobalIndexScanner;
+import org.apache.paimon.globalindex.sorted.SortedGlobalIndexTestUtils;
+import org.apache.paimon.index.DataEvolutionIndexSourceMeta;
import org.apache.paimon.index.GlobalIndexMeta;
import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.io.CompactIncrement;
@@ -1627,6 +1631,27 @@ public class DataEvolutionRowIdReassignerTest extends TableTestBase {
assertThat(readPayloads(table, predicate)).containsExactly("v4");
}
+ @Test
+ public void testReassignPreservesGlobalIndexSourceMeta() throws Exception {
+ FileStoreTable table = createTableWithInterleavedPartitions();
+ createBTreeIndex(table);
+ long scanSnapshotId = table.snapshotManager().latestSnapshot().id();
+ setGlobalIndexSourceMeta(table, scanSnapshotId);
+
+ new DataEvolutionRowIdReassigner(table).reassign("test-preserve-index-source-meta");
+
+ List<IndexManifestEntry> entries = table.store().newIndexFileHandler().scanEntries();
+ assertThat(entries).isNotEmpty();
+ assertThat(entries)
+ .allSatisfy(
+ entry ->
+ assertThat(
+ DataEvolutionIndexSourceMeta.fromIndexFile(
+ entry.indexFile())
+ .scanSnapshotId())
+ .isEqualTo(scanSnapshotId));
+ }
+
@Test
public void testReassignKeepsConcurrentDisjointGlobalIndexRange() throws Exception {
FileStoreTable table = createTableWithInterleavedPartitions();
@@ -2763,24 +2788,60 @@ public class DataEvolutionRowIdReassignerTest extends TableTestBase {
}
private void createBTreeIndex(FileStoreTable table) throws Exception {
- SortedGlobalIndexBuilder builder =
- new SortedGlobalIndexBuilder(table, "btree").withIndexField("id");
- List<DataSplit> dataSplits =
+ SortedGlobalIndexScanner builder =
+ new SortedGlobalIndexScanner(table, "btree").withIndexField("id");
+ ScanResult<DataSplit> scanResult =
builder.scan()
- .map(Pair::getRight)
.orElseThrow(
() ->
new IllegalStateException(
"Expected scan result when building index."));
List<CommitMessage> commitMessages = new ArrayList<>();
- for (DataSplit dataSplit : SortedGlobalIndexBuilder.splitByContiguousRowRange(dataSplits)) {
- commitMessages.addAll(builder.build(dataSplit, ioManager));
+ for (DataSplit dataSplit :
+ GlobalIndexBuilderUtils.splitByContiguousRowRange(scanResult.entries())) {
+ commitMessages.addAll(
+ SortedGlobalIndexTestUtils.buildIndex(
+ table, "btree", "id", dataSplit, scanResult.scanSnapshotId()));
}
try (BatchTableCommit commit = table.newBatchWriteBuilder().newCommit()) {
commit.commit(commitMessages);
}
}
+ private void setGlobalIndexSourceMeta(FileStoreTable table, long scanSnapshotId)
+ throws Exception {
+ Snapshot latest = table.snapshotManager().latestSnapshot();
+ IndexManifestFile indexManifestFile = table.store().indexManifestFileFactory().create();
+ byte[] sourceMeta = new DataEvolutionIndexSourceMeta(scanSnapshotId).serialize();
+ List<IndexManifestEntry> rewritten = new ArrayList<>();
+ for (IndexManifestEntry entry : indexManifestFile.read(latest.indexManifest())) {
+ IndexFileMeta indexFile = entry.indexFile();
+ GlobalIndexMeta globalIndex = indexFile.globalIndexMeta();
+ assertThat(globalIndex).isNotNull();
+ rewritten.add(
+ new IndexManifestEntry(
+ entry.kind(),
+ entry.partition(),
+ entry.bucket(),
+ new IndexFileMeta(
+ indexFile.indexType(),
+ indexFile.fileName(),
+ indexFile.fileSize(),
+ indexFile.rowCount(),
+ indexFile.dvRanges(),
+ indexFile.externalPath(),
+ new GlobalIndexMeta(
+ globalIndex.rowRangeStart(),
+ globalIndex.rowRangeEnd(),
+ globalIndex.indexFieldId(),
+ globalIndex.extraFieldIds(),
+ globalIndex.indexMeta(),
+ sourceMeta))));
+ }
+ replaceLatestSnapshotIndexManifest(
+ table, latest, indexManifestFile.writeWithoutRolling(rewritten));
+ }
+
private void replaceGlobalIndexRangesWithPartitionSpanningRanges(FileStoreTable table)
throws Exception {
Snapshot latest = table.snapshotManager().latestSnapshot();
diff --git a/paimon-core/src/test/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexRefreshPlannerTest.java b/paimon-core/src/test/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexRefreshPlannerTest.java
new file mode 100644
index 0000000000..68d1a26218
--- /dev/null
+++ b/paimon-core/src/test/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexRefreshPlannerTest.java
@@ -0,0 +1,494 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.globalindex;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.index.DataEvolutionIndexSourceMeta;
+import org.apache.paimon.index.GlobalIndexMeta;
+import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.manifest.FileKind;
+import org.apache.paimon.manifest.IndexManifestEntry;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.stats.SimpleStats;
+import org.apache.paimon.table.SpecialFields;
+import org.apache.paimon.types.ArrayType;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.FloatType;
+import org.apache.paimon.types.IntType;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Random;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/** Tests for {@link DataEvolutionGlobalIndexRefreshPlanner}. */
+class DataEvolutionGlobalIndexRefreshPlannerTest {
+
+ private static final DataField VECTOR_FIELD =
+ new DataField(1, "vector", new ArrayType(new FloatType()));
+ private static final DataField OTHER_FIELD = new DataField(2, "other", new IntType());
+ private static final DataField UNRELATED_FIELD = new DataField(3, "unrelated", new IntType());
+
+ private SchemaManager schemaManager;
+
+ @BeforeEach
+ void beforeEach() {
+ schemaManager = mock(SchemaManager.class);
+ when(schemaManager.schema(0L))
+ .thenReturn(
+ new TableSchema(
+ 0L,
+ Collections.singletonList(OTHER_FIELD),
+ 1,
+ Collections.emptyList(),
+ Collections.emptyList(),
+ new HashMap<>(),
+ ""));
+ when(schemaManager.schema(1L))
+ .thenReturn(
+ new TableSchema(
+ 1L,
+ Arrays.asList(VECTOR_FIELD, OTHER_FIELD, UNRELATED_FIELD),
+ 3,
+ Collections.emptyList(),
+ Collections.emptyList(),
+ new HashMap<>(),
+ ""));
+ when(schemaManager.schema(2L))
+ .thenReturn(
+ new TableSchema(
+ 2L,
+ Arrays.asList(
+ new DataField(
+ 1,
+ "renamed_vector",
+ new ArrayType(new FloatType())),
+ OTHER_FIELD,
+ UNRELATED_FIELD),
+ 3,
+ Collections.emptyList(),
+ Collections.emptyList(),
+ new HashMap<>(),
+ ""));
+ }
+
+ @Test
+ void testRefreshesOnlyForNewPhysicalIndexColumnFile() {
+ IndexManifestEntry index = index("index", 0, 99, 5L, BinaryRow.EMPTY_ROW, 0);
+
+ assertThat(
+ plan(
+ Collections.singletonList(
+ data("vector-update", 0, 100, 6, 1, "vector")),
+ index))
+ .containsExactly(index);
+ assertThat(
+ plan(
+ Collections.singletonList(
+ data("old-vector", 0, 100, 5, 1, "vector")),
+ index))
+ .isEmpty();
+ assertThat(
+ plan(
+ Collections.singletonList(
+ data("other-update", 0, 100, 6, 1, "other")),
+ index))
+ .isEmpty();
+ assertThat(
+ plan(
+ Collections.singletonList(
+ data("outside", 100, 100, 6, 1, "vector")),
+ index))
+ .isEmpty();
+ }
+
+ @Test
+ void testLegacyIndexIsNotRefreshedAutomatically() {
+ IndexManifestEntry legacy = index("legacy", 0, 99, null, BinaryRow.EMPTY_ROW, 0);
+
+ assertThat(plan(Collections.singletonList(data("base", 0, 100, 0, 1, "vector")), legacy))
+ .isEmpty();
+ }
+
+ @Test
+ void testUsesStableFieldIdAcrossRenameAndFullWrites() {
+ IndexManifestEntry index = index("index", 0, 99, 5L, BinaryRow.EMPTY_ROW, 0);
+
+ assertThat(
+ plan(
+ Collections.singletonList(
+ data("renamed", 0, 100, 6, 2, "renamed_vector")),
+ index))
+ .containsExactly(index);
+ assertThat(plan(Collections.singletonList(data("full", 0, 100, 6, 1)), index))
+ .containsExactly(index);
+ }
+
+ @Test
+ void testRefreshesFromUpdateLayerOverBaseSchemaWithoutIndexColumn() {
+ IndexManifestEntry index = index("index", 0, 99, 5L, BinaryRow.EMPTY_ROW, 0);
+ ManifestEntry baseWithoutVector = data("base", 0, 100, 1, 0);
+ ManifestEntry vectorUpdate = data("vector-update", 0, 100, 6, 1, "vector");
+
+ assertThat(plan(Arrays.asList(baseWithoutVector, vectorUpdate), index))
+ .containsExactly(index);
+ }
+
+ @Test
+ void testHandlesSystemAndEmptyPhysicalColumns() {
+ IndexManifestEntry index = index("index", 0, 99, 5L, BinaryRow.EMPTY_ROW, 0);
+
+ assertThat(
+ plan(
+ Collections.singletonList(
+ data(
+ "vector-with-system-fields",
+ 0,
+ 100,
+ 6,
+ 1,
+ SpecialFields.ROW_ID.name(),
+ "vector",
+ SpecialFields.SEQUENCE_NUMBER.name())),
+ index))
+ .containsExactly(index);
+ assertThat(
+ plan(
+ Collections.singletonList(
+ data(
+ "system-only",
+ 0,
+ 100,
+ 6,
+ 1,
+ SpecialFields.ROW_ID.name(),
+ SpecialFields.SEQUENCE_NUMBER.name())),
+ index))
+ .isEmpty();
+ assertThat(
+ plan(
+ Collections.singletonList(
+ dataWithWriteCols(
+ "empty", 0, 100, 6, 1, Collections.emptyList())),
+ index))
+ .isEmpty();
+ }
+
+ @Test
+ void testMultiColumnIndexRefreshesForEitherIndexedField() {
+ List<DataField> indexedFields = Arrays.asList(VECTOR_FIELD, OTHER_FIELD);
+ IndexManifestEntry index =
+ index(
+ "multi-column",
+ 0,
+ 99,
+ 5L,
+ BinaryRow.EMPTY_ROW,
+ 0,
+ new int[] {OTHER_FIELD.id()});
+
+ assertThat(
+ plan(
+ Collections.singletonList(
+ data("vector-update", 0, 100, 6, 1, "vector")),
+ Collections.singletonList(index),
+ indexedFields))
+ .containsExactly(index);
+ assertThat(
+ plan(
+ Collections.singletonList(
+ data("extra-field-update", 0, 100, 6, 1, "other")),
+ Collections.singletonList(index),
+ indexedFields))
+ .containsExactly(index);
+ assertThat(
+ plan(
+ Collections.singletonList(
+ data("unrelated-update", 0, 100, 6, 1, "unrelated")),
+ Collections.singletonList(index),
+ indexedFields))
+ .isEmpty();
+ }
+
+ @Test
+ void testIsolatesPartitionAndBucket() {
+ BinaryRow partition = BinaryRow.singleColumn(1);
+ IndexManifestEntry index = index("index", 0, 99, 5L, partition, 2);
+
+ ManifestEntry wrongPartition =
+ data("wrong-partition", 0, 100, 6, 1, BinaryRow.singleColumn(2), 2, "vector");
+ ManifestEntry wrongBucket = data("wrong-bucket", 0, 100, 6, 1, partition, 1, "vector");
+ assertThat(plan(Arrays.asList(wrongPartition, wrongBucket), index)).isEmpty();
+
+ ManifestEntry matching = data("matching", 0, 100, 6, 1, partition, 2, "vector");
+ assertThat(plan(Collections.singletonList(matching), index)).containsExactly(index);
+ }
+
+ @Test
+ void testSequenceSweepHandlesOverlappingRangesAndPreservesOrder() {
+ IndexManifestEntry equalSequence = index("equal", 20, 29, 5L, BinaryRow.EMPTY_ROW, 0);
+ IndexManifestEntry boundary = index("boundary", 10, 19, 6L, BinaryRow.EMPTY_ROW, 0);
+ IndexManifestEntry first = index("first", 0, 9, 8L, BinaryRow.EMPTY_ROW, 0);
+
+ List<ManifestEntry> dataEntries =
+ Arrays.asList(
+ data("wide-equal", 0, 30, 5, 1, "vector"),
+ data("high-outside", 30, 10, 100, 1, "vector"),
+ data("boundary-update", 9, 2, 7, 1, "vector"),
+ data("first-update", 0, 10, 9, 1, "vector"));
+
+ assertThat(plan(dataEntries, Arrays.asList(equalSequence, boundary, first)))
+ .containsExactly(boundary, first);
+ }
+
+ @Test
+ void testRefreshesAllBTreeKeyShardsForOneRowRange() {
+ IndexManifestEntry first =
+ index("btree-first", "btree", OTHER_FIELD, 0, 99, 5L, BinaryRow.EMPTY_ROW, 0);
+ IndexManifestEntry second =
+ index("btree-second", "btree", OTHER_FIELD, 0, 99, 5L, BinaryRow.EMPTY_ROW, 0);
+ IndexManifestEntry nextRange =
+ index("btree-next", "btree", OTHER_FIELD, 100, 199, 5L, BinaryRow.EMPTY_ROW, 0);
+
+ assertThat(
+ plan(
+ Collections.singletonList(data("key-update", 10, 1, 6, 1, "other")),
+ Arrays.asList(first, second, nextRange),
+ Collections.singletonList(OTHER_FIELD)))
+ .containsExactly(first, second);
+ }
+
+ @Test
+ void testSequenceSweepMatchesBruteForceForOverlappingRanges() {
+ Random random = new Random(123456L);
+ for (int round = 0; round < 10; round++) {
+ List<ManifestEntry> dataEntries = new ArrayList<>();
+ for (int i = 0; i < 100; i++) {
+ dataEntries.add(
+ data(
+ "data-" + round + "-" + i,
+ random.nextInt(1000),
+ random.nextInt(100) + 1,
+ random.nextInt(25),
+ 1,
+ "vector"));
+ }
+
+ List<IndexManifestEntry> indexEntries = new ArrayList<>();
+ for (int i = 0; i < 100; i++) {
+ long from = random.nextInt(1000);
+ indexEntries.add(
+ index(
+ "index-" + round + "-" + i,
+ from,
+ from + random.nextInt(100),
+ (long) random.nextInt(24) + 1,
+ BinaryRow.EMPTY_ROW,
+ 0));
+ }
+ Collections.shuffle(dataEntries, random);
+ Collections.shuffle(indexEntries, random);
+
+ assertThat(plan(dataEntries, indexEntries))
+ .containsExactlyElementsOf(bruteForcePlan(dataEntries, indexEntries));
+ }
+ }
+
+ private List<IndexManifestEntry> plan(
+ List<ManifestEntry> dataEntries, IndexManifestEntry indexEntry) {
+ return plan(dataEntries, Collections.singletonList(indexEntry));
+ }
+
+ private List<IndexManifestEntry> plan(
+ List<ManifestEntry> dataEntries, List<IndexManifestEntry> indexEntries) {
+ return plan(dataEntries, indexEntries, Collections.singletonList(VECTOR_FIELD));
+ }
+
+ private List<IndexManifestEntry> plan(
+ List<ManifestEntry> dataEntries,
+ List<IndexManifestEntry> indexEntries,
+ List<DataField> indexedFields) {
+ return DataEvolutionGlobalIndexRefreshPlanner.findIndexesToRefresh(
+ schemaManager, dataEntries, indexEntries, indexedFields);
+ }
+
+ private List<IndexManifestEntry> bruteForcePlan(
+ List<ManifestEntry> dataEntries, List<IndexManifestEntry> indexEntries) {
+ List<IndexManifestEntry> result = new ArrayList<>();
+ for (IndexManifestEntry indexEntry : indexEntries) {
+ GlobalIndexMeta indexMeta = indexEntry.indexFile().globalIndexMeta();
+ long scanSnapshotId =
+ DataEvolutionIndexSourceMeta.deserialize(indexMeta.sourceMeta())
+ .scanSnapshotId();
+ for (ManifestEntry dataEntry : dataEntries) {
+ DataFileMeta file = dataEntry.file();
+ if (file.maxSequenceNumber() > scanSnapshotId
+ && file.nonNullRowIdRange().hasIntersection(indexMeta.rowRange())) {
+ result.add(indexEntry);
+ break;
+ }
+ }
+ }
+ return result;
+ }
+
+ private IndexManifestEntry index(
+ String fileName,
+ long from,
+ long to,
+ Long scanSnapshotId,
+ BinaryRow partition,
+ int bucket) {
+ return index(fileName, from, to, scanSnapshotId, partition, bucket, null);
+ }
+
+ private IndexManifestEntry index(
+ String fileName,
+ long from,
+ long to,
+ Long scanSnapshotId,
+ BinaryRow partition,
+ int bucket,
+ int[] extraFieldIds) {
+ byte[] sourceMeta =
+ scanSnapshotId == null
+ ? null
+ : new DataEvolutionIndexSourceMeta(scanSnapshotId).serialize();
+ return new IndexManifestEntry(
+ FileKind.ADD,
+ partition,
+ bucket,
+ new IndexFileMeta(
+ "lumina",
+ fileName,
+ 1L,
+ to - from + 1,
+ new GlobalIndexMeta(
+ from, to, VECTOR_FIELD.id(), extraFieldIds, null, sourceMeta),
+ null));
+ }
+
+ private IndexManifestEntry index(
+ String fileName,
+ String indexType,
+ DataField indexField,
+ long from,
+ long to,
+ long scanSnapshotId,
+ BinaryRow partition,
+ int bucket) {
+ byte[] sourceMeta = new DataEvolutionIndexSourceMeta(scanSnapshotId).serialize();
+ return new IndexManifestEntry(
+ FileKind.ADD,
+ partition,
+ bucket,
+ new IndexFileMeta(
+ indexType,
+ fileName,
+ 1L,
+ to - from + 1,
+ new GlobalIndexMeta(from, to, indexField.id(), null, null, sourceMeta),
+ null));
+ }
+
+ private ManifestEntry data(
+ String fileName,
+ long firstRowId,
+ long rowCount,
+ long maxSequenceNumber,
+ long schemaId,
+ String... writeCols) {
+ return data(
+ fileName,
+ firstRowId,
+ rowCount,
+ maxSequenceNumber,
+ schemaId,
+ BinaryRow.EMPTY_ROW,
+ 0,
+ writeCols);
+ }
+
+ private ManifestEntry data(
+ String fileName,
+ long firstRowId,
+ long rowCount,
+ long maxSequenceNumber,
+ long schemaId,
+ BinaryRow partition,
+ int bucket,
+ String... writeCols) {
+ List<String> physicalColumns = writeCols.length == 0 ? null : Arrays.asList(writeCols);
+ DataFileMeta file =
+ DataFileMeta.forAppend(
+ fileName,
+ 1L,
+ rowCount,
+ SimpleStats.EMPTY_STATS,
+ maxSequenceNumber,
+ maxSequenceNumber,
+ schemaId,
+ Collections.emptyList(),
+ null,
+ null,
+ null,
+ null,
+ firstRowId,
+ physicalColumns);
+ return ManifestEntry.create(FileKind.ADD, partition, bucket, 1, file);
+ }
+
+ private ManifestEntry dataWithWriteCols(
+ String fileName,
+ long firstRowId,
+ long rowCount,
+ long maxSequenceNumber,
+ long schemaId,
+ List<String> writeCols) {
+ DataFileMeta file =
+ DataFileMeta.forAppend(
+ fileName,
+ 1L,
+ rowCount,
+ SimpleStats.EMPTY_STATS,
+ maxSequenceNumber,
+ maxSequenceNumber,
+ schemaId,
+ Collections.emptyList(),
+ null,
+ null,
+ null,
+ null,
+ firstRowId,
+ writeCols);
+ return ManifestEntry.create(FileKind.ADD, BinaryRow.EMPTY_ROW, 0, 1, file);
+ }
+}
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 11c8ddb64c..cf0e6820ad 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
@@ -23,19 +23,24 @@ 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.DataEvolutionIndexSourceMeta;
import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.index.IndexPathFactory;
+import org.apache.paimon.io.DataFileMeta;
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.table.source.DataSplit;
+import org.apache.paimon.table.source.Split;
import org.apache.paimon.types.ArrayType;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.FloatType;
import org.apache.paimon.types.IntType;
import org.apache.paimon.types.VarCharType;
import org.apache.paimon.utils.Range;
+import org.apache.paimon.utils.RowRangeIndex;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -45,6 +50,7 @@ import java.io.IOException;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
+import java.util.Map;
import java.util.UUID;
import static org.assertj.core.api.Assertions.assertThat;
@@ -94,7 +100,14 @@ class GlobalIndexBuilderUtilsTest {
List<IndexFileMeta> metas =
GlobalIndexBuilderUtils.toIndexFileMetas(
- fileIO, indexPathFactory, coreOptions, range, fields, "test-type", entries);
+ fileIO,
+ indexPathFactory,
+ coreOptions,
+ range,
+ fields,
+ "test-type",
+ entries,
+ null);
assertThat(metas).hasSize(1);
assertThat(metas.get(0).globalIndexMeta().indexFieldId()).isEqualTo(1);
@@ -115,13 +128,39 @@ class GlobalIndexBuilderUtilsTest {
List<IndexFileMeta> metas =
GlobalIndexBuilderUtils.toIndexFileMetas(
- fileIO, indexPathFactory, coreOptions, range, fields, "test-type", entries);
+ fileIO,
+ indexPathFactory,
+ coreOptions,
+ range,
+ fields,
+ "test-type",
+ entries,
+ null);
assertThat(metas).hasSize(1);
assertThat(metas.get(0).globalIndexMeta().indexFieldId()).isEqualTo(1);
assertThat(metas.get(0).globalIndexMeta().extraFieldIds()).isNull();
}
+ @Test
+ void testToIndexFileMetasWithSourceMeta() throws IOException {
+ DataField field = new DataField(1, "vec", new ArrayType(new FloatType()));
+ byte[] sourceMeta = new DataEvolutionIndexSourceMeta(7L).serialize();
+
+ List<IndexFileMeta> metas =
+ GlobalIndexBuilderUtils.toIndexFileMetas(
+ fileIO,
+ indexPathFactory,
+ coreOptions,
+ new Range(0, 9),
+ Collections.singletonList(field),
+ "lumina",
+ createDummyResultEntries(),
+ sourceMeta);
+
+ assertThat(metas.get(0).globalIndexMeta().sourceMeta()).containsExactly(sourceMeta);
+ }
+
// Test: 3 columns (title + vec + id), primary column title is indexFieldId, rest in
// extraFieldIds
@Test
@@ -136,7 +175,14 @@ class GlobalIndexBuilderUtilsTest {
List<IndexFileMeta> metas =
GlobalIndexBuilderUtils.toIndexFileMetas(
- fileIO, indexPathFactory, coreOptions, range, fields, "test-type", entries);
+ fileIO,
+ indexPathFactory,
+ coreOptions,
+ range,
+ fields,
+ "test-type",
+ entries,
+ null);
assertThat(metas).hasSize(1);
assertThat(metas.get(0).globalIndexMeta().indexFieldId()).isEqualTo(1);
@@ -176,6 +222,63 @@ class GlobalIndexBuilderUtilsTest {
assertThat(splits.get(0).dataSplit()).isEqualTo(splits.get(1).dataSplit());
}
+ @Test
+ void testSplitByContiguousRowRangeFromDataFiles() {
+ DataFileMeta file1 = createDataFileMeta(0L, 100L);
+ DataFileMeta file2 = createDataFileMeta(300L, 100L);
+ DataFileMeta file3 = createDataFileMeta(100L, 100L);
+ DataSplit split =
+ DataSplit.builder()
+ .withSnapshot(1L)
+ .withPartition(BinaryRow.EMPTY_ROW)
+ .withBucket(0)
+ .withBucketPath("bucket-0")
+ .withDataFiles(Arrays.asList(file1, file2, file3))
+ .isStreaming(false)
+ .rawConvertible(false)
+ .build();
+
+ List<DataSplit> rebuilt =
+ GlobalIndexBuilderUtils.splitByContiguousRowRange(Collections.singletonList(split));
+
+ assertThat(rebuilt).hasSize(2);
+ assertThat(rebuilt.get(0).dataFiles()).containsExactly(file1, file3);
+ assertThat(rebuilt.get(1).dataFiles()).containsExactly(file2);
+ assertThat(GlobalIndexBuilderUtils.calcRowRange(rebuilt.get(0)))
+ .isEqualTo(new Range(0, 199));
+ assertThat(GlobalIndexBuilderUtils.calcRowRange(rebuilt.get(1)))
+ .isEqualTo(new Range(300, 399));
+ }
+
+ @Test
+ void testGroupSplitsByDiscontiguousRowRangeIndex() {
+ DataFileMeta file1 = createDataFileMeta(4750L, 151L);
+ DataFileMeta file2 = createDataFileMeta(4901L, 1037L);
+ DataFileMeta file3 = createDataFileMeta(5938L, 1662L);
+ DataSplit split =
+ DataSplit.builder()
+ .withSnapshot(1L)
+ .withPartition(BinaryRow.EMPTY_ROW)
+ .withBucket(0)
+ .withBucketPath("bucket-0")
+ .withDataFiles(Arrays.asList(file1, file2, file3))
+ .isStreaming(false)
+ .rawConvertible(false)
+ .build();
+
+ Map<BinaryRow, Map<Range, List<Split>>> result =
+ GlobalIndexBuilderUtils.groupSplitsByRange(
+ RowRangeIndex.create(
+ Arrays.asList(new Range(4750, 4900), new Range(5938, 7599))),
+ Collections.singletonList(split));
+
+ assertThat(result).containsOnlyKeys(BinaryRow.EMPTY_ROW);
+ Map<Range, List<Split>> ranges = result.get(BinaryRow.EMPTY_ROW);
+ assertThat(ranges).containsOnlyKeys(new Range(4750, 4900), new Range(5938, 7599));
+ assertIndexedSplitRowRanges(ranges.get(new Range(4750, 4900)), new Range(4750, 4900));
+ assertIndexedSplitRowRanges(ranges.get(new Range(5938, 7599)), new Range(5938, 7599));
+ }
+
private List<ResultEntry> createDummyResultEntries() throws IOException {
String fileName = "test-index-" + UUID.randomUUID();
Path filePath = indexPathFactory.toPath(fileName);
@@ -208,4 +311,34 @@ class GlobalIndexBuilderUtilsTest {
null);
return ManifestEntry.create(FileKind.ADD, BinaryRow.EMPTY_ROW, 0, 1, file);
}
+
+ private static void assertIndexedSplitRowRanges(List<Split> splits, Range rowRange) {
+ assertThat(splits).hasSize(1);
+ assertThat(splits.get(0)).isInstanceOf(IndexedSplit.class);
+ assertThat(((IndexedSplit) splits.get(0)).rowRanges()).containsExactly(rowRange);
+ }
+
+ private static DataFileMeta createDataFileMeta(long firstRowId, long rowCount) {
+ return 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);
+ }
}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/globalindex/GenericGlobalIndexBuilderDeletionVectorTest.java b/paimon-core/src/test/java/org/apache/paimon/globalindex/generic/GenericGlobalIndexScannerDeletionVectorTest.java
similarity index 89%
rename from paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/globalindex/GenericGlobalIndexBuilderDeletionVectorTest.java
rename to paimon-core/src/test/java/org/apache/paimon/globalindex/generic/GenericGlobalIndexScannerDeletionVectorTest.java
index 62de5eee68..99cc244bc3 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/globalindex/GenericGlobalIndexBuilderDeletionVectorTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/globalindex/generic/GenericGlobalIndexScannerDeletionVectorTest.java
@@ -16,7 +16,7 @@
* limitations under the License.
*/
-package org.apache.paimon.flink.globalindex;
+package org.apache.paimon.globalindex.generic;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.table.FileStoreTable;
@@ -28,8 +28,8 @@ import static org.assertj.core.api.Assertions.assertThatCode;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
-/** Tests deletion-vector table support in {@link GenericGlobalIndexBuilder}. */
-class GenericGlobalIndexBuilderDeletionVectorTest {
+/** Tests deletion-vector table support in {@link GenericGlobalIndexScanner}. */
+class GenericGlobalIndexScannerDeletionVectorTest {
private static FileStoreTable table(boolean deletionVectorsEnabled) {
FileStoreTable table = mock(FileStoreTable.class);
@@ -46,13 +46,13 @@ class GenericGlobalIndexBuilderDeletionVectorTest {
@Test
void testGlobalIndexAllowedOnDeletionVectorTable() {
- assertThatCode(() -> new GenericGlobalIndexBuilder(table(true)).scan())
+ assertThatCode(() -> new GenericGlobalIndexScanner(table(true)).scan())
.doesNotThrowAnyException();
}
@Test
void testGlobalIndexAllowedWithoutDeletionVectors() {
- assertThatCode(() -> new GenericGlobalIndexBuilder(table(false)).scan())
+ assertThatCode(() -> new GenericGlobalIndexScanner(table(false)).scan())
.doesNotThrowAnyException();
}
}
diff --git a/paimon-core/src/test/java/org/apache/paimon/globalindex/generic/GenericGlobalIndexScannerTest.java b/paimon-core/src/test/java/org/apache/paimon/globalindex/generic/GenericGlobalIndexScannerTest.java
new file mode 100644
index 0000000000..9b3b00c4d2
--- /dev/null
+++ b/paimon-core/src/test/java/org/apache/paimon/globalindex/generic/GenericGlobalIndexScannerTest.java
@@ -0,0 +1,111 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.globalindex.generic;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.globalindex.ScanResult;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.TableTestBase;
+import org.apache.paimon.table.sink.BatchTableCommit;
+import org.apache.paimon.table.sink.BatchTableWrite;
+import org.apache.paimon.table.sink.BatchWriteBuilder;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.utils.Range;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link GenericGlobalIndexScanner}. */
+public class GenericGlobalIndexScannerTest extends TableTestBase {
+
+ @Override
+ public Schema schemaDefault() {
+ return Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("v", DataTypes.STRING())
+ .option(CoreOptions.BUCKET.key(), "-1")
+ .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true")
+ .option(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true")
+ .build();
+ }
+
+ @Test
+ public void testScan() throws Exception {
+ FileStoreTable table = writeRows();
+ long snapshotId = table.snapshotManager().latestSnapshot().id();
+
+ ScanResult<ManifestEntry> scanResult =
+ new GenericGlobalIndexScanner(table)
+ .scan()
+ .orElseThrow(() -> new IllegalStateException("Expected scan result."));
+
+ assertThat(scanResult.scanSnapshotId()).isEqualTo(snapshotId);
+ assertThat(scanResult.entries()).isNotEmpty();
+ assertThat(scanResult.rowRangeIndex().ranges()).containsExactly(new Range(0, 9));
+ assertThat(scanResult.deletedIndexEntries()).isEmpty();
+ }
+
+ @Test
+ public void testIncrementalScanWithoutExistingIndex() throws Exception {
+ FileStoreTable table = writeRows();
+
+ ScanResult<ManifestEntry> scanResult =
+ new GenericGlobalIndexScanner(table)
+ .withIndex("test-index", Collections.singletonList("v"), new Options())
+ .incrementalScan()
+ .orElseThrow(
+ () ->
+ new IllegalStateException(
+ "Expected incremental scan result."));
+
+ assertThat(scanResult.entries()).isNotEmpty();
+ assertThat(scanResult.rowRangeIndex().ranges()).containsExactly(new Range(0, 9));
+ assertThat(scanResult.deletedIndexEntries()).isEmpty();
+ }
+
+ @Test
+ public void testScanEmptyTable() throws Exception {
+ createTableDefault();
+
+ assertThat(new GenericGlobalIndexScanner(getTableDefault()).scan()).isEmpty();
+ }
+
+ private FileStoreTable writeRows() throws Exception {
+ createTableDefault();
+ FileStoreTable table = getTableDefault();
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = builder.newWrite()) {
+ for (int i = 0; i < 10; i++) {
+ write.write(GenericRow.of(i, BinaryString.fromString("v-" + i)));
+ }
+ try (BatchTableCommit commit = builder.newCommit()) {
+ commit.commit(write.prepareCommit());
+ }
+ }
+ return table;
+ }
+}
diff --git a/paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexBuilderSplitTest.java b/paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexBuilderSplitTest.java
deleted file mode 100644
index 2d64ce9db5..0000000000
--- a/paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexBuilderSplitTest.java
+++ /dev/null
@@ -1,131 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.paimon.globalindex.sorted;
-
-import org.apache.paimon.data.BinaryRow;
-import org.apache.paimon.globalindex.IndexedSplit;
-import org.apache.paimon.io.DataFileMeta;
-import org.apache.paimon.io.PojoDataFileMeta;
-import org.apache.paimon.stats.SimpleStats;
-import org.apache.paimon.table.source.DataSplit;
-import org.apache.paimon.table.source.Split;
-import org.apache.paimon.utils.Range;
-import org.apache.paimon.utils.RowRangeIndex;
-
-import org.junit.jupiter.api.Test;
-
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.List;
-import java.util.Map;
-import java.util.UUID;
-
-import static org.assertj.core.api.Assertions.assertThat;
-
-/** Tests for split regrouping in {@link SortedGlobalIndexBuilder}. */
-public class SortedGlobalIndexBuilderSplitTest {
-
- @Test
- public void testSplitByContiguousRowRangeFromDataFiles() {
- DataFileMeta file1 = createDataFileMeta(0L, 100L);
- DataFileMeta file2 = createDataFileMeta(300L, 100L);
- DataFileMeta file3 = createDataFileMeta(100L, 100L);
- DataSplit split =
- DataSplit.builder()
- .withSnapshot(1L)
- .withPartition(BinaryRow.EMPTY_ROW)
- .withBucket(0)
- .withBucketPath("bucket-0")
- .withDataFiles(Arrays.asList(file1, file2, file3))
- .isStreaming(false)
- .rawConvertible(false)
- .build();
-
- List<DataSplit> rebuilt =
- SortedGlobalIndexBuilder.splitByContiguousRowRange(
- Collections.singletonList(split));
-
- assertThat(rebuilt).hasSize(2);
- assertThat(rebuilt.get(0).dataFiles()).containsExactly(file1, file3);
- assertThat(rebuilt.get(1).dataFiles()).containsExactly(file2);
- assertThat(SortedGlobalIndexBuilder.calcRowRange(rebuilt.get(0)))
- .isEqualTo(new Range(0, 199));
- assertThat(SortedGlobalIndexBuilder.calcRowRange(rebuilt.get(1)))
- .isEqualTo(new Range(300, 399));
- }
-
- @Test
- public void testGroupSplitsByDiscontiguousRowRangeIndex() {
- DataFileMeta file1 = createDataFileMeta(4750L, 151L);
- DataFileMeta file2 = createDataFileMeta(4901L, 1037L);
- DataFileMeta file3 = createDataFileMeta(5938L, 1662L);
- DataSplit split =
- DataSplit.builder()
- .withSnapshot(1L)
- .withPartition(BinaryRow.EMPTY_ROW)
- .withBucket(0)
- .withBucketPath("bucket-0")
- .withDataFiles(Arrays.asList(file1, file2, file3))
- .isStreaming(false)
- .rawConvertible(false)
- .build();
-
- Map<BinaryRow, Map<Range, List<Split>>> result =
- SortedGlobalIndexBuilder.groupSplitsByRange(
- RowRangeIndex.create(
- Arrays.asList(new Range(4750, 4900), new Range(5938, 7599))),
- Collections.singletonList(split));
-
- assertThat(result).containsOnlyKeys(BinaryRow.EMPTY_ROW);
- Map<Range, List<Split>> ranges = result.get(BinaryRow.EMPTY_ROW);
- assertThat(ranges).containsOnlyKeys(new Range(4750, 4900), new Range(5938, 7599));
- assertIndexedSplitRowRanges(ranges.get(new Range(4750, 4900)), new Range(4750, 4900));
- assertIndexedSplitRowRanges(ranges.get(new Range(5938, 7599)), new Range(5938, 7599));
- }
-
- private static void assertIndexedSplitRowRanges(List<Split> splits, Range rowRange) {
- assertThat(splits).hasSize(1);
- assertThat(splits.get(0)).isInstanceOf(IndexedSplit.class);
- assertThat(((IndexedSplit) splits.get(0)).rowRanges()).containsExactly(rowRange);
- }
-
- private static DataFileMeta createDataFileMeta(long firstRowId, long rowCount) {
- return 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);
- }
-}
diff --git a/paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexBuilderTest.java b/paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexScannerTest.java
similarity index 78%
rename from paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexBuilderTest.java
rename to paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexScannerTest.java
index 84aca44201..f918b17289 100644
--- a/paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexBuilderTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexScannerTest.java
@@ -24,11 +24,10 @@ import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.BlobData;
import org.apache.paimon.data.GenericRow;
-import org.apache.paimon.data.InternalRow;
-import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter;
import org.apache.paimon.globalindex.KeySerializer;
-import org.apache.paimon.globalindex.ResultEntry;
+import org.apache.paimon.globalindex.ScanResult;
import org.apache.paimon.globalindex.btree.BTreeIndexOptions;
+import org.apache.paimon.index.DataEvolutionIndexSourceMeta;
import org.apache.paimon.index.GlobalIndexMeta;
import org.apache.paimon.index.IndexFileHandler;
import org.apache.paimon.index.IndexFileMeta;
@@ -53,105 +52,23 @@ import org.apache.paimon.utils.Pair;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
-import java.io.Closeable;
-import java.util.ArrayDeque;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
-import java.util.Queue;
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.when;
-/** Test class for {@link SortedGlobalIndexBuilder}. */
-public class SortedGlobalIndexBuilderTest extends TableTestBase {
+/** Test class for {@link SortedGlobalIndexScanner}. */
+public class SortedGlobalIndexScannerTest extends TableTestBase {
private static final long PART_ROW_NUM = 1000L;
private static final KeySerializer KEY_SERIALIZER = KeySerializer.create(DataTypes.INT());
private static final Comparator<Object> COMPARATOR = KEY_SERIALIZER.createComparator();
- @Test
- public void testSingleColumnWriterRotationPreservesResultGroups() throws Exception {
- GlobalIndexSingleColumnWriter first = mock(GlobalIndexSingleColumnWriter.class);
- GlobalIndexSingleColumnWriter second = mock(GlobalIndexSingleColumnWriter.class);
- when(first.finish())
- .thenReturn(Collections.singletonList(new ResultEntry("index-1", 2, null)));
- when(second.finish())
- .thenReturn(Collections.singletonList(new ResultEntry("index-2", 1, null)));
- Queue<GlobalIndexSingleColumnWriter> writers =
- new ArrayDeque<>(Arrays.asList(first, second));
- SortedSingleColumnIndexWriter rotatingWriter =
- new SortedSingleColumnIndexWriter(2, writers::remove);
-
- rotatingWriter.write(10, 0);
- rotatingWriter.write(20, 1);
- rotatingWriter.write(30, 2);
- List<List<ResultEntry>> results = rotatingWriter.finish();
-
- verify(first).write(10, 0);
- verify(first).write(20, 1);
- verify(second).write(30, 2);
- assertThat(results).hasSize(2);
- assertThat(results.get(0)).extracting(ResultEntry::fileName).containsExactly("index-1");
- assertThat(results.get(1)).extracting(ResultEntry::fileName).containsExactly("index-2");
- }
-
- @Test
- public void testSingleColumnWriterClosesActiveWriter() throws Exception {
- GlobalIndexSingleColumnWriter activeWriter =
- mock(
- GlobalIndexSingleColumnWriter.class,
- org.mockito.Mockito.withSettings().extraInterfaces(Closeable.class));
- SortedSingleColumnIndexWriter rotatingWriter =
- new SortedSingleColumnIndexWriter(2, () -> activeWriter);
- rotatingWriter.write(10, 0);
-
- assertThat(rotatingWriter).isInstanceOf(AutoCloseable.class);
- ((AutoCloseable) rotatingWriter).close();
-
- verify((Closeable) activeWriter).close();
- }
-
- @Test
- public void testBuildForSinglePartitionClosesWriterAfterFailure() throws Exception {
- createTableDefault();
- GlobalIndexSingleColumnWriter activeWriter =
- mock(
- GlobalIndexSingleColumnWriter.class,
- org.mockito.Mockito.withSettings().extraInterfaces(Closeable.class));
- org.mockito.Mockito.doThrow(new RuntimeException("write failed"))
- .when(activeWriter)
- .write(10, 0);
- SortedGlobalIndexBuilder builder =
- new SortedGlobalIndexBuilder(getTableDefault(), "btree") {
- @Override
- public GlobalIndexSingleColumnWriter createWriter() {
- return activeWriter;
- }
- };
- builder.withIndexField("f0");
-
- org.assertj.core.api.Assertions.assertThatThrownBy(
- () ->
- builder.buildForSinglePartition(
- new org.apache.paimon.utils.Range(0, 0),
- null,
- Collections.<InternalRow>singletonList(
- GenericRow.of(10, 0L))
- .iterator()))
- .isInstanceOf(RuntimeException.class)
- .hasMessage("write failed");
-
- verify((Closeable) activeWriter).close();
- }
-
@Override
public Schema schemaDefault() {
Schema.Builder schemaBuilder = Schema.newBuilder();
@@ -197,19 +114,20 @@ public class SortedGlobalIndexBuilderTest extends TableTestBase {
private void createIndex(PartitionPredicate partitionPredicate) throws Exception {
FileStoreTable table = getTableDefault();
- SortedGlobalIndexBuilder builder = new SortedGlobalIndexBuilder(table, "btree");
+ SortedGlobalIndexScanner builder = new SortedGlobalIndexScanner(table, "btree");
builder.withIndexField("f0");
builder.withPartitionPredicate(partitionPredicate);
- List<DataSplit> dataSplits =
+ ScanResult<DataSplit> scanResult =
builder.scan()
- .map(Pair::getRight)
.orElseThrow(
() ->
new IllegalStateException(
"Expected scan result when building index."));
List<CommitMessage> commitMessages = new ArrayList<>();
- for (DataSplit dataSplit : dataSplits) {
- commitMessages.addAll(builder.build(dataSplit, ioManager));
+ for (DataSplit dataSplit : scanResult.entries()) {
+ commitMessages.addAll(
+ SortedGlobalIndexTestUtils.buildIndex(
+ table, "btree", "f0", dataSplit, scanResult.scanSnapshotId()));
}
try (BatchTableCommit commit = table.newBatchWriteBuilder().newCommit()) {
@@ -251,12 +169,51 @@ public class SortedGlobalIndexBuilderTest extends TableTestBase {
metasByParts.forEach(this::assertFilesNonOverlapping);
}
+ @Test
+ public void testBuildStoresScanSnapshotInSourceMeta() throws Exception {
+ write();
+
+ long scanSnapshotId = getTableDefault().snapshotManager().latestSnapshot().id();
+ createIndex(null);
+
+ List<IndexManifestEntry> entries =
+ getTableDefault()
+ .store()
+ .newIndexFileHandler()
+ .scan(getTableDefault().snapshotManager().latestSnapshot(), "btree");
+ assertThat(entries).isNotEmpty();
+ assertThat(entries)
+ .allSatisfy(
+ entry ->
+ assertThat(
+ DataEvolutionIndexSourceMeta.fromIndexFile(
+ entry.indexFile())
+ .scanSnapshotId())
+ .isEqualTo(scanSnapshotId));
+ }
+
+ @Test
+ public void testScanResultContainsScanSnapshot() throws Exception {
+ write();
+ FileStoreTable table = getTableDefault();
+ long expectedSnapshotId = table.snapshotManager().latestSnapshot().id();
+
+ ScanResult<DataSplit> scanResult =
+ new SortedGlobalIndexScanner(table, "btree")
+ .withIndexField("f0")
+ .scan()
+ .orElseThrow(() -> new IllegalStateException("Expected scan result."));
+
+ assertThat(scanResult.scanSnapshotId()).isEqualTo(expectedSnapshotId);
+ assertThat(scanResult.deletedIndexEntries()).isEmpty();
+ }
+
@Test
public void testIncrementalScanNoData() throws Exception {
createTableDefault();
FileStoreTable table = getTableDefault();
- SortedGlobalIndexBuilder builder = new SortedGlobalIndexBuilder(table, "btree");
+ SortedGlobalIndexScanner builder = new SortedGlobalIndexScanner(table, "btree");
builder.withIndexField("f0");
Assertions.assertFalse(
@@ -270,7 +227,7 @@ public class SortedGlobalIndexBuilderTest extends TableTestBase {
createIndex(null);
FileStoreTable table = getTableDefault();
- SortedGlobalIndexBuilder builder = new SortedGlobalIndexBuilder(table, "btree");
+ SortedGlobalIndexScanner builder = new SortedGlobalIndexScanner(table, "btree");
builder.withIndexField("f0");
Assertions.assertFalse(
@@ -300,15 +257,14 @@ public class SortedGlobalIndexBuilderTest extends TableTestBase {
}
table = getTableDefault();
- SortedGlobalIndexBuilder builder = new SortedGlobalIndexBuilder(table, "btree");
+ SortedGlobalIndexScanner builder = new SortedGlobalIndexScanner(table, "btree");
builder.withIndexField("f0");
- Optional<Pair<org.apache.paimon.utils.RowRangeIndex, List<DataSplit>>> incrementalScan =
- builder.incrementalScan();
+ Optional<ScanResult<DataSplit>> incrementalScan = builder.incrementalScan();
Assertions.assertTrue(
incrementalScan.isPresent(),
"incrementalScan should return non-empty splits for newly written data");
- List<DataSplit> splits = incrementalScan.get().getRight();
+ List<DataSplit> splits = incrementalScan.get().entries();
Assertions.assertFalse(
splits.isEmpty(),
"incrementalScan should return non-empty splits for newly written data");
@@ -358,13 +314,13 @@ public class SortedGlobalIndexBuilderTest extends TableTestBase {
predicate =
PartitionPredicate.createPartitionPredicate(
partType, Collections.singletonMap("dt", BinaryString.fromString("p0")));
- SortedGlobalIndexBuilder builder = new SortedGlobalIndexBuilder(table, "btree");
+ SortedGlobalIndexScanner builder = new SortedGlobalIndexScanner(table, "btree");
builder.withIndexField("f0");
builder.withPartitionPredicate(PartitionPredicate.fromPredicate(partType, predicate));
List<DataSplit> splits =
builder.incrementalScan()
- .map(Pair::getRight)
+ .map(ScanResult::entries)
.orElseThrow(
() ->
new IllegalStateException(
@@ -418,18 +374,18 @@ public class SortedGlobalIndexBuilderTest extends TableTestBase {
containsBlobFile(table.store().newScan().plan().files()),
"Test table should contain blob manifest entries.");
- SortedGlobalIndexBuilder builder =
- new SortedGlobalIndexBuilder(table, "btree").withIndexField("f0");
+ SortedGlobalIndexScanner builder =
+ new SortedGlobalIndexScanner(table, "btree").withIndexField("f0");
assertNoBlobFiles(
builder.scan()
- .map(Pair::getRight)
+ .map(ScanResult::entries)
.orElseThrow(
() ->
new IllegalStateException(
"Expected scan result for blob table.")));
assertNoBlobFiles(
builder.incrementalScan()
- .map(Pair::getRight)
+ .map(ScanResult::entries)
.orElseThrow(
() ->
new IllegalStateException(
diff --git a/paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexTestUtils.java b/paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexTestUtils.java
new file mode 100644
index 0000000000..ed9238068f
--- /dev/null
+++ b/paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexTestUtils.java
@@ -0,0 +1,96 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.globalindex.sorted;
+
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.globalindex.KeySerializer;
+import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.SpecialFields;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.CloseableIterator;
+import org.apache.paimon.utils.InternalRowUtils;
+import org.apache.paimon.utils.Pair;
+import org.apache.paimon.utils.Range;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.List;
+
+import static org.apache.paimon.globalindex.GlobalIndexBuilderUtils.calcRowRange;
+
+/** Utilities for building sorted indexes in tests without exposing a test-only production API. */
+public final class SortedGlobalIndexTestUtils {
+
+ private SortedGlobalIndexTestUtils() {}
+
+ public static List<CommitMessage> buildIndex(
+ FileStoreTable table,
+ String indexType,
+ String indexFieldName,
+ DataSplit dataSplit,
+ long scanSnapshotId)
+ throws Exception {
+ SortedGlobalIndexWriter writer =
+ new SortedGlobalIndexWriter(table, indexType).withIndexField(indexFieldName);
+ DataField indexField = table.rowType().getField(indexFieldName);
+ RowType readRowType =
+ SpecialFields.rowTypeWithRowId(table.rowType())
+ .project(Arrays.asList(indexFieldName, SpecialFields.ROW_ID.name()));
+ InternalRow.FieldGetter fieldGetter = InternalRow.createFieldGetter(indexField.type(), 0);
+ List<Pair<Object, Long>> rows = new ArrayList<>();
+ try (RecordReader<InternalRow> reader =
+ table.newReadBuilder()
+ .withReadType(readRowType)
+ .newRead()
+ .createReader(Collections.singletonList(dataSplit));
+ CloseableIterator<InternalRow> iterator = reader.toCloseableIterator()) {
+ while (iterator.hasNext()) {
+ InternalRow row = iterator.next();
+ Object value = fieldGetter.getFieldOrNull(row);
+ rows.add(Pair.of(InternalRowUtils.copy(value, indexField.type()), row.getLong(1)));
+ }
+ }
+
+ Comparator<Object> comparator = KeySerializer.create(indexField.type()).createComparator();
+ rows.sort(
+ (left, right) -> {
+ if (left.getKey() == null) {
+ return right.getKey() == null ? 0 : -1;
+ }
+ return right.getKey() == null
+ ? 1
+ : comparator.compare(left.getKey(), right.getKey());
+ });
+
+ List<InternalRow> sortedRows = new ArrayList<>(rows.size());
+ for (Pair<Object, Long> row : rows) {
+ sortedRows.add(GenericRow.of(row.getKey(), row.getValue()));
+ }
+ Range rowRange = calcRowRange(dataSplit);
+ return writer.buildForSinglePartition(
+ rowRange, dataSplit.partition(), sortedRows.iterator(), scanSnapshotId);
+ }
+}
diff --git a/paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexWriterTest.java b/paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexWriterTest.java
new file mode 100644
index 0000000000..e449d63211
--- /dev/null
+++ b/paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexWriterTest.java
@@ -0,0 +1,128 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.globalindex.sorted;
+
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter;
+import org.apache.paimon.globalindex.ResultEntry;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.table.TableTestBase;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.utils.Range;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.Closeable;
+import java.util.ArrayDeque;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Queue;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/** Tests for {@link SortedGlobalIndexWriter}. */
+public class SortedGlobalIndexWriterTest extends TableTestBase {
+
+ @Test
+ public void testSingleColumnWriterRotationPreservesResultGroups() throws Exception {
+ GlobalIndexSingleColumnWriter first = mock(GlobalIndexSingleColumnWriter.class);
+ GlobalIndexSingleColumnWriter second = mock(GlobalIndexSingleColumnWriter.class);
+ when(first.finish())
+ .thenReturn(Collections.singletonList(new ResultEntry("index-1", 2, null)));
+ when(second.finish())
+ .thenReturn(Collections.singletonList(new ResultEntry("index-2", 1, null)));
+ Queue<GlobalIndexSingleColumnWriter> writers =
+ new ArrayDeque<>(Arrays.asList(first, second));
+ SortedSingleColumnIndexWriter rotatingWriter =
+ new SortedSingleColumnIndexWriter(2, writers::remove);
+
+ rotatingWriter.write(10, 0);
+ rotatingWriter.write(20, 1);
+ rotatingWriter.write(30, 2);
+ List<List<ResultEntry>> results = rotatingWriter.finish();
+
+ verify(first).write(10, 0);
+ verify(first).write(20, 1);
+ verify(second).write(30, 2);
+ assertThat(results).hasSize(2);
+ assertThat(results.get(0)).extracting(ResultEntry::fileName).containsExactly("index-1");
+ assertThat(results.get(1)).extracting(ResultEntry::fileName).containsExactly("index-2");
+ }
+
+ @Test
+ public void testSingleColumnWriterClosesActiveWriter() throws Exception {
+ GlobalIndexSingleColumnWriter activeWriter =
+ mock(
+ GlobalIndexSingleColumnWriter.class,
+ org.mockito.Mockito.withSettings().extraInterfaces(Closeable.class));
+ SortedSingleColumnIndexWriter rotatingWriter =
+ new SortedSingleColumnIndexWriter(2, () -> activeWriter);
+ rotatingWriter.write(10, 0);
+
+ assertThat(rotatingWriter).isInstanceOf(AutoCloseable.class);
+ ((AutoCloseable) rotatingWriter).close();
+
+ verify((Closeable) activeWriter).close();
+ }
+
+ @Test
+ public void testBuildForSinglePartitionClosesWriterAfterFailure() throws Exception {
+ createTableDefault();
+ GlobalIndexSingleColumnWriter activeWriter =
+ mock(
+ GlobalIndexSingleColumnWriter.class,
+ org.mockito.Mockito.withSettings().extraInterfaces(Closeable.class));
+ org.mockito.Mockito.doThrow(new RuntimeException("write failed"))
+ .when(activeWriter)
+ .write(10, 0);
+ SortedGlobalIndexWriter writer =
+ new SortedGlobalIndexWriter(getTableDefault(), "btree") {
+ @Override
+ public GlobalIndexSingleColumnWriter createWriter() {
+ return activeWriter;
+ }
+ };
+ writer.withIndexField("f0");
+
+ assertThatThrownBy(
+ () ->
+ writer.buildForSinglePartition(
+ new Range(0, 0),
+ null,
+ Collections.<InternalRow>singletonList(
+ GenericRow.of(10, 0L))
+ .iterator(),
+ 1L))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessage("write failed");
+
+ verify((Closeable) activeWriter).close();
+ }
+
+ @Override
+ public Schema schemaDefault() {
+ return Schema.newBuilder().column("f0", DataTypes.INT()).build();
+ }
+}
diff --git a/paimon-core/src/test/java/org/apache/paimon/index/DataEvolutionIndexSourceMetaTest.java b/paimon-core/src/test/java/org/apache/paimon/index/DataEvolutionIndexSourceMetaTest.java
new file mode 100644
index 0000000000..c281ebdd2f
--- /dev/null
+++ b/paimon-core/src/test/java/org/apache/paimon/index/DataEvolutionIndexSourceMetaTest.java
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.index;
+
+import org.apache.paimon.io.DataOutputSerializer;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link DataEvolutionIndexSourceMeta}. */
+class DataEvolutionIndexSourceMetaTest {
+
+ @Test
+ void testRoundTripAndDetection() {
+ DataEvolutionIndexSourceMeta sourceMeta = new DataEvolutionIndexSourceMeta(42L);
+
+ byte[] serialized = sourceMeta.serialize();
+
+ assertThat(DataEvolutionIndexSourceMeta.isDataEvolutionMeta(serialized)).isTrue();
+ assertThat(DataEvolutionIndexSourceMeta.deserialize(serialized).scanSnapshotId())
+ .isEqualTo(42L);
+ assertThat(DataEvolutionIndexSourceMeta.isDataEvolutionMeta(null)).isFalse();
+ assertThat(DataEvolutionIndexSourceMeta.isDataEvolutionMeta(new byte[] {1})).isFalse();
+ }
+
+ @Test
+ void testRejectsInvalidSnapshotId() {
+ assertThatThrownBy(() -> new DataEvolutionIndexSourceMeta(0L))
+ .hasMessageContaining("snapshot id must be positive");
+ }
+
+ @Test
+ void testRejectsWrongMagicAndVersion() throws Exception {
+ DataOutputSerializer wrongMagic = new DataOutputSerializer(16);
+ wrongMagic.writeInt(1);
+ wrongMagic.writeInt(1);
+ wrongMagic.writeLong(1L);
+ assertThatThrownBy(
+ () ->
+ DataEvolutionIndexSourceMeta.deserialize(
+ wrongMagic.getCopyOfBuffer()))
+ .hasMessageContaining("Not data-evolution index source metadata");
+
+ byte[] wrongVersion = new DataEvolutionIndexSourceMeta(1L).serialize();
+ wrongVersion[7] = 2;
+ assertThatThrownBy(() -> DataEvolutionIndexSourceMeta.deserialize(wrongVersion))
+ .hasMessageContaining("Unsupported data-evolution index source version");
+ }
+
+ @Test
+ void testRejectsTruncatedAndTrailingBytes() {
+ byte[] serialized = new DataEvolutionIndexSourceMeta(1L).serialize();
+
+ assertThatThrownBy(
+ () ->
+ DataEvolutionIndexSourceMeta.deserialize(
+ Arrays.copyOf(serialized, serialized.length - 1)))
+ .hasMessageContaining("Failed to deserialize data-evolution index source metadata");
+
+ byte[] trailing = Arrays.copyOf(serialized, serialized.length + 1);
+ assertThatThrownBy(() -> DataEvolutionIndexSourceMeta.deserialize(trailing))
+ .hasMessageContaining("Unexpected trailing bytes");
+ }
+
+ @Test
+ void testReadsFromIndexFile() {
+ DataEvolutionIndexSourceMeta sourceMeta = new DataEvolutionIndexSourceMeta(9L);
+ IndexFileMeta indexFile =
+ new IndexFileMeta(
+ "lumina",
+ "index-1",
+ 1L,
+ 10L,
+ new GlobalIndexMeta(0, 9, 1, null, null, sourceMeta.serialize()),
+ null);
+
+ assertThat(DataEvolutionIndexSourceMeta.fromIndexFile(indexFile).scanSnapshotId())
+ .isEqualTo(9L);
+ }
+}
diff --git a/paimon-core/src/test/java/org/apache/paimon/manifest/IndexManifestFileHandlerTest.java b/paimon-core/src/test/java/org/apache/paimon/manifest/IndexManifestFileHandlerTest.java
index f706972b6d..3c0002c2ad 100644
--- a/paimon-core/src/test/java/org/apache/paimon/manifest/IndexManifestFileHandlerTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/manifest/IndexManifestFileHandlerTest.java
@@ -24,6 +24,7 @@ import org.apache.paimon.data.InternalRow;
import org.apache.paimon.format.FileFormat;
import org.apache.paimon.format.FormatReaderFactory;
import org.apache.paimon.fs.Path;
+import org.apache.paimon.index.DataEvolutionIndexSourceMeta;
import org.apache.paimon.index.GlobalIndexMeta;
import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.table.BucketMode;
@@ -212,6 +213,49 @@ public class IndexManifestFileHandlerTest {
assertThat(entries).containsExactly(added);
}
+ @Test
+ public void testMissingGlobalIndexDeleteRejected() throws Exception {
+ TestAppendFileStore fileStore =
+ TestAppendFileStore.createAppendStore(tempDir, new HashMap<>());
+ IndexManifestFile indexManifestFile = createIndexManifestFile(fileStore);
+
+ IndexManifestEntry previous = globalIndexEntry("prev-index", 0, 99, 1);
+ String manifest =
+ indexManifestFile.writeIndexFiles(
+ null, Arrays.asList(previous), BucketMode.BUCKET_UNAWARE);
+ IndexManifestEntry missing = globalIndexEntry("missing-index", 100, 199, 1);
+
+ assertThatThrownBy(
+ () ->
+ indexManifestFile.writeIndexFiles(
+ manifest,
+ Arrays.asList(missing.toDeleteEntry()),
+ BucketMode.BUCKET_UNAWARE))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining(
+ "Trying to delete global index file missing-index which does not exist.");
+ }
+
+ @Test
+ public void testDataEvolutionSourceMetaDoesNotDisableRangeValidation() throws Exception {
+ TestAppendFileStore fileStore =
+ TestAppendFileStore.createAppendStore(tempDir, new HashMap<>());
+ IndexManifestFile indexManifestFile = createIndexManifestFile(fileStore);
+ IndexManifestFileHandler handler =
+ new IndexManifestFileHandler(indexManifestFile, BucketMode.BUCKET_UNAWARE);
+
+ IndexManifestEntry previous = dataEvolutionIndexEntry("old-index", 0, 99, 1, 1);
+ String manifest = handler.write(null, Arrays.asList(previous));
+ IndexManifestEntry replacement = dataEvolutionIndexEntry("new-index", 0, 99, 1, 2);
+
+ assertThatThrownBy(() -> handler.write(manifest, Arrays.asList(replacement)))
+ .hasMessageContaining("overlapping row range");
+
+ String replaced =
+ handler.write(manifest, Arrays.asList(previous.toDeleteEntry(), replacement));
+ assertThat(indexManifestFile.read(replaced)).containsExactly(replacement);
+ }
+
@Test
public void testGlobalIndexOverlappingRangeAllowedForDifferentFieldId() throws Exception {
TestAppendFileStore fileStore =
@@ -332,4 +376,29 @@ public class IndexManifestFileHandlerTest {
new GlobalIndexMeta(0, 1, 1, null, null, new byte[] {1}),
null));
}
+
+ private IndexManifestEntry dataEvolutionIndexEntry(
+ String fileName,
+ long rowRangeStart,
+ long rowRangeEnd,
+ int indexFieldId,
+ long scanSnapshotId) {
+ return new IndexManifestEntry(
+ FileKind.ADD,
+ BinaryRow.EMPTY_ROW,
+ 0,
+ new IndexFileMeta(
+ "lumina",
+ fileName,
+ 1L,
+ rowRangeEnd - rowRangeStart + 1,
+ new GlobalIndexMeta(
+ rowRangeStart,
+ rowRangeEnd,
+ indexFieldId,
+ null,
+ null,
+ new DataEvolutionIndexSourceMeta(scanSnapshotId).serialize()),
+ null));
+ }
}
diff --git a/paimon-core/src/test/java/org/apache/paimon/metastore/VisibilityWaitCallbackTest.java b/paimon-core/src/test/java/org/apache/paimon/metastore/VisibilityWaitCallbackTest.java
index 9440988e88..3b9581f9c9 100644
--- a/paimon-core/src/test/java/org/apache/paimon/metastore/VisibilityWaitCallbackTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/metastore/VisibilityWaitCallbackTest.java
@@ -22,7 +22,9 @@ import org.apache.paimon.CoreOptions;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.GenericRow;
-import org.apache.paimon.globalindex.sorted.SortedGlobalIndexBuilder;
+import org.apache.paimon.globalindex.ScanResult;
+import org.apache.paimon.globalindex.sorted.SortedGlobalIndexScanner;
+import org.apache.paimon.globalindex.sorted.SortedGlobalIndexTestUtils;
import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.schema.Schema;
@@ -35,8 +37,6 @@ import org.apache.paimon.table.sink.CommitMessage;
import org.apache.paimon.table.source.DataSplit;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
-import org.apache.paimon.utils.Pair;
-import org.apache.paimon.utils.RowRangeIndex;
import org.junit.jupiter.api.Test;
@@ -97,8 +97,8 @@ public class VisibilityWaitCallbackTest extends TableTestBase {
buildIndex(getTableDefault(), true);
writeFuture.get(10, TimeUnit.SECONDS);
- SortedGlobalIndexBuilder builder =
- new SortedGlobalIndexBuilder(getTableDefault(), "btree").withIndexField("f1");
+ SortedGlobalIndexScanner builder =
+ new SortedGlobalIndexScanner(getTableDefault(), "btree").withIndexField("f1");
assertThat(builder.incrementalScan()).isNotPresent();
} finally {
executor.shutdownNow();
@@ -159,15 +159,17 @@ public class VisibilityWaitCallbackTest extends TableTestBase {
}
private void buildIndex(FileStoreTable table, boolean incremental) throws Exception {
- SortedGlobalIndexBuilder builder =
- new SortedGlobalIndexBuilder(table, "btree").withIndexField("f1");
- Optional<Pair<RowRangeIndex, List<DataSplit>>> scan =
+ SortedGlobalIndexScanner builder =
+ new SortedGlobalIndexScanner(table, "btree").withIndexField("f1");
+ Optional<ScanResult<DataSplit>> scan =
incremental ? builder.incrementalScan() : builder.scan();
assertThat(scan).isPresent();
List<CommitMessage> commitMessages = new ArrayList<>();
- for (DataSplit dataSplit : scan.get().getRight()) {
- commitMessages.addAll(builder.build(dataSplit, ioManager));
+ for (DataSplit dataSplit : scan.get().entries()) {
+ commitMessages.addAll(
+ SortedGlobalIndexTestUtils.buildIndex(
+ table, "btree", "f1", dataSplit, scan.get().scanSnapshotId()));
}
try (BatchTableCommit commit = table.newBatchWriteBuilder().newCommit()) {
@@ -176,16 +178,18 @@ public class VisibilityWaitCallbackTest extends TableTestBase {
}
private void buildPartitionIndex(FileStoreTable table, String partition) throws Exception {
- SortedGlobalIndexBuilder builder =
- new SortedGlobalIndexBuilder(table, "btree")
+ SortedGlobalIndexScanner builder =
+ new SortedGlobalIndexScanner(table, "btree")
.withIndexField("f1")
.withPartitionPredicate(partitionPredicate(table, partition));
- Optional<Pair<RowRangeIndex, List<DataSplit>>> scan = builder.scan();
+ Optional<ScanResult<DataSplit>> scan = builder.scan();
assertThat(scan).isPresent();
List<CommitMessage> commitMessages = new ArrayList<>();
- for (DataSplit dataSplit : scan.get().getRight()) {
- commitMessages.addAll(builder.build(dataSplit, ioManager));
+ for (DataSplit dataSplit : scan.get().entries()) {
+ commitMessages.addAll(
+ SortedGlobalIndexTestUtils.buildIndex(
+ table, "btree", "f1", dataSplit, scan.get().scanSnapshotId()));
}
try (BatchTableCommit commit = table.newBatchWriteBuilder().newCommit()) {
diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
index 6458f82fab..f326be18dc 100644
--- a/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
@@ -1457,6 +1457,27 @@ public class FileStoreCommitTest {
}
}
+ @Test
+ public void testMissingGlobalIndexDeleteRejected() throws Exception {
+ TestFileStore store = createStore(false);
+ KeyValue record = gen.next();
+ BinaryRow partition = gen.getPartition(record);
+ store.commitData(Collections.singletonList(record), s -> partition, kv -> 0);
+
+ assertThatThrownBy(
+ () -> {
+ try (FileStoreCommitImpl commit = store.newCommit()) {
+ commit.commit(
+ deleteIndexCommittable(partition, "missing-index", 0, 0),
+ false);
+ }
+ })
+ .satisfies(
+ anyCauseMatches(
+ IllegalStateException.class,
+ "Trying to delete global index file missing-index which does not exist."));
+ }
+
@Test
public void testCommitTwiceWithDifferentKind() throws Exception {
TestFileStore store = createStore(false);
@@ -2015,6 +2036,28 @@ public class FileStoreCommitTest {
return committable;
}
+ private ManifestCommittable deleteIndexCommittable(
+ BinaryRow partition, String fileName, long rowRangeStart, long rowRangeEnd) {
+ ManifestCommittable committable = new ManifestCommittable(0);
+ committable.addFileCommittable(
+ new CommitMessageImpl(
+ partition,
+ 0,
+ null,
+ DataIncrement.deleteIndexIncrement(
+ Collections.singletonList(
+ new IndexFileMeta(
+ "btree",
+ fileName,
+ 1,
+ 1,
+ new GlobalIndexMeta(
+ rowRangeStart, rowRangeEnd, 0, null, null),
+ null))),
+ CompactIncrement.emptyIncrement()));
+ return committable;
+ }
+
private static List<ManifestEntry> tableFilesFrom(
ManifestCommittable committable, CoreOptions options) {
ManifestEntryChanges changes = new ManifestEntryChanges(options.bucket());
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/BtreeGlobalIndexTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/BtreeGlobalIndexTableTest.java
index 9d85570130..0349b3290a 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/BtreeGlobalIndexTableTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/BtreeGlobalIndexTableTest.java
@@ -27,8 +27,10 @@ import org.apache.paimon.globalindex.DataEvolutionGlobalIndexCoverage;
import org.apache.paimon.globalindex.DataEvolutionGlobalIndexScanner;
import org.apache.paimon.globalindex.GlobalIndexResult;
import org.apache.paimon.globalindex.IndexedSplit;
+import org.apache.paimon.globalindex.ScanResult;
import org.apache.paimon.globalindex.btree.BTreeIndexOptions;
-import org.apache.paimon.globalindex.sorted.SortedGlobalIndexBuilder;
+import org.apache.paimon.globalindex.sorted.SortedGlobalIndexScanner;
+import org.apache.paimon.globalindex.sorted.SortedGlobalIndexTestUtils;
import org.apache.paimon.index.GlobalIndexMeta;
import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.manifest.IndexManifestEntry;
@@ -793,18 +795,19 @@ public class BtreeGlobalIndexTableTest extends DataEvolutionTestBase {
private void createIndexIncremental(String fieldName) throws Exception {
FileStoreTable table = (FileStoreTable) catalog.getTable(identifier());
- SortedGlobalIndexBuilder builder =
- new SortedGlobalIndexBuilder(table, "btree").withIndexField(fieldName);
- List<DataSplit> dataSplits =
+ SortedGlobalIndexScanner builder =
+ new SortedGlobalIndexScanner(table, "btree").withIndexField(fieldName);
+ ScanResult<DataSplit> scanResult =
builder.incrementalScan()
- .map(org.apache.paimon.utils.Pair::getRight)
.orElseThrow(
() ->
new IllegalStateException(
"Expected incremental scan result when building index."));
List<CommitMessage> commitMessages = new ArrayList<>();
- for (DataSplit dataSplit : dataSplits) {
- commitMessages.addAll(builder.build(dataSplit, ioManager));
+ for (DataSplit dataSplit : scanResult.entries()) {
+ commitMessages.addAll(
+ SortedGlobalIndexTestUtils.buildIndex(
+ table, "btree", fieldName, dataSplit, scanResult.scanSnapshotId()));
}
try (BatchTableCommit commit = table.newBatchWriteBuilder().newCommit()) {
commit.commit(commitMessages);
@@ -817,18 +820,19 @@ public class BtreeGlobalIndexTableTest extends DataEvolutionTestBase {
private void createIndex(String fieldName, List<Range> rowRanges) throws Exception {
FileStoreTable table = (FileStoreTable) catalog.getTable(identifier());
- SortedGlobalIndexBuilder builder =
- new SortedGlobalIndexBuilder(table, "btree").withIndexField(fieldName);
- List<DataSplit> dataSplits =
+ SortedGlobalIndexScanner builder =
+ new SortedGlobalIndexScanner(table, "btree").withIndexField(fieldName);
+ ScanResult<DataSplit> scanResult =
builder.scan()
- .map(org.apache.paimon.utils.Pair::getRight)
.orElseThrow(
() ->
new IllegalStateException(
"Expected scan result when building index."));
List<CommitMessage> commitMessages = new ArrayList<>();
- for (DataSplit dataSplit : indexSplits(table, rowRanges, dataSplits)) {
- commitMessages.addAll(builder.build(dataSplit, ioManager));
+ for (DataSplit dataSplit : indexSplits(table, rowRanges, scanResult.entries())) {
+ commitMessages.addAll(
+ SortedGlobalIndexTestUtils.buildIndex(
+ table, "btree", fieldName, dataSplit, scanResult.scanSnapshotId()));
}
try (BatchTableCommit commit = table.newBatchWriteBuilder().newCommit()) {
commit.commit(commitMessages);
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/source/FullTextSearchBuilderTest.java b/paimon-core/src/test/java/org/apache/paimon/table/source/FullTextSearchBuilderTest.java
index 0cc025e040..04c1fd6848 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/source/FullTextSearchBuilderTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/source/FullTextSearchBuilderTest.java
@@ -1013,7 +1013,8 @@ public class FullTextSearchBuilderTest extends TableTestBase {
rowRange,
indexFields,
TestFullTextGlobalIndexerFactory.IDENTIFIER,
- entries);
+ entries,
+ null);
DataIncrement dataIncrement = DataIncrement.indexIncrement(indexFiles);
CommitMessage message =
@@ -1051,7 +1052,8 @@ public class FullTextSearchBuilderTest extends TableTestBase {
new Range(0, documents.length - 1),
Collections.singletonList(textField),
TestFullTextGlobalIndexerFactory.IDENTIFIER,
- writer.finish());
+ writer.finish(),
+ null);
byte[] sourceMeta =
new PrimaryKeyIndexSourceMeta(
1, new PrimaryKeyIndexSourceFile("data-file", documents.length))
@@ -1122,7 +1124,8 @@ public class FullTextSearchBuilderTest extends TableTestBase {
rowRange,
indexFields,
TestFullTextGlobalIndexerFactory.IDENTIFIER,
- entries);
+ entries,
+ null);
DataIncrement dataIncrement = DataIncrement.indexIncrement(indexFiles);
CommitMessage message =
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/source/VectorSearchBuilderTest.java b/paimon-core/src/test/java/org/apache/paimon/table/source/VectorSearchBuilderTest.java
index f287f1875e..ec116a87a7 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/source/VectorSearchBuilderTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/source/VectorSearchBuilderTest.java
@@ -2071,7 +2071,8 @@ public class VectorSearchBuilderTest extends TableTestBase {
rowRange,
indexFields,
TestVectorGlobalIndexerFactory.IDENTIFIER,
- entries);
+ entries,
+ null);
DataIncrement dataIncrement = DataIncrement.indexIncrement(indexFiles);
CommitMessage message =
diff --git a/paimon-core/src/test/java/org/apache/paimon/utils/DataEvolutionUtilsTest.java b/paimon-core/src/test/java/org/apache/paimon/utils/DataEvolutionUtilsTest.java
index 6f51235362..33feb9d850 100644
--- a/paimon-core/src/test/java/org/apache/paimon/utils/DataEvolutionUtilsTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/utils/DataEvolutionUtilsTest.java
@@ -19,12 +19,18 @@
package org.apache.paimon.utils;
import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.stats.SimpleStats;
+import org.apache.paimon.table.SpecialFields;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.IntType;
import org.junit.jupiter.api.Test;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
import java.util.function.Function;
import static org.assertj.core.api.Assertions.assertThat;
@@ -33,6 +39,76 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Test for {@link DataEvolutionUtils}. */
public class DataEvolutionUtilsTest {
+ @Test
+ public void testFileFieldIdsIgnoresSystemFields() {
+ TableSchema schema =
+ new TableSchema(
+ 1L,
+ Arrays.asList(
+ new DataField(1, "indexed", new IntType()),
+ new DataField(2, "other", new IntType())),
+ 2,
+ Collections.emptyList(),
+ Collections.emptyList(),
+ new HashMap<>(),
+ "");
+
+ assertThat(
+ DataEvolutionUtils.fileFieldIds(
+ ignored -> schema,
+ dataFile(
+ "mixed.parquet",
+ 1,
+ Arrays.asList(
+ SpecialFields.ROW_ID.name(),
+ "indexed",
+ SpecialFields.SEQUENCE_NUMBER.name()))))
+ .containsExactly(1);
+ assertThat(
+ DataEvolutionUtils.fileFieldIds(
+ ignored -> schema,
+ dataFile(
+ "system-only.parquet",
+ 1,
+ Arrays.asList(
+ SpecialFields.ROW_ID.name(),
+ SpecialFields.SEQUENCE_NUMBER.name()))))
+ .isEmpty();
+ }
+
+ @Test
+ public void testFileFieldIdsHandlesFullEmptyAndUnrelatedWrites() {
+ TableSchema schema =
+ new TableSchema(
+ 1L,
+ Arrays.asList(
+ new DataField(1, "indexed", new IntType()),
+ new DataField(2, "other", new IntType())),
+ 2,
+ Collections.emptyList(),
+ Collections.emptyList(),
+ new HashMap<>(),
+ "");
+
+ assertThat(
+ DataEvolutionUtils.fileFieldIds(
+ ignored -> schema, dataFile("full.parquet", 1, null)))
+ .containsExactlyInAnyOrder(1, 2);
+ assertThat(
+ DataEvolutionUtils.fileFieldIds(
+ ignored -> schema,
+ dataFile("empty.parquet", 1, Collections.emptyList())))
+ .isEmpty();
+ assertThat(
+ DataEvolutionUtils.fileFieldIds(
+ ignored -> schema,
+ dataFile(
+ "unrelated.parquet",
+ 1,
+ Collections.singletonList("other"))))
+ .containsExactly(2);
+ }
+
@Test
public void testRetrieveAnchorFileSkipsSpecialFiles() {
DataFileMeta blobFile = dataFile("blob-file.blob", 1);
@@ -74,6 +150,11 @@ public class DataEvolutionUtilsTest {
}
private static DataFileMeta dataFile(String fileName, long maxSequenceNumber) {
+ return dataFile(fileName, maxSequenceNumber, Collections.emptyList());
+ }
+
+ private static DataFileMeta dataFile(
+ String fileName, long maxSequenceNumber, List<String> writeCols) {
return DataFileMeta.forAppend(
fileName,
1L,
@@ -88,6 +169,6 @@ public class DataEvolutionUtilsTest {
null,
null,
0L,
- Collections.emptyList());
+ writeCols);
}
}
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
deleted file mode 100644
index 2872ac591d..0000000000
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/globalindex/GenericGlobalIndexBuilder.java
+++ /dev/null
@@ -1,92 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.paimon.flink.globalindex;
-
-import org.apache.paimon.Snapshot;
-import org.apache.paimon.manifest.IndexManifestEntry;
-import org.apache.paimon.manifest.ManifestEntry;
-import org.apache.paimon.partition.PartitionPredicate;
-import org.apache.paimon.table.FileStoreTable;
-
-import javax.annotation.Nullable;
-
-import java.io.Serializable;
-import java.util.Collections;
-import java.util.List;
-
-import static org.apache.paimon.utils.Preconditions.checkArgument;
-
-/** Builder for generic (non-btree) global index. */
-public class GenericGlobalIndexBuilder implements Serializable {
-
- private static final long serialVersionUID = 1L;
-
- protected final FileStoreTable table;
-
- @Nullable protected PartitionPredicate partitionPredicate;
- @Nullable private Snapshot scanSnapshot;
-
- public GenericGlobalIndexBuilder(FileStoreTable table) {
- this.table = table;
- }
-
- public GenericGlobalIndexBuilder withPartitionPredicate(PartitionPredicate partitionPredicate) {
- this.partitionPredicate = partitionPredicate;
- return this;
- }
-
- public FileStoreTable table() {
- return table;
- }
-
- /**
- * Scans manifest entries to determine which files need to be indexed.
- *
- * @return manifest entries to build index from
- */
- public List<ManifestEntry> scan() {
- checkArgument(
- table.coreOptions().bucket() == -1,
- "Generic global index only supports unaware-bucket tables (bucket = -1), "
- + "but table '%s' has bucket = %d.",
- table.name(),
- table.coreOptions().bucket());
- scanSnapshot = table.snapshotManager().latestSnapshot();
- if (scanSnapshot == null) {
- return Collections.emptyList();
- }
-
- return table.store()
- .newScan()
- .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 34461d7f75..4740a3ea80 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,7 +19,6 @@
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;
@@ -35,6 +34,9 @@ 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.globalindex.ScanResult;
+import org.apache.paimon.globalindex.generic.GenericGlobalIndexScanner;
+import org.apache.paimon.index.DataEvolutionIndexSourceMeta;
import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.manifest.IndexManifestEntry;
import org.apache.paimon.manifest.ManifestEntry;
@@ -68,17 +70,14 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
+import java.util.Optional;
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;
@@ -94,48 +93,6 @@ import static org.apache.paimon.utils.Preconditions.checkArgument;
public class GenericIndexTopoBuilder {
private static final Logger LOG = LoggerFactory.getLogger(GenericIndexTopoBuilder.class);
- public static final long NO_MAX_INDEXED_ROW_ID = -1L;
-
- public static void buildIndexAndExecute(
- StreamExecutionEnvironment env,
- FileStoreTable table,
- String indexColumn,
- String indexType,
- PartitionPredicate partitionPredicate,
- Options userOptions)
- throws Exception {
- buildIndexAndExecuteInternal(
- env,
- table,
- indexColumn,
- Collections.emptyList(),
- indexType,
- partitionPredicate,
- userOptions,
- NO_MAX_INDEXED_ROW_ID,
- true);
- }
-
- public static void buildIndexAndExecute(
- StreamExecutionEnvironment env,
- FileStoreTable table,
- String indexColumn,
- String indexType,
- PartitionPredicate partitionPredicate,
- Options userOptions,
- long maxIndexedRowId)
- throws Exception {
- buildIndexAndExecuteInternal(
- env,
- table,
- indexColumn,
- Collections.emptyList(),
- indexType,
- partitionPredicate,
- userOptions,
- maxIndexedRowId,
- false);
- }
public static void buildIndexAndExecute(
StreamExecutionEnvironment env,
@@ -147,37 +104,7 @@ public class GenericIndexTopoBuilder {
Options userOptions)
throws Exception {
buildIndexAndExecuteInternal(
- env,
- table,
- indexColumn,
- extraColumns,
- indexType,
- partitionPredicate,
- userOptions,
- NO_MAX_INDEXED_ROW_ID,
- true);
- }
-
- public static void buildIndexAndExecute(
- StreamExecutionEnvironment env,
- FileStoreTable table,
- String indexColumn,
- List<String> extraColumns,
- String indexType,
- PartitionPredicate partitionPredicate,
- Options userOptions,
- long maxIndexedRowId)
- throws Exception {
- buildIndexAndExecuteInternal(
- env,
- table,
- indexColumn,
- extraColumns,
- indexType,
- partitionPredicate,
- userOptions,
- maxIndexedRowId,
- false);
+ env, table, indexColumn, extraColumns, indexType, partitionPredicate, userOptions);
}
private static void buildIndexAndExecuteInternal(
@@ -187,22 +114,18 @@ public class GenericIndexTopoBuilder {
List<String> extraColumns,
String indexType,
PartitionPredicate partitionPredicate,
- Options userOptions,
- long maxIndexedRowId,
- boolean autoIncremental)
+ Options userOptions)
throws Exception {
boolean hasIndexToBuild =
buildIndexInternal(
env,
- () -> new GenericGlobalIndexBuilder(table),
+ () -> new GenericGlobalIndexScanner(table),
table,
indexColumn,
extraColumns,
indexType,
partitionPredicate,
- userOptions,
- maxIndexedRowId,
- autoIncremental);
+ userOptions);
if (hasIndexToBuild) {
env.execute("Create " + indexType + " global index for table: " + table.name());
} else {
@@ -212,7 +135,7 @@ public class GenericIndexTopoBuilder {
public static boolean buildIndex(
StreamExecutionEnvironment env,
- Supplier<GenericGlobalIndexBuilder> indexBuilderSupplier,
+ Supplier<GenericGlobalIndexScanner> indexBuilderSupplier,
FileStoreTable table,
String indexColumn,
String indexType,
@@ -227,83 +150,33 @@ public class GenericIndexTopoBuilder {
Collections.emptyList(),
indexType,
partitionPredicate,
- userOptions,
- NO_MAX_INDEXED_ROW_ID,
- true);
- }
-
- public static boolean buildIndex(
- StreamExecutionEnvironment env,
- Supplier<GenericGlobalIndexBuilder> indexBuilderSupplier,
- FileStoreTable table,
- String indexColumn,
- String indexType,
- PartitionPredicate partitionPredicate,
- Options userOptions,
- long maxIndexedRowId)
- throws Exception {
- return buildIndexInternal(
- env,
- indexBuilderSupplier,
- table,
- indexColumn,
- Collections.emptyList(),
- indexType,
- partitionPredicate,
- userOptions,
- maxIndexedRowId,
- false);
- }
-
- /**
- * Builds a generic global index topology using a {@link GenericGlobalIndexBuilder} supplier.
- *
- * @return {@code true} if a Flink topology was built and is ready to execute, {@code false} if
- * there was nothing to index
- */
- public static boolean buildIndex(
- StreamExecutionEnvironment env,
- Supplier<GenericGlobalIndexBuilder> indexBuilderSupplier,
- FileStoreTable table,
- String indexColumn,
- List<String> extraColumns,
- String indexType,
- PartitionPredicate partitionPredicate,
- Options userOptions,
- long maxIndexedRowId)
- throws Exception {
- return buildIndexInternal(
- env,
- indexBuilderSupplier,
- table,
- indexColumn,
- extraColumns,
- indexType,
- partitionPredicate,
- userOptions,
- maxIndexedRowId,
- false);
+ userOptions);
}
private static boolean buildIndexInternal(
StreamExecutionEnvironment env,
- Supplier<GenericGlobalIndexBuilder> indexBuilderSupplier,
+ Supplier<GenericGlobalIndexScanner> indexBuilderSupplier,
FileStoreTable table,
String indexColumn,
List<String> extraColumns,
String indexType,
PartitionPredicate partitionPredicate,
- Options userOptions,
- long maxIndexedRowId,
- boolean autoIncremental)
+ Options userOptions)
throws Exception {
- GenericGlobalIndexBuilder indexBuilder = indexBuilderSupplier.get();
+ List<String> indexColumns = new ArrayList<>(1 + extraColumns.size());
+ indexColumns.add(indexColumn);
+ indexColumns.addAll(extraColumns);
+
+ GenericGlobalIndexScanner indexBuilder =
+ indexBuilderSupplier.get().withIndex(indexType, indexColumns, userOptions);
if (partitionPredicate != null) {
indexBuilder.withPartitionPredicate(partitionPredicate);
}
- List<ManifestEntry> entries = indexBuilder.scan();
- List<IndexManifestEntry> deletedIndexEntries = indexBuilder.deletedIndexEntries();
+ Optional<ScanResult<ManifestEntry>> optionalScanResult = indexBuilder.incrementalScan();
+ if (!optionalScanResult.isPresent()) {
+ return false;
+ }
return buildTopology(
env,
@@ -312,12 +185,7 @@ public class GenericIndexTopoBuilder {
extraColumns,
indexType,
userOptions,
- entries,
- deletedIndexEntries,
- partitionPredicate,
- indexBuilder.scanSnapshot(),
- maxIndexedRowId,
- autoIncremental);
+ optionalScanResult.get());
}
/**
@@ -333,13 +201,9 @@ public class GenericIndexTopoBuilder {
List<String> extraColumns,
String indexType,
Options userOptions,
- List<ManifestEntry> entries,
- List<IndexManifestEntry> deletedIndexEntries,
- PartitionPredicate partitionPredicate,
- @Nullable Snapshot scanSnapshot,
- long maxIndexedRowId,
- boolean autoIncremental)
+ ScanResult<ManifestEntry> scanResult)
throws Exception {
+ List<ManifestEntry> entries = scanResult.entries();
// The primary column followed by the extra columns, in index order.
List<String> indexColumns = new ArrayList<>(1 + extraColumns.size());
indexColumns.add(indexColumn);
@@ -354,51 +218,30 @@ public class GenericIndexTopoBuilder {
indexType,
indexColumns);
- long minNonIndexableRowId =
- findMinNonIndexableRowId(table.schemaManager(), entries, indexColumns);
- entries = filterEntriesBefore(entries, minNonIndexableRowId);
-
RowType rowType = table.rowType();
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());
RowType projectedRowType = SpecialFields.rowTypeWithRowId(rowType).project(readColumns);
Options mergedOptions = new Options(table.options(), userOptions.toMap());
+ byte[] sourceMeta =
+ new DataEvolutionIndexSourceMeta(scanResult.scanSnapshotId()).serialize();
long rowsPerShard = mergedOptions.get(CoreOptions.GLOBAL_INDEX_ROW_COUNT_PER_SHARD);
checkArgument(
rowsPerShard > 0,
"Option 'global-index.row-count-per-shard' must be greater than 0.");
- 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;
- }
+ List<IndexManifestEntry> deletedIndexEntries = scanResult.deletedIndexEntries();
+ List<Range> rowRangesToBuild = scanResult.rowRangeIndex().ranges();
+ LOG.info("Selected row ranges to build: {}.", rowRangesToBuild);
+ if (rowRangesToBuild.isEmpty()) {
+ LOG.info("No row ranges found, nothing to index.");
+ return false;
}
// Compute shard tasks at file level from the provided entries
@@ -441,7 +284,8 @@ public class GenericIndexTopoBuilder {
indexField,
extraFields,
projectedRowType,
- mergedOptions))
+ mergedOptions,
+ sourceMeta))
.setParallelism(parallelism);
if (!deletedIndexEntries.isEmpty()) {
@@ -465,14 +309,6 @@ public class GenericIndexTopoBuilder {
return createShardIndexedSplits(table, entries, rowsPerShard);
}
- static List<IndexedSplit> computeShardTasks(
- FileStoreTable table,
- List<ManifestEntry> entries,
- long rowsPerShard,
- long maxIndexedRowId) {
- return computeShardTasks(table, entries, rowsPerShard, rowRangesAfter(maxIndexedRowId));
- }
-
static List<IndexedSplit> computeShardTasks(
FileStoreTable table,
List<ManifestEntry> entries,
@@ -531,6 +367,7 @@ public class GenericIndexTopoBuilder {
private final List<DataField> extraFields;
private final RowType projectedRowType;
private final Options mergedOptions;
+ private final byte[] sourceMeta;
private transient TableRead tableRead;
private transient List<DataField> indexedFields;
@@ -546,7 +383,8 @@ public class GenericIndexTopoBuilder {
DataField indexField,
List<DataField> extraFields,
RowType projectedRowType,
- Options mergedOptions) {
+ Options mergedOptions,
+ byte[] sourceMeta) {
this.readBuilder = readBuilder;
this.table = table;
this.indexType = indexType;
@@ -554,6 +392,7 @@ public class GenericIndexTopoBuilder {
this.extraFields = extraFields;
this.projectedRowType = projectedRowType;
this.mergedOptions = mergedOptions;
+ this.sourceMeta = sourceMeta;
}
@Override
@@ -678,7 +517,8 @@ public class GenericIndexTopoBuilder {
shardRange,
indexedFields,
indexType,
- resultEntries);
+ resultEntries,
+ sourceMeta);
output.collect(
new StreamRecord<>(
new Committable(
@@ -702,7 +542,8 @@ public class GenericIndexTopoBuilder {
Range rowRange,
List<DataField> indexFields,
String indexType,
- List<ResultEntry> resultEntries)
+ List<ResultEntry> resultEntries,
+ byte[] sourceMeta)
throws IOException {
List<IndexFileMeta> indexFileMetas =
toIndexFileMetas(
@@ -712,7 +553,8 @@ public class GenericIndexTopoBuilder {
rowRange,
indexFields,
indexType,
- resultEntries);
+ resultEntries,
+ sourceMeta);
return new CommitMessageImpl(
partition, 0, null, indexIncrement(indexFileMetas), emptyIncrement());
}
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 a361687f38..775ad0722e 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
@@ -38,8 +38,11 @@ import org.apache.paimon.flink.utils.BoundedOneInputOperator;
import org.apache.paimon.flink.utils.JavaTypeInfo;
import org.apache.paimon.flink.utils.StreamExecutionEnvironmentUtils;
import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter;
-import org.apache.paimon.globalindex.sorted.SortedGlobalIndexBuilder;
+import org.apache.paimon.globalindex.ScanResult;
+import org.apache.paimon.globalindex.sorted.SortedGlobalIndexScanner;
+import org.apache.paimon.globalindex.sorted.SortedGlobalIndexWriter;
import org.apache.paimon.globalindex.sorted.SortedIndexOptions;
+import org.apache.paimon.manifest.IndexManifestEntry;
import org.apache.paimon.options.Options;
import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.reader.RecordReader;
@@ -47,6 +50,7 @@ import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.SpecialFields;
import org.apache.paimon.table.sink.BatchWriteBuilder;
import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.CommitMessageImpl;
import org.apache.paimon.table.source.DataSplit;
import org.apache.paimon.table.source.ReadBuilder;
import org.apache.paimon.table.source.Split;
@@ -55,9 +59,7 @@ import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataType;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
-import org.apache.paimon.utils.Pair;
import org.apache.paimon.utils.Range;
-import org.apache.paimon.utils.RowRangeIndex;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
@@ -78,8 +80,10 @@ import java.util.Map;
import java.util.Optional;
import java.util.function.Supplier;
-import static org.apache.paimon.globalindex.sorted.SortedGlobalIndexBuilder.groupSplitsByRange;
-import static org.apache.paimon.globalindex.sorted.SortedGlobalIndexBuilder.splitByContiguousRowRange;
+import static org.apache.paimon.globalindex.GlobalIndexBuilderUtils.groupSplitsByRange;
+import static org.apache.paimon.globalindex.GlobalIndexBuilderUtils.splitByContiguousRowRange;
+import static org.apache.paimon.io.CompactIncrement.emptyIncrement;
+import static org.apache.paimon.io.DataIncrement.deleteIndexIncrement;
/** The topology builder for sorted indexes in Flink. */
public class SortedIndexTopoBuilder {
@@ -95,18 +99,20 @@ public class SortedIndexTopoBuilder {
public static boolean buildIndex(
StreamExecutionEnvironment env,
- Supplier<SortedGlobalIndexBuilder> indexBuilderSupplier,
+ Supplier<SortedGlobalIndexScanner> indexScannerSupplier,
FileStoreTable table,
List<String> indexColumns,
+ String indexType,
PartitionPredicate partitionPredicate,
Options userOptions)
throws Exception {
Optional<DataStream<Committable>> written =
buildIndexStream(
env,
- indexBuilderSupplier,
+ indexScannerSupplier,
table,
indexColumns,
+ indexType,
partitionPredicate,
userOptions);
if (!written.isPresent()) {
@@ -125,33 +131,33 @@ public class SortedIndexTopoBuilder {
*/
public static Optional<DataStream<Committable>> buildIndexStream(
StreamExecutionEnvironment env,
- Supplier<SortedGlobalIndexBuilder> indexBuilderSupplier,
+ Supplier<SortedGlobalIndexScanner> indexScannerSupplier,
FileStoreTable table,
List<String> indexColumns,
+ String indexType,
PartitionPredicate partitionPredicate,
Options userOptions)
throws Exception {
List<DataStream<Committable>> allStreams = new ArrayList<>();
for (String indexColumn : indexColumns) {
- SortedGlobalIndexBuilder indexBuilder =
- indexBuilderSupplier.get().withIndexField(indexColumn);
+ SortedGlobalIndexScanner indexScanner =
+ indexScannerSupplier.get().withIndexField(indexColumn);
if (partitionPredicate != null) {
- indexBuilder = indexBuilder.withPartitionPredicate(partitionPredicate);
+ indexScanner = indexScanner.withPartitionPredicate(partitionPredicate);
}
- Optional<Pair<RowRangeIndex, List<DataSplit>>> indexRangeAndSplits =
- indexBuilder.incrementalScan();
- if (!indexRangeAndSplits.isPresent()) {
+ Optional<ScanResult<DataSplit>> optionalScanResult = indexScanner.incrementalScan();
+ if (!optionalScanResult.isPresent()) {
continue;
}
- Pair<RowRangeIndex, List<DataSplit>> scanResult = indexRangeAndSplits.get();
- List<DataSplit> splits = splitByContiguousRowRange(scanResult.getRight());
+ ScanResult<DataSplit> scanResult = optionalScanResult.get();
+ List<DataSplit> splits = splitByContiguousRowRange(scanResult.entries());
if (splits.isEmpty()) {
continue;
}
Map<BinaryRow, Map<Range, List<Split>>> partitionRangeSplits =
- groupSplitsByRange(scanResult.getLeft(), splits);
+ groupSplitsByRange(scanResult.rowRangeIndex(), splits);
if (partitionRangeSplits.isEmpty()) {
continue;
}
@@ -214,7 +220,9 @@ public class SortedIndexTopoBuilder {
buildTasks,
splitTasks,
readBuilder,
- indexBuilder,
+ new SortedGlobalIndexWriter(table, indexType, userOptions)
+ .withIndexField(indexColumn),
+ scanResult.scanSnapshotId(),
partitionFieldSize,
taskIdPos,
indexFieldPos,
@@ -226,6 +234,19 @@ public class SortedIndexTopoBuilder {
recordsPerRange,
maxParallelism);
+ List<IndexManifestEntry> deletedIndexEntries = scanResult.deletedIndexEntries();
+ if (!deletedIndexEntries.isEmpty()) {
+ DataStream<Committable> deletes =
+ StreamExecutionEnvironmentUtils.fromData(
+ env,
+ new CommittableTypeInfo(),
+ createDeleteCommittables(deletedIndexEntries)
+ .toArray(new Committable[0]))
+ .name("Index Delete Source")
+ .setParallelism(1);
+ commitMessages = commitMessages.union(deletes);
+ }
+
allStreams.add(commitMessages);
}
if (allStreams.isEmpty()) {
@@ -237,6 +258,22 @@ public class SortedIndexTopoBuilder {
return Optional.of(allStreams.get(0).union(rest));
}
+ private static List<Committable> createDeleteCommittables(
+ List<IndexManifestEntry> deletedEntries) {
+ List<Committable> committables = new ArrayList<>();
+ for (IndexManifestEntry entry : deletedEntries) {
+ CommitMessage message =
+ new CommitMessageImpl(
+ entry.partition(),
+ entry.bucket(),
+ null,
+ deleteIndexIncrement(Collections.singletonList(entry.indexFile())),
+ emptyIncrement());
+ committables.add(new Committable(BatchWriteBuilder.COMMIT_IDENTIFIER, message));
+ }
+ return committables;
+ }
+
public static void buildIndexAndExecute(
StreamExecutionEnvironment env,
FileStoreTable table,
@@ -247,9 +284,10 @@ public class SortedIndexTopoBuilder {
throws Exception {
if (buildIndex(
env,
- () -> new SortedGlobalIndexBuilder(table, indexType, userOptions),
+ () -> new SortedGlobalIndexScanner(table, indexType, userOptions),
table,
Collections.singletonList(indexColumn),
+ indexType,
partitionPredicate,
userOptions)) {
env.execute("Create " + indexType + " global index for table: " + table.name());
@@ -261,7 +299,8 @@ public class SortedIndexTopoBuilder {
List<SortedBuildTask> buildTasks,
List<SortedSplitTask> splitTasks,
ReadBuilder readBuilder,
- SortedGlobalIndexBuilder indexBuilder,
+ SortedGlobalIndexWriter indexWriter,
+ long scanSnapshotId,
int partitionFieldSize,
int taskIdPos,
int indexFieldPos,
@@ -309,7 +348,8 @@ public class SortedIndexTopoBuilder {
new WriteIndexOperator(
buildTasks,
partitionFieldSize,
- indexBuilder,
+ indexWriter,
+ scanSnapshotId,
taskIdPos,
indexFieldPos,
rowIdPos,
@@ -406,7 +446,8 @@ public class SortedIndexTopoBuilder {
private final List<SortedBuildTask> buildTasks;
private final int partitionFieldSize;
- private final SortedGlobalIndexBuilder builder;
+ private final SortedGlobalIndexWriter writer;
+ private final long scanSnapshotId;
private final int taskIdPos;
private final int indexFieldPos;
private final int rowIdPos;
@@ -424,14 +465,16 @@ public class SortedIndexTopoBuilder {
public WriteIndexOperator(
List<SortedBuildTask> buildTasks,
int partitionFieldSize,
- SortedGlobalIndexBuilder builder,
+ SortedGlobalIndexWriter writer,
+ long scanSnapshotId,
int taskIdPos,
int indexFieldPos,
int rowIdPos,
DataType indexFieldType) {
this.buildTasks = buildTasks;
this.partitionFieldSize = partitionFieldSize;
- this.builder = builder;
+ this.writer = writer;
+ this.scanSnapshotId = scanSnapshotId;
this.taskIdPos = taskIdPos;
this.indexFieldPos = indexFieldPos;
this.rowIdPos = rowIdPos;
@@ -465,14 +508,14 @@ public class SortedIndexTopoBuilder {
currentPartition = binaryRowSerializer.deserializeFromBytes(task.partition);
}
- if (currentWriter != null && counter >= builder.recordsPerRange()) {
+ if (currentWriter != null && counter >= writer.recordsPerRange()) {
flushCurrentWriter();
}
counter++;
if (currentWriter == null) {
- currentWriter = builder.createWriter();
+ currentWriter = writer.createWriter();
}
long localRowId = row.getLong(rowIdPos) - currentTask.rowRange.from;
@@ -507,8 +550,11 @@ public class SortedIndexTopoBuilder {
private void flushCurrentWriter() throws IOException {
if (counter > 0 && currentWriter != null) {
commitMessages.add(
- builder.flushIndex(
- currentTask.rowRange, currentWriter.finish(), currentPartition));
+ writer.flushIndex(
+ currentTask.rowRange,
+ currentWriter.finish(),
+ currentPartition,
+ scanSnapshotId));
}
currentWriter = null;
counter = 0;
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/LuminaVectorGlobalIndexITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/LuminaVectorGlobalIndexITCase.java
index c7e11a5cd8..4bc4056486 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/LuminaVectorGlobalIndexITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/LuminaVectorGlobalIndexITCase.java
@@ -19,8 +19,10 @@
package org.apache.paimon.flink;
import org.apache.paimon.catalog.Catalog;
-import org.apache.paimon.flink.globalindex.GenericGlobalIndexBuilder;
import org.apache.paimon.flink.globalindex.GenericIndexTopoBuilder;
+import org.apache.paimon.globalindex.ScanResult;
+import org.apache.paimon.globalindex.generic.GenericGlobalIndexScanner;
+import org.apache.paimon.index.DataEvolutionIndexSourceMeta;
import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.manifest.IndexManifestEntry;
import org.apache.paimon.manifest.ManifestEntry;
@@ -34,6 +36,7 @@ import org.junit.jupiter.api.Test;
import java.lang.reflect.Method;
import java.util.List;
+import java.util.Optional;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
@@ -46,6 +49,11 @@ public class LuminaVectorGlobalIndexITCase extends CatalogITCaseBase {
private static final String INDEX_TYPE = "lumina";
private static final String LEGACY_INDEX_TYPE = "lumina-vector-ann";
+ @Override
+ protected Boolean sqlSyncMode() {
+ return true;
+ }
+
@BeforeAll
static void checkLuminaAvailable() {
try {
@@ -430,16 +438,20 @@ public class LuminaVectorGlobalIndexITCase extends CatalogITCaseBase {
GenericIndexTopoBuilder.buildIndex(
env,
() ->
- new GenericGlobalIndexBuilder(table) {
+ new GenericGlobalIndexScanner(table) {
@Override
- public List<ManifestEntry> scan() {
- return super.scan().stream()
- .filter(
- e ->
- e.file().firstRowId() != null
- && e.file().firstRowId()
- < 50)
- .collect(Collectors.toList());
+ public Optional<ScanResult<ManifestEntry>> scan() {
+ return super.scan()
+ .map(
+ result ->
+ result.withEntries(
+ result.entries().stream()
+ .filter(
+ LuminaVectorGlobalIndexITCase
+ ::hasFirstRowIdBefore50)
+ .collect(
+ Collectors
+ .toList())));
}
},
table,
@@ -458,87 +470,89 @@ public class LuminaVectorGlobalIndexITCase extends CatalogITCaseBase {
}
@Test
- public void testDeletedIndexEntriesDefault() throws Catalog.TableNotExistException {
+ public void testDataEvolutionUpdateRefreshesIndexAtomically() throws Exception {
sql(
- "CREATE TABLE T_DEL_DEFAULT (id INT, v ARRAY<FLOAT>) WITH ("
+ "CREATE TABLE T_REFRESH (id INT, v ARRAY<FLOAT>) WITH ("
+ "'bucket' = '-1', "
+ "'row-tracking.enabled' = 'true', "
+ "'data-evolution.enabled' = 'true', "
+ + "'global-index.column-update-action' = 'IGNORE', "
+ "'lumina.index.dimension' = '3', "
+ "'lumina.distance.metric' = 'l2'"
+ ")");
+ String nullValues =
+ IntStream.range(0, 10)
+ .mapToObj(i -> "(" + i + ", CAST(NULL AS ARRAY<FLOAT>))")
+ .collect(Collectors.joining(","));
+ sql("INSERT INTO T_REFRESH VALUES " + nullValues + "," + vectorValues(10, 20, 3));
- FileStoreTable table = paimonTable("T_DEL_DEFAULT");
- GenericGlobalIndexBuilder builder = new GenericGlobalIndexBuilder(table);
- assertThat(builder.deletedIndexEntries()).isEmpty();
- }
-
- @Test
- public void testDeletedIndexEntriesAtomicCommit() throws Exception {
- sql(
- "CREATE TABLE T_DEL_ATOMIC (id INT, v ARRAY<FLOAT>) WITH ("
- + "'bucket' = '-1', "
- + "'row-tracking.enabled' = 'true', "
- + "'data-evolution.enabled' = 'true', "
- + "'lumina.index.dimension' = '3', "
- + "'lumina.distance.metric' = 'l2'"
- + ")");
-
- sql("INSERT INTO T_DEL_ATOMIC VALUES " + vectorValues(0, 100, 3));
-
- // First build: create initial index
+ FileStoreTable table = paimonTable("T_REFRESH");
+ long firstDataSnapshotId = table.snapshotManager().latestSnapshot().id();
sql(
"CALL sys.create_global_index("
- + "`table` => 'default.T_DEL_ATOMIC', "
+ + "`table` => 'default.T_REFRESH', "
+ "index_column => 'v', "
+ "index_type => '"
+ INDEX_TYPE
+ "')");
- FileStoreTable table = paimonTable("T_DEL_ATOMIC");
- List<IndexManifestEntry> oldEntries =
- table.store().newIndexFileHandler().scanEntries().stream()
- .filter(e -> INDEX_TYPE.equals(e.indexFile().indexType()))
- .collect(Collectors.toList());
+ List<IndexManifestEntry> oldEntries = table.store().newIndexFileHandler().scan(INDEX_TYPE);
assertThat(oldEntries).isNotEmpty();
-
+ assertThat(oldEntries)
+ .allSatisfy(
+ entry ->
+ assertThat(
+ DataEvolutionIndexSourceMeta.fromIndexFile(
+ entry.indexFile())
+ .scanSnapshotId())
+ .isEqualTo(firstDataSnapshotId));
List<String> oldFileNames =
- oldEntries.stream().map(e -> e.indexFile().fileName()).collect(Collectors.toList());
+ oldEntries.stream()
+ .map(entry -> entry.indexFile().fileName())
+ .collect(Collectors.toList());
- // Second build: use a custom builder that reports old entries as deletedIndexEntries
- StreamExecutionEnvironment env = streamExecutionEnvironmentBuilder().batchMode().build();
- boolean hasIndex =
- GenericIndexTopoBuilder.buildIndex(
- env,
- () ->
- new GenericGlobalIndexBuilder(table) {
- @Override
- public List<IndexManifestEntry> deletedIndexEntries() {
- return oldEntries;
- }
- },
- table,
- "v",
- INDEX_TYPE,
- null,
- new Options(table.options()));
- assertThat(hasIndex).isTrue();
- env.execute("test-rebuild-index");
+ sql("CREATE TABLE S_REFRESH (id INT, v ARRAY<FLOAT>)");
+ sql(
+ "INSERT INTO S_REFRESH VALUES "
+ + "(1, ARRAY[CAST(0.0 AS FLOAT), CAST(1.0 AS FLOAT), "
+ + "CAST(2.0 AS FLOAT)])");
+ sql(
+ "CALL sys.data_evolution_merge_into("
+ + "'default.T_REFRESH', '', '', 'S_REFRESH', "
+ + "'T_REFRESH._ROW_ID=S_REFRESH.id', 'v=S_REFRESH.v', 2)");
+ assertThat(sql("SELECT id FROM T_REFRESH WHERE id = 1 AND v IS NOT NULL")).hasSize(1);
- // After the atomic commit, old index files should be deleted and new ones created
- List<IndexManifestEntry> newEntries =
- table.store().newIndexFileHandler().scanEntries().stream()
- .filter(e -> INDEX_TYPE.equals(e.indexFile().indexType()))
- .collect(Collectors.toList());
+ long updatedSnapshotId = table.snapshotManager().latestSnapshot().id();
+ assertThat(
+ table.store().newIndexFileHandler().scan(INDEX_TYPE).stream()
+ .map(entry -> entry.indexFile().fileName()))
+ .containsExactlyInAnyOrderElementsOf(oldFileNames);
- assertThat(newEntries).isNotEmpty();
- long totalRowCount = newEntries.stream().mapToLong(e -> e.indexFile().rowCount()).sum();
- assertThat(totalRowCount).isEqualTo(100L);
+ sql(
+ "CALL sys.create_global_index("
+ + "`table` => 'default.T_REFRESH', "
+ + "index_column => 'v', "
+ + "index_type => '"
+ + INDEX_TYPE
+ + "')");
- // New files should be different from old files (old ones were deleted)
- List<String> newFileNames =
- newEntries.stream().map(e -> e.indexFile().fileName()).collect(Collectors.toList());
- assertThat(newFileNames).doesNotContainAnyElementsOf(oldFileNames);
+ assertThat(table.snapshotManager().latestSnapshot().id()).isEqualTo(updatedSnapshotId + 1);
+ List<IndexManifestEntry> refreshedEntries =
+ table.store().newIndexFileHandler().scan(INDEX_TYPE);
+ assertThat(refreshedEntries).isNotEmpty();
+ assertThat(
+ refreshedEntries.stream()
+ .map(entry -> entry.indexFile().fileName())
+ .collect(Collectors.toList()))
+ .doesNotContainAnyElementsOf(oldFileNames);
+ assertThat(refreshedEntries)
+ .allSatisfy(
+ entry ->
+ assertThat(
+ DataEvolutionIndexSourceMeta.fromIndexFile(
+ entry.indexFile())
+ .scanSnapshotId())
+ .isEqualTo(updatedSnapshotId));
}
// -- Helpers --
@@ -550,6 +564,10 @@ public class LuminaVectorGlobalIndexITCase extends CatalogITCaseBase {
.collect(Collectors.toList());
}
+ private static boolean hasFirstRowIdBefore50(ManifestEntry entry) {
+ return entry.file().firstRowId() != null && entry.file().firstRowId() < 50;
+ }
+
private static String vectorValues(int startId, int count, int dim) {
return IntStream.range(startId, startId + count)
.mapToObj(
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/SortedGlobalIndexITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/SortedGlobalIndexITCase.java
index 8b730819bb..14f768c1f7 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/SortedGlobalIndexITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/SortedGlobalIndexITCase.java
@@ -19,17 +19,20 @@
package org.apache.paimon.flink;
import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.index.DataEvolutionIndexSourceMeta;
import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.manifest.IndexManifestEntry;
import org.apache.paimon.table.FileStoreTable;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.config.TableConfigOptions;
import org.apache.flink.types.Row;
import org.junit.jupiter.api.Test;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
@@ -104,6 +107,209 @@ public class SortedGlobalIndexITCase extends CatalogITCaseBase {
.containsOnly(Row.of(100, "name_100"));
}
+ @Test
+ public void testBitmapRefreshesUpdatedDataEvolutionRange() throws Exception {
+ tEnv.getConfig().set(TableConfigOptions.TABLE_DML_SYNC, true);
+ sql(
+ "CREATE TABLE T_BITMAP_REFRESH (id INT, idx INT) WITH ("
+ + "'bucket' = '-1', "
+ + "'global-index.enabled' = 'true', "
+ + "'row-tracking.enabled' = 'true', "
+ + "'data-evolution.enabled' = 'true', "
+ + "'global-index.column-update-action' = 'IGNORE'"
+ + ")");
+ sql(
+ "INSERT INTO T_BITMAP_REFRESH VALUES "
+ + IntStream.range(0, 10)
+ .mapToObj(i -> String.format("(%d, %d)", i, i))
+ .collect(Collectors.joining(",")));
+ buildBitmapIndexForTable("T_BITMAP_REFRESH", "idx");
+
+ FileStoreTable table = paimonTable("T_BITMAP_REFRESH");
+ List<IndexManifestEntry> initial = indexEntries(table, "bitmap");
+ assertThat(initial).isNotEmpty();
+ Set<String> initialFiles = fileNames(initial);
+
+ sql("CREATE TABLE S_BITMAP_REFRESH (id INT, idx INT)");
+ sql("INSERT INTO S_BITMAP_REFRESH VALUES (1, 1001)");
+ sql(
+ "CALL sys.data_evolution_merge_into("
+ + "'default.T_BITMAP_REFRESH', '', '', 'S_BITMAP_REFRESH', "
+ + "'T_BITMAP_REFRESH.id=S_BITMAP_REFRESH.id', "
+ + "'idx=S_BITMAP_REFRESH.idx', 2)");
+ table = paimonTable("T_BITMAP_REFRESH");
+ long updateSnapshotId = table.snapshotManager().latestSnapshot().id();
+ assertThat(fileNames(indexEntries(table, "bitmap"))).isEqualTo(initialFiles);
+
+ buildBitmapIndexForTable("T_BITMAP_REFRESH", "idx");
+ table = paimonTable("T_BITMAP_REFRESH");
+ List<IndexManifestEntry> refreshed = indexEntries(table, "bitmap");
+ assertThat(fileNames(refreshed)).doesNotContainAnyElementsOf(initialFiles);
+ assertThat(refreshed)
+ .allSatisfy(
+ entry ->
+ assertThat(
+ DataEvolutionIndexSourceMeta.fromIndexFile(
+ entry.indexFile())
+ .scanSnapshotId())
+ .isEqualTo(updateSnapshotId));
+ assertThat(sql("SELECT id FROM T_BITMAP_REFRESH WHERE idx = 1001")).containsOnly(Row.of(1));
+ }
+
+ @Test
+ public void testBTreeRefreshesUpdatedDataEvolutionRangeAtomically() throws Exception {
+ tEnv.getConfig().set(TableConfigOptions.TABLE_DML_SYNC, true);
+ sql(
+ "CREATE TABLE T_REFRESH (id INT, idx INT, payload STRING) WITH ("
+ + "'bucket' = '-1', "
+ + "'global-index.enabled' = 'true', "
+ + "'row-tracking.enabled' = 'true', "
+ + "'data-evolution.enabled' = 'true', "
+ + "'global-index.column-update-action' = 'IGNORE', "
+ + "'btree-index.records-per-range' = '2'"
+ + ")");
+ sql(
+ "INSERT INTO T_REFRESH VALUES "
+ + IntStream.range(0, 10)
+ .mapToObj(i -> String.format("(%d, %d, 'p%d')", i, i, i))
+ .collect(Collectors.joining(",")));
+ buildBTreeIndexForTable("T_REFRESH", "idx");
+
+ sql(
+ "INSERT INTO T_REFRESH VALUES "
+ + IntStream.range(10, 20)
+ .mapToObj(i -> String.format("(%d, %d, 'p%d')", i, i, i))
+ .collect(Collectors.joining(",")));
+ buildBTreeIndexForTable("T_REFRESH", "idx");
+
+ FileStoreTable table = paimonTable("T_REFRESH");
+ Map<String, List<IndexManifestEntry>> initial = btreeEntriesByRange(table);
+ assertThat(initial.keySet()).containsExactlyInAnyOrder("0:9", "10:19");
+ assertThat(initial.get("0:9")).hasSizeGreaterThan(1);
+ assertThat(initial.get("10:19")).hasSizeGreaterThan(1);
+ assertThat(flatten(initial)).allSatisfy(entry -> assertThat(entry.bucket()).isZero());
+ assertThat(table.store().newScan().plan().files())
+ .allSatisfy(entry -> assertThat(entry.bucket()).isZero());
+ Set<String> initialFirstFiles = fileNames(initial.get("0:9"));
+ Set<String> initialSecondFiles = fileNames(initial.get("10:19"));
+ Set<String> initialFiles = fileNames(flatten(initial));
+ long beforeUpdateSnapshotId = table.snapshotManager().latestSnapshot().id();
+
+ sql("CREATE TABLE S_REFRESH (id INT, idx INT)");
+ sql("INSERT INTO S_REFRESH VALUES (1, 1001)");
+ sql(
+ "CALL sys.data_evolution_merge_into("
+ + "'default.T_REFRESH', '', '', 'S_REFRESH', "
+ + "'T_REFRESH.id=S_REFRESH.id', 'idx=S_REFRESH.idx', 2)");
+ table = paimonTable("T_REFRESH");
+ long updateSnapshotId = table.snapshotManager().latestSnapshot().id();
+ assertThat(updateSnapshotId).isGreaterThan(beforeUpdateSnapshotId);
+ assertThat(fileNames(flatten(btreeEntriesByRange(table)))).isEqualTo(initialFiles);
+ assertThat(sql("SELECT id FROM T_REFRESH WHERE idx = 1")).isEmpty();
+ assertThat(sql("SELECT id FROM T_REFRESH WHERE idx = 1001")).isEmpty();
+ assertThat(sql("SELECT idx FROM T_REFRESH WHERE id = 1")).containsOnly(Row.of(1001));
+ assertThat(sql("SELECT id FROM T_REFRESH WHERE idx = 2")).containsOnly(Row.of(2));
+
+ buildBTreeIndexForTable("T_REFRESH", "idx");
+ table = paimonTable("T_REFRESH");
+ assertThat(table.snapshotManager().latestSnapshot().id()).isEqualTo(updateSnapshotId + 1);
+
+ Map<String, List<IndexManifestEntry>> refreshed = btreeEntriesByRange(table);
+ assertThat(refreshed.keySet()).containsExactlyInAnyOrder("0:9", "10:19");
+ assertThat(fileNames(refreshed.get("0:9"))).doesNotContainAnyElementsOf(initialFirstFiles);
+ assertThat(fileNames(refreshed.get("10:19"))).isEqualTo(initialSecondFiles);
+ assertThat(refreshed.get("0:9"))
+ .allSatisfy(
+ entry ->
+ assertThat(
+ DataEvolutionIndexSourceMeta.fromIndexFile(
+ entry.indexFile())
+ .scanSnapshotId())
+ .isEqualTo(updateSnapshotId));
+ assertThat(sql("SELECT id FROM T_REFRESH WHERE idx = 1")).isEmpty();
+ assertThat(sql("SELECT id FROM T_REFRESH WHERE idx = 1001")).containsOnly(Row.of(1));
+
+ long refreshedSnapshotId = table.snapshotManager().latestSnapshot().id();
+ Set<String> refreshedFiles = fileNames(flatten(refreshed));
+ buildBTreeIndexForTable("T_REFRESH", "idx");
+ table = paimonTable("T_REFRESH");
+ assertThat(table.snapshotManager().latestSnapshot().id()).isEqualTo(refreshedSnapshotId);
+ assertThat(fileNames(flatten(btreeEntriesByRange(table)))).isEqualTo(refreshedFiles);
+
+ sql("CREATE TABLE S_PAYLOAD (id INT, payload STRING)");
+ sql("INSERT INTO S_PAYLOAD VALUES (1, 'new-payload')");
+ sql(
+ "CALL sys.data_evolution_merge_into("
+ + "'default.T_REFRESH', '', '', 'S_PAYLOAD', "
+ + "'T_REFRESH.id=S_PAYLOAD.id', 'payload=S_PAYLOAD.payload', 1)");
+ table = paimonTable("T_REFRESH");
+ long payloadUpdateSnapshotId = table.snapshotManager().latestSnapshot().id();
+ assertThat(payloadUpdateSnapshotId).isGreaterThan(refreshedSnapshotId);
+ assertThat(sql("SELECT payload FROM T_REFRESH WHERE id = 1"))
+ .containsOnly(Row.of("new-payload"));
+ buildBTreeIndexForTable("T_REFRESH", "idx");
+ table = paimonTable("T_REFRESH");
+ assertThat(table.snapshotManager().latestSnapshot().id())
+ .isEqualTo(payloadUpdateSnapshotId);
+ assertThat(fileNames(flatten(btreeEntriesByRange(table)))).isEqualTo(refreshedFiles);
+ }
+
+ @Test
+ public void testBTreeRefreshesOnlyUpdatedPartition() throws Exception {
+ tEnv.getConfig().set(TableConfigOptions.TABLE_DML_SYNC, true);
+ sql(
+ "CREATE TABLE T_PARTITION_REFRESH (pt INT, id INT, idx INT) "
+ + "PARTITIONED BY (pt) WITH ("
+ + "'bucket' = '-1', "
+ + "'global-index.enabled' = 'true', "
+ + "'row-tracking.enabled' = 'true', "
+ + "'data-evolution.enabled' = 'true', "
+ + "'global-index.column-update-action' = 'IGNORE', "
+ + "'btree-index.records-per-range' = '2'"
+ + ")");
+ sql(
+ "INSERT INTO T_PARTITION_REFRESH VALUES "
+ + IntStream.range(0, 10)
+ .mapToObj(i -> String.format("(0, %d, %d)", i, i))
+ .collect(Collectors.joining(",")));
+ sql(
+ "INSERT INTO T_PARTITION_REFRESH VALUES "
+ + IntStream.range(10, 20)
+ .mapToObj(i -> String.format("(1, %d, %d)", i, i))
+ .collect(Collectors.joining(",")));
+ buildBTreeIndexForTable("T_PARTITION_REFRESH", "idx");
+
+ FileStoreTable table = paimonTable("T_PARTITION_REFRESH");
+ Map<String, List<IndexManifestEntry>> initial = btreeEntriesByRange(table);
+ assertThat(initial.keySet()).containsExactlyInAnyOrder("0:9", "10:19");
+ Set<String> firstPartitionFiles = fileNames(initial.get("0:9"));
+ Set<String> secondPartitionFiles = fileNames(initial.get("10:19"));
+ assertThat(initial.get("0:9"))
+ .allSatisfy(entry -> assertThat(entry.partition().getInt(0)).isZero());
+ assertThat(initial.get("10:19"))
+ .allSatisfy(entry -> assertThat(entry.partition().getInt(0)).isEqualTo(1));
+
+ sql("CREATE TABLE S_PARTITION_REFRESH (id INT, idx INT)");
+ sql("INSERT INTO S_PARTITION_REFRESH VALUES (11, 1011)");
+ sql(
+ "CALL sys.data_evolution_merge_into("
+ + "'default.T_PARTITION_REFRESH', '', '', 'S_PARTITION_REFRESH', "
+ + "'T_PARTITION_REFRESH.id=S_PARTITION_REFRESH.id', "
+ + "'idx=S_PARTITION_REFRESH.idx', 2)");
+ long updateSnapshotId =
+ paimonTable("T_PARTITION_REFRESH").snapshotManager().latestSnapshot().id();
+
+ buildBTreeIndexForTable("T_PARTITION_REFRESH", "idx");
+ table = paimonTable("T_PARTITION_REFRESH");
+ assertThat(table.snapshotManager().latestSnapshot().id()).isEqualTo(updateSnapshotId + 1);
+ Map<String, List<IndexManifestEntry>> refreshed = btreeEntriesByRange(table);
+ assertThat(fileNames(refreshed.get("0:9"))).isEqualTo(firstPartitionFiles);
+ assertThat(fileNames(refreshed.get("10:19")))
+ .doesNotContainAnyElementsOf(secondPartitionFiles);
+ assertThat(sql("SELECT pt, id FROM T_PARTITION_REFRESH WHERE idx = 1011"))
+ .containsOnly(Row.of(1, 11));
+ }
+
@Test
public void testBTreeIndexWithMultiPartition() throws Catalog.TableNotExistException {
sql(
@@ -177,6 +383,43 @@ public class SortedGlobalIndexITCase extends CatalogITCaseBase {
tableName, indexColumn);
}
+ private void buildBitmapIndexForTable(String tableName, String indexColumn) {
+ sql(
+ "CALL sys.create_global_index(`table` => 'default.%s', "
+ + "index_column => '%s', index_type => 'bitmap', "
+ + "options => 'sorted-index.records-per-range=20')",
+ tableName, indexColumn);
+ }
+
+ private List<IndexManifestEntry> indexEntries(FileStoreTable table, String indexType) {
+ return table.store().newIndexFileHandler().scanEntries().stream()
+ .filter(entry -> indexType.equals(entry.indexFile().indexType()))
+ .collect(Collectors.toList());
+ }
+
+ private Map<String, List<IndexManifestEntry>> btreeEntriesByRange(FileStoreTable table) {
+ return table.store().newIndexFileHandler().scanEntries().stream()
+ .filter(entry -> "btree".equals(entry.indexFile().indexType()))
+ .collect(
+ Collectors.groupingBy(
+ entry ->
+ entry.indexFile().globalIndexMeta().rowRangeStart()
+ + ":"
+ + entry.indexFile()
+ .globalIndexMeta()
+ .rowRangeEnd()));
+ }
+
+ private List<IndexManifestEntry> flatten(Map<String, List<IndexManifestEntry>> entriesByRange) {
+ return entriesByRange.values().stream().flatMap(List::stream).collect(Collectors.toList());
+ }
+
+ private Set<String> fileNames(List<IndexManifestEntry> entries) {
+ return entries.stream()
+ .map(entry -> entry.indexFile().fileName())
+ .collect(Collectors.toSet());
+ }
+
@Test
void testBTreeIndexWithSingleRangeAndParallelWriters() throws Catalog.TableNotExistException {
sql(
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 916d6c2578..57d58b34c6 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
@@ -23,26 +23,26 @@ import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.BinaryRowWriter;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.fs.Path;
-import org.apache.paimon.globalindex.GlobalIndexBuilderUtils;
import org.apache.paimon.globalindex.IndexedSplit;
+import org.apache.paimon.globalindex.generic.GenericGlobalIndexScanner;
import org.apache.paimon.io.PojoDataFileMeta;
import org.apache.paimon.manifest.FileKind;
import org.apache.paimon.manifest.ManifestEntry;
-import org.apache.paimon.schema.SchemaManager;
-import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.options.Options;
import org.apache.paimon.stats.SimpleStats;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.utils.FileStorePathFactory;
import org.apache.paimon.utils.Range;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import java.io.IOException;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.Collections;
import java.util.List;
+import java.util.Optional;
import java.util.UUID;
import static org.assertj.core.api.Assertions.assertThat;
@@ -172,6 +172,24 @@ class GenericIndexTopoBuilderTest {
assertThat(tasks).isEmpty();
}
+ @Test
+ void testNoIncrementalScanResult() throws Exception {
+ GenericGlobalIndexScanner indexBuilder = mock(GenericGlobalIndexScanner.class);
+ when(indexBuilder.withIndex(any(), any(), any())).thenReturn(indexBuilder);
+ when(indexBuilder.incrementalScan()).thenReturn(Optional.empty());
+
+ assertThat(
+ GenericIndexTopoBuilder.buildIndex(
+ StreamExecutionEnvironment.getExecutionEnvironment(),
+ () -> indexBuilder,
+ table,
+ "id",
+ "test-index",
+ null,
+ new Options()))
+ .isFalse();
+ }
+
@Test
void testAllFilesNullRowId() throws IOException {
List<ManifestEntry> entries = new ArrayList<>();
@@ -215,326 +233,8 @@ class GenericIndexTopoBuilderTest {
assertThat(tasks.get(1).rowRanges().get(0)).isEqualTo(new Range(100, 149));
}
- // ========== Incremental build scenarios (maxIndexedRowId) ==========
-
- @Test
- void testIncrementalFirstBuildNoIndex() {
- // First build: no existing index, two files.
- // maxIndexedRowId=-1 → all shards created normally.
- List<ManifestEntry> entries = new ArrayList<>();
- entries.add(createEntry(BinaryRow.EMPTY_ROW, 0L, 100));
- entries.add(createEntry(BinaryRow.EMPTY_ROW, 100L, 100));
-
- List<IndexedSplit> tasks =
- GenericIndexTopoBuilder.computeShardTasks(table, entries, 200, -1);
-
- assertThat(tasks).hasSize(1);
- assertThat(tasks.get(0).rowRanges().get(0)).isEqualTo(new Range(0, 199));
- assertThat(tasks.get(0).dataSplit().dataFiles()).hasSize(2);
- }
-
- @Test
- void testIncrementalNormalNoCompaction() {
- // Indexed [0,199], new file [200,399]. No compaction.
- List<ManifestEntry> entries = new ArrayList<>();
- entries.add(createEntry(BinaryRow.EMPTY_ROW, 200L, 200));
-
- List<IndexedSplit> tasks =
- GenericIndexTopoBuilder.computeShardTasks(table, entries, 200, 199);
-
- assertThat(tasks).hasSize(1);
- 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
- void testIncrementalNoNewDataAllIndexed() {
- // All data [0,399] already indexed. All shards should be skipped.
- List<ManifestEntry> entries = new ArrayList<>();
- entries.add(createEntry(BinaryRow.EMPTY_ROW, 0L, 400));
-
- List<IndexedSplit> tasks =
- GenericIndexTopoBuilder.computeShardTasks(table, entries, 200, 399);
-
- assertThat(tasks).isEmpty();
- }
-
- @Test
- void testIncrementalCompactMergesIndexedAndUnindexed() {
- // Files A[0,99], B[100,199] indexed, new C[200,299], compact B+C → D[100,299]
- // Shard 0 [0,199]: effectiveStart=200 > 199 → skip
- // Shard 1 [200,299]: effectiveStart=200 → [200,299]
- List<ManifestEntry> entries = new ArrayList<>();
- entries.add(createEntry(BinaryRow.EMPTY_ROW, 100L, 200)); // D[100,299]
-
- List<IndexedSplit> tasks =
- GenericIndexTopoBuilder.computeShardTasks(table, entries, 200, 199);
-
- assertThat(tasks).hasSize(1);
- assertThat(tasks.get(0).rowRanges().get(0)).isEqualTo(new Range(200, 299));
- }
-
- @Test
- void testIncrementalCompactOnlyIndexedFiles() {
- // Compact two indexed files → empty entries → no tasks.
- List<IndexedSplit> tasks =
- GenericIndexTopoBuilder.computeShardTasks(table, Collections.emptyList(), 200, 199);
-
- assertThat(tasks).isEmpty();
- }
-
- @Test
- void testIncrementalCompactPartialWithUntouchedFiles() {
- // Indexed [0,399]. Compact [200,399]+[400,599] → D[200,599].
- // Shard 1 [200,399]: effectiveStart=400 > 399 → skip
- // Shard 2 [400,599]: effectiveStart=400 → [400,599]
- List<ManifestEntry> entries = new ArrayList<>();
- entries.add(createEntry(BinaryRow.EMPTY_ROW, 200L, 400)); // D[200,599]
-
- List<IndexedSplit> tasks =
- GenericIndexTopoBuilder.computeShardTasks(table, entries, 200, 399);
-
- assertThat(tasks).hasSize(1);
- assertThat(tasks.get(0).rowRanges().get(0)).isEqualTo(new Range(400, 599));
- }
-
- @Test
- void testIncrementalMultipleWritesThenCompact() {
- // Write 200 (indexed), write 200 more, compact → big file [0,399].
- // Shard 0 [0,199]: effectiveStart=200 > 199 → skip
- // Shard 1 [200,399]: effectiveStart=200 → [200,399]
- List<ManifestEntry> entries = new ArrayList<>();
- entries.add(createEntry(BinaryRow.EMPTY_ROW, 0L, 400));
-
- List<IndexedSplit> tasks =
- GenericIndexTopoBuilder.computeShardTasks(table, entries, 200, 199);
-
- assertThat(tasks).hasSize(1);
- assertThat(tasks.get(0).rowRanges().get(0)).isEqualTo(new Range(200, 399));
- }
-
- @Test
- void testIncrementalMergeAllSmallShards() {
- // All entries small and deleted by merge. maxIndexedRowId=-1 → full rebuild.
- List<ManifestEntry> entries = new ArrayList<>();
- entries.add(createEntry(BinaryRow.EMPTY_ROW, 0L, 250));
-
- List<IndexedSplit> tasks =
- GenericIndexTopoBuilder.computeShardTasks(table, entries, 200, -1);
-
- assertThat(tasks).hasSize(2);
- 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
- void testIncrementalMergePartialKeepLargeShard() {
- // Entry [0,199] kept. Small shards [200-399] merged. maxIndexedRowId=199.
- // Shard 0 [0,199]: effectiveStart=200 > 199 → skip
- // Shard 1 [200,399]: effectiveStart=200 → [200,399]
- // Shard 2 [400,599]: effectiveStart=200 → [400,599]
- List<ManifestEntry> entries = new ArrayList<>();
- entries.add(createEntry(BinaryRow.EMPTY_ROW, 0L, 600));
-
- List<IndexedSplit> tasks =
- GenericIndexTopoBuilder.computeShardTasks(table, entries, 200, 199);
-
- assertThat(tasks).hasSize(2);
- 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
- void testIncrementalShardBoundaryExactAlign() {
- // maxIndexedRowId=199, new file starts exactly at shard boundary.
- List<ManifestEntry> entries = new ArrayList<>();
- entries.add(createEntry(BinaryRow.EMPTY_ROW, 200L, 200));
-
- List<IndexedSplit> tasks =
- GenericIndexTopoBuilder.computeShardTasks(table, entries, 200, 199);
-
- assertThat(tasks).hasSize(1);
- assertThat(tasks.get(0).rowRanges().get(0)).isEqualTo(new Range(200, 399));
- }
-
- @Test
- void testIncrementalShardBoundaryNotAligned() {
- // maxIndexedRowId=149. Compacted file [0,349].
- // Shard 0 [0,199]: effectiveStart=150 → [150,199]
- // Shard 1 [200,349]: → [200,349]
- List<ManifestEntry> entries = new ArrayList<>();
- entries.add(createEntry(BinaryRow.EMPTY_ROW, 0L, 350));
-
- List<IndexedSplit> tasks =
- GenericIndexTopoBuilder.computeShardTasks(table, entries, 200, 149);
-
- assertThat(tasks).hasSize(2);
- 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
- void testIncrementalFileSpansMultipleShards() {
- // One large file [0,599] spanning 3 shards, indexed [0,199].
- // Shard 0: skip. Shard 1: [200,399]. Shard 2: [400,599].
- List<ManifestEntry> entries = new ArrayList<>();
- entries.add(createEntry(BinaryRow.EMPTY_ROW, 0L, 600));
-
- List<IndexedSplit> tasks =
- GenericIndexTopoBuilder.computeShardTasks(table, entries, 200, 199);
-
- assertThat(tasks).hasSize(2);
- 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
- void testIncrementalNullFirstRowIdFileSkipped() {
- List<ManifestEntry> entries = new ArrayList<>();
- entries.add(createEntry(BinaryRow.EMPTY_ROW, null, 100));
- entries.add(createEntry(BinaryRow.EMPTY_ROW, 200L, 100));
-
- List<IndexedSplit> tasks =
- GenericIndexTopoBuilder.computeShardTasks(table, entries, 200, 199);
-
- assertThat(tasks).hasSize(1);
- assertThat(tasks.get(0).rowRanges().get(0)).isEqualTo(new Range(200, 299));
- assertThat(tasks.get(0).dataSplit().dataFiles()).hasSize(1);
- }
-
- @Test
- void testIncrementalMultipleFilesInOneShard() {
- // Two contiguous new files in same shard.
- List<ManifestEntry> entries = new ArrayList<>();
- entries.add(createEntry(BinaryRow.EMPTY_ROW, 200L, 100));
- entries.add(createEntry(BinaryRow.EMPTY_ROW, 300L, 100));
-
- List<IndexedSplit> tasks =
- GenericIndexTopoBuilder.computeShardTasks(table, entries, 400, 199);
-
- assertThat(tasks).hasSize(1);
- assertThat(tasks.get(0).rowRanges().get(0)).isEqualTo(new Range(200, 399));
- assertThat(tasks.get(0).dataSplit().dataFiles()).hasSize(2);
- }
-
- @Test
- void testIncrementalGapBetweenFilesProducesSeparateTasks() {
- // Two files with a gap, same shard. maxIndexedRowId=-1.
- List<ManifestEntry> entries = new ArrayList<>();
- entries.add(createEntry(BinaryRow.EMPTY_ROW, 0L, 50));
- entries.add(createEntry(BinaryRow.EMPTY_ROW, 150L, 50));
-
- List<IndexedSplit> tasks =
- GenericIndexTopoBuilder.computeShardTasks(table, entries, 200, -1);
-
- assertThat(tasks).hasSize(2);
- 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
- void testIncrementalFileStartsAfterEffectiveStart() {
- // maxIndexedRowId=250. New file [300,499].
- // Shard 1 [200,399]: effectiveStart=251, file starts at 300 → [300,399]
- // Shard 2 [400,499]: effectiveStart=251 → [400,499]
- List<ManifestEntry> entries = new ArrayList<>();
- entries.add(createEntry(BinaryRow.EMPTY_ROW, 300L, 200));
-
- List<IndexedSplit> tasks =
- GenericIndexTopoBuilder.computeShardTasks(table, entries, 200, 250);
-
- assertThat(tasks).hasSize(2);
- 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
- void testAppendFilterOldFilesBeforeNewFiles() {
- // Typical append: write file0[0,99](schema1), file1[100,199](schema1),
- // then file2[200,299](schema0) arrives (old schema).
- // Boundary = 200, keep files with firstRowId < 200.
- SchemaManager schemaManager = mock(SchemaManager.class);
- TableSchema oldSchema = mock(TableSchema.class);
- TableSchema newSchema = mock(TableSchema.class);
- when(schemaManager.schema(0L)).thenReturn(oldSchema);
- when(schemaManager.schema(1L)).thenReturn(newSchema);
- when(oldSchema.fieldNames()).thenReturn(Arrays.asList("id", "name"));
- when(newSchema.fieldNames()).thenReturn(Arrays.asList("id", "name", "vec"));
-
- List<ManifestEntry> entries = new ArrayList<>();
- entries.add(createEntryWithSchemaId(BinaryRow.EMPTY_ROW, 0L, 100, 1L));
- entries.add(createEntryWithSchemaId(BinaryRow.EMPTY_ROW, 100L, 100, 1L));
- entries.add(createEntryWithSchemaId(BinaryRow.EMPTY_ROW, 200L, 100, 0L));
-
- List<ManifestEntry> result =
- GlobalIndexBuilderUtils.filterEntriesBefore(
- entries,
- GlobalIndexBuilderUtils.findMinNonIndexableRowId(
- schemaManager, entries, Collections.singletonList("vec")));
-
- assertThat(result).hasSize(2);
- assertThat(result.get(0).file().nonNullFirstRowId()).isEqualTo(0L);
- assertThat(result.get(1).file().nonNullFirstRowId()).isEqualTo(100L);
- }
-
// -- Helpers --
- private static ManifestEntry createEntryWithSchemaId(
- BinaryRow partition, Long firstRowId, long rowCount, long schemaId) {
- PojoDataFileMeta file =
- new PojoDataFileMeta(
- "test-file-" + UUID.randomUUID(),
- 1024L,
- rowCount,
- BinaryRow.EMPTY_ROW,
- BinaryRow.EMPTY_ROW,
- SimpleStats.EMPTY_STATS,
- SimpleStats.EMPTY_STATS,
- 0L,
- 0L,
- schemaId,
- 0,
- Collections.emptyList(),
- null,
- null,
- null,
- null,
- null,
- null,
- firstRowId,
- null);
- return ManifestEntry.create(FileKind.ADD, partition, 0, 1, file);
- }
-
private static ManifestEntry createEntry(BinaryRow partition, Long firstRowId, long rowCount) {
PojoDataFileMeta file =
new PojoDataFileMeta(
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 19b80e5bd5..6a33b2288c 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
@@ -20,7 +20,8 @@ package org.apache.paimon.flink.globalindex;
import org.apache.paimon.flink.globalindex.SortedIndexTopoBuilder.SortedBuildTask;
import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter;
-import org.apache.paimon.globalindex.sorted.SortedGlobalIndexBuilder;
+import org.apache.paimon.globalindex.sorted.SortedGlobalIndexScanner;
+import org.apache.paimon.globalindex.sorted.SortedGlobalIndexWriter;
import org.apache.paimon.options.Options;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.types.DataTypes;
@@ -68,7 +69,8 @@ public class SortedIndexTopoBuilderTest {
operatorClass.getDeclaredConstructor(
List.class,
int.class,
- SortedGlobalIndexBuilder.class,
+ SortedGlobalIndexWriter.class,
+ long.class,
int.class,
int.class,
int.class,
@@ -78,7 +80,8 @@ public class SortedIndexTopoBuilderTest {
constructor.newInstance(
Collections.emptyList(),
0,
- mock(SortedGlobalIndexBuilder.class),
+ mock(SortedGlobalIndexWriter.class),
+ 1L,
0,
0,
0,
@@ -99,41 +102,43 @@ public class SortedIndexTopoBuilderTest {
@Test
public void testBuildIndexReturnsFalseWhenNoBuildTask() throws Exception {
- SortedGlobalIndexBuilder indexBuilder = mock(SortedGlobalIndexBuilder.class);
- when(indexBuilder.withIndexField("id")).thenReturn(indexBuilder);
- when(indexBuilder.incrementalScan()).thenReturn(Optional.empty());
+ SortedGlobalIndexScanner indexScanner = mock(SortedGlobalIndexScanner.class);
+ when(indexScanner.withIndexField("id")).thenReturn(indexScanner);
+ when(indexScanner.incrementalScan()).thenReturn(Optional.empty());
StreamExecutionEnvironment env = mock(StreamExecutionEnvironment.class);
assertThat(
SortedIndexTopoBuilder.buildIndex(
env,
- () -> indexBuilder,
+ () -> indexScanner,
mock(FileStoreTable.class),
Collections.singletonList("id"),
+ "btree",
null,
new Options()))
.isFalse();
- verify(indexBuilder).incrementalScan();
+ verify(indexScanner).incrementalScan();
verifyNoInteractions(env);
}
@Test
public void testBuildIndexStreamReturnsEmptyWhenNoBuildTask() throws Exception {
- SortedGlobalIndexBuilder indexBuilder = mock(SortedGlobalIndexBuilder.class);
- when(indexBuilder.withIndexField("id")).thenReturn(indexBuilder);
- when(indexBuilder.incrementalScan()).thenReturn(Optional.empty());
+ SortedGlobalIndexScanner indexScanner = mock(SortedGlobalIndexScanner.class);
+ when(indexScanner.withIndexField("id")).thenReturn(indexScanner);
+ when(indexScanner.incrementalScan()).thenReturn(Optional.empty());
StreamExecutionEnvironment env = mock(StreamExecutionEnvironment.class);
assertThat(
SortedIndexTopoBuilder.buildIndexStream(
env,
- () -> indexBuilder,
+ () -> indexScanner,
mock(FileStoreTable.class),
Collections.singletonList("id"),
+ "btree",
null,
new Options()))
.isEmpty();
- verify(indexBuilder).incrementalScan();
+ verify(indexScanner).incrementalScan();
verifyNoInteractions(env);
}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/FullTextSearchProcedureITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/FullTextSearchProcedureITCase.java
index b621d0bc9f..a5fbc060ac 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/FullTextSearchProcedureITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/FullTextSearchProcedureITCase.java
@@ -19,7 +19,10 @@
package org.apache.paimon.flink.procedure;
import org.apache.paimon.flink.CatalogITCaseBase;
+import org.apache.paimon.index.DataEvolutionIndexSourceMeta;
import org.apache.paimon.index.pkfulltext.PkFullTextIndexFile;
+import org.apache.paimon.manifest.IndexManifestEntry;
+import org.apache.paimon.table.FileStoreTable;
import org.apache.flink.table.api.config.TableConfigOptions;
import org.apache.flink.types.Row;
@@ -34,6 +37,85 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** IT cases for {@link FullTextSearchProcedure}. */
public class FullTextSearchProcedureITCase extends CatalogITCaseBase {
+ @Test
+ public void testGenericFullTextRefreshesUpdatedDataEvolutionRange() throws Exception {
+ sql(
+ "CREATE TABLE T_DE (id INT, content STRING) WITH ("
+ + "'bucket' = '-1', "
+ + "'row-tracking.enabled' = 'true', "
+ + "'data-evolution.enabled' = 'true', "
+ + "'global-index.column-update-action' = 'IGNORE'"
+ + ")");
+ sql("INSERT INTO T_DE VALUES (1, 'apache paimon'), (2, 'lake format')");
+
+ FileStoreTable table = paimonTable("T_DE");
+ long scanSnapshotId = table.snapshotManager().latestSnapshot().id();
+ tEnv.getConfig().set(TableConfigOptions.TABLE_DML_SYNC, true);
+ buildFullTextIndex("T_DE");
+
+ List<IndexManifestEntry> entries = table.store().newIndexFileHandler().scan("full-text");
+ assertThat(entries).isNotEmpty();
+ assertThat(entries)
+ .allSatisfy(
+ entry ->
+ assertThat(
+ DataEvolutionIndexSourceMeta.fromIndexFile(
+ entry.indexFile())
+ .scanSnapshotId())
+ .isEqualTo(scanSnapshotId));
+ List<String> initialFiles =
+ entries.stream()
+ .map(entry -> entry.indexFile().fileName())
+ .collect(Collectors.toList());
+
+ sql("CREATE TABLE S_DE (id INT, content STRING)");
+ sql("INSERT INTO S_DE VALUES (1, 'refreshed content')");
+ sql(
+ "CALL sys.data_evolution_merge_into("
+ + "'default.T_DE', '', '', 'S_DE', "
+ + "'T_DE.id=S_DE.id', 'content=S_DE.content', 2)");
+ table = paimonTable("T_DE");
+ long updateSnapshotId = table.snapshotManager().latestSnapshot().id();
+
+ buildFullTextIndex("T_DE");
+
+ table = paimonTable("T_DE");
+ entries = table.store().newIndexFileHandler().scan("full-text");
+ assertThat(entries).isNotEmpty();
+ assertThat(
+ entries.stream()
+ .map(entry -> entry.indexFile().fileName())
+ .collect(Collectors.toList()))
+ .doesNotContainAnyElementsOf(initialFiles);
+ assertThat(entries)
+ .allSatisfy(
+ entry ->
+ assertThat(
+ DataEvolutionIndexSourceMeta.fromIndexFile(
+ entry.indexFile())
+ .scanSnapshotId())
+ .isEqualTo(updateSnapshotId));
+
+ List<String> rows =
+ sql(
+ "CALL sys.full_text_search("
+ + "`table` => 'default.T_DE', "
+ + "`column` => 'content', "
+ + "query => '{\"match\":{\"column\":\"content\",\"terms\":\"refreshed\"}}', "
+ + "top_k => 10, "
+ + "projection => 'id,content')")
+ .stream()
+ .map(row -> row.getField(0).toString())
+ .collect(Collectors.toList());
+ assertThat(rows)
+ .singleElement()
+ .satisfies(
+ row ->
+ assertThat(row)
+ .contains("\"id\":\"1\"")
+ .contains("\"content\":\"refreshed content\""));
+ }
+
@Test
public void testPrimaryKeyFullTextSearchWithScoreProjection() throws Exception {
createPrimaryKeyFullTextTable("T");
@@ -132,6 +214,15 @@ public class FullTextSearchProcedureITCase extends CatalogITCaseBase {
table, query, topK, projection);
}
+ private void buildFullTextIndex(String table) {
+ sql(
+ "CALL sys.create_global_index("
+ + "`table` => 'default.%s', "
+ + "index_column => 'content', "
+ + "index_type => 'full-text')",
+ table);
+ }
+
private void createPrimaryKeyFullTextTable(String tableName) {
sql(
"CREATE TABLE %s ("
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/VectorSearchProcedureITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/VectorSearchProcedureITCase.java
index 9fb817ba10..31324f2ec7 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/VectorSearchProcedureITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/VectorSearchProcedureITCase.java
@@ -1114,7 +1114,8 @@ public class VectorSearchProcedureITCase extends CatalogITCaseBase {
rowRange,
Arrays.asList(vectorField, idField),
TestVectorGlobalIndexerFactory.IDENTIFIER,
- entries);
+ entries,
+ null);
DataIncrement dataIncrement = DataIncrement.indexIncrement(indexFiles);
CommitMessage message =
diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/DefaultGlobalIndexBuilder.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/DefaultGlobalIndexBuilder.java
index fd13f5ee0a..5734cf84e3 100644
--- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/DefaultGlobalIndexBuilder.java
+++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/DefaultGlobalIndexBuilder.java
@@ -39,6 +39,8 @@ import org.apache.paimon.utils.LongCounter;
import org.apache.paimon.utils.ProjectedRow;
import org.apache.paimon.utils.Range;
+import javax.annotation.Nullable;
+
import java.io.IOException;
import java.io.Serializable;
import java.util.ArrayList;
@@ -61,6 +63,7 @@ public class DefaultGlobalIndexBuilder implements Serializable {
private final String indexType;
private final Range rowRange;
private final Options options;
+ private final @Nullable byte[] sourceMeta;
public DefaultGlobalIndexBuilder(
FileStoreTable table,
@@ -78,7 +81,8 @@ public class DefaultGlobalIndexBuilder implements Serializable {
Collections.emptyList(),
indexType,
rowRange,
- options);
+ options,
+ null);
}
public DefaultGlobalIndexBuilder(
@@ -90,6 +94,28 @@ public class DefaultGlobalIndexBuilder implements Serializable {
String indexType,
Range rowRange,
Options options) {
+ this(
+ table,
+ partition,
+ readType,
+ indexField,
+ extraFields,
+ indexType,
+ rowRange,
+ options,
+ null);
+ }
+
+ public DefaultGlobalIndexBuilder(
+ FileStoreTable table,
+ BinaryRow partition,
+ RowType readType,
+ DataField indexField,
+ List<DataField> extraFields,
+ String indexType,
+ Range rowRange,
+ Options options,
+ @Nullable byte[] sourceMeta) {
this.table = table;
this.partition = partition;
this.readType = readType;
@@ -102,6 +128,7 @@ public class DefaultGlobalIndexBuilder implements Serializable {
this.indexType = indexType;
this.rowRange = rowRange;
this.options = options;
+ this.sourceMeta = sourceMeta;
}
/** The primary index column followed by the extra columns, in index order. */
@@ -131,7 +158,8 @@ public class DefaultGlobalIndexBuilder implements Serializable {
rowRange,
indexedFields(),
indexType,
- resultEntries);
+ resultEntries,
+ sourceMeta);
DataIncrement dataIncrement = DataIncrement.indexIncrement(indexFileMetas);
return new CommitMessageImpl(
partition, 0, null, dataIncrement, CompactIncrement.emptyIncrement());
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 faa85f6588..7e5a401df3 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
@@ -22,15 +22,20 @@ 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.DataEvolutionGlobalIndexRefreshPlanner;
import org.apache.paimon.globalindex.GlobalIndexBuilderUtils;
import org.apache.paimon.globalindex.IndexedSplit;
+import org.apache.paimon.index.DataEvolutionIndexSourceMeta;
+import org.apache.paimon.io.CompactIncrement;
+import org.apache.paimon.io.DataIncrement;
+import org.apache.paimon.manifest.IndexManifestEntry;
import org.apache.paimon.manifest.ManifestEntry;
import org.apache.paimon.options.Options;
import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.reader.RecordReader;
-import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.CommitMessageImpl;
import org.apache.paimon.table.sink.CommitMessageSerializer;
import org.apache.paimon.table.source.ReadBuilder;
import org.apache.paimon.types.DataField;
@@ -55,7 +60,9 @@ import java.util.function.BiFunction;
import java.util.stream.Collectors;
import static org.apache.paimon.CoreOptions.GLOBAL_INDEX_BUILD_MAX_PARALLELISM;
+import static org.apache.paimon.CoreOptions.GLOBAL_INDEX_COLUMN_UPDATE_ACTION;
import static org.apache.paimon.CoreOptions.GLOBAL_INDEX_ROW_COUNT_PER_SHARD;
+import static org.apache.paimon.CoreOptions.GlobalIndexColumnUpdateAction.IGNORE;
import static org.apache.paimon.utils.Preconditions.checkArgument;
/** Default topology builder. */
@@ -102,6 +109,22 @@ public class DefaultGlobalIndexTopoBuilder implements GlobalIndexTopologyBuilder
if (snapshot == null) {
return Collections.emptyList();
}
+ List<DataField> indexFields = new ArrayList<>();
+ indexFields.add(indexField);
+ indexFields.addAll(extraFields);
+ List<IndexManifestEntry> currentIndexes =
+ GlobalIndexBuilderUtils.currentIndexEntries(
+ table, snapshot, indexType, indexFields, partitionPredicate);
+ List<Range> rowRangesToBuild =
+ new ArrayList<>(
+ GlobalIndexBuilderUtils.unindexedRowRanges(snapshot, currentIndexes));
+ byte[] sourceMeta = new DataEvolutionIndexSourceMeta(snapshot.id()).serialize();
+ boolean detectDataFileChange =
+ new Options(table.options(), options.toMap()).get(GLOBAL_INDEX_COLUMN_UPDATE_ACTION)
+ == IGNORE;
+ if (rowRangesToBuild.isEmpty() && !detectDataFileChange) {
+ return Collections.emptyList();
+ }
List<ManifestEntry> entries =
table.store()
.newScan()
@@ -109,19 +132,16 @@ public class DefaultGlobalIndexTopoBuilder implements GlobalIndexTopologyBuilder
.withPartitionFilter(partitionPredicate)
.plan()
.files();
- List<DataField> indexFields = new ArrayList<>();
- indexFields.add(indexField);
- indexFields.addAll(extraFields);
- List<String> indexColumns =
- indexFields.stream().map(DataField::name).collect(Collectors.toList());
- SchemaManager schemaManager = new SchemaManager(table.fileIO(), table.location());
- long boundaryRowId =
- GlobalIndexBuilderUtils.findMinNonIndexableRowId(
- schemaManager, entries, indexColumns);
- entries = GlobalIndexBuilderUtils.filterEntriesBefore(entries, boundaryRowId);
- List<Range> rowRangesToBuild =
- GlobalIndexBuilderUtils.unindexedRowRanges(
- table, snapshot, indexType, indexFields, partitionPredicate);
+ List<IndexManifestEntry> indexesToRefresh = Collections.emptyList();
+ if (detectDataFileChange) {
+ indexesToRefresh =
+ DataEvolutionGlobalIndexRefreshPlanner.findIndexesToRefresh(
+ table.schemaManager(), entries, currentIndexes, indexFields);
+ for (IndexManifestEntry index : indexesToRefresh) {
+ rowRangesToBuild.add(index.indexFile().globalIndexMeta().rowRange());
+ }
+ }
+ rowRangesToBuild = Range.sortAndMergeOverlap(rowRangesToBuild, true);
if (rowRangesToBuild.isEmpty()) {
return Collections.emptyList();
}
@@ -145,23 +165,34 @@ public class DefaultGlobalIndexTopoBuilder implements GlobalIndexTopologyBuilder
extraFields,
indexType,
indexedSplit.rowRanges().get(0),
- options);
+ options,
+ sourceMeta);
byte[] builderBytes = InstantiationUtil.serializeObject(builder);
byte[] splitBytes = InstantiationUtil.serializeObject(indexedSplit);
taskList.add(Pair.of(builderBytes, splitBytes));
}
- if (taskList.isEmpty()) {
- return Collections.emptyList();
+ List<CommitMessage> commitMessages = new ArrayList<>();
+ if (!taskList.isEmpty()) {
+ int parallelism = parallelism(taskList.size(), options);
+ List<byte[]> commitMessageBytes =
+ javaSparkContext
+ .parallelize(taskList, parallelism)
+ .map(DefaultGlobalIndexTopoBuilder::buildIndex)
+ .collect();
+ commitMessages.addAll(CommitMessageSerializer.deserializeAll(commitMessageBytes));
}
-
- int parallelism = parallelism(taskList.size(), options);
- List<byte[]> commitMessageBytes =
- javaSparkContext
- .parallelize(taskList, parallelism)
- .map(DefaultGlobalIndexTopoBuilder::buildIndex)
- .collect();
- return CommitMessageSerializer.deserializeAll(commitMessageBytes);
+ for (IndexManifestEntry index : indexesToRefresh) {
+ commitMessages.add(
+ new CommitMessageImpl(
+ index.partition(),
+ index.bucket(),
+ null,
+ DataIncrement.deleteIndexIncrement(
+ Collections.singletonList(index.indexFile())),
+ CompactIncrement.emptyIncrement()));
+ }
+ return commitMessages;
}
static long rowsPerShard(Options options) {
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 9a23d1944a..57fa28a47a 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
@@ -21,8 +21,13 @@ package org.apache.paimon.spark.globalindex.sorted;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.data.serializer.BinaryRowSerializer;
-import org.apache.paimon.globalindex.sorted.SortedGlobalIndexBuilder;
+import org.apache.paimon.globalindex.ScanResult;
+import org.apache.paimon.globalindex.sorted.SortedGlobalIndexScanner;
+import org.apache.paimon.globalindex.sorted.SortedGlobalIndexWriter;
import org.apache.paimon.globalindex.sorted.SortedIndexOptions;
+import org.apache.paimon.io.CompactIncrement;
+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.spark.SparkRow;
@@ -30,15 +35,14 @@ import org.apache.paimon.spark.globalindex.GlobalIndexTopologyBuilder;
import org.apache.paimon.spark.util.ScanPlanHelper$;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.CommitMessageImpl;
import org.apache.paimon.table.sink.CommitMessageSerializer;
import org.apache.paimon.table.source.DataSplit;
import org.apache.paimon.table.source.Split;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.InstantiationUtil;
-import org.apache.paimon.utils.Pair;
import org.apache.paimon.utils.Range;
-import org.apache.paimon.utils.RowRangeIndex;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.FlatMapFunction;
@@ -60,8 +64,8 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
-import static org.apache.paimon.globalindex.sorted.SortedGlobalIndexBuilder.groupSplitsByRange;
-import static org.apache.paimon.globalindex.sorted.SortedGlobalIndexBuilder.splitByContiguousRowRange;
+import static org.apache.paimon.globalindex.GlobalIndexBuilderUtils.groupSplitsByRange;
+import static org.apache.paimon.globalindex.GlobalIndexBuilderUtils.splitByContiguousRowRange;
/** The {@link GlobalIndexTopologyBuilder} for sorted indexes. */
public class SortedIndexTopoBuilder implements GlobalIndexTopologyBuilder {
@@ -84,27 +88,27 @@ public class SortedIndexTopoBuilder implements GlobalIndexTopologyBuilder {
DataField indexField,
Options options)
throws IOException {
- SortedGlobalIndexBuilder indexBuilder =
- new SortedGlobalIndexBuilder(table, indexType, options)
+ SortedGlobalIndexScanner indexScanner =
+ new SortedGlobalIndexScanner(table, indexType, options)
.withIndexField(indexField.name());
if (partitionPredicate != null) {
- indexBuilder = indexBuilder.withPartitionPredicate(partitionPredicate);
+ indexScanner = indexScanner.withPartitionPredicate(partitionPredicate);
}
- Optional<Pair<RowRangeIndex, List<DataSplit>>> indexRangeAndSplits =
- indexBuilder.incrementalScan();
- if (!indexRangeAndSplits.isPresent()) {
+ Optional<ScanResult<DataSplit>> optionalScanResult = indexScanner.incrementalScan();
+ if (!optionalScanResult.isPresent()) {
return Collections.emptyList();
}
- Pair<RowRangeIndex, List<DataSplit>> scanResult = indexRangeAndSplits.get();
- List<DataSplit> splits = splitByContiguousRowRange(scanResult.getRight());
+ ScanResult<DataSplit> scanResult = optionalScanResult.get();
+ long scanSnapshotId = scanResult.scanSnapshotId();
+ List<DataSplit> splits = splitByContiguousRowRange(scanResult.entries());
if (splits.isEmpty()) {
return Collections.emptyList();
}
Map<BinaryRow, Map<Range, List<Split>>> partitionRangeSplits =
- groupSplitsByRange(scanResult.getKey(), splits);
+ groupSplitsByRange(scanResult.rowRangeIndex(), splits);
if (partitionRangeSplits.isEmpty()) {
return Collections.emptyList();
}
@@ -120,6 +124,9 @@ public class SortedIndexTopoBuilder implements GlobalIndexTopologyBuilder {
sortColumns.add(indexField.name());
final int partitionKeyNum = table.partitionKeys().size();
BinaryRowSerializer binaryRowSerializer = new BinaryRowSerializer(partitionKeyNum);
+ SortedGlobalIndexWriter indexWriter =
+ new SortedGlobalIndexWriter(table, indexType, options)
+ .withIndexField(indexField.name());
for (Map.Entry<BinaryRow, Map<Range, List<Split>>> partitionEntry :
partitionRangeSplits.entrySet()) {
for (Map.Entry<Range, List<Split>> entry : partitionEntry.getValue().entrySet()) {
@@ -150,7 +157,7 @@ public class SortedIndexTopoBuilder implements GlobalIndexTopologyBuilder {
selected.repartitionByRange(partitionNum, sortFields)
.sortWithinPartitions(sortFields);
- final byte[] serializedBuilder = InstantiationUtil.serializeObject(indexBuilder);
+ final byte[] serializedWriter = InstantiationUtil.serializeObject(indexWriter);
final byte[] partitionBytes =
binaryRowSerializer.serializeToBytes(partitionEntry.getKey());
JavaRDD<byte[]> written =
@@ -162,31 +169,43 @@ public class SortedIndexTopoBuilder implements GlobalIndexTopologyBuilder {
iter ->
buildSortedIndex(
iter,
- serializedBuilder,
+ serializedWriter,
range,
partitionKeyNum,
- partitionBytes));
+ partitionBytes,
+ scanSnapshotId));
List<byte[]> commitBytes = written.collect();
allMessages.addAll(CommitMessageSerializer.deserializeAll(commitBytes));
}
}
+ for (IndexManifestEntry entry : scanResult.deletedIndexEntries()) {
+ allMessages.add(
+ new CommitMessageImpl(
+ entry.partition(),
+ entry.bucket(),
+ null,
+ DataIncrement.deleteIndexIncrement(
+ Collections.singletonList(entry.indexFile())),
+ CompactIncrement.emptyIncrement()));
+ }
return allMessages;
}
private static Iterator<byte[]> buildSortedIndex(
Iterator<InternalRow> input,
- byte[] serializedBuilder,
+ byte[] serializedWriter,
Range range,
int partitionKeyNum,
- byte[] partitionBytes)
+ byte[] partitionBytes,
+ long scanSnapshotId)
throws IOException, ClassNotFoundException {
final BinaryRowSerializer binaryRowSerializer = new BinaryRowSerializer(partitionKeyNum);
BinaryRow partition = binaryRowSerializer.deserializeFromBytes(partitionBytes);
- SortedGlobalIndexBuilder builder =
+ SortedGlobalIndexWriter writer =
InstantiationUtil.deserializeObject(
- serializedBuilder, SortedGlobalIndexBuilder.class.getClassLoader());
+ serializedWriter, SortedGlobalIndexWriter.class.getClassLoader());
return CommitMessageSerializer.serializeAll(
- builder.buildForSinglePartition(range, partition, input))
+ writer.buildForSinglePartition(range, partition, input, scanSnapshotId))
.iterator();
}
}
diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CreateGlobalIndexProcedureTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CreateGlobalIndexProcedureTest.scala
index 9182ff70d5..9f4910cbe0 100644
--- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CreateGlobalIndexProcedureTest.scala
+++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CreateGlobalIndexProcedureTest.scala
@@ -19,11 +19,14 @@
package org.apache.paimon.spark.procedure
import org.apache.paimon.globalindex.{KeySerializer, SortedIndexFileMeta}
+import org.apache.paimon.index.DataEvolutionIndexSourceMeta
+import org.apache.paimon.manifest.IndexManifestEntry
import org.apache.paimon.memory.MemorySlice
import org.apache.paimon.spark.PaimonSparkTestBase
import org.apache.paimon.types.VarCharType
import org.apache.paimon.utils.Range
+import org.apache.spark.sql.Row
import org.apache.spark.sql.paimon.Utils
import org.apache.spark.sql.streaming.StreamTest
@@ -34,6 +37,99 @@ import scala.collection.immutable
class CreateGlobalIndexProcedureTest extends PaimonSparkTestBase with StreamTest {
+ test("refresh btree index after data evolution update") {
+ withTable("T", "S", "P") {
+ spark.sql("""
+ |CREATE TABLE T (id INT, idx INT, payload STRING)
+ |TBLPROPERTIES (
+ | 'bucket' = '-1',
+ | 'global-index.enabled' = 'true',
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true',
+ | 'global-index.column-update-action' = 'IGNORE',
+ | 'btree-index.records-per-range' = '2')
+ |""".stripMargin)
+
+ spark.sql(
+ s"INSERT INTO T VALUES ${(0 until 10).map(i => s"($i, $i, 'p$i')").mkString(",")}"
+ )
+ createBTreeIndex("T", "idx")
+ spark.sql(
+ s"INSERT INTO T VALUES ${(10 until 20).map(i => s"($i, $i, 'p$i')").mkString(",")}"
+ )
+ createBTreeIndex("T", "idx")
+
+ def entriesByRange: Map[String, Seq[IndexManifestEntry]] = {
+ loadTable("T")
+ .store()
+ .newIndexFileHandler()
+ .scan("btree")
+ .asScala
+ .groupBy(
+ entry =>
+ s"${entry.indexFile().globalIndexMeta().rowRangeStart()}:" +
+ s"${entry.indexFile().globalIndexMeta().rowRangeEnd()}")
+ .map { case (range, entries) => range -> entries.toList }
+ }
+
+ def fileNames(entries: Seq[IndexManifestEntry]): Set[String] =
+ entries.map(_.indexFile().fileName()).toSet
+
+ val initial = entriesByRange
+ assert(initial.keySet == Set("0:9", "10:19"))
+ assert(initial("0:9").size > 1)
+ assert(initial("10:19").size > 1)
+ val initialFirstFiles = fileNames(initial("0:9"))
+ val initialSecondFiles = fileNames(initial("10:19"))
+
+ spark.sql("CREATE TABLE S (id INT, idx INT)")
+ spark.sql("INSERT INTO S VALUES (1, 1001)")
+ spark.sql("""
+ |MERGE INTO T
+ |USING S
+ |ON T.id = S.id
+ |WHEN MATCHED THEN UPDATE SET T.idx = S.idx
+ |""".stripMargin)
+ val updateSnapshotId = loadTable("T").snapshotManager().latestSnapshot().id()
+
+ createBTreeIndex("T", "idx")
+ assert(loadTable("T").snapshotManager().latestSnapshot().id() == updateSnapshotId + 1)
+
+ val refreshed = entriesByRange
+ assert(refreshed.keySet == Set("0:9", "10:19"))
+ assert((fileNames(refreshed("0:9")).intersect(initialFirstFiles)).isEmpty)
+ assert(fileNames(refreshed("10:19")) == initialSecondFiles)
+ refreshed("0:9").foreach(
+ entry =>
+ assert(
+ DataEvolutionIndexSourceMeta
+ .fromIndexFile(entry.indexFile())
+ .scanSnapshotId() == updateSnapshotId
+ ))
+ checkAnswer(sql("SELECT id FROM T WHERE idx = 1"), Seq.empty)
+ checkAnswer(sql("SELECT id FROM T WHERE idx = 1001"), Seq(Row(1)))
+
+ val refreshedSnapshotId = loadTable("T").snapshotManager().latestSnapshot().id()
+ val refreshedFiles = refreshed.values.flatten.map(_.indexFile().fileName()).toSet
+ createBTreeIndex("T", "idx")
+ assert(loadTable("T").snapshotManager().latestSnapshot().id() == refreshedSnapshotId)
+ assert(entriesByRange.values.flatten.map(_.indexFile().fileName()).toSet == refreshedFiles)
+
+ spark.sql("CREATE TABLE P (id INT, payload STRING)")
+ spark.sql("INSERT INTO P VALUES (1, 'new-payload')")
+ spark.sql("""
+ |MERGE INTO T
+ |USING P
+ |ON T.id = P.id
+ |WHEN MATCHED THEN UPDATE SET T.payload = P.payload
+ |""".stripMargin)
+ val payloadUpdateSnapshotId = loadTable("T").snapshotManager().latestSnapshot().id()
+ createBTreeIndex("T", "idx")
+ assert(loadTable("T").snapshotManager().latestSnapshot().id() == payloadUpdateSnapshotId)
+ assert(entriesByRange.values.flatten.map(_.indexFile().fileName()).toSet == refreshedFiles)
+ }
+ }
+
test("create btree global index") {
withTable("T") {
spark.sql("""
@@ -286,4 +382,12 @@ class CreateGlobalIndexProcedureTest extends PaimonSparkTestBase with StreamTest
}
}
}
+
+ private def createBTreeIndex(tableName: String, column: String): Unit = {
+ spark
+ .sql(
+ s"CALL sys.create_global_index(table => 'test.$tableName', " +
+ s"index_column => '$column', index_type => 'btree')")
+ .collect()
+ }
}
diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/FullTextSearchTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/FullTextSearchTest.scala
index 5777f867fb..4c131e1f03 100644
--- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/FullTextSearchTest.scala
+++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/FullTextSearchTest.scala
@@ -18,6 +18,7 @@
package org.apache.paimon.spark.sql
+import org.apache.paimon.index.DataEvolutionIndexSourceMeta
import org.apache.paimon.spark.PaimonSparkTestBase
import scala.collection.JavaConverters._
@@ -44,6 +45,7 @@ class FullTextSearchTest extends PaimonSparkTestBase {
.map(i => s"($i, 'document number $i about paimon lake format')")
.mkString(",")
spark.sql(s"INSERT INTO T VALUES $values")
+ val scanSnapshotId = loadTable("T").snapshotManager().latestSnapshot().id()
val output = spark
.sql(
@@ -61,6 +63,11 @@ class FullTextSearchTest extends PaimonSparkTestBase {
.filter(_.indexFile().indexType() == indexType)
assert(indexEntries.nonEmpty)
+ assert(
+ indexEntries.forall(
+ entry =>
+ DataEvolutionIndexSourceMeta.fromIndexFile(entry.indexFile()).scanSnapshotId() ==
+ scanSnapshotId))
val totalRowCount = indexEntries.map(_.indexFile().rowCount()).sum
assert(totalRowCount == 100L)
}
diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/LuminaVectorIndexTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/LuminaVectorIndexTest.scala
index 6f079d4dcf..3cc2339704 100644
--- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/LuminaVectorIndexTest.scala
+++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/LuminaVectorIndexTest.scala
@@ -18,6 +18,7 @@
package org.apache.paimon.spark.sql
+import org.apache.paimon.index.DataEvolutionIndexSourceMeta
import org.apache.paimon.spark.PaimonSparkTestBase
import scala.collection.JavaConverters._
@@ -69,6 +70,54 @@ class LuminaVectorIndexTest extends PaimonSparkTestBase {
}
}
+ test("create lumina vector index after adding the index column") {
+ withTable("T") {
+ spark.sql("""
+ |CREATE TABLE T (id INT)
+ |TBLPROPERTIES (
+ | 'bucket' = '-1',
+ | 'global-index.row-count-per-shard' = '10000',
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true')
+ |""".stripMargin)
+ spark.sql("INSERT INTO T VALUES (0), (1), (2), (3), (4)")
+
+ spark.sql("ALTER TABLE T ADD COLUMNS (v ARRAY<FLOAT>)")
+ spark.sql("""
+ |INSERT INTO T VALUES
+ | (5, array(5.0f, 6.0f, 7.0f)),
+ | (6, array(6.0f, 7.0f, 8.0f)),
+ | (7, array(7.0f, 8.0f, 9.0f)),
+ | (8, array(8.0f, 9.0f, 10.0f)),
+ | (9, array(9.0f, 10.0f, 11.0f))
+ |""".stripMargin)
+
+ val output = spark
+ .sql(
+ s"CALL sys.create_global_index(table => 'test.T', index_column => 'v', index_type => '$indexType', options => '$defaultOptions')")
+ .collect()
+ .head
+ assert(output.getBoolean(0))
+
+ val indexEntries =
+ loadTable("T").store().newIndexFileHandler().scan(indexType).asScala
+ assert(indexEntries.size == 1)
+ val indexFile = indexEntries.head.indexFile()
+ assert(indexFile.rowCount() == 10L)
+ assert(indexFile.globalIndexMeta().rowRangeStart() == 0L)
+ assert(indexFile.globalIndexMeta().rowRangeEnd() == 9L)
+
+ val searchResult = spark
+ .sql("""
+ |SELECT id FROM vector_search(
+ | 'T', 'v', array(5.0f, 6.0f, 7.0f), 5,
+ | map('refine_factor', '5'))
+ |""".stripMargin)
+ .collect()
+ assert(searchResult.map(_.getInt(0)).toSet == (5 to 9).toSet)
+ }
+ }
+
test("create lumina vector index - legacy index type") {
withTable("T") {
spark.sql("""
@@ -406,4 +455,90 @@ class LuminaVectorIndexTest extends PaimonSparkTestBase {
assert(searchResult.length == 10)
}
}
+
+ test("incremental build atomically refreshes vectors updated inside an indexed range") {
+ withTable("T") {
+ spark.sql("""
+ |CREATE TABLE T (id INT, v ARRAY<FLOAT>)
+ |TBLPROPERTIES (
+ | 'bucket' = '-1',
+ | 'global-index.row-count-per-shard' = '10000',
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true',
+ | 'lumina.distance.metric' = 'l2',
+ | 'global-index.column-update-action' = 'IGNORE')
+ |""".stripMargin)
+
+ val values = (0 until 200)
+ .map {
+ case i if i < 100 => s"($i, cast(NULL as ARRAY<FLOAT>))"
+ case i =>
+ s"($i, array(cast($i as float), cast(${i + 1} as float), cast(${i + 2} as float)))"
+ }
+ .mkString(",")
+ spark.sql(s"INSERT INTO T VALUES $values")
+
+ val table = loadTable("T")
+ val firstDataSnapshotId = table.snapshotManager().latestSnapshot().id()
+ spark.sql(
+ s"CALL sys.create_global_index(table => 'test.T', index_column => 'v', " +
+ s"index_type => '$indexType', options => '$defaultOptions')")
+
+ val oldEntries = table.store().newIndexFileHandler().scan(indexType).asScala
+ assert(oldEntries.nonEmpty)
+ assert(
+ oldEntries.forall(
+ entry =>
+ DataEvolutionIndexSourceMeta.fromIndexFile(entry.indexFile()).scanSnapshotId() ==
+ firstDataSnapshotId))
+ val oldFileNames = oldEntries.map(_.indexFile().fileName()).toSet
+
+ spark.sql("""
+ |MERGE INTO T
+ |USING (SELECT 0 AS id,
+ | array(cast(0 as float), cast(1 as float), cast(2 as float)) AS v) S
+ |ON T.id = S.id
+ |WHEN MATCHED THEN UPDATE SET v = S.v
+ |""".stripMargin)
+
+ val updatedSnapshot = table.snapshotManager().latestSnapshot()
+ val stillVisible =
+ table.store().newIndexFileHandler().scan(updatedSnapshot, indexType).asScala
+ assert(stillVisible.map(_.indexFile().fileName()).toSet == oldFileNames)
+ // Retrieve every indexed candidate before exact reranking so ANN recall is deterministic.
+ val beforeRefresh = spark
+ .sql("""
+ |SELECT id FROM vector_search(
+ | 'T', 'v', array(0.0f, 1.0f, 2.0f), 1,
+ | map('refine_factor', '200'))
+ |""".stripMargin)
+ .collect()
+ assert(beforeRefresh.head.getInt(0) != 0)
+
+ spark.sql(
+ s"CALL sys.create_global_index(table => 'test.T', index_column => 'v', " +
+ s"index_type => '$indexType', options => '$defaultOptions')")
+
+ val indexCommitSnapshot = table.snapshotManager().latestSnapshot()
+ assert(indexCommitSnapshot.id() == updatedSnapshot.id() + 1)
+ val refreshedEntries =
+ table.store().newIndexFileHandler().scan(indexCommitSnapshot, indexType).asScala
+ assert(refreshedEntries.nonEmpty)
+ assert(refreshedEntries.map(_.indexFile().fileName()).toSet.intersect(oldFileNames).isEmpty)
+ assert(
+ refreshedEntries.forall(
+ entry =>
+ DataEvolutionIndexSourceMeta.fromIndexFile(entry.indexFile()).scanSnapshotId() ==
+ updatedSnapshot.id()))
+
+ val afterRefresh = spark
+ .sql("""
+ |SELECT id FROM vector_search(
+ | 'T', 'v', array(0.0f, 1.0f, 2.0f), 1,
+ | map('refine_factor', '200'))
+ |""".stripMargin)
+ .collect()
+ assert(afterRefresh.head.getInt(0) == 0)
+ }
+ }
}