Revert "use sequentialBatchedExecute"

This reverts commit 43350a6.

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

@@ -55,7 +55,6 @@
 import static org.apache.paimon.utils.ThreadPoolUtils.createCachedThreadPool;
 import static org.apache.paimon.utils.ThreadPoolUtils.randomlyExecute;
 import static org.apache.paimon.utils.ThreadPoolUtils.randomlyOnlyExecute;
-import static org.apache.paimon.utils.ThreadPoolUtils.sequentialBatchedExecute;
 
 /**
  * Local {@link OrphanFilesClean}, it will use thread pool to execute deletion.
@@ -71,8 +70,6 @@ public class LocalOrphanFilesClean extends OrphanFilesClean {
 
  private Set<String> candidateDeletes;
 
- private final int parallelism;
-
  public LocalOrphanFilesClean(FileStoreTable table) {
  this(table, System.currentTimeMillis() - TimeUnit.DAYS.toMillis(1));
  }
@@ -85,8 +82,9 @@ public LocalOrphanFilesClean(
  FileStoreTable table, long olderThanMillis, SerializableConsumer<Path> fileCleaner) {
  super(table, olderThanMillis, fileCleaner);
  this.deleteFiles = new ArrayList<>();
- this.parallelism = table.coreOptions().deleteFileThreadNum();
- this.executor = createCachedThreadPool(parallelism, "ORPHAN_FILES_CLEAN");
+ this.executor =
+ createCachedThreadPool(
+ table.coreOptions().deleteFileThreadNum(), "ORPHAN_FILES_CLEAN");
  }
 
  public List<Path> clean() throws IOException, ExecutionException, InterruptedException {
@@ -141,39 +139,29 @@ private Set<String> getUsedFiles(String branch) {
  try {
  Set<String> manifests = ConcurrentHashMap.newKeySet();
  collectWithoutDataFile(branch, usedFiles::add, manifests::add);
- Iterable<String> dataFiles =
- sequentialBatchedExecute(
- executor,
- manifestName -> {
- try {
- List<String> dataFilesInBatch = new ArrayList<>();
- retryReadingFiles(
- () ->
- manifestFile.readWithIOException(
- manifestName),
- Collections.<ManifestEntry>emptyList())
- .stream()
- .map(ManifestEntry::file)
- .forEach(
- f -> {
- if (candidateDeletes.contains(
- f.fileName())) {
- dataFilesInBatch.add(f.fileName());
- }
- f.extraFiles().stream()
- .filter(candidateDeletes::contains)
- .forEach(dataFilesInBatch::add);
- });
- return dataFilesInBatch;
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- },
- new ArrayList<>(manifests),
- parallelism);
- for (String fileName : dataFiles) {
- usedFiles.add(fileName);
- }
+ randomlyOnlyExecute(
+ executor,
+ manifestName -> {
+ try {
+ retryReadingFiles(
+ () -> manifestFile.readWithIOException(manifestName),
+ Collections.<ManifestEntry>emptyList())
+ .stream()
+ .map(ManifestEntry::file)
+ .forEach(
+ f -> {
+ if (candidateDeletes.contains(f.fileName())) {
+ usedFiles.add(f.fileName());
+ }
+ f.extraFiles().stream()
+ .filter(candidateDeletes::contains)
+ .forEach(usedFiles::add);
+ });
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ },
+ manifests);
  } catch (IOException e) {
  throw new RuntimeException(e);
  }