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 d55fabffcf [core] Reduce memory for update index (#9092)
d55fabffcf is described below
commit d55fabffcfd7038ca33e1f2b9eda912e79da175b
Author: YeJunHao <[email protected]>
AuthorDate: Fri Aug 7 19:20:17 2026 +0800
[core] Reduce memory for update index (#9092)
---
.../DataEvolutionGlobalIndexRefreshPlanner.java | 325 ++++++++++++++++++---
.../sorted/SortedGlobalIndexScanner.java | 15 +-
.../sorted/SortedGlobalIndexScannerTest.java | 124 ++++++++
3 files changed, 419 insertions(+), 45 deletions(-)
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
index 44ed39685d..36e11b38f4 100644
---
a/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexRefreshPlanner.java
+++
b/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexRefreshPlanner.java
@@ -18,21 +18,34 @@
package org.apache.paimon.globalindex;
+import org.apache.paimon.Snapshot;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.index.DataEvolutionIndexSourceMeta;
import org.apache.paimon.index.GlobalIndexMeta;
+import org.apache.paimon.io.BinaryDataFileMeta;
import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.manifest.BinaryManifestEntry;
+import org.apache.paimon.manifest.DeletedIdentifierSet;
import org.apache.paimon.manifest.FileKind;
import org.apache.paimon.manifest.IndexManifestEntry;
import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.manifest.ManifestFile;
+import org.apache.paimon.manifest.ManifestFileMeta;
+import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.source.ScanMode;
import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.CloseableIterator;
import org.apache.paimon.utils.Pair;
import org.apache.paimon.utils.Range;
+import javax.annotation.Nullable;
+
import java.util.ArrayList;
import java.util.Arrays;
-import java.util.Comparator;
+import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -48,16 +61,91 @@ public final class DataEvolutionGlobalIndexRefreshPlanner {
private DataEvolutionGlobalIndexRefreshPlanner() {}
+ /**
+ * Consumes data entries one by one. Entries are never retained; only
merged row ranges bucketed
+ * by distinct scan sequence numbers are kept in memory.
+ */
public static List<IndexManifestEntry> findIndexesToRefresh(
SchemaManager schemaManager,
- List<ManifestEntry> dataEntries,
+ Iterable<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 =
+ collectRefreshGroups(indexEntries, indexedFields);
+ if (groups.isEmpty()) {
+ return Collections.emptyList();
+ }
+
+ Set<Integer> indexedFieldIds = indexedFieldIds(indexedFields);
+ 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;
+ }
+
+ addIfUpdatesIndexedFields(
+ schemaManager, fileFieldIdsCache, indexedFieldIds, group,
file);
+ }
+
+ return collectMarkedIndexes(groups, indexEntries);
+ }
+
+ /**
+ * Scans data manifests through reusable {@link BinaryManifestEntry} views
and plans indexes to
+ * refresh. Neither {@link ManifestEntry} nor {@link DataFileMeta} POJOs
are materialized: one
+ * narrow DELETE pass tracks removed files in a primitive identifier set,
then one projected ADD
+ * pass merges updated row ranges directly into the refresh groups.
+ */
+ public static List<IndexManifestEntry> findIndexesToRefresh(
+ FileStoreTable table,
+ Snapshot snapshot,
+ @Nullable PartitionPredicate partitionPredicate,
+ List<IndexManifestEntry> indexEntries,
+ List<DataField> indexedFields) {
+ Map<Pair<BinaryRow, Integer>, RefreshGroup> groups =
+ collectRefreshGroups(indexEntries, indexedFields);
+ if (groups.isEmpty()) {
+ return Collections.emptyList();
+ }
+
+ List<ManifestFileMeta> manifests =
+ table.store()
+ .newScan()
+ .withPartitionFilter(partitionPredicate)
+ .manifestsReader()
+ .read(snapshot, ScanMode.ALL)
+ .filteredManifests;
+ ManifestFile manifestFile =
table.store().manifestFileFactory().create();
+ Set<BinaryRow> groupPartitions = new HashSet<>();
+ for (Pair<BinaryRow, Integer> key : groups.keySet()) {
+ groupPartitions.add(key.getLeft());
}
+ DeletedIdentifierSet deleted = new DeletedIdentifierSet();
+ try {
+ collectDeletedIdentifiers(manifestFile, manifests,
groupPartitions, deleted);
+ collectUpdatedRanges(
+ table.schemaManager(),
+ manifestFile,
+ manifests,
+ deleted,
+ groups,
+ indexedFieldIds(indexedFields));
+ } finally {
+ deleted.release();
+ }
+
+ return collectMarkedIndexes(groups, indexEntries);
+ }
+
+ private static Map<Pair<BinaryRow, Integer>, RefreshGroup>
collectRefreshGroups(
+ List<IndexManifestEntry> indexEntries, List<DataField>
indexedFields) {
Map<Pair<BinaryRow, Integer>, RefreshGroup> groups = new HashMap<>();
for (int i = 0; i < indexEntries.size(); i++) {
IndexManifestEntry indexEntry = indexEntries.get(i);
@@ -80,28 +168,135 @@ public final class DataEvolutionGlobalIndexRefreshPlanner
{
key -> new RefreshGroup())
.addIndex(i, indexMeta.rowRange(), scanSnapshotId);
}
+ for (RefreshGroup group : groups.values()) {
+ group.finishAddingIndexes();
+ }
+ return groups;
+ }
- 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)
{
+ private static void collectDeletedIdentifiers(
+ ManifestFile manifestFile,
+ List<ManifestFileMeta> manifests,
+ Set<BinaryRow> groupPartitions,
+ DeletedIdentifierSet deleted) {
+ for (ManifestFileMeta manifest : manifests) {
+ if (manifest.numDeletedFiles() <= 0) {
continue;
}
+ try (CloseableIterator<BinaryManifestEntry> entries =
+ manifestFile.scan(
+ manifest.fileName(),
+ manifest.fileSize(),
+ BinaryManifestEntry.DELETE_ENTRY_PROJECTION)) {
+ while (entries.hasNext()) {
+ BinaryManifestEntry entry = entries.next();
+ if (entry.isDelete() &&
groupPartitions.contains(entry.partition())) {
+ deleted.add(entry);
+ }
+ }
+ } catch (Exception e) {
+ throw manifestScanException(manifest, e);
+ }
+ }
+ }
- RefreshGroup group = groups.get(Pair.of(dataEntry.partition(),
dataEntry.bucket()));
- if (group == null || !group.mayContainUpdate(file)) {
+ private static void collectUpdatedRanges(
+ SchemaManager schemaManager,
+ ManifestFile manifestFile,
+ List<ManifestFileMeta> manifests,
+ DeletedIdentifierSet deleted,
+ Map<Pair<BinaryRow, Integer>, RefreshGroup> groups,
+ Set<Integer> indexedFieldIds) {
+ Map<Pair<Long, List<String>>, Set<Integer>> fileFieldIdsCache = new
HashMap<>();
+ BinaryManifestEntry.Projection projection =
addedEntryProjection(!deleted.isEmpty());
+ for (ManifestFileMeta manifest : manifests) {
+ if (manifest.numAddedFiles() <= 0) {
continue;
}
-
- Set<Integer> physicalFieldIds =
- fileFieldIdsCache.computeIfAbsent(
- Pair.of(file.schemaId(), file.writeCols()),
- key -> fileFieldIds(schemaManager::schema, file));
- if (!disjoint(indexedFieldIds, physicalFieldIds)) {
- group.addDataFile(file);
+ try (CloseableIterator<BinaryManifestEntry> entries =
+ manifestFile.scan(manifest.fileName(),
manifest.fileSize(), projection)) {
+ while (entries.hasNext()) {
+ BinaryManifestEntry entry = entries.next();
+ if (!entry.isAdd()) {
+ continue;
+ }
+ BinaryDataFileMeta file = entry.file();
+ if (!file.hasFirstRowId()) {
+ continue;
+ }
+ RefreshGroup group = groups.get(Pair.of(entry.partition(),
entry.bucket()));
+ if (group == null || !group.mayContainUpdate(file)) {
+ continue;
+ }
+ if (!deleted.isEmpty() && deleted.contains(entry)) {
+ continue;
+ }
+ addIfUpdatesIndexedFields(
+ schemaManager, fileFieldIdsCache, indexedFieldIds,
group, file);
+ }
+ } catch (Exception e) {
+ throw manifestScanException(manifest, e);
}
}
+ }
+
+ private static void addIfUpdatesIndexedFields(
+ SchemaManager schemaManager,
+ Map<Pair<Long, List<String>>, Set<Integer>> fileFieldIdsCache,
+ Set<Integer> indexedFieldIds,
+ RefreshGroup group,
+ DataFileMeta file) {
+ Set<Integer> physicalFieldIds =
+ fileFieldIdsCache.computeIfAbsent(
+ Pair.of(file.schemaId(), file.writeCols()),
+ key -> fileFieldIds(schemaManager::schema, file));
+ if (!disjoint(indexedFieldIds, physicalFieldIds)) {
+ group.addUpdatedFile(file.maxSequenceNumber(),
file.nonNullRowIdRange());
+ }
+ }
+
+ /**
+ * Projects only the fields the refresh planner consumes; identifier
fields are included only
+ * when deleted files must be recognized.
+ */
+ private static BinaryManifestEntry.Projection addedEntryProjection(
+ boolean includeIdentifierFields) {
+ List<DataField> fileFields = new ArrayList<>();
+ fileFields.add(DataFileMeta.SCHEMA.getField(DataFileMeta.ROW_COUNT));
+
fileFields.add(DataFileMeta.SCHEMA.getField(DataFileMeta.MAX_SEQUENCE_NUMBER));
+ fileFields.add(DataFileMeta.SCHEMA.getField(DataFileMeta.SCHEMA_ID));
+
fileFields.add(DataFileMeta.SCHEMA.getField(DataFileMeta.FIRST_ROW_ID));
+ fileFields.add(DataFileMeta.SCHEMA.getField(DataFileMeta.WRITE_COLS));
+ if (includeIdentifierFields) {
+
fileFields.add(DataFileMeta.SCHEMA.getField(DataFileMeta.FILE_NAME));
+ fileFields.add(DataFileMeta.SCHEMA.getField(DataFileMeta.LEVEL));
+
fileFields.add(DataFileMeta.SCHEMA.getField(DataFileMeta.EXTRA_FILES));
+
fileFields.add(DataFileMeta.SCHEMA.getField(DataFileMeta.EMBEDDED_FILE_INDEX));
+
fileFields.add(DataFileMeta.SCHEMA.getField(DataFileMeta.EXTERNAL_PATH));
+ }
+
+ List<DataField> fields = new ArrayList<>();
+
fields.add(ManifestEntry.MANIFEST_ROW_TYPE.getField(ManifestEntry.KIND));
+
fields.add(ManifestEntry.MANIFEST_ROW_TYPE.getField(ManifestEntry.PARTITION));
+
fields.add(ManifestEntry.MANIFEST_ROW_TYPE.getField(ManifestEntry.BUCKET));
+ fields.add(
+ ManifestEntry.MANIFEST_ROW_TYPE
+ .getField(ManifestEntry.FILE)
+ .newType(new RowType(false, fileFields)));
+ return BinaryManifestEntry.Projection.create(new RowType(false,
fields));
+ }
+
+ private static Set<Integer> indexedFieldIds(List<DataField> indexedFields)
{
+ Set<Integer> indexedFieldIds = new HashSet<>();
+ for (DataField field : indexedFields) {
+ indexedFieldIds.add(field.id());
+ }
+ return indexedFieldIds;
+ }
+ private static List<IndexManifestEntry> collectMarkedIndexes(
+ Map<Pair<BinaryRow, Integer>, RefreshGroup> groups,
+ List<IndexManifestEntry> indexEntries) {
boolean[] indexesToRefresh = new boolean[indexEntries.size()];
for (RefreshGroup group : groups.values()) {
group.markIndexesToRefresh(indexesToRefresh);
@@ -116,42 +311,88 @@ public final class DataEvolutionGlobalIndexRefreshPlanner
{
return result;
}
+ private static RuntimeException manifestScanException(
+ ManifestFileMeta manifest, Exception cause) {
+ return new RuntimeException("Failed to scan manifest " +
manifest.fileName(), cause);
+ }
+
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;
+ // Distinct scan sequence numbers in descending order. Bucket i merges
row ranges of updated
+ // data files whose max sequence number lies in
+ // (sequenceNumbers[i], sequenceNumbers[i - 1]].
+ private long[] sequenceNumbers;
+ private MergedRanges[] updatedRangesPerSequenceNumber;
+
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 void finishAddingIndexes() {
+ indexes.sort((left, right) -> Long.compare(right.scanSnapshotId,
left.scanSnapshotId));
+ long[] distinct = new long[indexes.size()];
+ int size = 0;
+ for (IndexQuery index : indexes) {
+ if (size == 0 || distinct[size - 1] != index.scanSnapshotId) {
+ distinct[size++] = index.scanSnapshotId;
+ }
+ }
+ this.sequenceNumbers = Arrays.copyOf(distinct, size);
+ this.updatedRangesPerSequenceNumber = new MergedRanges[size];
+ }
+
private boolean mayContainUpdate(DataFileMeta file) {
return file.maxSequenceNumber() > minScanSnapshotId
&& indexedRanges.intersects(file.nonNullRowIdRange());
}
- private void addDataFile(DataFileMeta file) {
- dataFiles.add(file);
+ private void addUpdatedFile(long maxSequenceNumber, Range rowRange) {
+ // Merge the range eagerly instead of retaining the file metadata.
+ int sequenceNumberIndex =
firstIndexWithSequenceNumberBelow(maxSequenceNumber);
+ if (updatedRangesPerSequenceNumber[sequenceNumberIndex] == null) {
+ updatedRangesPerSequenceNumber[sequenceNumberIndex] = new
MergedRanges();
+ }
+ updatedRangesPerSequenceNumber[sequenceNumberIndex].add(rowRange);
}
- 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));
+ /** Returns the first position whose scan sequence number is below the
maximum sequence. */
+ private int firstIndexWithSequenceNumberBelow(long maxSequenceNumber) {
+ // mayContainUpdate guarantees the last sequence number qualifies.
+ int low = 0;
+ int high = sequenceNumbers.length - 1;
+ while (low < high) {
+ int mid = (low + high) >>> 1;
+ if (sequenceNumbers[mid] < maxSequenceNumber) {
+ high = mid;
+ } else {
+ low = mid + 1;
+ }
+ }
+ return low;
+ }
- MergedRanges updatedRanges = new MergedRanges();
- int nextFile = 0;
+ private void markIndexesToRefresh(boolean[] result) {
+ // As scan sequence numbers decrease, eligible updated ranges only
grow.
+ MergedRanges updatedRanges = null;
+ int nextSequenceNumberIndex = 0;
for (IndexQuery index : indexes) {
- while (nextFile < dataFiles.size()
- && dataFiles.get(nextFile).maxSequenceNumber() >
index.scanSnapshotId) {
-
updatedRanges.add(dataFiles.get(nextFile).nonNullRowIdRange());
- nextFile++;
+ while (nextSequenceNumberIndex < sequenceNumbers.length
+ && sequenceNumbers[nextSequenceNumberIndex] >=
index.scanSnapshotId) {
+ MergedRanges ranges =
updatedRangesPerSequenceNumber[nextSequenceNumberIndex];
+ if (ranges != null) {
+ updatedRanges =
+ updatedRanges == null ? ranges :
updatedRanges.merge(ranges);
+
updatedRangesPerSequenceNumber[nextSequenceNumberIndex] = null;
+ }
+ nextSequenceNumberIndex++;
}
- if (updatedRanges.intersects(index.rowRange)) {
+ if (updatedRanges != null &&
updatedRanges.intersects(index.rowRange)) {
result[index.ordinal] = true;
}
}
@@ -177,9 +418,10 @@ public final class DataEvolutionGlobalIndexRefreshPlanner {
private final NavigableMap<Long, Long> ranges = new TreeMap<>();
private void add(Range range) {
- long from = range.from;
- long to = range.to;
+ add(range.from, range.to);
+ }
+ private void add(long from, long to) {
Map.Entry<Long, Long> floor = ranges.floorEntry(from);
if (floor != null && floor.getValue() >= from) {
from = floor.getKey();
@@ -196,6 +438,21 @@ public final class DataEvolutionGlobalIndexRefreshPlanner {
ranges.put(from, to);
}
+ /** Merges the smaller range set into the larger one and clears the
source. */
+ private MergedRanges merge(MergedRanges other) {
+ MergedRanges target = this;
+ MergedRanges source = other;
+ if (target.ranges.size() < source.ranges.size()) {
+ target = other;
+ source = this;
+ }
+ for (Map.Entry<Long, Long> range : source.ranges.entrySet()) {
+ target.add(range.getKey(), range.getValue());
+ }
+ source.ranges.clear();
+ return target;
+ }
+
private boolean intersects(Range range) {
Map.Entry<Long, Long> floor = ranges.floorEntry(range.to);
return floor != null && floor.getValue() >= range.from;
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
index e1417dd90c..7f44a33451 100644
---
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
@@ -23,7 +23,6 @@ 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;
@@ -144,18 +143,12 @@ public class SortedGlobalIndexScanner implements
Serializable {
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();
+ // Scans data manifests through reusable binary views without
materializing entries.
deletedIndexEntries =
DataEvolutionGlobalIndexRefreshPlanner.findIndexesToRefresh(
- table.schemaManager(),
- dataEntries,
+ table,
+ snapshot,
+ partitionPredicate,
currentIndexes,
Collections.singletonList(indexField));
for (IndexManifestEntry entry : deletedIndexEntries) {
diff --git
a/paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexScannerTest.java
b/paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexScannerTest.java
index f918b17289..ab2b87d93b 100644
---
a/paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexScannerTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexScannerTest.java
@@ -20,10 +20,13 @@ package org.apache.paimon.globalindex.sorted;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.Snapshot;
+import org.apache.paimon.append.dataevolution.DataEvolutionCompactCoordinator;
+import org.apache.paimon.append.dataevolution.DataEvolutionCompactTask;
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.globalindex.DataEvolutionGlobalIndexRefreshPlanner;
import org.apache.paimon.globalindex.KeySerializer;
import org.apache.paimon.globalindex.ScanResult;
import org.apache.paimon.globalindex.btree.BTreeIndexOptions;
@@ -44,7 +47,10 @@ 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.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.CommitMessageImpl;
import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.EndOfScanException;
+import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.Pair;
@@ -53,6 +59,7 @@ import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
@@ -344,6 +351,123 @@ public class SortedGlobalIndexScannerTest extends
TableTestBase {
"incrementalScan should only return the new rows in partition
p0");
}
+ @Test
+ public void testIncrementalScanRefreshesIndexAfterColumnUpdate() throws
Exception {
+ write();
+ createIndex(null);
+
+ FileStoreTable table =
+ getTableDefault()
+ .copy(
+ Collections.singletonMap(
+
CoreOptions.GLOBAL_INDEX_COLUMN_UPDATE_ACTION.key(),
+ "IGNORE"));
+ updateColumn(table, 0L, 5);
+
+ SortedGlobalIndexScanner scanner =
+ new SortedGlobalIndexScanner(table,
"btree").withIndexField("f0");
+ ScanResult<DataSplit> scanResult =
+ scanner.incrementalScan()
+ .orElseThrow(
+ () ->
+ new IllegalStateException(
+ "Expected incremental scan
result after column update."));
+
+ assertThat(scanResult.deletedIndexEntries()).isNotEmpty();
+ assertThat(scanResult.deletedIndexEntries())
+ .allSatisfy(
+ entry ->
+
assertThat(entry.indexFile().globalIndexMeta().rowRange().from)
+ .isEqualTo(0L));
+ }
+
+ @Test
+ public void testFindIndexesToRefreshBinaryScanMatchesEntryScan() throws
Exception {
+ write();
+ createIndex(null);
+ FileStoreTable table = getTableDefault();
+
+ // Column update overlapping indexed rows must be planned identically
by both paths.
+ // Cover the whole base file so the follow-up compaction can merge the
two files.
+ updateColumn(table, 0L, (int) PART_ROW_NUM);
+ assertThat(refreshPlanParity(table)).isNotEmpty();
+
+ // Compaction produces DELETE manifest entries which the binary scan
must recognize.
+ doDataEvolutionCompact(table);
+ updateColumn(table, 100L, 5);
+ refreshPlanParity(table);
+ }
+
+ private List<IndexManifestEntry> refreshPlanParity(FileStoreTable table) {
+ Snapshot snapshot = table.snapshotManager().latestSnapshot();
+ List<IndexManifestEntry> currentIndexes =
+ table.store().newIndexFileHandler().scan(snapshot, "btree");
+ List<DataField> indexedFields =
Collections.singletonList(table.rowType().getField("f0"));
+
+ List<IndexManifestEntry> viaEntries =
+ DataEvolutionGlobalIndexRefreshPlanner.findIndexesToRefresh(
+ table.schemaManager(),
+
table.store().newScan().withSnapshot(snapshot).plan().files(),
+ currentIndexes,
+ indexedFields);
+ List<IndexManifestEntry> viaBinaryScan =
+ DataEvolutionGlobalIndexRefreshPlanner.findIndexesToRefresh(
+ table, snapshot, null, currentIndexes, indexedFields);
+
+ assertThat(viaBinaryScan).containsExactlyElementsOf(viaEntries);
+ return viaBinaryScan;
+ }
+
+ private void updateColumn(FileStoreTable table, long firstRowId, int
rowCount)
+ throws Exception {
+ RowType writeType = table.rowType().project(Arrays.asList("dt", "f0"));
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write =
builder.newWrite().withWriteType(writeType)) {
+ for (int i = 0; i < rowCount; i++) {
+ write.write(GenericRow.of(BinaryString.fromString("p0"),
-(int) (firstRowId + i)));
+ }
+ List<CommitMessage> messages = write.prepareCommit();
+ setFirstRowId(messages, firstRowId);
+ try (BatchTableCommit commit = builder.newCommit()) {
+ commit.commit(messages);
+ }
+ }
+ }
+
+ private void setFirstRowId(List<CommitMessage> messages, long firstRowId) {
+ for (CommitMessage message : messages) {
+ CommitMessageImpl commitMessage = (CommitMessageImpl) message;
+ List<DataFileMeta> newFiles =
+ new
ArrayList<>(commitMessage.newFilesIncrement().newFiles());
+ commitMessage.newFilesIncrement().newFiles().clear();
+ for (DataFileMeta newFile : newFiles) {
+ commitMessage
+ .newFilesIncrement()
+ .newFiles()
+ .add(newFile.assignFirstRowId(firstRowId));
+ }
+ }
+ }
+
+ private void doDataEvolutionCompact(FileStoreTable table) throws Exception
{
+ DataEvolutionCompactCoordinator coordinator =
+ new DataEvolutionCompactCoordinator(
+ table, false, false, table.latestSnapshot().get());
+ List<CommitMessage> messages = new ArrayList<>();
+ try {
+ List<DataEvolutionCompactTask> tasks;
+ while (!(tasks = coordinator.plan()).isEmpty()) {
+ for (DataEvolutionCompactTask task : tasks) {
+ messages.add(task.doCompact(table, "test-compact"));
+ }
+ }
+ } catch (EndOfScanException ignore) {
+ }
+ if (!messages.isEmpty()) {
+ table.newBatchWriteBuilder().newCommit().commit(messages);
+ }
+ }
+
@Test
public void testScanFiltersBlobFilesByManifestEntryFilter() throws
Exception {
Schema.Builder schemaBuilder = Schema.newBuilder();
