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