[core] Validate chain partition drops against the post-commit state

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 @@
  * <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 void call(
         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 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<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 void call(
                                                     nextSnapshotPartition,
                                                     chainComparator,
                                                     projector))
+                            .filter(
+                                    deltaPartition ->
+                                            !freshlyWrittenDeltaPartitions.contains(deltaPartition))
                             .collect(Collectors.toList());
             boolean canDrop =
                     deltaFollowingPartitions.isEmpty() || preSnapshotPartition.isPresent();
@@ -181,6 +203,37 @@ public void call(
         }
     }
 
+    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)

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,18 @@ 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<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(

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);
+    }
+}

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);
     }