From 3046b299a55ff8245331675242d454ba6981e29a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BB=9F=E5=BC=8B?= Date: Wed, 28 Oct 2026 15:54:31 +0800 Subject: [PATCH 01/13] [core] Advance index builds across row ID reassignment --- .../paimon/operation/FileStoreCommitImpl.java | 92 ++++++++- .../commit/CommitChangesProvider.java | 2 +- .../ReassignCompactChangesProvider.java | 81 +++++++- .../operation/commit/StrictModeChecker.java | 15 +- .../CompactReassignReuseTest.java | 189 ++++++++++++++++-- 5 files changed, 358 insertions(+), 21 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java index a460a5234a5f..2d1de4c24160 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java @@ -28,6 +28,7 @@ import org.apache.paimon.disk.IOManager; import org.apache.paimon.fs.FileIO; import org.apache.paimon.fs.Path; +import org.apache.paimon.globalindex.DataEvolutionGlobalIndexRefreshPlanner; import org.apache.paimon.index.IndexFileMeta; import org.apache.paimon.index.IndexPathFactory; import org.apache.paimon.io.DataFileMeta; @@ -75,6 +76,8 @@ import org.apache.paimon.table.sink.CommitMessage; import org.apache.paimon.table.sink.CommitMessageImpl; import org.apache.paimon.table.sink.CommitPreCallback; +import org.apache.paimon.table.source.ScanMode; +import org.apache.paimon.types.DataField; import org.apache.paimon.types.RowType; import org.apache.paimon.utils.DataFilePathFactories; import org.apache.paimon.utils.FileStorePathFactory; @@ -83,6 +86,7 @@ import org.apache.paimon.utils.InternalRowPartitionComputer; import org.apache.paimon.utils.ListUtils; import org.apache.paimon.utils.Pair; +import org.apache.paimon.utils.Range; import org.apache.paimon.utils.RetryWaiter; import org.apache.paimon.utils.SnapshotManager; @@ -120,6 +124,7 @@ import static org.apache.paimon.types.VectorType.isVectorStoreFile; import static org.apache.paimon.utils.Preconditions.checkArgument; import static org.apache.paimon.utils.Preconditions.checkNotNull; +import static org.apache.paimon.utils.Preconditions.checkState; /** * Default implementation of {@link FileStoreCommit}. @@ -159,6 +164,7 @@ public class FileStoreCommitImpl implements FileStoreCommit { private final IndexManifestFile indexManifestFile; @Nullable private final CommitRollback rollback; private final CommitScanner scanner; + private final Supplier scanSupplier; private final List commitPreCallbacks; private final List commitCallbacks; private final StatsFileHandler statsFileHandler; @@ -211,6 +217,7 @@ public FileStoreCommitImpl( this.indexManifestFile = indexManifestFileFactory.create(); this.rollback = rollback; this.scanner = new CommitScanner(scanSupplier, snapshotManager, indexManifestFile, options); + this.scanSupplier = scanSupplier; this.commitPreCallbacks = commitPreCallbacks; this.commitCallbacks = commitCallbacks; this.retryWaiter = @@ -379,12 +386,27 @@ public int commit(ManifestCommittable committable, boolean checkAppendFiles) { checkAppendFiles = true; } + CommitChangesProvider changesProvider = + CommitChangesProvider.provider( + changes.appendTableFiles, + changes.appendChangelog, + changes.appendIndexFiles); + if (options.dataEvolutionEnabled() + && commitKind == CommitKind.APPEND + && changes.appendTableFiles.isEmpty() + && changes.appendChangelog.isEmpty()) { + changesProvider = + ReassignCompactChangesProvider.forIndexBuild( + fileIO, + pathFactory, + snapshotManager, + changes.appendIndexFiles, + options.commitLastSafeSnapshot().orElse(null), + this::checkIndexBuildChanges); + } attempts += tryCommit( - CommitChangesProvider.provider( - changes.appendTableFiles, - changes.appendChangelog, - changes.appendIndexFiles), + changesProvider, committable.identifier(), committable.watermark(), committable.properties(), @@ -835,6 +857,68 @@ private void configureRowIdCheckFromMessages(List commitMessages) } } + private void checkIndexBuildChanges(Snapshot snapshot, List indexes) { + if (indexes.isEmpty()) { + return; + } + List partitions = + indexes.stream() + .map(IndexManifestEntry::partition) + .distinct() + .collect(Collectors.toList()); + if (snapshot.commitKind() == CommitKind.OVERWRITE) { + new StrictModeChecker( + snapshotManager, commitUser, scanSupplier, indexManifestFile, true, 0) + .checkIndexBuildOverwrite(snapshot, partitions); + return; + } + Map schemaFields = + schemaManager.schema(snapshot.schemaId()).idToFieldMap(); + Map, List> definitions = + indexes.stream() + .collect( + Collectors.groupingBy( + entry -> + entry.indexFile() + .globalIndexMeta() + .getIndexedFieldIds())); + for (Map.Entry, List> definition : + definitions.entrySet()) { + List fields = + definition.getKey().stream() + .map( + id -> { + DataField field = schemaFields.get(id); + checkState( + field != null, + "Indexed field %s no longer exists.", + id); + return field; + }) + .collect(Collectors.toList()); + List ranges = + definition.getValue().stream() + .map(entry -> entry.indexFile().globalIndexMeta().rowRange()) + .collect(Collectors.toList()); + Iterable entries = + () -> + scanSupplier + .get() + .withSnapshot(snapshot) + .withKind(ScanMode.DELTA) + .withPartitionFilter(partitions) + .withRowRanges(ranges) + .dropStats() + .readFileIterator(); + checkState( + DataEvolutionGlobalIndexRefreshPlanner.findIndexesToRefresh( + schemaManager, entries, definition.getValue(), fields) + .isEmpty(), + "Cannot reuse index build because snapshot %s modified indexed columns.", + snapshot.id()); + } + } + private Set> materializedBuckets(List commitMessages) { if (!options.dataEvolutionEnabled() || !options.deletionVectorsEnabled()) { return Collections.emptySet(); diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/commit/CommitChangesProvider.java b/paimon-core/src/main/java/org/apache/paimon/operation/commit/CommitChangesProvider.java index 7759225d3e9d..ca57068cd8d5 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/commit/CommitChangesProvider.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/commit/CommitChangesProvider.java @@ -34,7 +34,7 @@ public interface CommitChangesProvider { CommitChanges provide(@Nullable Snapshot latestSnapshot); - /** Reassignments already validated and applied to this provider's compact results. */ + /** Reassignments already validated and applied to this provider's results. */ default Set rebasedReassignments() { return Collections.emptySet(); } diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/commit/ReassignCompactChangesProvider.java b/paimon-core/src/main/java/org/apache/paimon/operation/commit/ReassignCompactChangesProvider.java index 88d3dcf90670..f01755c7bed8 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/commit/ReassignCompactChangesProvider.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/commit/ReassignCompactChangesProvider.java @@ -21,9 +21,11 @@ import org.apache.paimon.Snapshot; import org.apache.paimon.append.dataevolution.SerializationAssignment; import org.apache.paimon.fs.FileIO; +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.table.SpecialFields; @@ -41,15 +43,18 @@ import java.util.List; import java.util.Objects; import java.util.Set; +import java.util.function.BiConsumer; +import java.util.stream.Collectors; import static org.apache.paimon.utils.Preconditions.checkState; -/** Reuses immutable compact outputs after verified, metadata-only row-id reassignments. */ +/** Reuses immutable compact or index-build outputs after metadata-only row-id reassignments. */ public final class ReassignCompactChangesProvider implements CommitChangesProvider { private final FileIO fileIO; private final FileStorePathFactory paths; private final SnapshotManager snapshots; + @Nullable private final BiConsumer> indexBuildCheck; private final Set reassignments = new HashSet<>(); private long checkedSnapshot; private CommitChanges changes; @@ -60,13 +65,63 @@ public ReassignCompactChangesProvider( SnapshotManager snapshots, long scanSnapshotId, CommitChanges changes) { + this(fileIO, paths, snapshots, scanSnapshotId, changes, null); + } + + private ReassignCompactChangesProvider( + FileIO fileIO, + FileStorePathFactory paths, + SnapshotManager snapshots, + long scanSnapshotId, + CommitChanges changes, + @Nullable BiConsumer> indexBuildCheck) { this.fileIO = fileIO; this.paths = paths; this.snapshots = snapshots; + this.indexBuildCheck = indexBuildCheck; this.checkedSnapshot = scanSnapshotId; this.changes = changes; } + /** Uses index scan baselines without changing the original source metadata. */ + public static CommitChangesProvider forIndexBuild( + FileIO fileIO, + FileStorePathFactory paths, + SnapshotManager snapshots, + List indexFiles, + @Nullable Long lastSafeSnapshot, + BiConsumer> buildCheck) { + CommitChanges changes = + new CommitChanges(Collections.emptyList(), Collections.emptyList(), indexFiles); + long scanSnapshotId = + lastSafeSnapshot == null || lastSafeSnapshot < 0 + ? Long.MAX_VALUE + : lastSafeSnapshot; + boolean hasBuildOutputs = false; + for (IndexManifestEntry entry : indexFiles) { + GlobalIndexMeta meta = entry.indexFile().globalIndexMeta(); + if (meta == null) { + return latest -> changes; + } + if (entry.kind() == FileKind.ADD) { + if (!DataEvolutionIndexSourceMeta.isDataEvolutionMeta(meta.sourceMeta())) { + return latest -> changes; + } + // DELETE source metadata belongs to the old index, not this build's scan. + hasBuildOutputs = true; + scanSnapshotId = + Math.min( + scanSnapshotId, + DataEvolutionIndexSourceMeta.fromIndexFile(entry.indexFile()) + .scanSnapshotId()); + } + } + return !hasBuildOutputs + ? latest -> changes + : new ReassignCompactChangesProvider( + fileIO, paths, snapshots, scanSnapshotId, changes, buildCheck); + } + @Override public CommitChanges provide(@Nullable Snapshot latest) { checkState( @@ -85,6 +140,23 @@ public CommitChanges provide(@Nullable Snapshot latest) { snapshot.commitKind() == Snapshot.CommitKind.OVERWRITE ? SerializationAssignment.planFile(snapshot) : null; + if (indexBuildCheck != null + && (snapshot.commitKind() == Snapshot.CommitKind.OVERWRITE + || snapshot.commitKind() == Snapshot.CommitKind.APPEND) + && plan == null) { + List indexes = + changes.indexFiles.stream() + .filter( + entry -> + entry.kind() == FileKind.ADD + && DataEvolutionIndexSourceMeta + .fromIndexFile( + entry.indexFile()) + .scanSnapshotId() + < snapshot.id()) + .collect(Collectors.toList()); + indexBuildCheck.accept(snapshot, indexes); + } if (plan != null) { SerializationAssignment assignment; try { @@ -134,6 +206,13 @@ private CommitChanges mapChanges(SerializationAssignment assignment) { List indexes = new ArrayList<>(changes.indexFiles.size()); for (IndexManifestEntry entry : changes.indexFiles) { IndexFileMeta file = entry.indexFile(); + if (indexBuildCheck != null + && entry.kind() == FileKind.ADD + && DataEvolutionIndexSourceMeta.fromIndexFile(file).scanSnapshotId() + >= assignment.snapshotId()) { + indexes.add(entry); + continue; + } GlobalIndexMeta meta = file.globalIndexMeta(); if (meta == null) { checkState( diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/commit/StrictModeChecker.java b/paimon-core/src/main/java/org/apache/paimon/operation/commit/StrictModeChecker.java index 092c2d331b05..4a8fb0a84d15 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/commit/StrictModeChecker.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/commit/StrictModeChecker.java @@ -71,7 +71,8 @@ public void check( Set rebasedReassignments) { Set newPartitions = new HashSet<>(newChangedPartitions); for (long id = lastSafeSnapshot + 1; id < newSnapshotId; id++) { - if (newCommitKind == CommitKind.COMPACT && rebasedReassignments.contains(id)) { + if ((newCommitKind == CommitKind.COMPACT || newCommitKind == CommitKind.APPEND) + && rebasedReassignments.contains(id)) { continue; } Snapshot snapshot = snapshotManager.snapshot(id); @@ -129,6 +130,18 @@ public void check( } } + /** Index builds must reject data changes even when optional strict mode is disabled. */ + public void checkIndexBuildOverwrite(Snapshot snapshot, List indexedPartitions) { + Set partitions = new HashSet<>(indexedPartitions); + if (hasOverlappedDataPartition(snapshot, partitions) + || hasOverlappedIndexPartition(snapshot, partitions)) { + throw new RuntimeException( + String.format( + "Cannot reuse index build because OVERWRITE snapshot %s modified indexed partitions.", + snapshot.id())); + } + } + private boolean hasOverlappedDataPartition(Snapshot snapshot, Set newPartitions) { if (newPartitions.isEmpty()) { return false; diff --git a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/CompactReassignReuseTest.java b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/CompactReassignReuseTest.java index 62d5a1e62ae4..fbc991bf30a6 100644 --- a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/CompactReassignReuseTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/CompactReassignReuseTest.java @@ -33,6 +33,7 @@ import org.apache.paimon.globalindex.GlobalIndexResult; 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.IndexFileMeta; import org.apache.paimon.io.CompactIncrement; import org.apache.paimon.io.DataFileMeta; @@ -158,19 +159,38 @@ private CommitMessageImpl compact(FileStoreTable table, boolean blob) throws Exc return message; } - private CommitMessageImpl compactIndex(FileStoreTable table) throws Exception { + private List buildIndex(FileStoreTable table, String field, String partition) + throws Exception { + return buildIndex(table, "btree", field, partition); + } + + private List buildIndex( + FileStoreTable table, String indexType, String field, String partition) + throws Exception { List builds = new ArrayList<>(); for (DataSplit split : GlobalIndexBuilderUtils.splitByContiguousRowRange( - new SortedGlobalIndexScanner(table, "btree") - .withIndexFields(Collections.singletonList("id")) + new SortedGlobalIndexScanner(table, indexType) + .withIndexFields(Collections.singletonList(field)) .scan() .get() .entries())) { - builds.addAll(SortedGlobalIndexTestUtils.buildIndex(table, "btree", "id", split, 4)); + if (partition == null || split.partition().getString(0).toString().equals(partition)) { + builds.addAll( + SortedGlobalIndexTestUtils.buildIndex( + table, + indexType, + field, + split, + table.snapshotManager().latestSnapshotId())); + } } + return builds; + } + + private CommitMessageImpl compactIndex(FileStoreTable table) throws Exception { try (BatchTableCommit commit = table.newBatchWriteBuilder().newCommit()) { - commit.commit(builds); + commit.commit(buildIndex(table, "id", null)); } Snapshot snapshot = table.snapshotManager().latestSnapshot(); List inputs = @@ -257,6 +277,95 @@ private void checkRows(FileStoreTable table, int... expected) throws Exception { Arrays.stream(expected).boxed().collect(Collectors.toList())); } + @ParameterizedTest + @CsvSource({"btree,false", "btree,true", "bitmap,false", "bitmap,true"}) + void testIndexBuildAcrossReassignment(String indexType, boolean strict) throws Exception { + FileStoreTable table = prepare(); + try (BatchTableCommit commit = table.newBatchWriteBuilder().newCommit()) { + commit.commit(buildIndex(table, "pt", null)); + } + long scanSnapshot = table.snapshotManager().latestSnapshotId(); + List messages = buildIndex(table, indexType, "id", "a"); + assertThat(new DataEvolutionRowIdReassigner(table).reassign().reassigned).isTrue(); + messages.addAll(buildIndex(table, indexType, "id", "b")); + List files = + messages.stream() + .flatMap( + m -> + ((CommitMessageImpl) m) + .newFilesIncrement().newIndexFiles().stream()) + .map(IndexFileMeta::fileName) + .collect(Collectors.toList()); + Map options = new HashMap<>(); + options.put(CoreOptions.COMMIT_STRICT_MODE_ENABLED.key(), Boolean.toString(strict)); + if (strict) { + options.put(CoreOptions.COMMIT_LAST_SAFE_SNAPSHOT.key(), Long.toString(scanSnapshot)); + } + try (BatchTableCommit commit = table.copy(options).newBatchWriteBuilder().newCommit()) { + commit.commit(messages); + } + assertThat(table.snapshotManager().latestSnapshot().commitKind()) + .isEqualTo(Snapshot.CommitKind.APPEND); + List current = + table.store().newIndexFileHandler().scanEntries().stream() + .filter(e -> e.indexFile().globalIndexMeta().indexFieldId() == 1) + .collect(Collectors.toList()); + assertThat(current) + .extracting(e -> e.indexFile().fileName()) + .containsExactlyInAnyOrderElementsOf(files); + for (IndexManifestEntry entry : current) { + assertThat( + DataEvolutionIndexSourceMeta.fromIndexFile(entry.indexFile()) + .scanSnapshotId()) + .isEqualTo( + entry.partition().getString(0).toString().equals("a") + ? scanSnapshot + : scanSnapshot + 1); + } + long[] rowIds = {4, 5, 2, 6}; + for (int id = 0; id < rowIds.length; id++) { + Predicate predicate = new PredicateBuilder(table.rowType()).equal(1, id); + try (DataEvolutionGlobalIndexScanner scanner = + DataEvolutionGlobalIndexScanner.create(table, null, predicate).get()) { + GlobalIndexResult result = scanner.scan(predicate).get(); + assertThat(result.results().getLongCardinality()).isEqualTo(1); + assertThat(result.results().contains(rowIds[id])).isTrue(); + } + } + } + + @Test + void testIndexBuildRejectsConcurrentColumnUpdate() throws Exception { + FileStoreTable table = prepare(); + List messages = buildIndex(table, "id", null); + BatchWriteBuilder builder = table.newBatchWriteBuilder(); + try (BatchTableWrite write = + builder.newWrite() + .withWriteType(table.rowType().project(Arrays.asList("pt", "id"))); + BatchTableCommit commit = builder.newCommit()) { + write.write(GenericRow.of(BinaryString.fromString("a"), 100)); + List updates = write.prepareCommit(); + for (CommitMessage update : updates) { + ((CommitMessageImpl) update) + .newFilesIncrement() + .newFiles() + .replaceAll(file -> file.assignFirstRowId(0L)); + } + commit.commit(updates); + } + new DataEvolutionRowIdReassigner(table).reassign(); + long snapshot = table.snapshotManager().latestSnapshotId(); + assertThatThrownBy( + () -> { + try (BatchTableCommit commit = + table.newBatchWriteBuilder().newCommit()) { + commit.commit(messages); + } + }) + .hasMessageContaining("modified indexed columns"); + assertThat(table.snapshotManager().latestSnapshotId()).isEqualTo(snapshot); + } + @ParameterizedTest @ValueSource(booleans = {false, true}) void testReuseNormalAndBlobAcrossRepeatedReassignments(boolean blob) throws Exception { @@ -291,12 +400,28 @@ void testReuseNormalAndBlobAcrossRepeatedReassignments(boolean blob) throws Exce } @ParameterizedTest - @ValueSource(booleans = {false, true}) - void testIndexReplacementKeepsPhysicalFilesAndPointLookup(boolean loseSuccessfulResponse) - throws Exception { + @CsvSource({"false,false", "false,true", "true,false", "true,true"}) + void testIndexReplacementKeepsPhysicalFilesAndPointLookup( + boolean indexBuild, boolean loseSuccessfulResponse) throws Exception { FileStoreTable table = prepare(); CommitMessageImpl message = compactIndex(table); List outputs = message.compactIncrement().newIndexFiles(); + long scanSnapshot = lastSafeSnapshots.get(message); + if (indexBuild) { + message = + new CommitMessageImpl( + message.partition(), + message.bucket(), + message.totalBuckets(), + new DataIncrement( + Collections.emptyList(), + Collections.emptyList(), + Collections.emptyList(), + outputs, + message.compactIncrement().deletedIndexFiles()), + CompactIncrement.emptyIncrement()); + lastSafeSnapshots.put(message, scanSnapshot); + } assertThat(new DataEvolutionRowIdReassigner(table).reassign().reassigned).isTrue(); long reassignedSnapshot = table.snapshotManager().latestSnapshotId(); AtomicInteger attempts = new AtomicInteger(); @@ -316,7 +441,10 @@ protected boolean commitImpl( throws Exception { boolean committed = delegate.commit(baseUuid, snapshot, branch, statistics); - if (snapshot.commitKind() == Snapshot.CommitKind.COMPACT) { + if (snapshot.commitKind() + == (indexBuild + ? Snapshot.CommitKind.APPEND + : Snapshot.CommitKind.COMPACT)) { attempts.incrementAndGet(); return committed && !loseSuccessfulResponse; } @@ -333,6 +461,16 @@ protected void closeImpl() throws Exception { FileStoreTable committingTable = FileStoreTableFactory.create( table.fileIO(), table.location(), table.schema(), environment); + if (indexBuild) { + // A replaced index's source baseline predates this refresh's planning snapshot. + assertThat( + table.fileIO() + .delete( + table.snapshotManager().snapshotPath(scanSnapshot), + false)) + .isTrue(); + table.snapshotManager().invalidateCache(); + } commit(committingTable, message, false); assertThat(attempts).hasValue(1); assertThat(table.snapshotManager().latestSnapshotId()).isEqualTo(reassignedSnapshot + 1); @@ -606,10 +744,15 @@ void testExpiredBoundaryAndLegacyOption() throws Exception { } @ParameterizedTest - @ValueSource(booleans = {false, true}) - void testReassignDuringCommitRetries(boolean loseSuccessfulResponse) throws Exception { + @CsvSource({"false,false", "false,true", "true,false", "true,true"}) + void testReassignDuringCommitRetries(boolean indexBuild, boolean loseSuccessfulResponse) + throws Exception { FileStoreTable original = prepare(); - CommitMessageImpl message = compact(original, false); + CommitMessageImpl message = + indexBuild + ? (CommitMessageImpl) buildIndex(original, "id", "a").get(0) + : compact(original, false); + lastSafeSnapshots.put(message, original.snapshotManager().latestSnapshotId()); AtomicInteger attempts = new AtomicInteger(); CatalogEnvironment environment = new CatalogEnvironment(null, null, null, null, null, null, false, false) { @@ -625,7 +768,10 @@ protected boolean commitImpl( String branch, List statistics) throws Exception { - if (snapshot.commitKind() != Snapshot.CommitKind.COMPACT) { + if (snapshot.commitKind() + != (indexBuild + ? Snapshot.CommitKind.APPEND + : Snapshot.CommitKind.COMPACT)) { return delegate.commit(baseUuid, snapshot, branch, statistics); } int attempt = attempts.getAndIncrement(); @@ -663,7 +809,22 @@ protected void closeImpl() throws Exception { assertThat(attempts).hasValue(3); assertThat(table.snapshotManager().latestSnapshotId()).isEqualTo(9); checkRows(table, 0, 1, 2, 3, 4, 5); - assertThat(message.compactIncrement().compactAfter().get(0).nonNullFirstRowId()).isZero(); + if (indexBuild) { + IndexFileMeta built = message.newFilesIncrement().newIndexFiles().get(0); + assertThat(built.globalIndexMeta().rowRange()).isEqualTo(new Range(0, 1)); + assertThat(table.store().newIndexFileHandler().scanEntries()) + .singleElement() + .satisfies( + entry -> { + assertThat(entry.indexFile().fileName()) + .isEqualTo(built.fileName()); + assertThat(entry.indexFile().globalIndexMeta().rowRange()) + .isEqualTo(new Range(9, 10)); + }); + } else { + assertThat(message.compactIncrement().compactAfter().get(0).nonNullFirstRowId()) + .isZero(); + } } @Test From 5e24b3e4c755a5b02f1beb42e7d03376121b8c45 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BB=9F=E5=BC=8B?= Date: Wed, 28 Oct 2026 16:03:31 +0800 Subject: [PATCH 02/13] [core] Extract append changes provider --- .../paimon/operation/FileStoreCommitImpl.java | 43 +++++++++++-------- 1 file changed, 24 insertions(+), 19 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java index 2d1de4c24160..b9a6601b471c 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java @@ -386,27 +386,14 @@ public int commit(ManifestCommittable committable, boolean checkAppendFiles) { checkAppendFiles = true; } - CommitChangesProvider changesProvider = - CommitChangesProvider.provider( - changes.appendTableFiles, - changes.appendChangelog, - changes.appendIndexFiles); - if (options.dataEvolutionEnabled() - && commitKind == CommitKind.APPEND - && changes.appendTableFiles.isEmpty() - && changes.appendChangelog.isEmpty()) { - changesProvider = - ReassignCompactChangesProvider.forIndexBuild( - fileIO, - pathFactory, - snapshotManager, - changes.appendIndexFiles, - options.commitLastSafeSnapshot().orElse(null), - this::checkIndexBuildChanges); - } attempts += tryCommit( - changesProvider, + appendChangesProvider( + new CommitChanges( + changes.appendTableFiles, + changes.appendChangelog, + changes.appendIndexFiles), + commitKind), committable.identifier(), committable.watermark(), committable.properties(), @@ -941,6 +928,24 @@ private Set> materializedBuckets(List co return result; } + private CommitChangesProvider appendChangesProvider( + CommitChanges changes, CommitKind commitKind) { + if (options.dataEvolutionEnabled() + && commitKind == CommitKind.APPEND + && changes.tableFiles.isEmpty() + && changes.changelogFiles.isEmpty()) { + return ReassignCompactChangesProvider.forIndexBuild( + fileIO, + pathFactory, + snapshotManager, + changes.indexFiles, + options.commitLastSafeSnapshot().orElse(null), + this::checkIndexBuildChanges); + } + return CommitChangesProvider.provider( + changes.tableFiles, changes.changelogFiles, changes.indexFiles); + } + @VisibleForTesting CommitChangesProvider compactChangesProvider( CommitChanges changes, Set> materializedBuckets) { From 36eb125ce05b5703cceaa7232da054765cbfca87 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BB=9F=E5=BC=8B?= Date: Wed, 28 Oct 2026 16:11:11 +0800 Subject: [PATCH 03/13] [core] Clarify append and overwrite provider name --- .../java/org/apache/paimon/operation/FileStoreCommitImpl.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java index b9a6601b471c..39cea35b875f 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java @@ -388,7 +388,7 @@ public int commit(ManifestCommittable committable, boolean checkAppendFiles) { attempts += tryCommit( - appendChangesProvider( + appendAndOverwriteChangesProvider( new CommitChanges( changes.appendTableFiles, changes.appendChangelog, @@ -928,7 +928,7 @@ private Set> materializedBuckets(List co return result; } - private CommitChangesProvider appendChangesProvider( + private CommitChangesProvider appendAndOverwriteChangesProvider( CommitChanges changes, CommitKind commitKind) { if (options.dataEvolutionEnabled() && commitKind == CommitKind.APPEND From defcc3d8e230d8210de3eb78e851ef828a1094d4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BB=9F=E5=BC=8B?= Date: Wed, 28 Oct 2026 16:24:33 +0800 Subject: [PATCH 04/13] [core] Rename index-only changes provider factory --- .../java/org/apache/paimon/operation/FileStoreCommitImpl.java | 2 +- .../paimon/operation/commit/ReassignCompactChangesProvider.java | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java index 39cea35b875f..eebb6086c448 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java @@ -934,7 +934,7 @@ private CommitChangesProvider appendAndOverwriteChangesProvider( && commitKind == CommitKind.APPEND && changes.tableFiles.isEmpty() && changes.changelogFiles.isEmpty()) { - return ReassignCompactChangesProvider.forIndexBuild( + return ReassignCompactChangesProvider.indexOnlyCommitChangesProvider( fileIO, pathFactory, snapshotManager, diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/commit/ReassignCompactChangesProvider.java b/paimon-core/src/main/java/org/apache/paimon/operation/commit/ReassignCompactChangesProvider.java index f01755c7bed8..bb73c2dd9448 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/commit/ReassignCompactChangesProvider.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/commit/ReassignCompactChangesProvider.java @@ -84,7 +84,7 @@ private ReassignCompactChangesProvider( } /** Uses index scan baselines without changing the original source metadata. */ - public static CommitChangesProvider forIndexBuild( + public static CommitChangesProvider indexOnlyCommitChangesProvider( FileIO fileIO, FileStorePathFactory paths, SnapshotManager snapshots, From d820c314d5d47a7f1604e22b8238c080ebdf6dce Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BB=9F=E5=BC=8B?= Date: Wed, 28 Oct 2026 17:03:37 +0800 Subject: [PATCH 05/13] [core] Advance row ID updates across reassignment plans --- .../paimon/operation/FileStoreCommitImpl.java | 55 ++++-- .../operation/commit/ConflictDetection.java | 9 + .../DataEvolutionConflictDetection.java | 22 +++ .../ReassignCompactChangesProvider.java | 87 +++++---- .../operation/commit/StrictModeChecker.java | 4 +- .../CompactReassignReuseTest.java | 177 +++++++++++++++--- .../DataEvolutionRowIdReassignerTest.java | 22 +-- 7 files changed, 290 insertions(+), 86 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java index eebb6086c448..8ef8ae8d1c59 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java @@ -22,6 +22,7 @@ import org.apache.paimon.Snapshot; import org.apache.paimon.Snapshot.CommitKind; import org.apache.paimon.annotation.VisibleForTesting; +import org.apache.paimon.append.dataevolution.SerializationAssignment; import org.apache.paimon.catalog.SnapshotCommit; import org.apache.paimon.data.BinaryRow; import org.apache.paimon.data.InternalRow; @@ -29,6 +30,7 @@ import org.apache.paimon.fs.FileIO; import org.apache.paimon.fs.Path; import org.apache.paimon.globalindex.DataEvolutionGlobalIndexRefreshPlanner; +import org.apache.paimon.index.DataEvolutionIndexSourceMeta; import org.apache.paimon.index.IndexFileMeta; import org.apache.paimon.index.IndexPathFactory; import org.apache.paimon.io.DataFileMeta; @@ -352,8 +354,15 @@ public int commit(ManifestCommittable committable, boolean checkAppendFiles) { int attempts = 0; List commitMessages = committable.fileCommittables(); - configureRowIdCheckFromMessages(commitMessages); + Long dataScanSnapshot = configureRowIdCheckFromMessages(commitMessages); ManifestEntryChanges changes = collectChanges(commitMessages); + if (options.dataEvolutionEnabled() + && dataScanSnapshot == null + && changes.appendTableFiles.stream() + .anyMatch(entry -> entry.file().firstRowId() != null)) { + dataScanSnapshot = options.commitLastSafeSnapshot().filter(id -> id >= 0).orElse(null); + conflictDetection.setRowIdCheckFromSnapshot(dataScanSnapshot); + } Set> materializedBuckets = materializedBuckets(commitMessages); try { List appendSimpleEntries = @@ -393,7 +402,8 @@ public int commit(ManifestCommittable committable, boolean checkAppendFiles) { changes.appendTableFiles, changes.appendChangelog, changes.appendIndexFiles), - commitKind), + commitKind, + dataScanSnapshot), committable.identifier(), committable.watermark(), committable.properties(), @@ -800,7 +810,8 @@ private ManifestEntryChanges collectChanges(List commitMessages) return changes; } - private void configureRowIdCheckFromMessages(List commitMessages) { + @Nullable + private Long configureRowIdCheckFromMessages(List commitMessages) { Long checkFromSnapshot = null; for (CommitMessage message : commitMessages) { Long snapshotId = message.checkFromSnapshot(); @@ -842,6 +853,29 @@ private void configureRowIdCheckFromMessages(List commitMessages) // A committer can be reused; an untagged commit must not inherit a previous baseline. conflictDetection.setRowIdCheckFromSnapshot(checkFromSnapshot); } + return checkFromSnapshot; + } + + private void checkReassignChanges( + Snapshot snapshot, CommitChanges changes, CommitKind commitKind) { + if (snapshot.commitKind() == CommitKind.OVERWRITE + && SerializationAssignment.planFile(snapshot) != null) { + conflictDetection.checkRowIdReassignment(schemaManager, snapshot, changes, commitKind); + } else if (snapshot.commitKind() == CommitKind.APPEND + || snapshot.commitKind() == CommitKind.OVERWRITE) { + checkIndexBuildChanges( + snapshot, + changes.indexFiles.stream() + .filter( + entry -> + entry.kind() == FileKind.ADD + && entry.indexFile().globalIndexMeta() != null + && DataEvolutionIndexSourceMeta.fromIndexFile( + entry.indexFile()) + .scanSnapshotId() + < snapshot.id()) + .collect(Collectors.toList())); + } } private void checkIndexBuildChanges(Snapshot snapshot, List indexes) { @@ -929,18 +963,17 @@ private Set> materializedBuckets(List co } private CommitChangesProvider appendAndOverwriteChangesProvider( - CommitChanges changes, CommitKind commitKind) { - if (options.dataEvolutionEnabled() - && commitKind == CommitKind.APPEND - && changes.tableFiles.isEmpty() - && changes.changelogFiles.isEmpty()) { - return ReassignCompactChangesProvider.indexOnlyCommitChangesProvider( + CommitChanges changes, CommitKind commitKind, @Nullable Long dataScanSnapshot) { + if (options.dataEvolutionEnabled()) { + return ReassignCompactChangesProvider.commitChangesProvider( fileIO, pathFactory, snapshotManager, - changes.indexFiles, + changes, + dataScanSnapshot, options.commitLastSafeSnapshot().orElse(null), - this::checkIndexBuildChanges); + (snapshot, currentChanges) -> + checkReassignChanges(snapshot, currentChanges, commitKind)); } return CommitChangesProvider.provider( changes.tableFiles, changes.changelogFiles, changes.indexFiles); diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/commit/ConflictDetection.java b/paimon-core/src/main/java/org/apache/paimon/operation/commit/ConflictDetection.java index 3685d96311da..fba34e614a4d 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/commit/ConflictDetection.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/commit/ConflictDetection.java @@ -173,6 +173,15 @@ public boolean shouldCheckRowIdFromSnapshot(CommitKind commitKind) { return false; } + /** Checks history in the old coordinates and advances the row-id check boundary. */ + public void checkRowIdReassignment( + SchemaManager schemaManager, + Snapshot snapshot, + CommitChanges changes, + CommitKind commitKind) { + // Only Data Evolution tables rebase row-id-based commits. + } + @Nullable public RowIdConflictChecker createRowIdConflictChecker( SchemaManager schemaManager, List deltaFiles, CommitKind commitKind) { diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionConflictDetection.java b/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionConflictDetection.java index b3a7c82faedb..9581c982d07e 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionConflictDetection.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionConflictDetection.java @@ -148,6 +148,28 @@ private RowIdConflictCheckStrategy rowIdConflictCheckStrategy() { return rowIdConflictCheckStrategy; } + @Override + public void checkRowIdReassignment( + SchemaManager schemaManager, + Snapshot snapshot, + CommitChanges changes, + CommitKind commitKind) { + if (!shouldCheckRowIdFromSnapshot(commitKind) || snapshot.id() <= rowIdCheckFromSnapshot) { + return; + } + // Check logical changes in the old coordinates before moving past this reassignment. + Optional exception = + checkForRowIdFromSnapshot( + snapshotManager.snapshot(snapshot.id() - 1), + SimpleFileEntry.from(changes.tableFiles), + changes.indexFiles, + createRowIdConflictChecker(schemaManager, changes.tableFiles, commitKind)); + if (exception.isPresent()) { + throw exception.get(); + } + rowIdCheckFromSnapshot = snapshot.id(); + } + @Override public boolean canSkipDataFileConflictDetection( Snapshot latestSnapshot, diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/commit/ReassignCompactChangesProvider.java b/paimon-core/src/main/java/org/apache/paimon/operation/commit/ReassignCompactChangesProvider.java index bb73c2dd9448..3c4974a1beaa 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/commit/ReassignCompactChangesProvider.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/commit/ReassignCompactChangesProvider.java @@ -44,17 +44,18 @@ import java.util.Objects; import java.util.Set; import java.util.function.BiConsumer; -import java.util.stream.Collectors; import static org.apache.paimon.utils.Preconditions.checkState; -/** Reuses immutable compact or index-build outputs after metadata-only row-id reassignments. */ +/** Reuses immutable outputs after verified, metadata-only row-id reassignments. */ public final class ReassignCompactChangesProvider implements CommitChangesProvider { private final FileIO fileIO; private final FileStorePathFactory paths; private final SnapshotManager snapshots; - @Nullable private final BiConsumer> indexBuildCheck; + @Nullable private final BiConsumer changesCheck; + private final long dataScanSnapshot; + private final long initialScanSnapshot; private final Set reassignments = new HashSet<>(); private long checkedSnapshot; private CommitChanges changes; @@ -65,7 +66,7 @@ public ReassignCompactChangesProvider( SnapshotManager snapshots, long scanSnapshotId, CommitChanges changes) { - this(fileIO, paths, snapshots, scanSnapshotId, changes, null); + this(fileIO, paths, snapshots, scanSnapshotId, changes, scanSnapshotId, null); } private ReassignCompactChangesProvider( @@ -74,33 +75,46 @@ private ReassignCompactChangesProvider( SnapshotManager snapshots, long scanSnapshotId, CommitChanges changes, - @Nullable BiConsumer> indexBuildCheck) { + long dataScanSnapshot, + @Nullable BiConsumer changesCheck) { this.fileIO = fileIO; this.paths = paths; this.snapshots = snapshots; - this.indexBuildCheck = indexBuildCheck; + this.changesCheck = changesCheck; + this.dataScanSnapshot = dataScanSnapshot; + this.initialScanSnapshot = scanSnapshotId; this.checkedSnapshot = scanSnapshotId; this.changes = changes; } - /** Uses index scan baselines without changing the original source metadata. */ - public static CommitChangesProvider indexOnlyCommitChangesProvider( + /** Uses data and index scan baselines without changing the original source metadata. */ + public static CommitChangesProvider commitChangesProvider( FileIO fileIO, FileStorePathFactory paths, SnapshotManager snapshots, - List indexFiles, + CommitChanges changes, + @Nullable Long dataScanSnapshot, @Nullable Long lastSafeSnapshot, - BiConsumer> buildCheck) { - CommitChanges changes = - new CommitChanges(Collections.emptyList(), Collections.emptyList(), indexFiles); + BiConsumer changesCheck) { + boolean hasRowIdFiles = + changes.tableFiles.stream().anyMatch(entry -> entry.file().firstRowId() != null); + if (!changes.changelogFiles.isEmpty() || (hasRowIdFiles && dataScanSnapshot == null)) { + return latest -> changes; + } long scanSnapshotId = lastSafeSnapshot == null || lastSafeSnapshot < 0 ? Long.MAX_VALUE : lastSafeSnapshot; + if (hasRowIdFiles) { + scanSnapshotId = Math.min(scanSnapshotId, dataScanSnapshot); + } boolean hasBuildOutputs = false; - for (IndexManifestEntry entry : indexFiles) { + for (IndexManifestEntry entry : changes.indexFiles) { GlobalIndexMeta meta = entry.indexFile().globalIndexMeta(); if (meta == null) { + if (entry.indexFile().dvRanges() != null) { + continue; + } return latest -> changes; } if (entry.kind() == FileKind.ADD) { @@ -116,14 +130,27 @@ public static CommitChangesProvider indexOnlyCommitChangesProvider( .scanSnapshotId()); } } - return !hasBuildOutputs + return !hasBuildOutputs && !hasRowIdFiles ? latest -> changes : new ReassignCompactChangesProvider( - fileIO, paths, snapshots, scanSnapshotId, changes, buildCheck); + fileIO, + paths, + snapshots, + scanSnapshotId, + changes, + dataScanSnapshot == null ? Long.MAX_VALUE : dataScanSnapshot, + changesCheck); } @Override public CommitChanges provide(@Nullable Snapshot latest) { + // An automatic rollback can retry older history while outputs are still unmapped. + if (changesCheck != null + && reassignments.isEmpty() + && latest != null + && latest.id() < checkedSnapshot) { + checkedSnapshot = initialScanSnapshot; + } checkState( latest != null && latest.id() >= checkedSnapshot, "Compaction planning snapshot %s is ahead of the current table.", @@ -140,23 +167,6 @@ public CommitChanges provide(@Nullable Snapshot latest) { snapshot.commitKind() == Snapshot.CommitKind.OVERWRITE ? SerializationAssignment.planFile(snapshot) : null; - if (indexBuildCheck != null - && (snapshot.commitKind() == Snapshot.CommitKind.OVERWRITE - || snapshot.commitKind() == Snapshot.CommitKind.APPEND) - && plan == null) { - List indexes = - changes.indexFiles.stream() - .filter( - entry -> - entry.kind() == FileKind.ADD - && DataEvolutionIndexSourceMeta - .fromIndexFile( - entry.indexFile()) - .scanSnapshotId() - < snapshot.id()) - .collect(Collectors.toList()); - indexBuildCheck.accept(snapshot, indexes); - } if (plan != null) { SerializationAssignment assignment; try { @@ -170,8 +180,13 @@ public CommitChanges provide(@Nullable Snapshot latest) { && Objects.equals(snapshot.nextRowId(), assignment.nextRowId()), "Reassignment plan does not match snapshot %s.", id); + if (changesCheck != null) { + changesCheck.accept(snapshot, changes); + } changes = mapChanges(assignment); reassignments.add(id); + } else if (changesCheck != null) { + changesCheck.accept(snapshot, changes); } checkedSnapshot = id; } @@ -190,6 +205,11 @@ private CommitChanges mapChanges(SerializationAssignment assignment) { List files = new ArrayList<>(changes.tableFiles.size()); for (ManifestEntry entry : changes.tableFiles) { DataFileMeta file = entry.file(); + if (changesCheck != null + && (file.firstRowId() == null || dataScanSnapshot >= assignment.snapshotId())) { + files.add(entry); + continue; + } checkState( file.firstRowId() != null, "Cannot reuse compaction across reassignment: file %s has no first row ID.", @@ -206,8 +226,9 @@ private CommitChanges mapChanges(SerializationAssignment assignment) { List indexes = new ArrayList<>(changes.indexFiles.size()); for (IndexManifestEntry entry : changes.indexFiles) { IndexFileMeta file = entry.indexFile(); - if (indexBuildCheck != null + if (changesCheck != null && entry.kind() == FileKind.ADD + && file.globalIndexMeta() != null && DataEvolutionIndexSourceMeta.fromIndexFile(file).scanSnapshotId() >= assignment.snapshotId()) { indexes.add(entry); diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/commit/StrictModeChecker.java b/paimon-core/src/main/java/org/apache/paimon/operation/commit/StrictModeChecker.java index 4a8fb0a84d15..92bacaef0600 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/commit/StrictModeChecker.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/commit/StrictModeChecker.java @@ -71,7 +71,9 @@ public void check( Set rebasedReassignments) { Set newPartitions = new HashSet<>(newChangedPartitions); for (long id = lastSafeSnapshot + 1; id < newSnapshotId; id++) { - if ((newCommitKind == CommitKind.COMPACT || newCommitKind == CommitKind.APPEND) + if ((newCommitKind == CommitKind.COMPACT + || newCommitKind == CommitKind.APPEND + || newCommitKind == CommitKind.OVERWRITE) && rebasedReassignments.contains(id)) { continue; } diff --git a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/CompactReassignReuseTest.java b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/CompactReassignReuseTest.java index fbc991bf30a6..4f5ba1573743 100644 --- a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/CompactReassignReuseTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/CompactReassignReuseTest.java @@ -338,20 +338,8 @@ void testIndexBuildAcrossReassignment(String indexType, boolean strict) throws E void testIndexBuildRejectsConcurrentColumnUpdate() throws Exception { FileStoreTable table = prepare(); List messages = buildIndex(table, "id", null); - BatchWriteBuilder builder = table.newBatchWriteBuilder(); - try (BatchTableWrite write = - builder.newWrite() - .withWriteType(table.rowType().project(Arrays.asList("pt", "id"))); - BatchTableCommit commit = builder.newCommit()) { - write.write(GenericRow.of(BinaryString.fromString("a"), 100)); - List updates = write.prepareCommit(); - for (CommitMessage update : updates) { - ((CommitMessageImpl) update) - .newFilesIncrement() - .newFiles() - .replaceAll(file -> file.assignFirstRowId(0L)); - } - commit.commit(updates); + try (BatchTableCommit commit = table.newBatchWriteBuilder().newCommit()) { + commit.commit(prepareUpdate(table, "id", 100, 0, false)); } new DataEvolutionRowIdReassigner(table).reassign(); long snapshot = table.snapshotManager().latestSnapshotId(); @@ -366,6 +354,134 @@ void testIndexBuildRejectsConcurrentColumnUpdate() throws Exception { assertThat(table.snapshotManager().latestSnapshotId()).isEqualTo(snapshot); } + private List prepareUpdate( + FileStoreTable table, String field, Object value, long rowId, boolean tagged) + throws Exception { + long scanSnapshot = table.snapshotManager().latestSnapshotId(); + BatchWriteBuilder builder = table.newBatchWriteBuilder(); + try (BatchTableWrite write = + builder.newWrite() + .withWriteType(table.rowType().project(Arrays.asList("pt", field)))) { + write.write(GenericRow.of(BinaryString.fromString("a"), value)); + List updates = write.prepareCommit(); + for (CommitMessage update : updates) { + ((CommitMessageImpl) update) + .newFilesIncrement() + .newFiles() + .replaceAll(file -> file.assignFirstRowId(rowId)); + } + if (tagged) { + updates.replaceAll( + update -> ((CommitMessageImpl) update).withCheckFromSnapshot(scanSnapshot)); + } + return updates; + } + } + + @ParameterizedTest + @CsvSource({"false,false,false", "false,true,true", "true,false,true", "true,true,false"}) + void testUpdateAndAppendAcrossReassignment(boolean tagged, boolean strict, boolean overwrite) + throws Exception { + FileStoreTable table = prepare(); + long scanSnapshot = table.snapshotManager().latestSnapshotId(); + List messages = prepareUpdate(table, "id", 100, 0, tagged); + CommitMessageImpl message = (CommitMessageImpl) messages.get(0); + DataIncrement increment = message.newFilesIncrement(); + String updatedFile = increment.newFiles().get(0).fileName(); + if (overwrite) { + DataFileMeta input = + table.store().newScan().plan().files().stream() + .filter(e -> !isBlobFile(e.file().fileName())) + .filter(e -> e.file().nonNullFirstRowId() == 0) + .findFirst() + .get() + .file(); + messages.set( + 0, + new CommitMessageImpl( + message.partition(), + message.bucket(), + message.totalBuckets(), + new DataIncrement( + increment.newFiles(), + Collections.singletonList(input), + increment.changelogFiles(), + increment.newIndexFiles(), + increment.deletedIndexFiles()), + message.compactIncrement(), + message.checkFromSnapshot())); + } + messages.addAll(buildIndex(table, "id", "b")); + try (BatchTableWrite write = table.newBatchWriteBuilder().newWrite()) { + write.write( + GenericRow.of( + BinaryString.fromString("b"), + 4, + new BlobData("image-4".getBytes(StandardCharsets.UTF_8)))); + List appends = write.prepareCommit(); + assertThat(appends) + .flatExtracting(m -> ((CommitMessageImpl) m).newFilesIncrement().newFiles()) + .extracting(DataFileMeta::firstRowId) + .containsOnlyNulls(); + messages.addAll(appends); + } + assertThat(new DataEvolutionRowIdReassigner(table).reassign().reassigned).isTrue(); + Map options = new HashMap<>(); + options.put(CoreOptions.COMMIT_LAST_SAFE_SNAPSHOT.key(), Long.toString(scanSnapshot)); + options.put(CoreOptions.COMMIT_STRICT_MODE_ENABLED.key(), Boolean.toString(strict)); + try (BatchTableCommit commit = table.copy(options).newBatchWriteBuilder().newCommit()) { + commit.commit(messages); + } + assertThat(table.snapshotManager().latestSnapshot().commitKind()) + .isEqualTo(overwrite ? Snapshot.CommitKind.OVERWRITE : Snapshot.CommitKind.APPEND); + assertThat(read(table)) + .extracting(row -> row.getInt(1)) + .containsExactlyInAnyOrder(100, 1, 2, 3, 4); + assertThat(table.store().newScan().plan().files()) + .filteredOn(entry -> entry.file().fileName().equals(updatedFile)) + .singleElement() + .satisfies(entry -> assertThat(entry.file().nonNullFirstRowId()).isEqualTo(4)); + } + + @ParameterizedTest + @CsvSource({"-1,true", "0,true", "1,true", "-1,false", "0,false", "1,false"}) + void testUpdateConflictsAcrossReassignment(int phase, boolean sameRow) throws Exception { + FileStoreTable table = prepare(); + List pending = prepareUpdate(table, "id", 100, 0, true); + if (phase >= 0) { + new DataEvolutionRowIdReassigner(table).reassign(); + } + try (BatchTableCommit commit = table.newBatchWriteBuilder().newCommit()) { + commit.commit( + prepareUpdate( + table, "id", 200, (phase < 0 ? 0 : 4) + (sameRow ? 0 : 1), false)); + } + if (phase <= 0) { + if (phase == 0) { + write(table, "b", 4, false); + write(table, "a", 5, false); + } + assertThat(new DataEvolutionRowIdReassigner(table).reassign().reassigned).isTrue(); + } + long snapshot = table.snapshotManager().latestSnapshotId(); + if (sameRow) { + assertThatThrownBy( + () -> { + try (BatchTableCommit commit = + table.newBatchWriteBuilder().newCommit()) { + commit.commit(pending); + } + }) + .hasMessageContaining("conflict"); + assertThat(table.snapshotManager().latestSnapshotId()).isEqualTo(snapshot); + } else { + try (BatchTableCommit commit = table.newBatchWriteBuilder().newCommit()) { + commit.commit(pending); + } + assertThat(read(table)).extracting(row -> row.getInt(1)).contains(100, 200); + } + } + @ParameterizedTest @ValueSource(booleans = {false, true}) void testReuseNormalAndBlobAcrossRepeatedReassignments(boolean blob) throws Exception { @@ -744,14 +860,25 @@ void testExpiredBoundaryAndLegacyOption() throws Exception { } @ParameterizedTest - @CsvSource({"false,false", "false,true", "true,false", "true,true"}) - void testReassignDuringCommitRetries(boolean indexBuild, boolean loseSuccessfulResponse) + @CsvSource({ + "compact,false", + "compact,true", + "index,false", + "index,true", + "update,false", + "update,true" + }) + void testReassignDuringCommitRetries(String kind, boolean loseSuccessfulResponse) throws Exception { + boolean indexBuild = kind.equals("index"); + boolean update = kind.equals("update"); FileStoreTable original = prepare(); CommitMessageImpl message = - indexBuild - ? (CommitMessageImpl) buildIndex(original, "id", "a").get(0) - : compact(original, false); + update + ? (CommitMessageImpl) prepareUpdate(original, "id", 100, 0, true).get(0) + : indexBuild + ? (CommitMessageImpl) buildIndex(original, "id", "a").get(0) + : compact(original, false); lastSafeSnapshots.put(message, original.snapshotManager().latestSnapshotId()); AtomicInteger attempts = new AtomicInteger(); CatalogEnvironment environment = @@ -769,7 +896,7 @@ protected boolean commitImpl( List statistics) throws Exception { if (snapshot.commitKind() - != (indexBuild + != (indexBuild || update ? Snapshot.CommitKind.APPEND : Snapshot.CommitKind.COMPACT)) { return delegate.commit(baseUuid, snapshot, branch, statistics); @@ -808,7 +935,13 @@ protected void closeImpl() throws Exception { commit(table, message, true); assertThat(attempts).hasValue(3); assertThat(table.snapshotManager().latestSnapshotId()).isEqualTo(9); - checkRows(table, 0, 1, 2, 3, 4, 5); + if (update) { + assertThat(read(table)) + .extracting(row -> row.getInt(1)) + .containsExactlyInAnyOrder(100, 1, 2, 3, 4, 5); + } else { + checkRows(table, 0, 1, 2, 3, 4, 5); + } if (indexBuild) { IndexFileMeta built = message.newFilesIncrement().newIndexFiles().get(0); assertThat(built.globalIndexMeta().rowRange()).isEqualTo(new Range(0, 1)); @@ -821,7 +954,7 @@ protected void closeImpl() throws Exception { assertThat(entry.indexFile().globalIndexMeta().rowRange()) .isEqualTo(new Range(9, 10)); }); - } else { + } else if (!update) { assertThat(message.compactIncrement().compactAfter().get(0).nonNullFirstRowId()) .isZero(); } 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 1b71781ed850..50009e8e921b 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 @@ -921,9 +921,8 @@ public void testReassignKeepsHistoricalEntriesConsistentAfterRetry(String compac @ParameterizedTest @ValueSource(strings = {"a", "b"}) - void testReassignBeforeVideoBackfillCommit(String reassignedPartition) throws Exception { + void testVideoBackfillAdvancesAcrossReassignment(String reassignedPartition) throws Exception { FileStoreTable table = createTableAndAddVideo(); - String originalFile = currentEntriesAtRange(table, "pt=a/", 0, 0).get(0).file().fileName(); List pending = prepareVideo(table, 0); DataEvolutionRowIdReassigner.Result result = new DataEvolutionRowIdReassigner( @@ -931,23 +930,8 @@ table, partitionPredicate(table, reassignedPartition)) .reassign(); assertThat(result.newSnapshotId).isGreaterThan(result.previousSnapshotId); long afterReassign = table.latestSnapshot().get().id(); - if (reassignedPartition.equals("a")) { - assertThatThrownBy(() -> commitDefault(pending)) - .hasMessageContaining("Row ID existence conflict"); - assertThat(table.latestSnapshot().get().id()).isEqualTo(afterReassign); - assertVideoBackfillRows(table, false); - long newRowId = - currentEntries(table).stream() - .map(ManifestEntry::file) - .filter(file -> file.fileName().equals(originalFile)) - .findFirst() - .get() - .nonNullFirstRowId(); - assertThat(newRowId).isGreaterThanOrEqualTo(5); - commitDefault(prepareVideo(table, newRowId)); - } else { - commitDefault(pending); - } + commitDefault(pending); + assertThat(table.latestSnapshot().get().id()).isEqualTo(afterReassign + 1); assertVideoBackfillRows(table, true); } From 9ec68f0a7e109badefb06683da07545e242fea59 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BB=9F=E5=BC=8B?= Date: Wed, 28 Oct 2026 17:55:07 +0800 Subject: [PATCH 06/13] [core] Limit reassignment advancement to index commits --- .../paimon/operation/FileStoreCommitImpl.java | 55 ++---- .../operation/commit/ConflictDetection.java | 9 - .../DataEvolutionConflictDetection.java | 22 --- .../ReassignCompactChangesProvider.java | 87 ++++----- .../operation/commit/StrictModeChecker.java | 4 +- .../CompactReassignReuseTest.java | 177 +++--------------- .../DataEvolutionRowIdReassignerTest.java | 22 ++- 7 files changed, 86 insertions(+), 290 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java index 8ef8ae8d1c59..eebb6086c448 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java @@ -22,7 +22,6 @@ import org.apache.paimon.Snapshot; import org.apache.paimon.Snapshot.CommitKind; import org.apache.paimon.annotation.VisibleForTesting; -import org.apache.paimon.append.dataevolution.SerializationAssignment; import org.apache.paimon.catalog.SnapshotCommit; import org.apache.paimon.data.BinaryRow; import org.apache.paimon.data.InternalRow; @@ -30,7 +29,6 @@ import org.apache.paimon.fs.FileIO; import org.apache.paimon.fs.Path; import org.apache.paimon.globalindex.DataEvolutionGlobalIndexRefreshPlanner; -import org.apache.paimon.index.DataEvolutionIndexSourceMeta; import org.apache.paimon.index.IndexFileMeta; import org.apache.paimon.index.IndexPathFactory; import org.apache.paimon.io.DataFileMeta; @@ -354,15 +352,8 @@ public int commit(ManifestCommittable committable, boolean checkAppendFiles) { int attempts = 0; List commitMessages = committable.fileCommittables(); - Long dataScanSnapshot = configureRowIdCheckFromMessages(commitMessages); + configureRowIdCheckFromMessages(commitMessages); ManifestEntryChanges changes = collectChanges(commitMessages); - if (options.dataEvolutionEnabled() - && dataScanSnapshot == null - && changes.appendTableFiles.stream() - .anyMatch(entry -> entry.file().firstRowId() != null)) { - dataScanSnapshot = options.commitLastSafeSnapshot().filter(id -> id >= 0).orElse(null); - conflictDetection.setRowIdCheckFromSnapshot(dataScanSnapshot); - } Set> materializedBuckets = materializedBuckets(commitMessages); try { List appendSimpleEntries = @@ -402,8 +393,7 @@ public int commit(ManifestCommittable committable, boolean checkAppendFiles) { changes.appendTableFiles, changes.appendChangelog, changes.appendIndexFiles), - commitKind, - dataScanSnapshot), + commitKind), committable.identifier(), committable.watermark(), committable.properties(), @@ -810,8 +800,7 @@ private ManifestEntryChanges collectChanges(List commitMessages) return changes; } - @Nullable - private Long configureRowIdCheckFromMessages(List commitMessages) { + private void configureRowIdCheckFromMessages(List commitMessages) { Long checkFromSnapshot = null; for (CommitMessage message : commitMessages) { Long snapshotId = message.checkFromSnapshot(); @@ -853,29 +842,6 @@ private Long configureRowIdCheckFromMessages(List commitMessages) // A committer can be reused; an untagged commit must not inherit a previous baseline. conflictDetection.setRowIdCheckFromSnapshot(checkFromSnapshot); } - return checkFromSnapshot; - } - - private void checkReassignChanges( - Snapshot snapshot, CommitChanges changes, CommitKind commitKind) { - if (snapshot.commitKind() == CommitKind.OVERWRITE - && SerializationAssignment.planFile(snapshot) != null) { - conflictDetection.checkRowIdReassignment(schemaManager, snapshot, changes, commitKind); - } else if (snapshot.commitKind() == CommitKind.APPEND - || snapshot.commitKind() == CommitKind.OVERWRITE) { - checkIndexBuildChanges( - snapshot, - changes.indexFiles.stream() - .filter( - entry -> - entry.kind() == FileKind.ADD - && entry.indexFile().globalIndexMeta() != null - && DataEvolutionIndexSourceMeta.fromIndexFile( - entry.indexFile()) - .scanSnapshotId() - < snapshot.id()) - .collect(Collectors.toList())); - } } private void checkIndexBuildChanges(Snapshot snapshot, List indexes) { @@ -963,17 +929,18 @@ private Set> materializedBuckets(List co } private CommitChangesProvider appendAndOverwriteChangesProvider( - CommitChanges changes, CommitKind commitKind, @Nullable Long dataScanSnapshot) { - if (options.dataEvolutionEnabled()) { - return ReassignCompactChangesProvider.commitChangesProvider( + CommitChanges changes, CommitKind commitKind) { + if (options.dataEvolutionEnabled() + && commitKind == CommitKind.APPEND + && changes.tableFiles.isEmpty() + && changes.changelogFiles.isEmpty()) { + return ReassignCompactChangesProvider.indexOnlyCommitChangesProvider( fileIO, pathFactory, snapshotManager, - changes, - dataScanSnapshot, + changes.indexFiles, options.commitLastSafeSnapshot().orElse(null), - (snapshot, currentChanges) -> - checkReassignChanges(snapshot, currentChanges, commitKind)); + this::checkIndexBuildChanges); } return CommitChangesProvider.provider( changes.tableFiles, changes.changelogFiles, changes.indexFiles); diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/commit/ConflictDetection.java b/paimon-core/src/main/java/org/apache/paimon/operation/commit/ConflictDetection.java index fba34e614a4d..3685d96311da 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/commit/ConflictDetection.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/commit/ConflictDetection.java @@ -173,15 +173,6 @@ public boolean shouldCheckRowIdFromSnapshot(CommitKind commitKind) { return false; } - /** Checks history in the old coordinates and advances the row-id check boundary. */ - public void checkRowIdReassignment( - SchemaManager schemaManager, - Snapshot snapshot, - CommitChanges changes, - CommitKind commitKind) { - // Only Data Evolution tables rebase row-id-based commits. - } - @Nullable public RowIdConflictChecker createRowIdConflictChecker( SchemaManager schemaManager, List deltaFiles, CommitKind commitKind) { diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionConflictDetection.java b/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionConflictDetection.java index 9581c982d07e..b3a7c82faedb 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionConflictDetection.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionConflictDetection.java @@ -148,28 +148,6 @@ private RowIdConflictCheckStrategy rowIdConflictCheckStrategy() { return rowIdConflictCheckStrategy; } - @Override - public void checkRowIdReassignment( - SchemaManager schemaManager, - Snapshot snapshot, - CommitChanges changes, - CommitKind commitKind) { - if (!shouldCheckRowIdFromSnapshot(commitKind) || snapshot.id() <= rowIdCheckFromSnapshot) { - return; - } - // Check logical changes in the old coordinates before moving past this reassignment. - Optional exception = - checkForRowIdFromSnapshot( - snapshotManager.snapshot(snapshot.id() - 1), - SimpleFileEntry.from(changes.tableFiles), - changes.indexFiles, - createRowIdConflictChecker(schemaManager, changes.tableFiles, commitKind)); - if (exception.isPresent()) { - throw exception.get(); - } - rowIdCheckFromSnapshot = snapshot.id(); - } - @Override public boolean canSkipDataFileConflictDetection( Snapshot latestSnapshot, diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/commit/ReassignCompactChangesProvider.java b/paimon-core/src/main/java/org/apache/paimon/operation/commit/ReassignCompactChangesProvider.java index 3c4974a1beaa..bb73c2dd9448 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/commit/ReassignCompactChangesProvider.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/commit/ReassignCompactChangesProvider.java @@ -44,18 +44,17 @@ import java.util.Objects; import java.util.Set; import java.util.function.BiConsumer; +import java.util.stream.Collectors; import static org.apache.paimon.utils.Preconditions.checkState; -/** Reuses immutable outputs after verified, metadata-only row-id reassignments. */ +/** Reuses immutable compact or index-build outputs after metadata-only row-id reassignments. */ public final class ReassignCompactChangesProvider implements CommitChangesProvider { private final FileIO fileIO; private final FileStorePathFactory paths; private final SnapshotManager snapshots; - @Nullable private final BiConsumer changesCheck; - private final long dataScanSnapshot; - private final long initialScanSnapshot; + @Nullable private final BiConsumer> indexBuildCheck; private final Set reassignments = new HashSet<>(); private long checkedSnapshot; private CommitChanges changes; @@ -66,7 +65,7 @@ public ReassignCompactChangesProvider( SnapshotManager snapshots, long scanSnapshotId, CommitChanges changes) { - this(fileIO, paths, snapshots, scanSnapshotId, changes, scanSnapshotId, null); + this(fileIO, paths, snapshots, scanSnapshotId, changes, null); } private ReassignCompactChangesProvider( @@ -75,46 +74,33 @@ private ReassignCompactChangesProvider( SnapshotManager snapshots, long scanSnapshotId, CommitChanges changes, - long dataScanSnapshot, - @Nullable BiConsumer changesCheck) { + @Nullable BiConsumer> indexBuildCheck) { this.fileIO = fileIO; this.paths = paths; this.snapshots = snapshots; - this.changesCheck = changesCheck; - this.dataScanSnapshot = dataScanSnapshot; - this.initialScanSnapshot = scanSnapshotId; + this.indexBuildCheck = indexBuildCheck; this.checkedSnapshot = scanSnapshotId; this.changes = changes; } - /** Uses data and index scan baselines without changing the original source metadata. */ - public static CommitChangesProvider commitChangesProvider( + /** Uses index scan baselines without changing the original source metadata. */ + public static CommitChangesProvider indexOnlyCommitChangesProvider( FileIO fileIO, FileStorePathFactory paths, SnapshotManager snapshots, - CommitChanges changes, - @Nullable Long dataScanSnapshot, + List indexFiles, @Nullable Long lastSafeSnapshot, - BiConsumer changesCheck) { - boolean hasRowIdFiles = - changes.tableFiles.stream().anyMatch(entry -> entry.file().firstRowId() != null); - if (!changes.changelogFiles.isEmpty() || (hasRowIdFiles && dataScanSnapshot == null)) { - return latest -> changes; - } + BiConsumer> buildCheck) { + CommitChanges changes = + new CommitChanges(Collections.emptyList(), Collections.emptyList(), indexFiles); long scanSnapshotId = lastSafeSnapshot == null || lastSafeSnapshot < 0 ? Long.MAX_VALUE : lastSafeSnapshot; - if (hasRowIdFiles) { - scanSnapshotId = Math.min(scanSnapshotId, dataScanSnapshot); - } boolean hasBuildOutputs = false; - for (IndexManifestEntry entry : changes.indexFiles) { + for (IndexManifestEntry entry : indexFiles) { GlobalIndexMeta meta = entry.indexFile().globalIndexMeta(); if (meta == null) { - if (entry.indexFile().dvRanges() != null) { - continue; - } return latest -> changes; } if (entry.kind() == FileKind.ADD) { @@ -130,27 +116,14 @@ public static CommitChangesProvider commitChangesProvider( .scanSnapshotId()); } } - return !hasBuildOutputs && !hasRowIdFiles + return !hasBuildOutputs ? latest -> changes : new ReassignCompactChangesProvider( - fileIO, - paths, - snapshots, - scanSnapshotId, - changes, - dataScanSnapshot == null ? Long.MAX_VALUE : dataScanSnapshot, - changesCheck); + fileIO, paths, snapshots, scanSnapshotId, changes, buildCheck); } @Override public CommitChanges provide(@Nullable Snapshot latest) { - // An automatic rollback can retry older history while outputs are still unmapped. - if (changesCheck != null - && reassignments.isEmpty() - && latest != null - && latest.id() < checkedSnapshot) { - checkedSnapshot = initialScanSnapshot; - } checkState( latest != null && latest.id() >= checkedSnapshot, "Compaction planning snapshot %s is ahead of the current table.", @@ -167,6 +140,23 @@ public CommitChanges provide(@Nullable Snapshot latest) { snapshot.commitKind() == Snapshot.CommitKind.OVERWRITE ? SerializationAssignment.planFile(snapshot) : null; + if (indexBuildCheck != null + && (snapshot.commitKind() == Snapshot.CommitKind.OVERWRITE + || snapshot.commitKind() == Snapshot.CommitKind.APPEND) + && plan == null) { + List indexes = + changes.indexFiles.stream() + .filter( + entry -> + entry.kind() == FileKind.ADD + && DataEvolutionIndexSourceMeta + .fromIndexFile( + entry.indexFile()) + .scanSnapshotId() + < snapshot.id()) + .collect(Collectors.toList()); + indexBuildCheck.accept(snapshot, indexes); + } if (plan != null) { SerializationAssignment assignment; try { @@ -180,13 +170,8 @@ public CommitChanges provide(@Nullable Snapshot latest) { && Objects.equals(snapshot.nextRowId(), assignment.nextRowId()), "Reassignment plan does not match snapshot %s.", id); - if (changesCheck != null) { - changesCheck.accept(snapshot, changes); - } changes = mapChanges(assignment); reassignments.add(id); - } else if (changesCheck != null) { - changesCheck.accept(snapshot, changes); } checkedSnapshot = id; } @@ -205,11 +190,6 @@ private CommitChanges mapChanges(SerializationAssignment assignment) { List files = new ArrayList<>(changes.tableFiles.size()); for (ManifestEntry entry : changes.tableFiles) { DataFileMeta file = entry.file(); - if (changesCheck != null - && (file.firstRowId() == null || dataScanSnapshot >= assignment.snapshotId())) { - files.add(entry); - continue; - } checkState( file.firstRowId() != null, "Cannot reuse compaction across reassignment: file %s has no first row ID.", @@ -226,9 +206,8 @@ private CommitChanges mapChanges(SerializationAssignment assignment) { List indexes = new ArrayList<>(changes.indexFiles.size()); for (IndexManifestEntry entry : changes.indexFiles) { IndexFileMeta file = entry.indexFile(); - if (changesCheck != null + if (indexBuildCheck != null && entry.kind() == FileKind.ADD - && file.globalIndexMeta() != null && DataEvolutionIndexSourceMeta.fromIndexFile(file).scanSnapshotId() >= assignment.snapshotId()) { indexes.add(entry); diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/commit/StrictModeChecker.java b/paimon-core/src/main/java/org/apache/paimon/operation/commit/StrictModeChecker.java index 92bacaef0600..4a8fb0a84d15 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/commit/StrictModeChecker.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/commit/StrictModeChecker.java @@ -71,9 +71,7 @@ public void check( Set rebasedReassignments) { Set newPartitions = new HashSet<>(newChangedPartitions); for (long id = lastSafeSnapshot + 1; id < newSnapshotId; id++) { - if ((newCommitKind == CommitKind.COMPACT - || newCommitKind == CommitKind.APPEND - || newCommitKind == CommitKind.OVERWRITE) + if ((newCommitKind == CommitKind.COMPACT || newCommitKind == CommitKind.APPEND) && rebasedReassignments.contains(id)) { continue; } diff --git a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/CompactReassignReuseTest.java b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/CompactReassignReuseTest.java index 4f5ba1573743..fbc991bf30a6 100644 --- a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/CompactReassignReuseTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/CompactReassignReuseTest.java @@ -338,8 +338,20 @@ void testIndexBuildAcrossReassignment(String indexType, boolean strict) throws E void testIndexBuildRejectsConcurrentColumnUpdate() throws Exception { FileStoreTable table = prepare(); List messages = buildIndex(table, "id", null); - try (BatchTableCommit commit = table.newBatchWriteBuilder().newCommit()) { - commit.commit(prepareUpdate(table, "id", 100, 0, false)); + BatchWriteBuilder builder = table.newBatchWriteBuilder(); + try (BatchTableWrite write = + builder.newWrite() + .withWriteType(table.rowType().project(Arrays.asList("pt", "id"))); + BatchTableCommit commit = builder.newCommit()) { + write.write(GenericRow.of(BinaryString.fromString("a"), 100)); + List updates = write.prepareCommit(); + for (CommitMessage update : updates) { + ((CommitMessageImpl) update) + .newFilesIncrement() + .newFiles() + .replaceAll(file -> file.assignFirstRowId(0L)); + } + commit.commit(updates); } new DataEvolutionRowIdReassigner(table).reassign(); long snapshot = table.snapshotManager().latestSnapshotId(); @@ -354,134 +366,6 @@ void testIndexBuildRejectsConcurrentColumnUpdate() throws Exception { assertThat(table.snapshotManager().latestSnapshotId()).isEqualTo(snapshot); } - private List prepareUpdate( - FileStoreTable table, String field, Object value, long rowId, boolean tagged) - throws Exception { - long scanSnapshot = table.snapshotManager().latestSnapshotId(); - BatchWriteBuilder builder = table.newBatchWriteBuilder(); - try (BatchTableWrite write = - builder.newWrite() - .withWriteType(table.rowType().project(Arrays.asList("pt", field)))) { - write.write(GenericRow.of(BinaryString.fromString("a"), value)); - List updates = write.prepareCommit(); - for (CommitMessage update : updates) { - ((CommitMessageImpl) update) - .newFilesIncrement() - .newFiles() - .replaceAll(file -> file.assignFirstRowId(rowId)); - } - if (tagged) { - updates.replaceAll( - update -> ((CommitMessageImpl) update).withCheckFromSnapshot(scanSnapshot)); - } - return updates; - } - } - - @ParameterizedTest - @CsvSource({"false,false,false", "false,true,true", "true,false,true", "true,true,false"}) - void testUpdateAndAppendAcrossReassignment(boolean tagged, boolean strict, boolean overwrite) - throws Exception { - FileStoreTable table = prepare(); - long scanSnapshot = table.snapshotManager().latestSnapshotId(); - List messages = prepareUpdate(table, "id", 100, 0, tagged); - CommitMessageImpl message = (CommitMessageImpl) messages.get(0); - DataIncrement increment = message.newFilesIncrement(); - String updatedFile = increment.newFiles().get(0).fileName(); - if (overwrite) { - DataFileMeta input = - table.store().newScan().plan().files().stream() - .filter(e -> !isBlobFile(e.file().fileName())) - .filter(e -> e.file().nonNullFirstRowId() == 0) - .findFirst() - .get() - .file(); - messages.set( - 0, - new CommitMessageImpl( - message.partition(), - message.bucket(), - message.totalBuckets(), - new DataIncrement( - increment.newFiles(), - Collections.singletonList(input), - increment.changelogFiles(), - increment.newIndexFiles(), - increment.deletedIndexFiles()), - message.compactIncrement(), - message.checkFromSnapshot())); - } - messages.addAll(buildIndex(table, "id", "b")); - try (BatchTableWrite write = table.newBatchWriteBuilder().newWrite()) { - write.write( - GenericRow.of( - BinaryString.fromString("b"), - 4, - new BlobData("image-4".getBytes(StandardCharsets.UTF_8)))); - List appends = write.prepareCommit(); - assertThat(appends) - .flatExtracting(m -> ((CommitMessageImpl) m).newFilesIncrement().newFiles()) - .extracting(DataFileMeta::firstRowId) - .containsOnlyNulls(); - messages.addAll(appends); - } - assertThat(new DataEvolutionRowIdReassigner(table).reassign().reassigned).isTrue(); - Map options = new HashMap<>(); - options.put(CoreOptions.COMMIT_LAST_SAFE_SNAPSHOT.key(), Long.toString(scanSnapshot)); - options.put(CoreOptions.COMMIT_STRICT_MODE_ENABLED.key(), Boolean.toString(strict)); - try (BatchTableCommit commit = table.copy(options).newBatchWriteBuilder().newCommit()) { - commit.commit(messages); - } - assertThat(table.snapshotManager().latestSnapshot().commitKind()) - .isEqualTo(overwrite ? Snapshot.CommitKind.OVERWRITE : Snapshot.CommitKind.APPEND); - assertThat(read(table)) - .extracting(row -> row.getInt(1)) - .containsExactlyInAnyOrder(100, 1, 2, 3, 4); - assertThat(table.store().newScan().plan().files()) - .filteredOn(entry -> entry.file().fileName().equals(updatedFile)) - .singleElement() - .satisfies(entry -> assertThat(entry.file().nonNullFirstRowId()).isEqualTo(4)); - } - - @ParameterizedTest - @CsvSource({"-1,true", "0,true", "1,true", "-1,false", "0,false", "1,false"}) - void testUpdateConflictsAcrossReassignment(int phase, boolean sameRow) throws Exception { - FileStoreTable table = prepare(); - List pending = prepareUpdate(table, "id", 100, 0, true); - if (phase >= 0) { - new DataEvolutionRowIdReassigner(table).reassign(); - } - try (BatchTableCommit commit = table.newBatchWriteBuilder().newCommit()) { - commit.commit( - prepareUpdate( - table, "id", 200, (phase < 0 ? 0 : 4) + (sameRow ? 0 : 1), false)); - } - if (phase <= 0) { - if (phase == 0) { - write(table, "b", 4, false); - write(table, "a", 5, false); - } - assertThat(new DataEvolutionRowIdReassigner(table).reassign().reassigned).isTrue(); - } - long snapshot = table.snapshotManager().latestSnapshotId(); - if (sameRow) { - assertThatThrownBy( - () -> { - try (BatchTableCommit commit = - table.newBatchWriteBuilder().newCommit()) { - commit.commit(pending); - } - }) - .hasMessageContaining("conflict"); - assertThat(table.snapshotManager().latestSnapshotId()).isEqualTo(snapshot); - } else { - try (BatchTableCommit commit = table.newBatchWriteBuilder().newCommit()) { - commit.commit(pending); - } - assertThat(read(table)).extracting(row -> row.getInt(1)).contains(100, 200); - } - } - @ParameterizedTest @ValueSource(booleans = {false, true}) void testReuseNormalAndBlobAcrossRepeatedReassignments(boolean blob) throws Exception { @@ -860,25 +744,14 @@ void testExpiredBoundaryAndLegacyOption() throws Exception { } @ParameterizedTest - @CsvSource({ - "compact,false", - "compact,true", - "index,false", - "index,true", - "update,false", - "update,true" - }) - void testReassignDuringCommitRetries(String kind, boolean loseSuccessfulResponse) + @CsvSource({"false,false", "false,true", "true,false", "true,true"}) + void testReassignDuringCommitRetries(boolean indexBuild, boolean loseSuccessfulResponse) throws Exception { - boolean indexBuild = kind.equals("index"); - boolean update = kind.equals("update"); FileStoreTable original = prepare(); CommitMessageImpl message = - update - ? (CommitMessageImpl) prepareUpdate(original, "id", 100, 0, true).get(0) - : indexBuild - ? (CommitMessageImpl) buildIndex(original, "id", "a").get(0) - : compact(original, false); + indexBuild + ? (CommitMessageImpl) buildIndex(original, "id", "a").get(0) + : compact(original, false); lastSafeSnapshots.put(message, original.snapshotManager().latestSnapshotId()); AtomicInteger attempts = new AtomicInteger(); CatalogEnvironment environment = @@ -896,7 +769,7 @@ protected boolean commitImpl( List statistics) throws Exception { if (snapshot.commitKind() - != (indexBuild || update + != (indexBuild ? Snapshot.CommitKind.APPEND : Snapshot.CommitKind.COMPACT)) { return delegate.commit(baseUuid, snapshot, branch, statistics); @@ -935,13 +808,7 @@ protected void closeImpl() throws Exception { commit(table, message, true); assertThat(attempts).hasValue(3); assertThat(table.snapshotManager().latestSnapshotId()).isEqualTo(9); - if (update) { - assertThat(read(table)) - .extracting(row -> row.getInt(1)) - .containsExactlyInAnyOrder(100, 1, 2, 3, 4, 5); - } else { - checkRows(table, 0, 1, 2, 3, 4, 5); - } + checkRows(table, 0, 1, 2, 3, 4, 5); if (indexBuild) { IndexFileMeta built = message.newFilesIncrement().newIndexFiles().get(0); assertThat(built.globalIndexMeta().rowRange()).isEqualTo(new Range(0, 1)); @@ -954,7 +821,7 @@ protected void closeImpl() throws Exception { assertThat(entry.indexFile().globalIndexMeta().rowRange()) .isEqualTo(new Range(9, 10)); }); - } else if (!update) { + } else { assertThat(message.compactIncrement().compactAfter().get(0).nonNullFirstRowId()) .isZero(); } 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 50009e8e921b..1b71781ed850 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 @@ -921,8 +921,9 @@ public void testReassignKeepsHistoricalEntriesConsistentAfterRetry(String compac @ParameterizedTest @ValueSource(strings = {"a", "b"}) - void testVideoBackfillAdvancesAcrossReassignment(String reassignedPartition) throws Exception { + void testReassignBeforeVideoBackfillCommit(String reassignedPartition) throws Exception { FileStoreTable table = createTableAndAddVideo(); + String originalFile = currentEntriesAtRange(table, "pt=a/", 0, 0).get(0).file().fileName(); List pending = prepareVideo(table, 0); DataEvolutionRowIdReassigner.Result result = new DataEvolutionRowIdReassigner( @@ -930,8 +931,23 @@ table, partitionPredicate(table, reassignedPartition)) .reassign(); assertThat(result.newSnapshotId).isGreaterThan(result.previousSnapshotId); long afterReassign = table.latestSnapshot().get().id(); - commitDefault(pending); - assertThat(table.latestSnapshot().get().id()).isEqualTo(afterReassign + 1); + if (reassignedPartition.equals("a")) { + assertThatThrownBy(() -> commitDefault(pending)) + .hasMessageContaining("Row ID existence conflict"); + assertThat(table.latestSnapshot().get().id()).isEqualTo(afterReassign); + assertVideoBackfillRows(table, false); + long newRowId = + currentEntries(table).stream() + .map(ManifestEntry::file) + .filter(file -> file.fileName().equals(originalFile)) + .findFirst() + .get() + .nonNullFirstRowId(); + assertThat(newRowId).isGreaterThanOrEqualTo(5); + commitDefault(prepareVideo(table, newRowId)); + } else { + commitDefault(pending); + } assertVideoBackfillRows(table, true); } From b175c0d54cf87eb464830def9d2b69f8f31adb34 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BB=9F=E5=BC=8B?= Date: Thu, 29 Oct 2026 10:23:56 +0800 Subject: [PATCH 07/13] [core] Preserve existing checks when advancing index commits --- .../paimon/operation/FileStoreCommitImpl.java | 72 +------------------ .../ReassignCompactChangesProvider.java | 34 ++------- .../operation/commit/StrictModeChecker.java | 12 ---- .../CompactReassignReuseTest.java | 25 +++---- 4 files changed, 21 insertions(+), 122 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java index eebb6086c448..8c8eb7788286 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java @@ -28,7 +28,6 @@ import org.apache.paimon.disk.IOManager; import org.apache.paimon.fs.FileIO; import org.apache.paimon.fs.Path; -import org.apache.paimon.globalindex.DataEvolutionGlobalIndexRefreshPlanner; import org.apache.paimon.index.IndexFileMeta; import org.apache.paimon.index.IndexPathFactory; import org.apache.paimon.io.DataFileMeta; @@ -76,8 +75,6 @@ import org.apache.paimon.table.sink.CommitMessage; import org.apache.paimon.table.sink.CommitMessageImpl; import org.apache.paimon.table.sink.CommitPreCallback; -import org.apache.paimon.table.source.ScanMode; -import org.apache.paimon.types.DataField; import org.apache.paimon.types.RowType; import org.apache.paimon.utils.DataFilePathFactories; import org.apache.paimon.utils.FileStorePathFactory; @@ -86,7 +83,6 @@ import org.apache.paimon.utils.InternalRowPartitionComputer; import org.apache.paimon.utils.ListUtils; import org.apache.paimon.utils.Pair; -import org.apache.paimon.utils.Range; import org.apache.paimon.utils.RetryWaiter; import org.apache.paimon.utils.SnapshotManager; @@ -124,7 +120,6 @@ import static org.apache.paimon.types.VectorType.isVectorStoreFile; import static org.apache.paimon.utils.Preconditions.checkArgument; import static org.apache.paimon.utils.Preconditions.checkNotNull; -import static org.apache.paimon.utils.Preconditions.checkState; /** * Default implementation of {@link FileStoreCommit}. @@ -164,7 +159,6 @@ public class FileStoreCommitImpl implements FileStoreCommit { private final IndexManifestFile indexManifestFile; @Nullable private final CommitRollback rollback; private final CommitScanner scanner; - private final Supplier scanSupplier; private final List commitPreCallbacks; private final List commitCallbacks; private final StatsFileHandler statsFileHandler; @@ -217,7 +211,6 @@ public FileStoreCommitImpl( this.indexManifestFile = indexManifestFileFactory.create(); this.rollback = rollback; this.scanner = new CommitScanner(scanSupplier, snapshotManager, indexManifestFile, options); - this.scanSupplier = scanSupplier; this.commitPreCallbacks = commitPreCallbacks; this.commitCallbacks = commitCallbacks; this.retryWaiter = @@ -844,68 +837,6 @@ private void configureRowIdCheckFromMessages(List commitMessages) } } - private void checkIndexBuildChanges(Snapshot snapshot, List indexes) { - if (indexes.isEmpty()) { - return; - } - List partitions = - indexes.stream() - .map(IndexManifestEntry::partition) - .distinct() - .collect(Collectors.toList()); - if (snapshot.commitKind() == CommitKind.OVERWRITE) { - new StrictModeChecker( - snapshotManager, commitUser, scanSupplier, indexManifestFile, true, 0) - .checkIndexBuildOverwrite(snapshot, partitions); - return; - } - Map schemaFields = - schemaManager.schema(snapshot.schemaId()).idToFieldMap(); - Map, List> definitions = - indexes.stream() - .collect( - Collectors.groupingBy( - entry -> - entry.indexFile() - .globalIndexMeta() - .getIndexedFieldIds())); - for (Map.Entry, List> definition : - definitions.entrySet()) { - List fields = - definition.getKey().stream() - .map( - id -> { - DataField field = schemaFields.get(id); - checkState( - field != null, - "Indexed field %s no longer exists.", - id); - return field; - }) - .collect(Collectors.toList()); - List ranges = - definition.getValue().stream() - .map(entry -> entry.indexFile().globalIndexMeta().rowRange()) - .collect(Collectors.toList()); - Iterable entries = - () -> - scanSupplier - .get() - .withSnapshot(snapshot) - .withKind(ScanMode.DELTA) - .withPartitionFilter(partitions) - .withRowRanges(ranges) - .dropStats() - .readFileIterator(); - checkState( - DataEvolutionGlobalIndexRefreshPlanner.findIndexesToRefresh( - schemaManager, entries, definition.getValue(), fields) - .isEmpty(), - "Cannot reuse index build because snapshot %s modified indexed columns.", - snapshot.id()); - } - } - private Set> materializedBuckets(List commitMessages) { if (!options.dataEvolutionEnabled() || !options.deletionVectorsEnabled()) { return Collections.emptySet(); @@ -939,8 +870,7 @@ private CommitChangesProvider appendAndOverwriteChangesProvider( pathFactory, snapshotManager, changes.indexFiles, - options.commitLastSafeSnapshot().orElse(null), - this::checkIndexBuildChanges); + options.commitLastSafeSnapshot().orElse(null)); } return CommitChangesProvider.provider( changes.tableFiles, changes.changelogFiles, changes.indexFiles); diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/commit/ReassignCompactChangesProvider.java b/paimon-core/src/main/java/org/apache/paimon/operation/commit/ReassignCompactChangesProvider.java index bb73c2dd9448..e977b8019bac 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/commit/ReassignCompactChangesProvider.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/commit/ReassignCompactChangesProvider.java @@ -43,8 +43,6 @@ import java.util.List; import java.util.Objects; import java.util.Set; -import java.util.function.BiConsumer; -import java.util.stream.Collectors; import static org.apache.paimon.utils.Preconditions.checkState; @@ -54,7 +52,7 @@ public final class ReassignCompactChangesProvider implements CommitChangesProvid private final FileIO fileIO; private final FileStorePathFactory paths; private final SnapshotManager snapshots; - @Nullable private final BiConsumer> indexBuildCheck; + private final boolean indexOnly; private final Set reassignments = new HashSet<>(); private long checkedSnapshot; private CommitChanges changes; @@ -65,7 +63,7 @@ public ReassignCompactChangesProvider( SnapshotManager snapshots, long scanSnapshotId, CommitChanges changes) { - this(fileIO, paths, snapshots, scanSnapshotId, changes, null); + this(fileIO, paths, snapshots, scanSnapshotId, changes, false); } private ReassignCompactChangesProvider( @@ -74,11 +72,11 @@ private ReassignCompactChangesProvider( SnapshotManager snapshots, long scanSnapshotId, CommitChanges changes, - @Nullable BiConsumer> indexBuildCheck) { + boolean indexOnly) { this.fileIO = fileIO; this.paths = paths; this.snapshots = snapshots; - this.indexBuildCheck = indexBuildCheck; + this.indexOnly = indexOnly; this.checkedSnapshot = scanSnapshotId; this.changes = changes; } @@ -89,8 +87,7 @@ public static CommitChangesProvider indexOnlyCommitChangesProvider( FileStorePathFactory paths, SnapshotManager snapshots, List indexFiles, - @Nullable Long lastSafeSnapshot, - BiConsumer> buildCheck) { + @Nullable Long lastSafeSnapshot) { CommitChanges changes = new CommitChanges(Collections.emptyList(), Collections.emptyList(), indexFiles); long scanSnapshotId = @@ -119,7 +116,7 @@ public static CommitChangesProvider indexOnlyCommitChangesProvider( return !hasBuildOutputs ? latest -> changes : new ReassignCompactChangesProvider( - fileIO, paths, snapshots, scanSnapshotId, changes, buildCheck); + fileIO, paths, snapshots, scanSnapshotId, changes, true); } @Override @@ -140,23 +137,6 @@ public CommitChanges provide(@Nullable Snapshot latest) { snapshot.commitKind() == Snapshot.CommitKind.OVERWRITE ? SerializationAssignment.planFile(snapshot) : null; - if (indexBuildCheck != null - && (snapshot.commitKind() == Snapshot.CommitKind.OVERWRITE - || snapshot.commitKind() == Snapshot.CommitKind.APPEND) - && plan == null) { - List indexes = - changes.indexFiles.stream() - .filter( - entry -> - entry.kind() == FileKind.ADD - && DataEvolutionIndexSourceMeta - .fromIndexFile( - entry.indexFile()) - .scanSnapshotId() - < snapshot.id()) - .collect(Collectors.toList()); - indexBuildCheck.accept(snapshot, indexes); - } if (plan != null) { SerializationAssignment assignment; try { @@ -206,7 +186,7 @@ private CommitChanges mapChanges(SerializationAssignment assignment) { List indexes = new ArrayList<>(changes.indexFiles.size()); for (IndexManifestEntry entry : changes.indexFiles) { IndexFileMeta file = entry.indexFile(); - if (indexBuildCheck != null + if (indexOnly && entry.kind() == FileKind.ADD && DataEvolutionIndexSourceMeta.fromIndexFile(file).scanSnapshotId() >= assignment.snapshotId()) { diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/commit/StrictModeChecker.java b/paimon-core/src/main/java/org/apache/paimon/operation/commit/StrictModeChecker.java index 4a8fb0a84d15..cfb700982b44 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/commit/StrictModeChecker.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/commit/StrictModeChecker.java @@ -130,18 +130,6 @@ public void check( } } - /** Index builds must reject data changes even when optional strict mode is disabled. */ - public void checkIndexBuildOverwrite(Snapshot snapshot, List indexedPartitions) { - Set partitions = new HashSet<>(indexedPartitions); - if (hasOverlappedDataPartition(snapshot, partitions) - || hasOverlappedIndexPartition(snapshot, partitions)) { - throw new RuntimeException( - String.format( - "Cannot reuse index build because OVERWRITE snapshot %s modified indexed partitions.", - snapshot.id())); - } - } - private boolean hasOverlappedDataPartition(Snapshot snapshot, Set newPartitions) { if (newPartitions.isEmpty()) { return false; diff --git a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/CompactReassignReuseTest.java b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/CompactReassignReuseTest.java index fbc991bf30a6..6aa1dd1bd4e4 100644 --- a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/CompactReassignReuseTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/CompactReassignReuseTest.java @@ -334,8 +334,9 @@ void testIndexBuildAcrossReassignment(String indexType, boolean strict) throws E } } - @Test - void testIndexBuildRejectsConcurrentColumnUpdate() throws Exception { + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void testIndexBuildKeepsExistingColumnUpdateBehavior(boolean reassign) throws Exception { FileStoreTable table = prepare(); List messages = buildIndex(table, "id", null); BatchWriteBuilder builder = table.newBatchWriteBuilder(); @@ -353,17 +354,17 @@ void testIndexBuildRejectsConcurrentColumnUpdate() throws Exception { } commit.commit(updates); } - new DataEvolutionRowIdReassigner(table).reassign(); + if (reassign) { + new DataEvolutionRowIdReassigner(table).reassign(); + } long snapshot = table.snapshotManager().latestSnapshotId(); - assertThatThrownBy( - () -> { - try (BatchTableCommit commit = - table.newBatchWriteBuilder().newCommit()) { - commit.commit(messages); - } - }) - .hasMessageContaining("modified indexed columns"); - assertThat(table.snapshotManager().latestSnapshotId()).isEqualTo(snapshot); + try (BatchTableCommit commit = table.newBatchWriteBuilder().newCommit()) { + commit.commit(messages); + } + assertThat(table.snapshotManager().latestSnapshotId()).isEqualTo(snapshot + 1); + assertThat(read(table)) + .extracting(row -> row.getInt(1)) + .containsExactlyInAnyOrder(100, 1, 2, 3); } @ParameterizedTest From f4e92cbf674ccfca3f96636764ef6f5b61475fd1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BB=9F=E5=BC=8B?= Date: Thu, 29 Oct 2026 10:42:39 +0800 Subject: [PATCH 08/13] [core] Reuse existing mapping for index reassignment --- .../ReassignCompactChangesProvider.java | 21 +------------------ .../CompactReassignReuseTest.java | 8 ++----- 2 files changed, 3 insertions(+), 26 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/commit/ReassignCompactChangesProvider.java b/paimon-core/src/main/java/org/apache/paimon/operation/commit/ReassignCompactChangesProvider.java index e977b8019bac..85b5fc823fc0 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/commit/ReassignCompactChangesProvider.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/commit/ReassignCompactChangesProvider.java @@ -52,7 +52,6 @@ public final class ReassignCompactChangesProvider implements CommitChangesProvid private final FileIO fileIO; private final FileStorePathFactory paths; private final SnapshotManager snapshots; - private final boolean indexOnly; private final Set reassignments = new HashSet<>(); private long checkedSnapshot; private CommitChanges changes; @@ -63,20 +62,9 @@ public ReassignCompactChangesProvider( SnapshotManager snapshots, long scanSnapshotId, CommitChanges changes) { - this(fileIO, paths, snapshots, scanSnapshotId, changes, false); - } - - private ReassignCompactChangesProvider( - FileIO fileIO, - FileStorePathFactory paths, - SnapshotManager snapshots, - long scanSnapshotId, - CommitChanges changes, - boolean indexOnly) { this.fileIO = fileIO; this.paths = paths; this.snapshots = snapshots; - this.indexOnly = indexOnly; this.checkedSnapshot = scanSnapshotId; this.changes = changes; } @@ -116,7 +104,7 @@ public static CommitChangesProvider indexOnlyCommitChangesProvider( return !hasBuildOutputs ? latest -> changes : new ReassignCompactChangesProvider( - fileIO, paths, snapshots, scanSnapshotId, changes, true); + fileIO, paths, snapshots, scanSnapshotId, changes); } @Override @@ -186,13 +174,6 @@ private CommitChanges mapChanges(SerializationAssignment assignment) { List indexes = new ArrayList<>(changes.indexFiles.size()); for (IndexManifestEntry entry : changes.indexFiles) { IndexFileMeta file = entry.indexFile(); - if (indexOnly - && entry.kind() == FileKind.ADD - && DataEvolutionIndexSourceMeta.fromIndexFile(file).scanSnapshotId() - >= assignment.snapshotId()) { - indexes.add(entry); - continue; - } GlobalIndexMeta meta = file.globalIndexMeta(); if (meta == null) { checkState( diff --git a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/CompactReassignReuseTest.java b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/CompactReassignReuseTest.java index 6aa1dd1bd4e4..4b69f895df75 100644 --- a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/CompactReassignReuseTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/CompactReassignReuseTest.java @@ -285,9 +285,8 @@ void testIndexBuildAcrossReassignment(String indexType, boolean strict) throws E commit.commit(buildIndex(table, "pt", null)); } long scanSnapshot = table.snapshotManager().latestSnapshotId(); - List messages = buildIndex(table, indexType, "id", "a"); + List messages = buildIndex(table, indexType, "id", null); assertThat(new DataEvolutionRowIdReassigner(table).reassign().reassigned).isTrue(); - messages.addAll(buildIndex(table, indexType, "id", "b")); List files = messages.stream() .flatMap( @@ -317,10 +316,7 @@ void testIndexBuildAcrossReassignment(String indexType, boolean strict) throws E assertThat( DataEvolutionIndexSourceMeta.fromIndexFile(entry.indexFile()) .scanSnapshotId()) - .isEqualTo( - entry.partition().getString(0).toString().equals("a") - ? scanSnapshot - : scanSnapshot + 1); + .isEqualTo(scanSnapshot); } long[] rowIds = {4, 5, 2, 6}; for (int id = 0; id < rowIds.length; id++) { From 73285ee2b2f26c583d8a1e29b00955c798852f05 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BB=9F=E5=BC=8B?= Date: Thu, 29 Oct 2026 11:01:01 +0800 Subject: [PATCH 09/13] [core] Reuse compact provider for index commit advancement --- .../paimon/operation/FileStoreCommitImpl.java | 13 ++---- .../commit/CommitChangesProvider.java | 2 +- .../ReassignCompactChangesProvider.java | 42 +------------------ .../CompactReassignReuseTest.java | 23 ++++++---- 4 files changed, 22 insertions(+), 58 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java index 8c8eb7788286..6fd795390035 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java @@ -861,16 +861,11 @@ private Set> materializedBuckets(List co private CommitChangesProvider appendAndOverwriteChangesProvider( CommitChanges changes, CommitKind commitKind) { - if (options.dataEvolutionEnabled() - && commitKind == CommitKind.APPEND + if (commitKind == CommitKind.APPEND && changes.tableFiles.isEmpty() - && changes.changelogFiles.isEmpty()) { - return ReassignCompactChangesProvider.indexOnlyCommitChangesProvider( - fileIO, - pathFactory, - snapshotManager, - changes.indexFiles, - options.commitLastSafeSnapshot().orElse(null)); + && changes.changelogFiles.isEmpty() + && !changes.indexFiles.isEmpty()) { + return compactChangesProvider(changes, Collections.emptySet()); } return CommitChangesProvider.provider( changes.tableFiles, changes.changelogFiles, changes.indexFiles); diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/commit/CommitChangesProvider.java b/paimon-core/src/main/java/org/apache/paimon/operation/commit/CommitChangesProvider.java index ca57068cd8d5..7759225d3e9d 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/commit/CommitChangesProvider.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/commit/CommitChangesProvider.java @@ -34,7 +34,7 @@ public interface CommitChangesProvider { CommitChanges provide(@Nullable Snapshot latestSnapshot); - /** Reassignments already validated and applied to this provider's results. */ + /** Reassignments already validated and applied to this provider's compact results. */ default Set rebasedReassignments() { return Collections.emptySet(); } diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/commit/ReassignCompactChangesProvider.java b/paimon-core/src/main/java/org/apache/paimon/operation/commit/ReassignCompactChangesProvider.java index 85b5fc823fc0..88d3dcf90670 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/commit/ReassignCompactChangesProvider.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/commit/ReassignCompactChangesProvider.java @@ -21,11 +21,9 @@ import org.apache.paimon.Snapshot; import org.apache.paimon.append.dataevolution.SerializationAssignment; import org.apache.paimon.fs.FileIO; -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.table.SpecialFields; @@ -46,7 +44,7 @@ import static org.apache.paimon.utils.Preconditions.checkState; -/** Reuses immutable compact or index-build outputs after metadata-only row-id reassignments. */ +/** Reuses immutable compact outputs after verified, metadata-only row-id reassignments. */ public final class ReassignCompactChangesProvider implements CommitChangesProvider { private final FileIO fileIO; @@ -69,44 +67,6 @@ public ReassignCompactChangesProvider( this.changes = changes; } - /** Uses index scan baselines without changing the original source metadata. */ - public static CommitChangesProvider indexOnlyCommitChangesProvider( - FileIO fileIO, - FileStorePathFactory paths, - SnapshotManager snapshots, - List indexFiles, - @Nullable Long lastSafeSnapshot) { - CommitChanges changes = - new CommitChanges(Collections.emptyList(), Collections.emptyList(), indexFiles); - long scanSnapshotId = - lastSafeSnapshot == null || lastSafeSnapshot < 0 - ? Long.MAX_VALUE - : lastSafeSnapshot; - boolean hasBuildOutputs = false; - for (IndexManifestEntry entry : indexFiles) { - GlobalIndexMeta meta = entry.indexFile().globalIndexMeta(); - if (meta == null) { - return latest -> changes; - } - if (entry.kind() == FileKind.ADD) { - if (!DataEvolutionIndexSourceMeta.isDataEvolutionMeta(meta.sourceMeta())) { - return latest -> changes; - } - // DELETE source metadata belongs to the old index, not this build's scan. - hasBuildOutputs = true; - scanSnapshotId = - Math.min( - scanSnapshotId, - DataEvolutionIndexSourceMeta.fromIndexFile(entry.indexFile()) - .scanSnapshotId()); - } - } - return !hasBuildOutputs - ? latest -> changes - : new ReassignCompactChangesProvider( - fileIO, paths, snapshots, scanSnapshotId, changes); - } - @Override public CommitChanges provide(@Nullable Snapshot latest) { checkState( diff --git a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/CompactReassignReuseTest.java b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/CompactReassignReuseTest.java index 4b69f895df75..f5fefe03c064 100644 --- a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/CompactReassignReuseTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/CompactReassignReuseTest.java @@ -297,9 +297,7 @@ void testIndexBuildAcrossReassignment(String indexType, boolean strict) throws E .collect(Collectors.toList()); Map options = new HashMap<>(); options.put(CoreOptions.COMMIT_STRICT_MODE_ENABLED.key(), Boolean.toString(strict)); - if (strict) { - options.put(CoreOptions.COMMIT_LAST_SAFE_SNAPSHOT.key(), Long.toString(scanSnapshot)); - } + options.put(CoreOptions.COMMIT_LAST_SAFE_SNAPSHOT.key(), Long.toString(scanSnapshot)); try (BatchTableCommit commit = table.copy(options).newBatchWriteBuilder().newCommit()) { commit.commit(messages); } @@ -334,6 +332,7 @@ void testIndexBuildAcrossReassignment(String indexType, boolean strict) throws E @ValueSource(booleans = {false, true}) void testIndexBuildKeepsExistingColumnUpdateBehavior(boolean reassign) throws Exception { FileStoreTable table = prepare(); + long scanSnapshot = table.snapshotManager().latestSnapshotId(); List messages = buildIndex(table, "id", null); BatchWriteBuilder builder = table.newBatchWriteBuilder(); try (BatchTableWrite write = @@ -354,7 +353,13 @@ void testIndexBuildKeepsExistingColumnUpdateBehavior(boolean reassign) throws Ex new DataEvolutionRowIdReassigner(table).reassign(); } long snapshot = table.snapshotManager().latestSnapshotId(); - try (BatchTableCommit commit = table.newBatchWriteBuilder().newCommit()) { + try (BatchTableCommit commit = + table.copy( + Collections.singletonMap( + CoreOptions.COMMIT_LAST_SAFE_SNAPSHOT.key(), + Long.toString(scanSnapshot))) + .newBatchWriteBuilder() + .newCommit()) { commit.commit(messages); } assertThat(table.snapshotManager().latestSnapshotId()).isEqualTo(snapshot + 1); @@ -714,10 +719,14 @@ void testCorruptPlanFailsClosed() throws Exception { checkRows(table, 0, 1, 2, 3); } - @Test - void testNoBoundaryKeepsExistingConflictCheck() throws Exception { + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void testNoBoundaryKeepsExistingConflictCheck(boolean indexBuild) throws Exception { FileStoreTable table = prepare(); - CommitMessageImpl message = compact(table, false); + CommitMessageImpl message = + indexBuild + ? (CommitMessageImpl) buildIndex(table, "id", "a").get(0) + : compact(table, false); new DataEvolutionRowIdReassigner(table).reassign(); long snapshot = table.snapshotManager().latestSnapshotId(); try (BatchTableCommit commit = table.newBatchWriteBuilder().newCommit()) { From c9e1799eb3cc582f9ad6214d3d2baf45d69ed67d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BB=9F=E5=BC=8B?= Date: Thu, 29 Oct 2026 11:16:28 +0800 Subject: [PATCH 10/13] [core] Unify data evolution commit changes provider --- .../paimon/operation/FileStoreCommitImpl.java | 34 +++++++++---------- .../paimon/operation/FileStoreCommitTest.java | 6 ++-- 2 files changed, 20 insertions(+), 20 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java index 6fd795390035..77044d0f7048 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java @@ -381,12 +381,13 @@ public int commit(ManifestCommittable committable, boolean checkAppendFiles) { attempts += tryCommit( - appendAndOverwriteChangesProvider( + dataEvolutionChangesProvider( new CommitChanges( changes.appendTableFiles, changes.appendChangelog, changes.appendIndexFiles), - commitKind), + commitKind, + Collections.emptySet()), committable.identifier(), committable.watermark(), committable.properties(), @@ -402,11 +403,12 @@ public int commit(ManifestCommittable committable, boolean checkAppendFiles) { || !changes.compactIndexFiles.isEmpty()) { attempts += tryCommit( - compactChangesProvider( + dataEvolutionChangesProvider( new CommitChanges( changes.compactTableFiles, changes.compactChangelog, changes.compactIndexFiles), + CommitKind.COMPACT, materializedBuckets), committable.identifier(), committable.watermark(), @@ -591,11 +593,12 @@ private int overwritePartition( if (withCompact) { attempts += tryCommit( - compactChangesProvider( + dataEvolutionChangesProvider( new CommitChanges( changes.compactTableFiles, emptyList(), changes.compactIndexFiles), + CommitKind.COMPACT, Collections.emptySet()), committable.identifier(), committable.watermark(), @@ -859,24 +862,19 @@ private Set> materializedBuckets(List co return result; } - private CommitChangesProvider appendAndOverwriteChangesProvider( - CommitChanges changes, CommitKind commitKind) { - if (commitKind == CommitKind.APPEND - && changes.tableFiles.isEmpty() - && changes.changelogFiles.isEmpty() - && !changes.indexFiles.isEmpty()) { - return compactChangesProvider(changes, Collections.emptySet()); - } - return CommitChangesProvider.provider( - changes.tableFiles, changes.changelogFiles, changes.indexFiles); - } - @VisibleForTesting - CommitChangesProvider compactChangesProvider( - CommitChanges changes, Set> materializedBuckets) { + CommitChangesProvider dataEvolutionChangesProvider( + CommitChanges changes, + CommitKind commitKind, + Set> materializedBuckets) { if (materializedBuckets.isEmpty()) { Long lastSafeSnapshot = options.commitLastSafeSnapshot().orElse(null); if (options.dataEvolutionEnabled() + && (commitKind == CommitKind.COMPACT + || (commitKind == CommitKind.APPEND + && changes.tableFiles.isEmpty() + && changes.changelogFiles.isEmpty() + && !changes.indexFiles.isEmpty())) && !conflictDetection.shouldCheckRowIdFromSnapshot(CommitKind.COMPACT) && lastSafeSnapshot != null && lastSafeSnapshot >= 0) { 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 99fd1f653277..b85329fd9025 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 @@ -1010,8 +1010,10 @@ public void testMaterializedCompactionOnlyRefreshesGlobalIndexInSameBucket() thr try (FileStoreCommitImpl commit = store.newCommit()) { CommitChanges refreshed = - commit.compactChangesProvider( - changes, Collections.singleton(Pair.of(partition, 0))) + commit.dataEvolutionChangesProvider( + changes, + Snapshot.CommitKind.COMPACT, + Collections.singleton(Pair.of(partition, 0))) .provide(null); assertThat(refreshed.indexFiles).containsExactly(otherBucketDelete); From f0b7aacc695bd1c0cf080d5b897f6812f261ede4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BB=9F=E5=BC=8B?= Date: Thu, 29 Oct 2026 13:27:35 +0800 Subject: [PATCH 11/13] [core] Rename data evolution reassignment changes provider --- .../org/apache/paimon/operation/FileStoreCommitImpl.java | 4 ++-- ...vider.java => DataEvolutionReassignChangesProvider.java} | 6 +++--- .../append/dataevolution/CompactReassignReuseTest.java | 6 +++--- 3 files changed, 8 insertions(+), 8 deletions(-) rename paimon-core/src/main/java/org/apache/paimon/operation/commit/{ReassignCompactChangesProvider.java => DataEvolutionReassignChangesProvider.java} (97%) diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java index 77044d0f7048..61d3f615b9b8 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java @@ -50,8 +50,8 @@ import org.apache.paimon.operation.commit.CommitRollback; import org.apache.paimon.operation.commit.CommitScanner; import org.apache.paimon.operation.commit.ConflictDetection; +import org.apache.paimon.operation.commit.DataEvolutionReassignChangesProvider; import org.apache.paimon.operation.commit.ManifestEntryChanges; -import org.apache.paimon.operation.commit.ReassignCompactChangesProvider; import org.apache.paimon.operation.commit.RetryCommitResult; import org.apache.paimon.operation.commit.RetryCommitResult.CommitFailRetryResult; import org.apache.paimon.operation.commit.RetryCommitResult.ManifestMergeResult; @@ -878,7 +878,7 @@ CommitChangesProvider dataEvolutionChangesProvider( && !conflictDetection.shouldCheckRowIdFromSnapshot(CommitKind.COMPACT) && lastSafeSnapshot != null && lastSafeSnapshot >= 0) { - return new ReassignCompactChangesProvider( + return new DataEvolutionReassignChangesProvider( fileIO, pathFactory, snapshotManager, lastSafeSnapshot, changes); } return CommitChangesProvider.provider( diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/commit/ReassignCompactChangesProvider.java b/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionReassignChangesProvider.java similarity index 97% rename from paimon-core/src/main/java/org/apache/paimon/operation/commit/ReassignCompactChangesProvider.java rename to paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionReassignChangesProvider.java index 88d3dcf90670..efd3351529b5 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/commit/ReassignCompactChangesProvider.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionReassignChangesProvider.java @@ -44,8 +44,8 @@ import static org.apache.paimon.utils.Preconditions.checkState; -/** Reuses immutable compact outputs after verified, metadata-only row-id reassignments. */ -public final class ReassignCompactChangesProvider implements CommitChangesProvider { +/** Reuses immutable outputs after verified, metadata-only row-id reassignments. */ +public final class DataEvolutionReassignChangesProvider implements CommitChangesProvider { private final FileIO fileIO; private final FileStorePathFactory paths; @@ -54,7 +54,7 @@ public final class ReassignCompactChangesProvider implements CommitChangesProvid private long checkedSnapshot; private CommitChanges changes; - public ReassignCompactChangesProvider( + public DataEvolutionReassignChangesProvider( FileIO fileIO, FileStorePathFactory paths, SnapshotManager snapshots, diff --git a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/CompactReassignReuseTest.java b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/CompactReassignReuseTest.java index f5fefe03c064..a79b1fe07482 100644 --- a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/CompactReassignReuseTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/CompactReassignReuseTest.java @@ -43,7 +43,7 @@ import org.apache.paimon.manifest.ManifestEntry; import org.apache.paimon.operation.Lock; import org.apache.paimon.operation.commit.CommitChanges; -import org.apache.paimon.operation.commit.ReassignCompactChangesProvider; +import org.apache.paimon.operation.commit.DataEvolutionReassignChangesProvider; import org.apache.paimon.partition.PartitionStatistics; import org.apache.paimon.predicate.Predicate; import org.apache.paimon.predicate.PredicateBuilder; @@ -849,8 +849,8 @@ void testNoReassignmentDoesNotValidateMappingInputs() throws Exception { Collections.emptyList(), Collections.emptyList(), Collections.singletonList(unsupported)); - ReassignCompactChangesProvider provider = - new ReassignCompactChangesProvider( + DataEvolutionReassignChangesProvider provider = + new DataEvolutionReassignChangesProvider( table.fileIO(), table.store().pathFactory(), table.snapshotManager(), From f60e106cac75f32f5d3a785985436d3f96b7b87c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BB=9F=E5=BC=8B?= Date: Thu, 29 Oct 2026 13:53:44 +0800 Subject: [PATCH 12/13] [core] Strengthen index reassignment regression coverage --- .../CompactReassignReuseTest.java | 149 +++++++++++++++--- 1 file changed, 127 insertions(+), 22 deletions(-) diff --git a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/CompactReassignReuseTest.java b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/CompactReassignReuseTest.java index a79b1fe07482..fbf3b0276def 100644 --- a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/CompactReassignReuseTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/CompactReassignReuseTest.java @@ -83,6 +83,7 @@ import java.util.List; import java.util.Map; import java.util.TreeSet; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; import java.util.stream.Collectors; @@ -167,11 +168,17 @@ private List buildIndex(FileStoreTable table, String field, Strin private List buildIndex( FileStoreTable table, String indexType, String field, String partition) throws Exception { + return buildIndex(table, indexType, Collections.singletonList(field), partition); + } + + private List buildIndex( + FileStoreTable table, String indexType, List fields, String partition) + throws Exception { List builds = new ArrayList<>(); for (DataSplit split : GlobalIndexBuilderUtils.splitByContiguousRowRange( new SortedGlobalIndexScanner(table, indexType) - .withIndexFields(Collections.singletonList(field)) + .withIndexFields(fields) .scan() .get() .entries())) { @@ -180,7 +187,7 @@ private List buildIndex( SortedGlobalIndexTestUtils.buildIndex( table, indexType, - field, + fields, split, table.snapshotManager().latestSnapshotId())); } @@ -278,15 +285,36 @@ private void checkRows(FileStoreTable table, int... expected) throws Exception { } @ParameterizedTest - @CsvSource({"btree,false", "btree,true", "bitmap,false", "bitmap,true"}) - void testIndexBuildAcrossReassignment(String indexType, boolean strict) throws Exception { + @CsvSource({ + "btree,false,after-scan", "btree,true,after-scan", + "bitmap,false,after-scan", "bitmap,true,after-scan", + "composite,false,after-scan", "composite,true,after-scan", + "btree,false,before-scan", "btree,true,before-scan", + "bitmap,false,before-scan", "bitmap,true,before-scan", + "composite,false,before-scan", "composite,true,before-scan", + "btree,false,index-first", "btree,true,index-first", + "bitmap,false,index-first", "bitmap,true,index-first", + "composite,false,index-first", "composite,true,index-first" + }) + void testIndexBuildAcrossReassignment(String indexType, boolean strict, String order) + throws Exception { FileStoreTable table = prepare(); try (BatchTableCommit commit = table.newBatchWriteBuilder().newCommit()) { commit.commit(buildIndex(table, "pt", null)); } + long lastSafeSnapshot = table.snapshotManager().latestSnapshotId(); + boolean reassignBeforeScan = order.equals("before-scan"); + if (reassignBeforeScan) { + assertThat(new DataEvolutionRowIdReassigner(table).reassign().reassigned).isTrue(); + } long scanSnapshot = table.snapshotManager().latestSnapshotId(); - List messages = buildIndex(table, indexType, "id", null); - assertThat(new DataEvolutionRowIdReassigner(table).reassign().reassigned).isTrue(); + boolean composite = indexType.equals("composite"); + List messages = + buildIndex( + table, + composite ? "btree" : indexType, + composite ? Arrays.asList("id", "pt") : Collections.singletonList("id"), + null); List files = messages.stream() .flatMap( @@ -297,12 +325,35 @@ void testIndexBuildAcrossReassignment(String indexType, boolean strict) throws E .collect(Collectors.toList()); Map options = new HashMap<>(); options.put(CoreOptions.COMMIT_STRICT_MODE_ENABLED.key(), Boolean.toString(strict)); - options.put(CoreOptions.COMMIT_LAST_SAFE_SNAPSHOT.key(), Long.toString(scanSnapshot)); - try (BatchTableCommit commit = table.copy(options).newBatchWriteBuilder().newCommit()) { - commit.commit(messages); + options.put(CoreOptions.COMMIT_LAST_SAFE_SNAPSHOT.key(), Long.toString(lastSafeSnapshot)); + FileStoreTable committingTable = table.copy(options); + AtomicBoolean indexCommitted = new AtomicBoolean(); + Runnable commitIndex = + () -> { + if (indexCommitted.compareAndSet(false, true)) { + try (BatchTableCommit commit = + committingTable.newBatchWriteBuilder().newCommit()) { + commit.commit(messages); + } catch (Exception e) { + throw new RuntimeException(e); + } + } + }; + boolean indexFirst = order.equals("index-first"); + if (indexFirst) { + DataEvolutionRowIdReassigner.Result result = + new DataEvolutionRowIdReassigner(table, null, commitIndex).reassign(); + assertThat(result.reassigned).isTrue(); + assertThat(result.previousSnapshotId).isEqualTo(scanSnapshot + 1); + } else { + if (!reassignBeforeScan) { + assertThat(new DataEvolutionRowIdReassigner(table).reassign().reassigned).isTrue(); + } + commitIndex.run(); } + assertThat(indexCommitted).isTrue(); assertThat(table.snapshotManager().latestSnapshot().commitKind()) - .isEqualTo(Snapshot.CommitKind.APPEND); + .isEqualTo(indexFirst ? Snapshot.CommitKind.OVERWRITE : Snapshot.CommitKind.APPEND); List current = table.store().newIndexFileHandler().scanEntries().stream() .filter(e -> e.indexFile().globalIndexMeta().indexFieldId() == 1) @@ -318,7 +369,14 @@ void testIndexBuildAcrossReassignment(String indexType, boolean strict) throws E } long[] rowIds = {4, 5, 2, 6}; for (int id = 0; id < rowIds.length; id++) { - Predicate predicate = new PredicateBuilder(table.rowType()).equal(1, id); + PredicateBuilder builder = new PredicateBuilder(table.rowType()); + Predicate predicate = builder.equal(1, id); + if (composite) { + predicate = + PredicateBuilder.and( + predicate, + builder.equal(0, BinaryString.fromString(id == 2 ? "b" : "a"))); + } try (DataEvolutionGlobalIndexScanner scanner = DataEvolutionGlobalIndexScanner.create(table, null, predicate).get()) { GlobalIndexResult result = scanner.scan(predicate).get(); @@ -408,6 +466,10 @@ void testIndexReplacementKeepsPhysicalFilesAndPointLookup( FileStoreTable table = prepare(); CommitMessageImpl message = compactIndex(table); List outputs = message.compactIncrement().newIndexFiles(); + List replacedFiles = + message.compactIncrement().deletedIndexFiles().stream() + .map(IndexFileMeta::fileName) + .collect(Collectors.toList()); long scanSnapshot = lastSafeSnapshots.get(message); if (indexBuild) { message = @@ -483,7 +545,8 @@ protected void closeImpl() throws Exception { .read(table.snapshotManager().latestSnapshot().indexManifest()); assertThat(current.stream().map(e -> e.indexFile().fileName())) .containsAll( - outputs.stream().map(IndexFileMeta::fileName).collect(Collectors.toList())); + outputs.stream().map(IndexFileMeta::fileName).collect(Collectors.toList())) + .doesNotContainAnyElementsOf(replacedFiles); Predicate predicate = new PredicateBuilder(table.rowType()).equal(1, 1); try (DataEvolutionGlobalIndexScanner scanner = DataEvolutionGlobalIndexScanner.create(table, null, predicate).get()) { @@ -684,10 +747,15 @@ void testReplacedInputsStillConflict() throws Exception { checkRows(table, 0, 1, 2, 3); } - @Test - void testMissingPlanFailsClosed() throws Exception { + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void testMissingPlanFailsClosed(boolean indexBuild) throws Exception { FileStoreTable table = prepare(); - CommitMessageImpl message = compact(table, false); + CommitMessageImpl message = + indexBuild + ? (CommitMessageImpl) buildIndex(table, "id", "a").get(0) + : compact(table, false); + lastSafeSnapshots.put(message, table.snapshotManager().latestSnapshotId()); new DataEvolutionRowIdReassigner(table).reassign(); Snapshot snapshot = table.snapshotManager().latestSnapshot(); Path plan = @@ -701,10 +769,15 @@ void testMissingPlanFailsClosed() throws Exception { checkRows(table, 0, 1, 2, 3); } - @Test - void testCorruptPlanFailsClosed() throws Exception { + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void testCorruptPlanFailsClosed(boolean indexBuild) throws Exception { FileStoreTable table = prepare(); - CommitMessageImpl message = compact(table, false); + CommitMessageImpl message = + indexBuild + ? (CommitMessageImpl) buildIndex(table, "id", "a").get(0) + : compact(table, false); + lastSafeSnapshots.put(message, table.snapshotManager().latestSnapshotId()); new DataEvolutionRowIdReassigner(table).reassign(); Snapshot snapshot = table.snapshotManager().latestSnapshot(); Path path = @@ -720,8 +793,9 @@ void testCorruptPlanFailsClosed() throws Exception { } @ParameterizedTest - @ValueSource(booleans = {false, true}) - void testNoBoundaryKeepsExistingConflictCheck(boolean indexBuild) throws Exception { + @CsvSource({"false,", "true,", "false,-1", "true,-1"}) + void testNoBoundaryKeepsExistingConflictCheck(boolean indexBuild, Long lastSafeSnapshot) + throws Exception { FileStoreTable table = prepare(); CommitMessageImpl message = indexBuild @@ -729,14 +803,45 @@ void testNoBoundaryKeepsExistingConflictCheck(boolean indexBuild) throws Excepti : compact(table, false); new DataEvolutionRowIdReassigner(table).reassign(); long snapshot = table.snapshotManager().latestSnapshotId(); - try (BatchTableCommit commit = table.newBatchWriteBuilder().newCommit()) { + Map options = new HashMap<>(); + options.put(CoreOptions.COMMIT_STRICT_MODE_ENABLED.key(), "false"); + if (lastSafeSnapshot != null) { + options.put(CoreOptions.COMMIT_LAST_SAFE_SNAPSHOT.key(), lastSafeSnapshot.toString()); + } + try (BatchTableCommit commit = table.copy(options).newBatchWriteBuilder().newCommit()) { assertThatThrownBy(() -> commit.commit(Collections.singletonList(message))) - .hasStackTraceContaining("conflict"); + .hasStackTraceContaining( + indexBuild + ? "Global index row ID existence conflict" + : "Row ID existence conflict"); } assertThat(table.snapshotManager().latestSnapshotId()).isEqualTo(snapshot); checkRows(table, 0, 1, 2, 3); } + @ParameterizedTest + @ValueSource(strings = {"snapshot", "manifest-list"}) + void testIndexBuildMissingMetadataFailsClosed(String missing) throws Exception { + FileStoreTable table = prepare(); + CommitMessageImpl message = (CommitMessageImpl) buildIndex(table, "id", "a").get(0); + long boundary = table.snapshotManager().latestSnapshotId(); + lastSafeSnapshots.put(message, boundary); + write(table, "b", 4, false); + new DataEvolutionRowIdReassigner(table).reassign(); + Snapshot latest = table.snapshotManager().latestSnapshot(); + boolean snapshotMissing = missing.equals("snapshot"); + Path path = + snapshotMissing + ? table.snapshotManager().snapshotPath(boundary + 1) + : table.store().pathFactory().toManifestListPath(latest.baseManifestList()); + assertThat(table.fileIO().delete(path, false)).isTrue(); + assertThatThrownBy(() -> commit(table, message, false)) + .hasStackTraceContaining( + snapshotMissing ? "missing snapshot" : latest.baseManifestList()); + assertThat(table.snapshotManager().latestSnapshotId()).isEqualTo(latest.id()); + assertThat(table.snapshotManager().snapshotExists(latest.id() + 1)).isFalse(); + } + @Test void testExpiredBoundaryAndLegacyOption() throws Exception { FileStoreTable table = prepare(); From 498197e7d7bcbc5e647c2753b472411fd84f1a2c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BB=9F=E5=BC=8B?= Date: Thu, 29 Oct 2026 14:37:45 +0800 Subject: [PATCH 13/13] [core] Rename changes provider to DataEvolutionChangesProvider --- .../org/apache/paimon/operation/FileStoreCommitImpl.java | 4 ++-- ...angesProvider.java => DataEvolutionChangesProvider.java} | 4 ++-- .../append/dataevolution/CompactReassignReuseTest.java | 6 +++--- 3 files changed, 7 insertions(+), 7 deletions(-) rename paimon-core/src/main/java/org/apache/paimon/operation/commit/{DataEvolutionReassignChangesProvider.java => DataEvolutionChangesProvider.java} (98%) diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java index 61d3f615b9b8..3354d4ab1a3b 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java @@ -50,7 +50,7 @@ import org.apache.paimon.operation.commit.CommitRollback; import org.apache.paimon.operation.commit.CommitScanner; import org.apache.paimon.operation.commit.ConflictDetection; -import org.apache.paimon.operation.commit.DataEvolutionReassignChangesProvider; +import org.apache.paimon.operation.commit.DataEvolutionChangesProvider; import org.apache.paimon.operation.commit.ManifestEntryChanges; import org.apache.paimon.operation.commit.RetryCommitResult; import org.apache.paimon.operation.commit.RetryCommitResult.CommitFailRetryResult; @@ -878,7 +878,7 @@ CommitChangesProvider dataEvolutionChangesProvider( && !conflictDetection.shouldCheckRowIdFromSnapshot(CommitKind.COMPACT) && lastSafeSnapshot != null && lastSafeSnapshot >= 0) { - return new DataEvolutionReassignChangesProvider( + return new DataEvolutionChangesProvider( fileIO, pathFactory, snapshotManager, lastSafeSnapshot, changes); } return CommitChangesProvider.provider( diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionReassignChangesProvider.java b/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionChangesProvider.java similarity index 98% rename from paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionReassignChangesProvider.java rename to paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionChangesProvider.java index efd3351529b5..620326f69e4c 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionReassignChangesProvider.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionChangesProvider.java @@ -45,7 +45,7 @@ import static org.apache.paimon.utils.Preconditions.checkState; /** Reuses immutable outputs after verified, metadata-only row-id reassignments. */ -public final class DataEvolutionReassignChangesProvider implements CommitChangesProvider { +public final class DataEvolutionChangesProvider implements CommitChangesProvider { private final FileIO fileIO; private final FileStorePathFactory paths; @@ -54,7 +54,7 @@ public final class DataEvolutionReassignChangesProvider implements CommitChanges private long checkedSnapshot; private CommitChanges changes; - public DataEvolutionReassignChangesProvider( + public DataEvolutionChangesProvider( FileIO fileIO, FileStorePathFactory paths, SnapshotManager snapshots, diff --git a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/CompactReassignReuseTest.java b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/CompactReassignReuseTest.java index fbf3b0276def..6346f25d0390 100644 --- a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/CompactReassignReuseTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/CompactReassignReuseTest.java @@ -43,7 +43,7 @@ import org.apache.paimon.manifest.ManifestEntry; import org.apache.paimon.operation.Lock; import org.apache.paimon.operation.commit.CommitChanges; -import org.apache.paimon.operation.commit.DataEvolutionReassignChangesProvider; +import org.apache.paimon.operation.commit.DataEvolutionChangesProvider; import org.apache.paimon.partition.PartitionStatistics; import org.apache.paimon.predicate.Predicate; import org.apache.paimon.predicate.PredicateBuilder; @@ -954,8 +954,8 @@ void testNoReassignmentDoesNotValidateMappingInputs() throws Exception { Collections.emptyList(), Collections.emptyList(), Collections.singletonList(unsupported)); - DataEvolutionReassignChangesProvider provider = - new DataEvolutionReassignChangesProvider( + DataEvolutionChangesProvider provider = + new DataEvolutionChangesProvider( table.fileIO(), table.store().pathFactory(), table.snapshotManager(),