From b8eaaaf7943ddee0e12bcd767a24647d2b2d19a4 Mon Sep 17 00:00:00 2001 From: henriquejsza Date: Fri, 30 Oct 2026 09:57:22 -0300 Subject: [PATCH] [core] Keep files referenced by other branches during snapshot expiration All branches of a table share the same physical data, manifest, index and statistics files; only snapshot, tag and schema metadata are branch scoped. Snapshot expiration on a branch built its skipping set only from that branch's own snapshots and tags, so after a branch created from a tag drops its copy of the tag, expiring it deletes files that main still references and main can no longer be read. Before deleting files, collect the data files and manifest/index/statistics files referenced by every other branch's live snapshots and tags, restricted to the current expiration batch, and skip any file still referenced. A failure to read another branch keeps the batch rather than risking deletion of in-use files. The common single-branch path is unchanged. Signed-off-by: henriquejsza --- .../paimon/operation/FileDeletionBase.java | 51 ++++++- .../paimon/table/AbstractFileStoreTable.java | 135 +++++++++++++++++- .../paimon/table/ExpireSnapshotsImpl.java | 92 +++++++++++- .../table/AppendOnlySimpleTableTest.java | 40 ++++++ 4 files changed, 315 insertions(+), 3 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/FileDeletionBase.java b/paimon-core/src/main/java/org/apache/paimon/operation/FileDeletionBase.java index b449e7bdf0d0..7e7dfa50ae32 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/FileDeletionBase.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/FileDeletionBase.java @@ -470,6 +470,34 @@ public Predicate createDataFileSkipperForTag( return entry -> containsDataFile(taggedCandidates, entry); } + /** + * Collect the data files referenced by {@code snapshot} that also appear in {@code candidates}, + * accumulating them into {@code referenced}. + * + *

All branches of a table share the same physical data directory; only snapshot, tag and + * manifest metadata are branch scoped. A file is therefore still live if any branch references + * it, even when the branch running expiration no longer does. The candidate restriction keeps + * memory proportional to the current expiration batch instead of the whole table, mirroring + * {@link #createDataFileSkipperForTag(Snapshot, Collection)}. + */ + public void addReferencedDataFiles( + Map>> referenced, + Snapshot snapshot, + Map>> candidates) + throws IOException { + if (candidates.isEmpty()) { + return; + } + Collection matchingEntries = + readMergedDataFiles( + manifestList.readDataManifests(snapshot), + createCandidateBucketFilter(candidates), + entry -> containsDataFile(candidates, entry)); + for (ExpireFileEntry entry : matchingEntries) { + addDataFile(referenced, entry); + } + } + /** * It is possible that a job was killed during expiration and some manifest files have been * deleted, so if the clean methods need to get manifests of a snapshot to be cleaned, we should @@ -550,7 +578,7 @@ public boolean mayContain(int minBucket, int maxBucket, int totalBuckets) { return new BucketFilter(false, null, null, filter); } - private void addDataFile( + protected void addDataFile( Map>> dataFiles, ExpireFileEntry entry) { dataFiles .computeIfAbsent(entry.partition(), p -> new HashMap<>()) @@ -558,6 +586,18 @@ private void addDataFile( .add(entry.fileName()); } + /** Add {@code entry} to a candidate data file index. */ + public void addCandidateDataFile( + Map>> dataFiles, ExpireFileEntry entry) { + addDataFile(dataFiles, entry); + } + + /** Whether {@code entry} is contained in the data file index. */ + public boolean containsDataFileEntry( + Map>> dataFiles, ExpireFileEntry entry) { + return containsDataFile(dataFiles, entry); + } + protected boolean containsDataFile( Map>> dataFiles, ExpireFileEntry entry) { return containsDataFile(dataFiles, entry.partition(), entry.bucket(), entry.fileName()); @@ -715,5 +755,14 @@ private DataFileDeletionPlan( private static DataFileDeletionPlan empty() { return new DataFileDeletionPlan(Collections.emptyMap()); } + + /** The {@link ExpireFileEntry}s that are candidates for deletion in this plan. */ + public Collection candidateEntries() { + List entries = new ArrayList<>(dataFileToDelete.size()); + for (Pair> pair : dataFileToDelete.values()) { + entries.add(pair.getLeft()); + } + return entries; + } } } diff --git a/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java b/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java index 459713687e88..735722e9d8db 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java @@ -19,9 +19,11 @@ package org.apache.paimon.table; import org.apache.paimon.CoreOptions; +import org.apache.paimon.FileStore; import org.apache.paimon.Snapshot; import org.apache.paimon.catalog.Identifier; import org.apache.paimon.consumer.ConsumerManager; +import org.apache.paimon.data.BinaryRow; import org.apache.paimon.fs.FileIO; import org.apache.paimon.fs.Path; import org.apache.paimon.iceberg.IcebergCommitCallback; @@ -30,6 +32,7 @@ import org.apache.paimon.manifest.ManifestEntry; import org.apache.paimon.manifest.ManifestFileMeta; import org.apache.paimon.operation.FileStoreScan; +import org.apache.paimon.operation.SnapshotDeletion; import org.apache.paimon.options.ConfigOption; import org.apache.paimon.options.ExpireConfig; import org.apache.paimon.options.FallbackKey; @@ -72,6 +75,9 @@ import org.apache.paimon.shade.caffeine2.com.github.benmanes.caffeine.cache.Cache; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + import javax.annotation.Nullable; import java.io.FileNotFoundException; @@ -90,6 +96,7 @@ import java.util.SortedMap; import java.util.function.BiConsumer; import java.util.function.LongConsumer; +import java.util.stream.Collectors; import static org.apache.paimon.CoreOptions.PATH; @@ -98,6 +105,8 @@ abstract class AbstractFileStoreTable implements FileStoreTable { private static final long serialVersionUID = 1L; + private static final Logger LOG = LoggerFactory.getLogger(AbstractFileStoreTable.class); + protected final FileIO fileIO; protected final Path path; protected final TableSchema tableSchema; @@ -507,7 +516,131 @@ public ExpireSnapshots newExpireSnapshots() { changelogManager(), store().newSnapshotDeletion(), store().newTagManager(), - store().options().scanManifestParallelism()); + store().options().scanManifestParallelism(), + newOtherBranchesLiveFiles()); + } + + /** + * Builds a collector for data files referenced by branches other than the one being expired. + * + *

Every branch shares the same physical data directory, so expiring snapshots on one branch + * must not delete files another branch (for example main) still references. Returns {@code + * null} when there is no other branch, keeping the common single-branch path untouched. + */ + @Nullable + private ExpireSnapshotsImpl.OtherBranchesLiveFiles newOtherBranchesLiveFiles() { + String currentBranch = BranchManager.normalizeBranch(currentBranch()); + List otherBranches; + try { + List allBranches = new ArrayList<>(branchManager().branches()); + allBranches.add(Identifier.DEFAULT_MAIN_BRANCH); + otherBranches = + allBranches.stream() + .map(BranchManager::normalizeBranch) + .distinct() + .filter(branch -> !branch.equals(currentBranch)) + .collect(Collectors.toList()); + } catch (Exception e) { + LOG.warn( + "Failed to list branches before snapshot expiration. " + + "Cross-branch data file protection is disabled for this run.", + e); + return null; + } + if (otherBranches.isEmpty()) { + return null; + } + + List branches = otherBranches; + return new ExpireSnapshotsImpl.OtherBranchesLiveFiles() { + @Override + public Map>> collect( + Map>> candidates) throws IOException { + Map>> referenced = new HashMap<>(); + for (String branch : branches) { + collectBranchReferencedDataFiles(branch, candidates, referenced); + } + return referenced; + } + + @Override + public Set manifestSkippingSet() { + Set skippingSet = new HashSet<>(); + for (String branch : branches) { + collectBranchManifestSkippingSet(branch, skippingSet); + } + return skippingSet; + } + }; + } + + private void collectBranchManifestSkippingSet(String branch, Set skippingSet) { + FileStoreTable branchTable; + try { + branchTable = switchToBranch(branch); + } catch (IllegalArgumentException e) { + return; + } + + FileStore branchStore = branchTable.store(); + SnapshotManager branchSnapshotManager = branchTable.snapshotManager(); + TagManager branchTagManager = branchStore.newTagManager(); + SnapshotDeletion branchDeletion = branchStore.newSnapshotDeletion(); + + List liveSnapshots = new ArrayList<>(); + Long earliest = branchSnapshotManager.earliestSnapshotId(); + Long latest = branchSnapshotManager.latestSnapshotId(); + if (earliest != null && latest != null) { + for (long id = earliest; id <= latest; id++) { + try { + liveSnapshots.add(branchSnapshotManager.tryGetSnapshot(id)); + } catch (FileNotFoundException ignored) { + // snapshot may have expired concurrently + } + } + } + liveSnapshots.addAll(branchTagManager.taggedSnapshots()); + if (!liveSnapshots.isEmpty()) { + skippingSet.addAll(branchDeletion.manifestSkippingSet(liveSnapshots)); + } + } + + private void collectBranchReferencedDataFiles( + String branch, + Map>> candidates, + Map>> referenced) + throws IOException { + FileStoreTable branchTable; + try { + branchTable = switchToBranch(branch); + } catch (IllegalArgumentException e) { + // The branch may have been dropped concurrently, or have no schema yet. + return; + } + + FileStore branchStore = branchTable.store(); + SnapshotManager branchSnapshotManager = branchTable.snapshotManager(); + TagManager branchTagManager = branchStore.newTagManager(); + SnapshotDeletion branchDeletion = branchStore.newSnapshotDeletion(); + + // live snapshots of the other branch + Long earliest = branchSnapshotManager.earliestSnapshotId(); + Long latest = branchSnapshotManager.latestSnapshotId(); + if (earliest != null && latest != null) { + for (long id = earliest; id <= latest; id++) { + try { + Snapshot snapshot = branchSnapshotManager.tryGetSnapshot(id); + branchDeletion.addReferencedDataFiles(referenced, snapshot, candidates); + } catch (FileNotFoundException ignored) { + // snapshot may have expired concurrently + } + } + } + + // tags of the other branch keep files alive independently of its snapshots + for (Snapshot taggedSnapshot : branchTagManager.taggedSnapshots()) { + branchDeletion.addReferencedDataFiles(referenced, taggedSnapshot, candidates); + } } @Override diff --git a/paimon-core/src/main/java/org/apache/paimon/table/ExpireSnapshotsImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/ExpireSnapshotsImpl.java index 6823b1ce3b5c..b09f74abf450 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/ExpireSnapshotsImpl.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/ExpireSnapshotsImpl.java @@ -22,6 +22,7 @@ import org.apache.paimon.Snapshot; import org.apache.paimon.annotation.VisibleForTesting; import org.apache.paimon.consumer.ConsumerManager; +import org.apache.paimon.data.BinaryRow; import org.apache.paimon.fs.Path; import org.apache.paimon.manifest.ExpireFileEntry; import org.apache.paimon.operation.FileDeletionBase.DataFileDeletionPlan; @@ -72,6 +73,7 @@ public class ExpireSnapshotsImpl implements ExpireSnapshots { private final Executor fileExecutor; private final TagManager tagManager; private final int snapshotExpireBatchSize; + @Nullable private final OtherBranchesLiveFiles otherBranchesLiveFiles; private ExpireConfig expireConfig; private Supplier currentTimeMillis = System::currentTimeMillis; @@ -82,6 +84,22 @@ public ExpireSnapshotsImpl( SnapshotDeletion snapshotDeletion, TagManager tagManager, @Nullable Integer scanManifestParallelism) { + this( + snapshotManager, + changelogManager, + snapshotDeletion, + tagManager, + scanManifestParallelism, + null); + } + + public ExpireSnapshotsImpl( + SnapshotManager snapshotManager, + ChangelogManager changelogManager, + SnapshotDeletion snapshotDeletion, + TagManager tagManager, + @Nullable Integer scanManifestParallelism, + @Nullable OtherBranchesLiveFiles otherBranchesLiveFiles) { this.snapshotManager = snapshotManager; this.changelogManager = changelogManager; this.consumerManager = @@ -93,6 +111,7 @@ public ExpireSnapshotsImpl( this.tagManager = tagManager; this.expireConfig = ExpireConfig.builder().build(); this.fileExecutor = snapshotDeletion.fileExecutor(); + this.otherBranchesLiveFiles = otherBranchesLiveFiles; this.snapshotExpireBatchSize = scanManifestParallelism == null || scanManifestParallelism <= 0 ? Runtime.getRuntime().availableProcessors() @@ -248,6 +267,11 @@ private int innerExpireUntil(long earliestId, long endExclusiveId) try { Set builtSkippingSet = ConcurrentHashMap.newKeySet(); builtSkippingSet.addAll(snapshotDeletion.manifestSkippingSet(skippingSnapshots)); + // Manifest, index and statistics files live in the shared table directory too, so + // keep every manifest referenced by another branch's live snapshots and tags. + if (otherBranchesLiveFiles != null) { + builtSkippingSet.addAll(otherBranchesLiveFiles.manifestSkippingSet()); + } skippingSet = builtSkippingSet; } catch (Exception e) { LOG.info("Skip cleaning manifest files due to failed to build skipping set.", e); @@ -339,6 +363,13 @@ private void cleanDataFileBatch(List snapshots, Map ta Map>> skippers = collectTagSkippers(tags, plansByTag); + + // All branches of a table share the same physical data files; a file this branch is + // about to delete may still be referenced by another branch (for example main, when this + // branch was created from a tag and its own copy of that tag has since been deleted). + // Build a skipper that keeps any candidate file still live on another branch. + Predicate retainedByOtherBranches = buildOtherBranchesSkipper(plans); + Predicate deleteAll = entry -> false; List paths = new ArrayList<>(); for (int i = 0; i < snapshots.size(); i++) { @@ -354,12 +385,71 @@ private void cleanDataFileBatch(List snapshots, Map ta snapshot.id()); continue; } - paths.addAll(snapshotDeletion.dataFilesToDelete(plans.get(i), skipper.get())); + Predicate effectiveSkipper = skipper.get(); + if (retainedByOtherBranches != null) { + Predicate ownSkipper = effectiveSkipper; + effectiveSkipper = + entry -> ownSkipper.test(entry) || retainedByOtherBranches.test(entry); + } + paths.addAll(snapshotDeletion.dataFilesToDelete(plans.get(i), effectiveSkipper)); } snapshotDeletion.cleanDataFiles(paths); snapshots.clear(); } + @Nullable + private Predicate buildOtherBranchesSkipper(List plans) { + if (otherBranchesLiveFiles == null) { + return null; + } + Map>> candidates = new HashMap<>(); + for (DataFileDeletionPlan plan : plans) { + for (ExpireFileEntry entry : plan.candidateEntries()) { + snapshotDeletion.addCandidateDataFile(candidates, entry); + } + } + if (candidates.isEmpty()) { + return null; + } + try { + Map>> referenced = + otherBranchesLiveFiles.collect(candidates); + if (referenced.isEmpty()) { + return null; + } + return entry -> snapshotDeletion.containsDataFileEntry(referenced, entry); + } catch (Exception e) { + // A failure to read another branch must never turn into deleting a file that branch + // may still reference. Keep the whole batch by skipping all candidate files. + LOG.info( + "Failed to collect data files referenced by other branches. " + + "Skip cleaning this batch to avoid deleting files still in use.", + e); + return entry -> true; + } + } + + /** + * Collects files referenced by all branches other than the one being expired. Every branch of a + * table shares the same physical data, manifest, index and statistics files, so expiring one + * branch must not delete files another branch still references. + */ + public interface OtherBranchesLiveFiles { + + /** + * Data files referenced by other branches, restricted to {@code candidates} to keep memory + * proportional to the current expiration batch. + */ + Map>> collect( + Map>> candidates) throws Exception; + + /** + * Names of manifest, manifest list, index and statistics files referenced by other + * branches' live snapshots and tags. + */ + Set manifestSkippingSet() throws Exception; + } + private Map>> collectTagSkippers( Map tags, Map> plansByTag) throws ExecutionException, InterruptedException { diff --git a/paimon-core/src/test/java/org/apache/paimon/table/AppendOnlySimpleTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/AppendOnlySimpleTableTest.java index df9323bf46e4..20feb71d5c76 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/AppendOnlySimpleTableTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/AppendOnlySimpleTableTest.java @@ -48,6 +48,7 @@ import org.apache.paimon.manifest.ManifestList; import org.apache.paimon.operation.BaseAppendFileStoreWrite; import org.apache.paimon.operation.FileStoreWrite; +import org.apache.paimon.options.ExpireConfig; import org.apache.paimon.options.MemorySize; import org.apache.paimon.options.Options; import org.apache.paimon.predicate.Equal; @@ -2884,6 +2885,45 @@ public void testMergeBranchAfterSnapshotExpiration() throws Exception { "Branch merge requires complete append-only snapshot history")); } + @Test + public void testExpireOnBranchKeepsFilesReferencedByMain() throws Exception { + FileStoreTable table = createFileStoreTable(); + + // main: write snapshot 1 and tag it; all branches share these physical data files + try (StreamTableWrite write = table.newWrite(commitUser); + StreamTableCommit commit = table.newCommit(commitUser)) { + write.write(rowData(1, 10, 100L)); + write.write(rowData(2, 20, 200L)); + commit.commit(0, write.prepareCommit(false, 1)); + } + table.createTag("t1", 1); + + // branch created from the tag starts on a copy of main's snapshot 1 + table.createBranch(BRANCH_NAME, "t1"); + FileStoreTable tableBranch = table.switchToBranch(BRANCH_NAME); + + // write a new snapshot on the branch, then drop the branch's own copy of the tag and + // expire down to a single snapshot: the branch no longer references main's files + try (StreamTableWrite write = tableBranch.newWrite(commitUser); + StreamTableCommit commit = tableBranch.newCommit(commitUser)) { + write.write(rowData(3, 30, 300L)); + commit.commit(1, write.prepareCommit(false, 2)); + } + tableBranch.deleteTag("t1"); + tableBranch + .newExpireSnapshots() + .config(ExpireConfig.builder().snapshotRetainMax(1).snapshotRetainMin(1).build()) + .expire(); + + // main still references snapshot 1 and tag t1, so its files must survive branch expiration + FileStoreTable mainTable = createFileStoreTable(); + List mainSplits = toSplits(mainTable.newSnapshotReader().read().dataSplits()); + assertThat(getResult(mainTable.newRead(), mainSplits, BATCH_ROW_TO_STRING)) + .containsExactlyInAnyOrder( + "1|10|100|binary|varbinary|mapKey:mapVal|multiset", + "2|20|200|binary|varbinary|mapKey:mapVal|multiset"); + } + @Test public void testMergeBranchRejectsNonAppendHistory() throws Exception { FileStoreTable table = createFileStoreTable();