From 033e5034f688ab5d402c5cab9fea36c0b53f19f2 Mon Sep 17 00:00:00 2001 From: yangjie01 Date: Tue, 15 Sep 2026 16:26:14 +0800 Subject: [PATCH 1/3] [core] Validate chain partition drops against the post-commit state ChainTableCommitPreCallback rejects dropping a snapshot partition whose delta followers would lose their baseline. The predecessor and successor candidates came from the pre-commit partition list, so when one commit dropped several partitions of a group at once (batch INSERT OVERWRITE or a rollback), a partition dropped by that same commit still counted as the predecessor of the next one: the check passed, the commit landed, and the delta partitions silently fell back to delta-only reads without their baseline rows. Validate the post-commit state instead. A partition counts as dropped only when the commit deletes all of its base files; a rollback deletes per-file and may leave a partition partially alive, and such a partition survives and still anchors its delta followers, so it is neither excluded from the candidates nor itself re-validated. Only fully dropped partitions are validated, and single-partition drops are unaffected. --- .../ChainTableCommitPreCallback.java | 54 +++++++++++-- .../ChainTablePartitionExpireTest.java | 78 +++++++++++++++++++ 2 files changed, 127 insertions(+), 5 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/metastore/ChainTableCommitPreCallback.java b/paimon-core/src/main/java/org/apache/paimon/metastore/ChainTableCommitPreCallback.java index 03399a4fde77..85a4473a9344 100644 --- a/paimon-core/src/main/java/org/apache/paimon/metastore/ChainTableCommitPreCallback.java +++ b/paimon-core/src/main/java/org/apache/paimon/metastore/ChainTableCommitPreCallback.java @@ -29,7 +29,6 @@ import org.apache.paimon.manifest.ManifestEntry; import org.apache.paimon.manifest.PartitionEntry; import org.apache.paimon.manifest.SimpleFileEntry; -import org.apache.paimon.operation.commit.ManifestEntryChanges; import org.apache.paimon.predicate.Predicate; import org.apache.paimon.predicate.PredicateBuilder; import org.apache.paimon.table.FileStoreTable; @@ -45,8 +44,12 @@ import org.slf4j.LoggerFactory; import java.util.ArrayList; +import java.util.HashMap; +import java.util.HashSet; import java.util.List; +import java.util.Map; import java.util.Optional; +import java.util.Set; import java.util.stream.Collectors; /** @@ -55,7 +58,8 @@ *

This callback performs a pre-check before dropping partitions on the snapshot branch of a * chain table. It verifies that a snapshot partition being dropped is either followed by no delta * partitions in the chain interval or has a previous snapshot partition that can serve as its - * predecessor. + * predecessor. The check considers the post-commit state, so partitions dropped by the same commit + * do not count as predecessors or successors. * *

The callback is only executed when all of following conditions are met: * @@ -99,8 +103,6 @@ public void call( if (!isPureDeleteCommit(deltaFiles, indexFiles)) { return; } - List changedPartitions = - ManifestEntryChanges.changedPartitions(deltaFiles, indexFiles); FileStoreTable candidateTable = ChainTableUtils.resolveChainPrimaryTable(table); FileStoreTable deltaTable = candidateTable.switchToBranch(coreOptions.scanFallbackDeltaBranch()); @@ -123,13 +125,24 @@ public void call( RecordComparator chainComparator = CodeGenUtils.newRecordComparator(projector.chainPartitionType().getFieldTypes()); + // The pure-delete commit may drop several partitions of a group at once (batch + // overwrite / rollback). Validation must consider the post-commit state: a partition + // dropped by this very commit can no longer serve as predecessor or successor, or a + // delta partition would silently lose its baseline rows once the commit lands. A + // partition counts as dropped only if the commit deletes ALL of its base files; a + // rollback deletes per-file and may leave a partition partially alive, and such a + // partition still serves as a baseline after the commit. + Set droppedPartitions = fullyDroppedPartitions(baseFiles, deltaFiles); List snapshotPartitions = table.newSnapshotReader().partitionEntries().stream() .map(PartitionEntry::partition) + .filter(partition -> !droppedPartitions.contains(partition)) .collect(Collectors.toList()); SnapshotReader deltaSnapshotReader = deltaTable.newSnapshotReader(); PredicateBuilder builder = new PredicateBuilder(partitionType); - for (BinaryRow partition : changedPartitions) { + // only fully dropped partitions can break the chain; a partially deleted partition + // survives the commit and keeps anchoring its delta followers + for (BinaryRow partition : droppedPartitions) { BinaryRow partitionGroup = projector.extractGroupPartition(partition); BinaryRow partitionChain = projector.extractChainPartition(partition); @@ -181,6 +194,37 @@ public void call( } } + private Set fullyDroppedPartitions( + List baseFiles, List deltaFiles) { + Map> deletedFilesByPartition = new HashMap<>(); + for (ManifestEntry entry : deltaFiles) { + if (entry.kind() == FileKind.DELETE) { + deletedFilesByPartition + .computeIfAbsent(entry.partition(), k -> new HashSet<>()) + .add(entry.bucket() + "/" + entry.file().fileName()); + } + } + Set droppedPartitions = new HashSet<>(); + for (Map.Entry> deleted : deletedFilesByPartition.entrySet()) { + BinaryRow partition = deleted.getKey(); + Set deletedFiles = deleted.getValue(); + List partitionBaseFiles = + baseFiles.stream() + .filter(base -> base.partition().equals(partition)) + .map(base -> base.bucket() + "/" + base.fileName()) + .collect(Collectors.toList()); + // A partition with no base files was never a baseline, so it cannot break the + // chain; treat it as dropped only when the commit deletes every one of its base + // files. Guarding the empty case keeps a caller that passes an incomplete base + // file list (e.g. a path that scans only changed partitions) from misclassifying + // a partially deleted survivor as fully dropped. + if (!partitionBaseFiles.isEmpty() && deletedFiles.containsAll(partitionBaseFiles)) { + droppedPartitions.add(partition); + } + } + return droppedPartitions; + } + private boolean isPureDeleteCommit( List deltaFiles, List indexFiles) { return deltaFiles.stream().allMatch(f -> f.kind() == FileKind.DELETE) diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/ChainTablePartitionExpireTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/ChainTablePartitionExpireTest.java index c06b09066202..ac7176dcbf82 100644 --- a/paimon-core/src/test/java/org/apache/paimon/operation/ChainTablePartitionExpireTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/operation/ChainTablePartitionExpireTest.java @@ -637,6 +637,84 @@ public void testRollbackToAsLatestRejectedWhenDroppingAnchorSnapshotPartition() .isEqualTo(2L); } + @Test + public void testRollbackRejectedWhenBatchDroppingBaselinesOfDelta() throws Exception { + Path tablePath = tablePath("rollback_reject_batch_baseline"); + createChainTable(tablePath, true); + + FileStoreTable snapshotTable = loadTable(tablePath).switchToBranch("snapshot"); + FileStoreTable deltaTable = loadTable(tablePath).switchToBranch("delta"); + + writeGrouped(snapshotTable, "US", "20250101", "v1"); // snapshot #1, unrelated group + writeGrouped(snapshotTable, "CN", "20250201", "v2"); // snapshot #2, CN baseline + writeGrouped(snapshotTable, "CN", "20250301", "v3"); // snapshot #3, CN baseline + // a delta partition anchored on CN/20250301 + writeGrouped(deltaTable, "CN", "20250315", "v4"); + + // Rolling back to snapshot #1 drops CN/20250201 and CN/20250301 in ONE pure-delete + // commit. Validating CN/20250301 against the PRE-commit partition list wrongly + // accepts it: CN/20250201 (dropped by the same commit) still counts as its + // predecessor, and after the commit the delta has no baseline at all. + FileStoreTable snapshotBranch = loadTable(tablePath).switchToBranch("snapshot"); + Snapshot target = snapshotBranch.snapshotManager().snapshot(1); + String protectionTag = "rollback-to-as-latest-" + target.id() + "-" + UUID.randomUUID(); + snapshotBranch + .tagManager() + .createTag(target, protectionTag, null, Collections.emptyList(), false); + try (TableCommitImpl commit = snapshotBranch.newCommit(commitUser)) { + assertThatThrownBy( + () -> + commit.rollbackToAsLatest( + snapshotBranch.tagManager().getOrThrow(protectionTag))) + .hasMessageContaining("Snapshot partition cannot be dropped"); + } + // The dangerous rollback was aborted, so the latest snapshot is unchanged. + assertThat( + loadTable(tablePath) + .switchToBranch("snapshot") + .snapshotManager() + .latestSnapshotId()) + .isEqualTo(3L); + } + + @Test + public void testRollbackAllowedWhenPartitionOnlyPartiallyDeleted() throws Exception { + Path tablePath = tablePath("rollback_partial_partition"); + createChainTable(tablePath, true); + + FileStoreTable snapshotTable = loadTable(tablePath).switchToBranch("snapshot"); + FileStoreTable deltaTable = loadTable(tablePath).switchToBranch("delta"); + + writeGrouped(snapshotTable, "US", "20250101", "v1"); // snapshot #1 + writeGrouped(snapshotTable, "CN", "20250201", "v2"); // snapshot #2, first file + writeGrouped(snapshotTable, "CN", "20250201", "v3"); // snapshot #3, second file + writeGrouped(snapshotTable, "CN", "20250301", "v4"); // snapshot #4 + // a delta partition anchored on CN/20250301 + writeGrouped(deltaTable, "CN", "20250315", "v5"); + + // Rolling back to snapshot #2 deletes the second CN/20250201 file (the partition + // itself survives with its first file) and fully drops CN/20250301. The surviving + // CN/20250201 must still count as the baseline of CN/20250315, so the rollback is + // safe and must not be vetoed. + FileStoreTable snapshotBranch = loadTable(tablePath).switchToBranch("snapshot"); + Snapshot target = snapshotBranch.snapshotManager().snapshot(2); + String protectionTag = "rollback-to-as-latest-" + target.id() + "-" + UUID.randomUUID(); + snapshotBranch + .tagManager() + .createTag(target, protectionTag, null, Collections.emptyList(), false); + try (TableCommitImpl commit = snapshotBranch.newCommit(commitUser)) { + commit.rollbackToAsLatest(snapshotBranch.tagManager().getOrThrow(protectionTag)); + } + assertThat( + loadTable(tablePath) + .switchToBranch("snapshot") + .snapshotManager() + .latestSnapshotId()) + .isEqualTo(5L); + assertThat(listGroupedPartitions(loadTable(tablePath).switchToBranch("snapshot"))) + .containsExactly("CN|20250201", "US|20250101"); + } + private Path tablePath(String tableName) { return new Path(tempDir.toUri().toString(), tableName); } From 8d2994cf197bf6108bd0b5a20c6994c8d070f14e Mon Sep 17 00:00:00 2001 From: yangjie01 Date: Wed, 23 Sep 2026 11:05:15 +0800 Subject: [PATCH 2/3] [core] Allow a chain OVERWRITE to truncate a whole group's snapshot partitions The post-commit-state validation rejected a legitimate batch INSERT OVERWRITE. ChainTableOverwriteCommitCallback truncates the snapshot partitions that the overwrite just rewrote on the delta branch, and the pre-callback saw the freshly rewritten delta followers as stranded and threw "Snapshot partition cannot be dropped" (SparkChainTableITCase). Those followers hold fresh, complete data and need no baseline. Pass the freshly rewritten delta partitions from the overwrite callback to the pre-callback via a thread-local scoped around the synchronous truncate, and exclude them from the follower check. A standalone drop or rollback rewrites no delta, so a genuinely stranded follower is still rejected. --- .../ChainTableCommitPreCallback.java | 9 +++ .../ChainTableOverwriteCommitCallback.java | 14 ++++- .../metastore/ChainTableOverwriteScope.java | 61 +++++++++++++++++++ 3 files changed, 83 insertions(+), 1 deletion(-) create mode 100644 paimon-core/src/main/java/org/apache/paimon/metastore/ChainTableOverwriteScope.java diff --git a/paimon-core/src/main/java/org/apache/paimon/metastore/ChainTableCommitPreCallback.java b/paimon-core/src/main/java/org/apache/paimon/metastore/ChainTableCommitPreCallback.java index 85a4473a9344..08f02b2e95b0 100644 --- a/paimon-core/src/main/java/org/apache/paimon/metastore/ChainTableCommitPreCallback.java +++ b/paimon-core/src/main/java/org/apache/paimon/metastore/ChainTableCommitPreCallback.java @@ -140,6 +140,12 @@ public void call( .collect(Collectors.toList()); SnapshotReader deltaSnapshotReader = deltaTable.newSnapshotReader(); PredicateBuilder builder = new PredicateBuilder(partitionType); + // Delta partitions that the triggering chain-table OVERWRITE just rewrote hold fresh, + // complete data and do not depend on a snapshot baseline, so dropping their baseline is + // intended rather than an orphan. A standalone drop or a rollback leaves this empty, so a + // genuinely stranded follower is still rejected below. + Set freshlyWrittenDeltaPartitions = + ChainTableOverwriteScope.freshlyWrittenDeltaPartitions(); // only fully dropped partitions can break the chain; a partially deleted partition // survives the commit and keeps anchoring its delta followers for (BinaryRow partition : droppedPartitions) { @@ -179,6 +185,9 @@ public void call( nextSnapshotPartition, chainComparator, projector)) + .filter( + deltaPartition -> + !freshlyWrittenDeltaPartitions.contains(deltaPartition)) .collect(Collectors.toList()); boolean canDrop = deltaFollowingPartitions.isEmpty() || preSnapshotPartition.isPresent(); diff --git a/paimon-core/src/main/java/org/apache/paimon/metastore/ChainTableOverwriteCommitCallback.java b/paimon-core/src/main/java/org/apache/paimon/metastore/ChainTableOverwriteCommitCallback.java index 2d1a0cd7050f..cafbe17db615 100644 --- a/paimon-core/src/main/java/org/apache/paimon/metastore/ChainTableOverwriteCommitCallback.java +++ b/paimon-core/src/main/java/org/apache/paimon/metastore/ChainTableOverwriteCommitCallback.java @@ -29,8 +29,10 @@ import org.apache.paimon.utils.ChainTableUtils; import org.apache.paimon.utils.InternalRowPartitionComputer; +import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.stream.Collectors; /** @@ -97,7 +99,17 @@ public void call(Context context) { .collect(Collectors.toList()); try (BatchTableCommit commit = snapshotTable.newBatchWriteBuilder().newCommit()) { - commit.truncatePartitions(candidatePartitions); + // The truncated snapshot partitions are exactly the partitions this overwrite just + // rewrote on the delta branch, so their surviving delta followers hold fresh data + // and do not depend on a snapshot baseline. Hand that set to the pre-callback that + // the truncate triggers so it does not reject dropping their baselines. + Set freshlyWritten = new HashSet<>(overwritePartitions); + ChainTableOverwriteScope.setFreshlyWrittenDeltaPartitions(freshlyWritten); + try { + commit.truncatePartitions(candidatePartitions); + } finally { + ChainTableOverwriteScope.clear(); + } } catch (Exception e) { throw new RuntimeException( String.format( diff --git a/paimon-core/src/main/java/org/apache/paimon/metastore/ChainTableOverwriteScope.java b/paimon-core/src/main/java/org/apache/paimon/metastore/ChainTableOverwriteScope.java new file mode 100644 index 000000000000..c883717ad87c --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/metastore/ChainTableOverwriteScope.java @@ -0,0 +1,61 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.metastore; + +import org.apache.paimon.data.BinaryRow; + +import java.util.Collections; +import java.util.Set; + +/** + * Carries the set of delta partitions that a chain-table OVERWRITE freshly (re)wrote from {@link + * ChainTableOverwriteCommitCallback} to the snapshot-branch truncate's {@link + * ChainTableCommitPreCallback}. + * + *

The overwrite callback truncates the snapshot branch synchronously, on the same thread, and + * that truncate is what invokes the pre-callback. A thread-local scoped around the truncate call + * therefore reaches the pre-callback without changing the generic commit path or the {@link + * org.apache.paimon.table.sink.CommitPreCallback} signature. + * + *

Why the pre-callback needs it: a delta partition that this overwrite just rewrote holds fresh, + * complete data and does not depend on a snapshot baseline, so dropping that baseline is intended, + * not an orphan. Only the trigger site knows which partitions were freshly written; a standalone + * drop or a rollback sets nothing here, so the pre-callback keeps rejecting genuinely stranded + * followers. + */ +final class ChainTableOverwriteScope { + + private static final ThreadLocal> FRESHLY_WRITTEN_DELTA_PARTITIONS = + new ThreadLocal<>(); + + private ChainTableOverwriteScope() {} + + static void setFreshlyWrittenDeltaPartitions(Set partitions) { + FRESHLY_WRITTEN_DELTA_PARTITIONS.set(partitions); + } + + static void clear() { + FRESHLY_WRITTEN_DELTA_PARTITIONS.remove(); + } + + static Set freshlyWrittenDeltaPartitions() { + Set partitions = FRESHLY_WRITTEN_DELTA_PARTITIONS.get(); + return partitions == null ? Collections.emptySet() : partitions; + } +} From 73342655dcaf62e8d5d4a3b909bf583bfc7fb710 Mon Sep 17 00:00:00 2001 From: yangjie01 Date: Wed, 23 Sep 2026 13:26:36 +0800 Subject: [PATCH 3/3] [core] Harden the chain-overwrite scope thread-local Follow-up from reviewing the freshly-written-partition handoff. Save and restore the previous scope value instead of unconditionally clearing it, so a future nested chain overwrite on the same thread cannot wipe an outer scope, and return an unmodifiable view of the set from the getter. --- .../ChainTableOverwriteCommitCallback.java | 5 +++-- .../metastore/ChainTableOverwriteScope.java | 22 +++++++++++++++---- 2 files changed, 21 insertions(+), 6 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/metastore/ChainTableOverwriteCommitCallback.java b/paimon-core/src/main/java/org/apache/paimon/metastore/ChainTableOverwriteCommitCallback.java index cafbe17db615..67255bd7f402 100644 --- a/paimon-core/src/main/java/org/apache/paimon/metastore/ChainTableOverwriteCommitCallback.java +++ b/paimon-core/src/main/java/org/apache/paimon/metastore/ChainTableOverwriteCommitCallback.java @@ -104,11 +104,12 @@ public void call(Context context) { // and do not depend on a snapshot baseline. Hand that set to the pre-callback that // the truncate triggers so it does not reject dropping their baselines. Set freshlyWritten = new HashSet<>(overwritePartitions); - ChainTableOverwriteScope.setFreshlyWrittenDeltaPartitions(freshlyWritten); + Set previous = + ChainTableOverwriteScope.setFreshlyWrittenDeltaPartitions(freshlyWritten); try { commit.truncatePartitions(candidatePartitions); } finally { - ChainTableOverwriteScope.clear(); + ChainTableOverwriteScope.restore(previous); } } catch (Exception e) { throw new RuntimeException( diff --git a/paimon-core/src/main/java/org/apache/paimon/metastore/ChainTableOverwriteScope.java b/paimon-core/src/main/java/org/apache/paimon/metastore/ChainTableOverwriteScope.java index c883717ad87c..3648102abeec 100644 --- a/paimon-core/src/main/java/org/apache/paimon/metastore/ChainTableOverwriteScope.java +++ b/paimon-core/src/main/java/org/apache/paimon/metastore/ChainTableOverwriteScope.java @@ -46,16 +46,30 @@ final class ChainTableOverwriteScope { private ChainTableOverwriteScope() {} - static void setFreshlyWrittenDeltaPartitions(Set partitions) { + /** + * Installs {@code partitions} as the freshly-written set and returns whatever was installed + * before, so the caller restores it in a finally rather than clearing unconditionally. + * Restoring keeps the scheme correct even if the truncate ever nests another chain overwrite on + * the same thread. + */ + static Set setFreshlyWrittenDeltaPartitions(Set partitions) { + Set previous = FRESHLY_WRITTEN_DELTA_PARTITIONS.get(); FRESHLY_WRITTEN_DELTA_PARTITIONS.set(partitions); + return previous; } - static void clear() { - FRESHLY_WRITTEN_DELTA_PARTITIONS.remove(); + static void restore(Set previous) { + if (previous == null) { + FRESHLY_WRITTEN_DELTA_PARTITIONS.remove(); + } else { + FRESHLY_WRITTEN_DELTA_PARTITIONS.set(previous); + } } static Set freshlyWrittenDeltaPartitions() { Set partitions = FRESHLY_WRITTEN_DELTA_PARTITIONS.get(); - return partitions == null ? Collections.emptySet() : partitions; + return partitions == null + ? Collections.emptySet() + : Collections.unmodifiableSet(partitions); } }