[core] Cleaning snapshots or deleting tags should skip those referenced by branches.

paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java

@@ -29,6 +29,7 @@
 import org.apache.paimon.manifest.ManifestList;
 import org.apache.paimon.metastore.AddPartitionTagCallback;
 import org.apache.paimon.metastore.MetastoreClient;
+import org.apache.paimon.operation.BranchDeletion;
 import org.apache.paimon.operation.ChangelogDeletion;
 import org.apache.paimon.operation.FileStoreCommitImpl;
 import org.apache.paimon.operation.PartitionExpire;
@@ -48,6 +49,7 @@
 import org.apache.paimon.table.sink.TagCallback;
 import org.apache.paimon.tag.TagAutoManager;
 import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.BranchManager;
 import org.apache.paimon.utils.FileStorePathFactory;
 import org.apache.paimon.utils.SegmentsCache;
 import org.apache.paimon.utils.SnapshotManager;
@@ -252,6 +254,30 @@ public TagManager newTagManager() {
  return new TagManager(fileIO, options.path());
  }
 
+ @Override
+ public BranchManager newBranchManager() {
+ return new BranchManager(
+ fileIO,
+ options.path(),
+ snapshotManager(),
+ newTagManager(),
+ schemaManager,
+ newBranchDeletion());
+ }
+
+ @Override
+ public BranchDeletion newBranchDeletion() {
+ return new BranchDeletion(
+ fileIO,
+ pathFactory(),
+ manifestFileFactory().create(),
+ manifestListFactory().create(),
+ newIndexFileHandler(),
+ newStatsFileHandler(),
+ options.cleanEmptyDirectories(),
+ options.deleteFileThreadNum());
+ }
+
  @Override
  public TagDeletion newTagDeletion() {
  return new TagDeletion(
@@ -297,6 +323,7 @@ public TagAutoManager newTagCreationManager() {
  options,
  snapshotManager(),
  newTagManager(),
+ newBranchManager(),
  newTagDeletion(),
  createTagCallbacks());
  }

paimon-core/src/main/java/org/apache/paimon/FileStore.java

@@ -23,6 +23,7 @@
 import org.apache.paimon.manifest.ManifestCacheFilter;
 import org.apache.paimon.manifest.ManifestFile;
 import org.apache.paimon.manifest.ManifestList;
+import org.apache.paimon.operation.BranchDeletion;
 import org.apache.paimon.operation.ChangelogDeletion;
 import org.apache.paimon.operation.FileStoreCommit;
 import org.apache.paimon.operation.FileStoreScan;
@@ -38,6 +39,7 @@
 import org.apache.paimon.table.sink.TagCallback;
 import org.apache.paimon.tag.TagAutoManager;
 import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.BranchManager;
 import org.apache.paimon.utils.FileStorePathFactory;
 import org.apache.paimon.utils.SegmentsCache;
 import org.apache.paimon.utils.SnapshotManager;
@@ -90,8 +92,12 @@ public interface FileStore<T> {
 
  TagManager newTagManager();
 
+ BranchManager newBranchManager();
+
  TagDeletion newTagDeletion();
 
+ BranchDeletion newBranchDeletion();
+
  @Nullable
  PartitionExpire newPartitionExpire(String commitUser);
 

paimon-core/src/main/java/org/apache/paimon/operation/BranchDeletion.java

@@ -0,0 +1,81 @@
+/*
+ * 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.operation;
+
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.index.IndexFileHandler;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.manifest.ManifestFile;
+import org.apache.paimon.manifest.ManifestList;
+import org.apache.paimon.stats.StatsFileHandler;
+import org.apache.paimon.utils.FileStorePathFactory;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.Set;
+import java.util.function.Predicate;
+
+/** Delete branch files. */
+public class BranchDeletion extends FileDeletionBase<Snapshot> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(BranchDeletion.class);
+
+ public BranchDeletion(
+ FileIO fileIO,
+ FileStorePathFactory pathFactory,
+ ManifestFile manifestFile,
+ ManifestList manifestList,
+ IndexFileHandler indexFileHandler,
+ StatsFileHandler statsFileHandler,
+ boolean cleanEmptyDirectories,
+ int deleteFileThreadNum) {
+ super(
+ fileIO,
+ pathFactory,
+ manifestFile,
+ manifestList,
+ indexFileHandler,
+ statsFileHandler,
+ cleanEmptyDirectories,
+ deleteFileThreadNum);
+ }
+
+ @Override
+ public void cleanUnusedDataFiles(Snapshot snapshotToClean, Predicate<ManifestEntry> skipper) {
+ Collection<ManifestEntry> manifestEntries;
+ try {
+ manifestEntries = readMergedDataFiles(snapshotToClean);
+ } catch (IOException e) {
+ LOG.info("Skip data file clean for the branch of id {}.", snapshotToClean.id(), e);
+ return;
+ }
+
+ deleteFiles(extractDataFilePaths(manifestEntries, skipper), fileIO::deleteQuietly);
+ }
+
+ @Override
+ public void cleanUnusedManifests(Snapshot snapshotToClean, Set<String> skippingSet) {
+ // doesn't clean changelog files because they are handled by SnapshotDeletion
+ cleanUnusedManifests(snapshotToClean, skippingSet, true, false);
+ }
+}

paimon-core/src/main/java/org/apache/paimon/operation/FileDeletionBase.java

@@ -344,6 +344,19 @@ public Predicate<ManifestEntry> dataFileSkipper(
  return entry -> false;
  }
 
+ public Predicate<ManifestEntry> dataFileSkipper(Snapshot skippingSnapshot) throws Exception {
+ return dataFileSkipper(Collections.singletonList(skippingSnapshot));
+ }
+
+ public Predicate<ManifestEntry> dataFileSkipper(List<Snapshot> skippingSnapshots)
+ throws Exception {
+ Map<BinaryRow, Map<Integer, Set<String>>> skipped = new HashMap<>();
+ for (Snapshot snapshot : skippingSnapshots) {
+ addMergedDataFiles(skipped, snapshot);
+ }
+ return entry -> containsDataFile(skipped, 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
@@ -475,4 +488,21 @@ protected <F> void deleteFiles(Collection<F> files, Consumer<F> deletion) {
  throw new RuntimeException(e);
  }
  }
+
+ public Set<Path> extractDataFilePaths(
+ Collection<ManifestEntry> manifestEntries, Predicate<ManifestEntry> skipper) {
+ Set<Path> dataFileToDelete = new HashSet<>();
+ for (ManifestEntry entry : manifestEntries) {
+ if (!skipper.test(entry)) {
+ Path bucketPath = pathFactory.bucketPath(entry.partition(), entry.bucket());
+ dataFileToDelete.add(new Path(bucketPath, entry.file().fileName()));
+ for (String file : entry.file().extraFiles()) {
+ dataFileToDelete.add(new Path(bucketPath, file));
+ }
+
+ recordDeletionBuckets(entry);
+ }
+ }
+ return dataFileToDelete;
+ }
 }

paimon-core/src/main/java/org/apache/paimon/operation/OrphanFilesClean.java

@@ -35,6 +35,7 @@
 import org.apache.paimon.manifest.ManifestList;
 import org.apache.paimon.table.FileStoreTable;
 import org.apache.paimon.table.Table;
+import org.apache.paimon.utils.BranchManager;
 import org.apache.paimon.utils.DateTimeUtils;
 import org.apache.paimon.utils.FileUtils;
 import org.apache.paimon.utils.SnapshotManager;
@@ -93,6 +94,7 @@ public class OrphanFilesClean {
 
  private final SnapshotManager snapshotManager;
  private final TagManager tagManager;
+ private final BranchManager branchManager;
  private final FileIO fileIO;
  private final Path location;
  private final int partitionKeysNum;
@@ -107,6 +109,7 @@ public class OrphanFilesClean {
  public OrphanFilesClean(FileStoreTable table) {
  this.snapshotManager = table.snapshotManager();
  this.tagManager = table.tagManager();
+ this.branchManager = table.branchManager();
  this.fileIO = table.fileIO();
  this.location = table.location();
  this.partitionKeysNum = table.partitionKeys().size();
@@ -169,13 +172,15 @@ public List<Path> clean() throws IOException, ExecutionException, InterruptedExc
  return deleteFiles;
  }
 
- /** Get all the files used by snapshots and tags. */
+ /** Get all the files used by snapshots and tags and branches. */
  private Set<String> getUsedFiles()
  throws IOException, ExecutionException, InterruptedException {
  // safely get all snapshots to be read
  Set<Snapshot> readSnapshots = new HashSet<>(snapshotManager.safelyGetAllSnapshots());
  List<Snapshot> taggedSnapshots = tagManager.taggedSnapshots();
+ Set<Snapshot> branchSnapshots = branchManager.branchesCreateSnapshots().keySet();
  readSnapshots.addAll(taggedSnapshots);
+ readSnapshots.addAll(branchSnapshots);
  readSnapshots.addAll(snapshotManager.safelyGetAllChangelogs());
 
  return FileUtils.COMMON_IO_FORK_JOIN_POOL

paimon-core/src/main/java/org/apache/paimon/operation/TagDeletion.java

@@ -19,9 +19,8 @@
 package org.apache.paimon.operation;
 
 import org.apache.paimon.Snapshot;
-import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.annotation.VisibleForTesting;
 import org.apache.paimon.fs.FileIO;
-import org.apache.paimon.fs.Path;
 import org.apache.paimon.index.IndexFileHandler;
 import org.apache.paimon.manifest.ManifestEntry;
 import org.apache.paimon.manifest.ManifestFile;
@@ -34,11 +33,6 @@
 
 import java.io.IOException;
 import java.util.Collection;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
 import java.util.Set;
 import java.util.function.Predicate;
 
@@ -77,19 +71,7 @@ public void cleanUnusedDataFiles(Snapshot taggedSnapshot, Predicate<ManifestEntr
  return;
  }
 
- Set<Path> dataFileToDelete = new HashSet<>();
- for (ManifestEntry entry : manifestEntries) {
- if (!skipper.test(entry)) {
- Path bucketPath = pathFactory.bucketPath(entry.partition(), entry.bucket());
- dataFileToDelete.add(new Path(bucketPath, entry.file().fileName()));
- for (String file : entry.file().extraFiles()) {
- dataFileToDelete.add(new Path(bucketPath, file));
- }
-
- recordDeletionBuckets(entry);
- }
- }
- deleteFiles(dataFileToDelete, fileIO::deleteQuietly);
+ deleteFiles(extractDataFilePaths(manifestEntries, skipper), fileIO::deleteQuietly);
  }
 
  @Override
@@ -98,15 +80,9 @@ public void cleanUnusedManifests(Snapshot taggedSnapshot, Set<String> skippingSe
  cleanUnusedManifests(taggedSnapshot, skippingSet, true, false);
  }
 
- public Predicate<ManifestEntry> dataFileSkipper(Snapshot fromSnapshot) throws Exception {
- return dataFileSkipper(Collections.singletonList(fromSnapshot));
- }
-
- public Predicate<ManifestEntry> dataFileSkipper(List<Snapshot> fromSnapshots) throws Exception {
- Map<BinaryRow, Map<Integer, Set<String>>> skipped = new HashMap<>();
- for (Snapshot snapshot : fromSnapshots) {
- addMergedDataFiles(skipped, snapshot);
- }
- return entry -> containsDataFile(skipped, entry);
+ @VisibleForTesting
+ public Collection<ManifestEntry> getDataFilesFromSnapshot(Snapshot snapshot)
+ throws IOException {
+ return readMergedDataFiles(snapshot);
  }
 }

paimon-core/src/main/java/org/apache/paimon/privilege/PrivilegedFileStore.java

@@ -26,6 +26,7 @@
 import org.apache.paimon.manifest.ManifestCacheFilter;
 import org.apache.paimon.manifest.ManifestFile;
 import org.apache.paimon.manifest.ManifestList;
+import org.apache.paimon.operation.BranchDeletion;
 import org.apache.paimon.operation.ChangelogDeletion;
 import org.apache.paimon.operation.FileStoreCommit;
 import org.apache.paimon.operation.FileStoreScan;
@@ -41,6 +42,7 @@
 import org.apache.paimon.table.sink.TagCallback;
 import org.apache.paimon.tag.TagAutoManager;
 import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.BranchManager;
 import org.apache.paimon.utils.FileStorePathFactory;
 import org.apache.paimon.utils.SegmentsCache;
 import org.apache.paimon.utils.SnapshotManager;
@@ -164,6 +166,17 @@ public TagManager newTagManager() {
  return wrapped.newTagManager();
  }
 
+ @Override
+ public BranchManager newBranchManager() {
+ privilegeChecker.assertCanInsert(identifier);
+ return wrapped.newBranchManager();
+ }
+
+ @Override
+ public BranchDeletion newBranchDeletion() {
+ return wrapped.newBranchDeletion();
+ }
+
  @Override
  public TagDeletion newTagDeletion() {
  privilegeChecker.assertCanInsert(identifier);

paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java

@@ -351,13 +351,16 @@ public SnapshotManager snapshotManager() {
  @Override
  public ExpireSnapshots newExpireSnapshots() {
  return new ExpireSnapshotsImpl(
- snapshotManager(), store().newSnapshotDeletion(), store().newTagManager());
+ snapshotManager(),
+ store().newSnapshotDeletion(),
+ store().newTagManager(),
+ branchManager());
  }
 
  @Override
  public ExpireSnapshots newExpireChangelog() {
  return new ExpireChangelogImpl(
- snapshotManager(), tagManager(), store().newChangelogDeletion());
+ snapshotManager(), tagManager(), store().newChangelogDeletion(), branchManager());
  }
 
  @Override
@@ -566,6 +569,7 @@ public void deleteTag(String tagName) {
  tagName,
  store().newTagDeletion(),
  snapshotManager(),
+ branchManager(),
  store().createTagCallbacks());
  }
 
@@ -627,7 +631,13 @@ public TagManager tagManager() {
 
  @Override
  public BranchManager branchManager() {
- return new BranchManager(fileIO, path, snapshotManager(), tagManager(), schemaManager());
+ return new BranchManager(
+ fileIO,
+ path,
+ snapshotManager(),
+ tagManager(),
+ schemaManager(),
+ store().newBranchDeletion());
  }
 
  private RollbackHelper rollbackHelper() {