[flink] filter by directory for orphan file clean to reduce memory usage

LsomeYeah · LsomeYeah · commit a4f359693e96 · 2025-09-12T15:37:59.000+08:00
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/LocalOrphanFilesClean.java b/paimon-core/src/main/java/org/apache/paimon/operation/LocalOrphanFilesClean.java
@@ -85,7 +85,7 @@ public LocalOrphanFilesClean(FileStoreTable table, long olderThanMillis) {
     }
 
     public LocalOrphanFilesClean(FileStoreTable table, long olderThanMillis, boolean dryRun) {
-        super(table, olderThanMillis, dryRun);
+        super(table, olderThanMillis, dryRun, false);
         this.deleteFiles = new ArrayList<>();
         this.executor =
                 createCachedThreadPool(
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/OrphanFilesClean.java b/paimon-core/src/main/java/org/apache/paimon/operation/OrphanFilesClean.java
@@ -96,14 +96,17 @@ public abstract class OrphanFilesClean implements Serializable {
     protected final boolean dryRun;
     protected final int partitionKeysNum;
     protected final Path location;
+    protected final boolean folderBasedCheck;
 
-    public OrphanFilesClean(FileStoreTable table, long olderThanMillis, boolean dryRun) {
+    public OrphanFilesClean(
+            FileStoreTable table, long olderThanMillis, boolean dryRun, boolean folderBasedCheck) {
         this.table = table;
         this.fileIO = table.fileIO();
         this.partitionKeysNum = table.partitionKeys().size();
         this.location = table.location();
         this.olderThanMillis = olderThanMillis;
         this.dryRun = dryRun;
+        this.folderBasedCheck = folderBasedCheck;
     }
 
     protected List<String> validBranches() {
@@ -368,6 +371,10 @@ private List<Path> listFileDirs(Path dir, int level) {
         List<FileStatus> dirs = tryBestListingDirs(dir);
 
         if (level == 0) {
+            if (folderBasedCheck) {
+                return filterDirs(
+                        dirs, p -> p.getName().startsWith(BUCKET_PATH_PREFIX), this::oldEnough);
+            }
             // return bucket paths
             return filterDirs(dirs, p -> p.getName().startsWith(BUCKET_PATH_PREFIX));
         }
@@ -395,6 +402,23 @@ private List<Path> filterDirs(List<FileStatus> statuses, Predicate<Path> filter)
         return filtered;
     }
 
+    private List<Path> filterDirs(
+            List<FileStatus> statuses,
+            Predicate<Path> nameFilter,
+            Predicate<FileStatus> timeFilter) {
+        List<Path> filtered = new ArrayList<>();
+
+        for (FileStatus status : statuses) {
+            Path path = status.getPath();
+            if (nameFilter.test(path) && timeFilter.test(status)) {
+                filtered.add(path);
+            }
+            // ignore unknown dirs
+        }
+
+        return filtered;
+    }
+
     /**
      * If failed to list directory, just return an empty result because it's OK to not delete them.
      */
diff --git a/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/RemoveOrphanFilesProcedure.java b/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/RemoveOrphanFilesProcedure.java
@@ -97,6 +97,7 @@ public String[] call(
                                     catalog,
                                     olderThanMillis(olderThan),
                                     dryRun,
+                                    false,
                                     parallelism,
                                     databaseName,
                                     tableName);
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RemoveOrphanFilesAction.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RemoveOrphanFilesAction.java
@@ -34,6 +34,7 @@ public class RemoveOrphanFilesAction extends ActionBase {
 
     private String olderThan = null;
     private boolean dryRun = false;
+    private boolean folderBasedCheck = false;
 
     public RemoveOrphanFilesAction(
             String databaseName,
@@ -54,13 +55,18 @@ public void dryRun() {
         this.dryRun = true;
     }
 
+    public void folderBasedCheck() {
+        this.folderBasedCheck = true;
+    }
+
     @Override
     public void run() throws Exception {
         executeDatabaseOrphanFiles(
                 env,
                 catalog,
                 olderThanMillis(olderThan),
                 dryRun,
+                folderBasedCheck,
                 parallelism == null ? null : Integer.parseInt(parallelism),
                 databaseName,
                 tableName);
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RemoveOrphanFilesActionFactory.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RemoveOrphanFilesActionFactory.java
@@ -26,6 +26,7 @@ public class RemoveOrphanFilesActionFactory implements ActionFactory {
     public static final String IDENTIFIER = "remove_orphan_files";
     private static final String OLDER_THAN = "older_than";
     private static final String DRY_RUN = "dry_run";
+    private static final String FOLDER_BASED_CHECK = "folder_based_check";
     private static final String PARALLELISM = "parallelism";
 
     @Override
@@ -50,6 +51,11 @@ public Optional<Action> create(MultipleParameterToolAdapter params) {
             action.dryRun();
         }
 
+        if (params.has(FOLDER_BASED_CHECK)
+                && Boolean.parseBoolean(params.get(FOLDER_BASED_CHECK))) {
+            action.folderBasedCheck();
+        }
+
         return Optional.of(action);
     }
 
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/orphan/FlinkOrphanFilesClean.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/orphan/FlinkOrphanFilesClean.java
@@ -81,8 +81,9 @@ public FlinkOrphanFilesClean(
             FileStoreTable table,
             long olderThanMillis,
             boolean dryRun,
+            boolean folderBasedCheck,
             @Nullable Integer parallelism) {
-        super(table, olderThanMillis, dryRun);
+        super(table, olderThanMillis, dryRun, folderBasedCheck);
         this.parallelism = parallelism;
     }
 
@@ -372,10 +373,13 @@ public static CleanOrphanFilesResult executeDatabaseOrphanFiles(
             Catalog catalog,
             long olderThanMillis,
             boolean dryRun,
+            boolean folderBasedCheck,
             @Nullable Integer parallelism,
             String databaseName,
             @Nullable String tableName)
             throws Catalog.DatabaseNotExistException, Catalog.TableNotExistException {
+        LOG.info("execute orphan file clean using folderBasedCheck:{}", folderBasedCheck);
+
         List<String> tableNames = Collections.singletonList(tableName);
         if (tableName == null || "*".equals(tableName)) {
             tableNames = catalog.listTables(databaseName);
@@ -393,7 +397,11 @@ public static CleanOrphanFilesResult executeDatabaseOrphanFiles(
 
             DataStream<CleanOrphanFilesResult> clean =
                     new FlinkOrphanFilesClean(
-                                    (FileStoreTable) table, olderThanMillis, dryRun, parallelism)
+                                    (FileStoreTable) table,
+                                    olderThanMillis,
+                                    dryRun,
+                                    folderBasedCheck,
+                                    parallelism)
                             .doOrphanClean(env);
             if (clean != null) {
                 orphanFilesCleans.add(clean);
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/RemoveOrphanFilesProcedure.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/RemoveOrphanFilesProcedure.java
@@ -85,6 +85,7 @@ public String[] call(
                                     catalog,
                                     olderThanMillis(olderThan),
                                     dryRun != null && dryRun,
+                                    false,
                                     parallelism,
                                     databaseName,
                                     tableName);
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/procedure/SparkOrphanFilesClean.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/procedure/SparkOrphanFilesClean.scala
@@ -26,7 +26,6 @@ import org.apache.paimon.operation.{CleanOrphanFilesResult, OrphanFilesClean}
 import org.apache.paimon.operation.OrphanFilesClean.retryReadingFiles
 import org.apache.paimon.table.FileStoreTable
 import org.apache.paimon.utils.FileStorePathFactory.BUCKET_PATH_PREFIX
-import org.apache.paimon.utils.SerializableConsumer
 
 import org.apache.spark.internal.Logging
 import org.apache.spark.sql.{functions, Dataset, PaimonSparkSession, SparkSession}
@@ -47,7 +46,7 @@ case class SparkOrphanFilesClean(
     parallelism: Int,
     dryRunPara: Boolean,
     @transient spark: SparkSession)
-  extends OrphanFilesClean(specifiedTable, specifiedOlderThanMillis, dryRunPara)
+  extends OrphanFilesClean(specifiedTable, specifiedOlderThanMillis, dryRunPara, false)
   with SQLConfHelper
   with Logging {
 

Original file line number	Diff line number	Diff line change
`@@ -85,7 +85,7 @@ public LocalOrphanFilesClean(FileStoreTable table, long olderThanMillis) {`
`85`	`85`	`}`
`86`	`86`
`87`	`87`	`public LocalOrphanFilesClean(FileStoreTable table, long olderThanMillis, boolean dryRun) {`
`88`		`- super(table, olderThanMillis, dryRun);`
	`88`	`+ super(table, olderThanMillis, dryRun, false);`
`89`	`89`	`this.deleteFiles = new ArrayList<>();`
`90`	`90`	`this.executor =`
`91`	`91`	`createCachedThreadPool(`