[AMORO-4277] imporove expire_snapshots performance (#4278)

* optimize expire snapshots
fix: logical error
remove some comments

* remove duplicate class and move test case to amoro-format-iceberg

* Update amoro-format-iceberg/src/main/java/org/apache/amoro/utils/TableFileUtil.java

Co-authored-by: Xu Bai <tocreationbai@gmail.com>

* fix review comment

---------

Co-authored-by: Xu Bai <tocreationbai@gmail.com>

Author: Aireed

amoro-ams/src/main/java/org/apache/amoro/server/utils/RollingFileCleaner.java

@@ -27,8 +27,11 @@
 import org.slf4j.LoggerFactory;
 
 import java.net.URI;
+import java.util.Comparator;
+import java.util.List;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
 
 /** Rolling file cleaner for Iceberg table maintenance operations. */
 public class RollingFileCleaner {
@@ -110,27 +113,34 @@ private void doCleanFiles() {
           }
         }
       }
-      // Try to delete empty parent directories. Skipped automatically by
-      // TableFileUtil.deleteEmptyDirectory when the underlying FileIO is an object store.
-      if (fileIO.supportFileSystemOperations()) {
-        for (String parentDir : parentDirectories) {
-          try {
-            TableFileUtil.deleteEmptyDirectory(fileIO, parentDir, excludeFiles);
-          } catch (Exception e) {
-            LOG.warn("Failed to delete empty parent directory: {}", parentDir, e);
-          }
-        }
-      }
 
       LOG.debug("Cleaned expired a file group, total files: {}", collectedFiles.size());
     } finally {
-      // Always drop both buffers so a failure in directory cleanup does not cause the
-      // already-deleted files to be re-submitted on the next batch.
-      parentDirectories.clear();
       collectedFiles.clear();
     }
   }
 
+  private void doCleanParentDirectory() {
+    // Try to delete empty parent directories. Skipped automatically by
+    // TableFileUtil.deleteEmptyDirectory when the underlying FileIO is an object store.
+    if (fileIO.supportFileSystemOperations()) {
+      List<String> parentDirectoriesSorted =
+          parentDirectories.stream()
+              .sorted(Comparator.comparingInt(String::length).reversed())
+              .collect(Collectors.toList());
+      for (String parentDir : parentDirectoriesSorted) {
+        try {
+          TableFileUtil.deleteEmptyDirectory(fileIO, parentDir, excludeFiles, parentDirectories);
+        } catch (Exception e) {
+          LOG.warn("Failed to delete empty parent directory: {}", parentDir, e);
+        }
+      }
+    }
+    // Always drop both buffers so a failure in directory cleanup does not cause the
+    // already-deleted files to be re-submitted on the next batch.
+    parentDirectories.clear();
+  }
+
   public int fileCount() {
     return fileCounter.get();
   }
@@ -143,6 +153,9 @@ public void clear() {
     if (!collectedFiles.isEmpty()) {
       doCleanFiles();
     }
+    if (!parentDirectories.isEmpty()) {
+      doCleanParentDirectory();
+    }
 
     collectedFiles.clear();
     excludeFiles.clear();

amoro-format-iceberg/src/main/java/org/apache/amoro/formats/iceberg/utils/RollingFileCleaner.java

@@ -28,6 +28,7 @@
 
 import java.io.File;
 import java.net.URI;
+import java.util.HashSet;
 import java.util.Set;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -75,15 +76,25 @@ public static String getFileDir(String filePath) {
     return filePath.substring(0, lastSlash);
   }
 
+  public static void deleteEmptyDirectory(
+      AuthenticatedFileIO io, String directoryPath, Set<String> exclude) {
+    deleteEmptyDirectory(io, directoryPath, exclude, new HashSet<>());
+  }
+
   /**
    * Try to recursiveDelete the empty directory
    *
    * @param io mixed-format file io
    * @param directoryPath directory location
    * @param exclude the directory will not be deleted
+   * @param directoriesToBeDeleted: all the directories that need to be deleted directories that
+   *     need to be deleted
    */
   public static void deleteEmptyDirectory(
-      AuthenticatedFileIO io, String directoryPath, Set<String> exclude) {
+      AuthenticatedFileIO io,
+      String directoryPath,
+      Set<String> exclude,
+      Set<String> directoriesToBeDeleted) {
     if (directoryPath == null || directoryPath.isEmpty()) {
       return;
     }
@@ -108,7 +119,10 @@ public static void deleteEmptyDirectory(
     if (io.asFileSystemIO().isEmptyDirectory(directoryPath)) {
       io.asFileSystemIO().deletePrefix(directoryPath);
       LOG.debug("success delete empty directory {}", directoryPath);
-      deleteEmptyDirectory(io, parent, exclude);
+      // for parent must be deleted after the sub-directory
+      if (directoriesToBeDeleted == null || !directoriesToBeDeleted.contains(parent)) {
+        deleteEmptyDirectory(io, parent, exclude);
+      }
     }
   }
 

amoro-format-iceberg/src/main/java/org/apache/amoro/utils/TableFileUtil.java

@@ -16,11 +16,10 @@
  * limitations under the License.
  */
 
-package org.apache.amoro.server.util;
+package org.apache.amoro.formats.iceberg.utils;
 
 import org.apache.amoro.io.AuthenticatedFileIO;
 import org.apache.amoro.io.AuthenticatedFileIOAdapter;
-import org.apache.amoro.server.utils.RollingFileCleaner;
 import org.apache.amoro.shade.guava32.com.google.common.collect.Sets;
 import org.apache.iceberg.inmemory.InMemoryFileIO;
 import org.junit.jupiter.api.Assertions;