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 cdf567523a [core] Avoid global index rebuilds for unrelated column
compaction (#9069)
cdf567523a is described below
commit cdf567523ab7fc0d63ca41921bd61411d509036a
Author: YeJunHao <[email protected]>
AuthorDate: Thu Aug 20 19:27:18 2026 +0800
[core] Avoid global index rebuilds for unrelated column compaction (#9069)
---
.../DataEvolutionGlobalIndexRefreshPlanner.java | 59 ++++++----
...DataEvolutionGlobalIndexRefreshPlannerTest.java | 121 +++++++++++++++++++++
.../sorted/SortedGlobalIndexScannerTest.java | 118 ++++++++++++++++++++
3 files changed, 276 insertions(+), 22 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 924749f4c8..d3dbb2c276 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
@@ -33,6 +33,7 @@ import org.apache.paimon.manifest.ManifestFileMeta;
import org.apache.paimon.manifest.ProjectedManifestEntry;
import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.source.ScanMode;
import org.apache.paimon.types.DataField;
@@ -53,8 +54,10 @@ import java.util.Map;
import java.util.NavigableMap;
import java.util.Set;
import java.util.TreeMap;
+import java.util.function.Function;
-import static org.apache.paimon.utils.DataEvolutionUtils.fileFieldIds;
+import static
org.apache.paimon.utils.DataEvolutionUtils.fieldMaxSequenceNumber;
+import static org.apache.paimon.utils.DataEvolutionUtils.fileFields;
/** Plans existing global index files which need refresh after data-evolution
updates. */
public final class DataEvolutionGlobalIndexRefreshPlanner {
@@ -77,7 +80,8 @@ public final class DataEvolutionGlobalIndexRefreshPlanner {
}
Set<Integer> indexedFieldIds = indexedFieldIds(indexedFields);
- Map<Pair<Long, List<String>>, Set<Integer>> fileFieldIdsCache = new
HashMap<>();
+ Function<Long, TableSchema> schemaLoader =
cachedSchemaLoader(schemaManager);
+ Map<Pair<Long, List<String>>, List<DataField>> fileFieldsCache = new
HashMap<>();
for (ManifestEntry dataEntry : dataEntries) {
DataFileMeta file = dataEntry.file();
if (dataEntry.kind() != FileKind.ADD || file.firstRowId() == null)
{
@@ -89,8 +93,7 @@ public final class DataEvolutionGlobalIndexRefreshPlanner {
continue;
}
- addIfUpdatesIndexedFields(
- schemaManager, fileFieldIdsCache, indexedFieldIds, group,
file);
+ addIfUpdatesIndexedFields(schemaLoader, fileFieldsCache,
indexedFieldIds, group, file);
}
return collectMarkedIndexes(groups, indexEntries);
@@ -205,7 +208,8 @@ public final class DataEvolutionGlobalIndexRefreshPlanner {
CompactFileIdentifierSet deleted,
Map<Pair<BinaryRow, Integer>, RefreshGroup> groups,
Set<Integer> indexedFieldIds) {
- Map<Pair<Long, List<String>>, Set<Integer>> fileFieldIdsCache = new
HashMap<>();
+ Function<Long, TableSchema> schemaLoader =
cachedSchemaLoader(schemaManager);
+ Map<Pair<Long, List<String>>, List<DataField>> fileFieldsCache = new
HashMap<>();
ProjectedManifestEntry.Projection projection =
addedEntryProjection(!deleted.isEmpty());
for (ManifestFileMeta manifest : manifests) {
if (manifest.numAddedFiles() <= 0) {
@@ -230,7 +234,7 @@ public final class DataEvolutionGlobalIndexRefreshPlanner {
continue;
}
addIfUpdatesIndexedFields(
- schemaManager, fileFieldIdsCache, indexedFieldIds,
group, file);
+ schemaLoader, fileFieldsCache, indexedFieldIds,
group, file);
}
} catch (Exception e) {
throw manifestScanException(manifest, e);
@@ -239,20 +243,36 @@ public final class DataEvolutionGlobalIndexRefreshPlanner
{
}
private static void addIfUpdatesIndexedFields(
- SchemaManager schemaManager,
- Map<Pair<Long, List<String>>, Set<Integer>> fileFieldIdsCache,
+ Function<Long, TableSchema> schemaLoader,
+ Map<Pair<Long, List<String>>, List<DataField>> fileFieldsCache,
Set<Integer> indexedFieldIds,
RefreshGroup group,
DataFileMeta file) {
- Set<Integer> physicalFieldIds =
- fileFieldIdsCache.computeIfAbsent(
+ List<DataField> physicalFields =
+ fileFieldsCache.computeIfAbsent(
Pair.of(file.schemaId(), file.writeCols()),
- key -> fileFieldIds(schemaManager::schema, file));
- if (!disjoint(indexedFieldIds, physicalFieldIds)) {
- group.addUpdatedFile(file.maxSequenceNumber(),
file.nonNullRowIdRange());
+ key -> fileFields(schemaLoader, file));
+ long[] columnSequences = file.columnMaxSequenceNumbers();
+ long indexedMaxSequence = Long.MIN_VALUE;
+ for (int position = 0; position < physicalFields.size(); position++) {
+ if (indexedFieldIds.contains(physicalFields.get(position).id())) {
+ indexedMaxSequence =
+ Math.max(
+ indexedMaxSequence,
+ fieldMaxSequenceNumber(
+ file, columnSequences, position,
physicalFields.size()));
+ }
+ }
+ if (indexedMaxSequence != Long.MIN_VALUE) {
+ group.addUpdatedFile(indexedMaxSequence, file.nonNullRowIdRange());
}
}
+ private static Function<Long, TableSchema>
cachedSchemaLoader(SchemaManager schemaManager) {
+ Map<Long, TableSchema> schemaCache = new HashMap<>();
+ return schemaId -> schemaCache.computeIfAbsent(schemaId,
schemaManager::schema);
+ }
+
/**
* Projects only the fields the refresh planner consumes; identifier
fields are included only
* when deleted files must be recognized.
@@ -265,6 +285,7 @@ public final class DataEvolutionGlobalIndexRefreshPlanner {
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));
+
fileFields.add(DataFileMeta.SCHEMA.getField(DataFileMeta.WRITE_COLS_SEQUENCES));
if (includeIdentifierFields) {
fileFields.add(DataFileMeta.SCHEMA.getField(DataFileMeta.FILE_NAME));
fileFields.add(DataFileMeta.SCHEMA.getField(DataFileMeta.LEVEL));
@@ -351,6 +372,9 @@ public final class DataEvolutionGlobalIndexRefreshPlanner {
}
private void addUpdatedFile(long maxSequenceNumber, Range rowRange) {
+ if (maxSequenceNumber <= minScanSnapshotId) {
+ return;
+ }
// Merge the range eagerly instead of retaining the file metadata.
int sequenceNumberIndex =
firstIndexWithSequenceNumberBelow(maxSequenceNumber);
if (updatedRangesPerSequenceNumber[sequenceNumberIndex] == null) {
@@ -473,13 +497,4 @@ public final class DataEvolutionGlobalIndexRefreshPlanner {
}
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/test/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexRefreshPlannerTest.java
b/paimon-core/src/test/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexRefreshPlannerTest.java
index 68d1a26218..f7bc10051c 100644
---
a/paimon-core/src/test/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexRefreshPlannerTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexRefreshPlannerTest.java
@@ -47,6 +47,8 @@ import java.util.Random;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
/** Tests for {@link DataEvolutionGlobalIndexRefreshPlanner}. */
@@ -152,6 +154,111 @@ class DataEvolutionGlobalIndexRefreshPlannerTest {
.containsExactly(index);
}
+ @Test
+ void testUsesColumnSequenceNumbersForCompactedFullFile() {
+ IndexManifestEntry index = index("index", 0, 99, 5L,
BinaryRow.EMPTY_ROW, 0);
+
+ assertThat(
+ plan(
+ Collections.singletonList(
+ dataWithColumnSequences(
+ "unrelated-compact",
+ 0,
+ 100,
+ 10,
+ new long[] {5L, 10L, 10L})),
+ index))
+ .isEmpty();
+ assertThat(
+ plan(
+ Collections.singletonList(
+ dataWithColumnSequences(
+ "index-compact",
+ 0,
+ 100,
+ 10,
+ new long[] {6L, 10L, 10L})),
+ index))
+ .containsExactly(index);
+
+ // Legacy compacted files have no column metadata and remain
conservative.
+ assertThat(plan(Collections.singletonList(data("legacy", 0, 100, 10,
1)), index))
+ .containsExactly(index);
+ }
+
+ @Test
+ void testColumnSequenceNumbersFollowWriteColsOrder() {
+ IndexManifestEntry index = index("index", 0, 99, 5L,
BinaryRow.EMPTY_ROW, 0);
+
+ assertThat(
+ plan(
+ Collections.singletonList(
+ dataWithColumnSequences(
+ "reordered-compact",
+ 0,
+ 100,
+ 10,
+ new long[] {10L, 5L},
+ "unrelated",
+ "vector")),
+ index))
+ .isEmpty();
+ }
+
+ @Test
+ void testColumnSequenceNumbersIgnoreRowTrackingFields() {
+ IndexManifestEntry index = index("index", 0, 99, 5L,
BinaryRow.EMPTY_ROW, 0);
+
+ assertThat(
+ plan(
+ Collections.singletonList(
+ dataWithColumnSequences(
+ "row-tracking-compact",
+ 0,
+ 100,
+ 10,
+ new long[] {5L, 10L, 10L},
+ "vector",
+ "other",
+ "unrelated",
+ SpecialFields.ROW_ID.name(),
+
SpecialFields.SEQUENCE_NUMBER.name())),
+ index))
+ .isEmpty();
+ }
+
+ @Test
+ void testCachesSchemaAcrossWriteColumnLayouts() {
+ IndexManifestEntry index = index("index", 0, 99, 5L,
BinaryRow.EMPTY_ROW, 0);
+
+ plan(
+ Arrays.asList(
+ data("vector-update", 0, 100, 6, 1, "vector"),
+ data("other-update", 0, 100, 6, 1, "other")),
+ index);
+
+ verify(schemaManager, times(1)).schema(1L);
+ }
+
+ @Test
+ void testMalformedColumnSequenceNumbersFallBackToFileSequence() {
+ IndexManifestEntry index = index("index", 0, 99, 5L,
BinaryRow.EMPTY_ROW, 0);
+
+ assertThat(
+ plan(
+ Collections.singletonList(
+ dataWithColumnSequences(
+ "malformed-compact",
+ 0,
+ 100,
+ 10,
+ new long[] {5L},
+ "vector",
+ "other")),
+ index))
+ .containsExactly(index);
+ }
+
@Test
void testRefreshesFromUpdateLayerOverBaseSchemaWithoutIndexColumn() {
IndexManifestEntry index = index("index", 0, 99, 5L,
BinaryRow.EMPTY_ROW, 0);
@@ -491,4 +598,18 @@ class DataEvolutionGlobalIndexRefreshPlannerTest {
writeCols);
return ManifestEntry.create(FileKind.ADD, BinaryRow.EMPTY_ROW, 0, 1,
file);
}
+
+ private ManifestEntry dataWithColumnSequences(
+ String fileName,
+ long firstRowId,
+ long rowCount,
+ long maxSequenceNumber,
+ long[] columnSequences,
+ String... writeCols) {
+ DataFileMeta file =
+ data(fileName, firstRowId, rowCount, maxSequenceNumber, 1,
writeCols)
+ .file()
+ .withColumnMaxSequenceNumbers(columnSequences);
+ return ManifestEntry.create(FileKind.ADD, BinaryRow.EMPTY_ROW, 0, 1,
file);
+ }
}
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 4770726e4b..524f54a85c 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
@@ -22,6 +22,7 @@ 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.append.dataevolution.DataEvolutionCompactionCommitPreparation;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.BlobData;
@@ -38,6 +39,8 @@ import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.manifest.IndexManifestEntry;
import org.apache.paimon.manifest.ManifestEntry;
import org.apache.paimon.memory.MemorySlice;
+import org.apache.paimon.options.ExpireConfig;
+import org.apache.paimon.options.Options;
import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.schema.Schema;
@@ -54,10 +57,12 @@ import org.apache.paimon.types.DataField;
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.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -66,6 +71,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.stream.Collectors;
import static org.assertj.core.api.Assertions.assertThat;
@@ -283,6 +289,118 @@ public class SortedGlobalIndexScannerTest extends
TableTestBase {
500, totalRowCount, "incrementalScan should only return the
newly written rows");
}
+ @Test
+ public void
testIncrementalScanIgnoresNonIndexColumnCompactionAfterSnapshotExpiration()
+ throws Exception {
+ write();
+ createIndex(null);
+
+ updateColumnAndCompact("f1", 1);
+ updateColumnAndCompact("f1", 2);
+
+ FileStoreTable table = getTableDefault();
+ assertThat(refreshPlanParity(table)).isEmpty();
+ table.newExpireSnapshots()
+ .config(
+ ExpireConfig.builder()
+ .snapshotRetainMax(1)
+ .snapshotRetainMin(1)
+ .snapshotTimeRetain(Duration.ZERO)
+ .build())
+ .expire();
+ assertThat(table.snapshotManager().earliestSnapshotId())
+ .isEqualTo(table.snapshotManager().latestSnapshotId());
+
+
assertThat(dataEvolutionScanner(table).withIndexField("f0").incrementalScan()).isEmpty();
+ }
+
+ @Test
+ public void testIncrementalScanRefreshesOnlyIndexColumnCompaction() throws
Exception {
+ write();
+ createIndex(null);
+
+ updateColumnAndCompact("f1", 1);
+
assertThat(dataEvolutionScanner(getTableDefault()).withIndexField("f0").incrementalScan())
+ .isEmpty();
+
+ DataFileMeta compacted = updateColumnAndCompact("f0", 2);
+ ScanResult<DataSplit> scanResult =
+ dataEvolutionScanner(getTableDefault())
+ .withIndexField("f0")
+ .incrementalScan()
+ .orElseThrow(
+ () ->
+ new IllegalStateException(
+ "Expected incremental index
build after indexed column compaction."));
+ Range expectedRange = new Range(0, PART_ROW_NUM - 1);
+ assertThat(compacted.nonNullRowIdRange()).isEqualTo(expectedRange);
+
assertThat(scanResult.rowRangeIndex().ranges()).containsExactly(expectedRange);
+ assertThat(scanResult.deletedIndexEntries())
+ .isNotEmpty()
+ .allSatisfy(
+ entry ->
+
assertThat(entry.indexFile().globalIndexMeta().rowRange())
+ .isEqualTo(expectedRange));
+ assertThat(scanResult.entries()).isNotEmpty();
+ }
+
+ private SortedGlobalIndexScanner dataEvolutionScanner(FileStoreTable
table) {
+ Options options = new Options();
+ options.set(
+ CoreOptions.GLOBAL_INDEX_COLUMN_UPDATE_ACTION,
+ CoreOptions.GlobalIndexColumnUpdateAction.IGNORE);
+ return new SortedGlobalIndexScanner(table, "btree", options);
+ }
+
+ private DataFileMeta updateColumnAndCompact(String column, int
updateRound) throws Exception {
+ Map<String, String> writeOptions = new HashMap<>();
+ writeOptions.put(
+ CoreOptions.GLOBAL_INDEX_COLUMN_UPDATE_ACTION.key(),
+ CoreOptions.GlobalIndexColumnUpdateAction.IGNORE.toString());
+ FileStoreTable table = getTableDefault().copy(writeOptions);
+ RowType writeType = table.rowType().project(Arrays.asList("dt",
column));
+ BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder();
+ try (BatchTableWrite batchWrite =
writeBuilder.newWrite().withWriteType(writeType)) {
+ for (int i = 0; i < PART_ROW_NUM; i++) {
+ Object value =
+ "f0".equals(column)
+ ? i + updateRound * (int) PART_ROW_NUM
+ : BinaryString.fromString("updated_" +
updateRound + "_" + i);
+ batchWrite.write(GenericRow.of(BinaryString.fromString("p0"),
value));
+ }
+ List<CommitMessage> messages = batchWrite.prepareCommit();
+ setFirstRowId(messages, 0L);
+ try (BatchTableCommit commit = writeBuilder.newCommit()) {
+ commit.commit(messages);
+ }
+ }
+
+ writeOptions.put(CoreOptions.COMPACTION_MIN_FILE_NUM.key(), "2");
+ table = getTableDefault().copy(writeOptions);
+ Snapshot compactSnapshot = table.snapshotManager().latestSnapshot();
+ DataEvolutionCompactCoordinator coordinator =
+ new DataEvolutionCompactCoordinator(table, false, false,
compactSnapshot);
+ List<CommitMessage> compactMessages = new ArrayList<>();
+ for (DataEvolutionCompactTask task : coordinator.plan()) {
+ compactMessages.add(task.doCompact(table, "test-compact"));
+ }
+ assertThat(compactMessages).isNotEmpty();
+ compactMessages.addAll(
+ new DataEvolutionCompactionCommitPreparation(table,
compactSnapshot)
+ .prepare(compactMessages));
+ try (BatchTableCommit commit =
table.newBatchWriteBuilder().newCommit()) {
+ commit.commit(compactMessages);
+ }
+
+ List<DataFileMeta> rowRangeFiles =
+ getTableDefault().store().newScan().plan().files().stream()
+ .map(ManifestEntry::file)
+ .filter(file -> file.firstRowId() != null &&
file.firstRowId() == 0L)
+ .collect(Collectors.toList());
+ assertThat(rowRangeFiles).hasSize(1);
+ return rowRangeFiles.get(0);
+ }
+
@Test
public void testIncrementalScanWithPartitionPredicate() throws Exception {
write();
