This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 157e02bd43 [core] Validate chain partition drops against the
post-commit state (#10127)
157e02bd43 is described below
commit 157e02bd432fbd332fdbbf0b8cfccd2897ca5675
Author: YangJie <[email protected]>
AuthorDate: Wed Sep 23 23:00:19 2026 -0400
[core] Validate chain partition drops against the post-commit state (#10127)
---
.../metastore/ChainTableCommitPreCallback.java | 63 +++++++++++++++--
.../ChainTableOverwriteCommitCallback.java | 15 ++++-
.../paimon/metastore/ChainTableOverwriteScope.java | 75 +++++++++++++++++++++
.../operation/ChainTablePartitionExpireTest.java | 78 ++++++++++++++++++++++
4 files changed, 225 insertions(+), 6 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 03399a4fde..08f02b2e95 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.IndexManifestEntry;
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.Logger;
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 @@ import java.util.stream.Collectors;
* <p>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.
*
* <p>The callback is only executed when all of following conditions are met:
*
@@ -99,8 +103,6 @@ public class ChainTableCommitPreCallback implements
CommitPreCallback {
if (!isPureDeleteCommit(deltaFiles, indexFiles)) {
return;
}
- List<BinaryRow> changedPartitions =
- ManifestEntryChanges.changedPartitions(deltaFiles, indexFiles);
FileStoreTable candidateTable =
ChainTableUtils.resolveChainPrimaryTable(table);
FileStoreTable deltaTable =
candidateTable.switchToBranch(coreOptions.scanFallbackDeltaBranch());
@@ -123,13 +125,30 @@ public class ChainTableCommitPreCallback implements
CommitPreCallback {
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<BinaryRow> droppedPartitions = fullyDroppedPartitions(baseFiles,
deltaFiles);
List<BinaryRow> 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) {
+ // 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<BinaryRow> 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) {
BinaryRow partitionGroup =
projector.extractGroupPartition(partition);
BinaryRow partitionChain =
projector.extractChainPartition(partition);
@@ -166,6 +185,9 @@ public class ChainTableCommitPreCallback implements
CommitPreCallback {
nextSnapshotPartition,
chainComparator,
projector))
+ .filter(
+ deltaPartition ->
+
!freshlyWrittenDeltaPartitions.contains(deltaPartition))
.collect(Collectors.toList());
boolean canDrop =
deltaFollowingPartitions.isEmpty() ||
preSnapshotPartition.isPresent();
@@ -181,6 +203,37 @@ public class ChainTableCommitPreCallback implements
CommitPreCallback {
}
}
+ private Set<BinaryRow> fullyDroppedPartitions(
+ List<SimpleFileEntry> baseFiles, List<ManifestEntry> deltaFiles) {
+ Map<BinaryRow, Set<String>> 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<BinaryRow> droppedPartitions = new HashSet<>();
+ for (Map.Entry<BinaryRow, Set<String>> deleted :
deletedFilesByPartition.entrySet()) {
+ BinaryRow partition = deleted.getKey();
+ Set<String> deletedFiles = deleted.getValue();
+ List<String> 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<ManifestEntry> deltaFiles, List<IndexManifestEntry>
indexFiles) {
return deltaFiles.stream().allMatch(f -> f.kind() == FileKind.DELETE)
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 2d1a0cd705..67255bd7f4 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.table.sink.CommitCallback;
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,18 @@ public class ChainTableOverwriteCommitCallback implements
CommitCallback {
.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<BinaryRow> freshlyWritten = new HashSet<>(overwritePartitions);
+ Set<BinaryRow> previous =
+
ChainTableOverwriteScope.setFreshlyWrittenDeltaPartitions(freshlyWritten);
+ try {
+ commit.truncatePartitions(candidatePartitions);
+ } finally {
+ ChainTableOverwriteScope.restore(previous);
+ }
} 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 0000000000..3648102abe
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/metastore/ChainTableOverwriteScope.java
@@ -0,0 +1,75 @@
+/*
+ * 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}.
+ *
+ * <p>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.
+ *
+ * <p>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<Set<BinaryRow>>
FRESHLY_WRITTEN_DELTA_PARTITIONS =
+ new ThreadLocal<>();
+
+ private ChainTableOverwriteScope() {}
+
+ /**
+ * 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<BinaryRow> setFreshlyWrittenDeltaPartitions(Set<BinaryRow>
partitions) {
+ Set<BinaryRow> previous = FRESHLY_WRITTEN_DELTA_PARTITIONS.get();
+ FRESHLY_WRITTEN_DELTA_PARTITIONS.set(partitions);
+ return previous;
+ }
+
+ static void restore(Set<BinaryRow> previous) {
+ if (previous == null) {
+ FRESHLY_WRITTEN_DELTA_PARTITIONS.remove();
+ } else {
+ FRESHLY_WRITTEN_DELTA_PARTITIONS.set(previous);
+ }
+ }
+
+ static Set<BinaryRow> freshlyWrittenDeltaPartitions() {
+ Set<BinaryRow> partitions = FRESHLY_WRITTEN_DELTA_PARTITIONS.get();
+ return partitions == null
+ ? Collections.emptySet()
+ : Collections.unmodifiableSet(partitions);
+ }
+}
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 c06b090662..ac7176dcbf 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 class ChainTablePartitionExpireTest {
.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);
}
