[core] Ensure enough tasks for scan.manifest.parallelism (apache#2722)

lmagic233 · Jan 17, 2024 · dbc06f7 · dbc06f7
1 parent 97d1314
commit dbc06f7
Show file tree

Hide file tree

Showing 4 changed files with 45 additions and 20 deletions.
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java b/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java
@@ -39,7 +39,7 @@
 import org.apache.paimon.utils.FileStorePathFactory;
 import org.apache.paimon.utils.Filter;
 import org.apache.paimon.utils.Pair;
-import org.apache.paimon.utils.ParallellyExecuteUtils;
+import org.apache.paimon.utils.ScanParallelExecutor;
 import org.apache.paimon.utils.SnapshotManager;
 
 import javax.annotation.Nullable;
@@ -261,7 +261,7 @@ private Pair<Snapshot, List<ManifestEntry>> doPlan(
 
         AtomicLong cntEntries = new AtomicLong(0);
         Iterable<ManifestEntry> entries =
-                ParallellyExecuteUtils.parallelismBatchIterable(
+                ScanParallelExecutor.parallelismBatchIterable(
                         files -> {
                             List<ManifestEntry> entryList =
                                     files.parallelStream()

diff --git a/paimon-core/src/main/java/org/apache/paimon/utils/FileUtils.java b/paimon-core/src/main/java/org/apache/paimon/utils/FileUtils.java
@@ -39,18 +39,35 @@ public class FileUtils {
 
     public static final ForkJoinPool COMMON_IO_FORK_JOIN_POOL;
 
-    // if we want to name threads in the fork join pool we need all these
-    // see https://stackoverflow.com/questions/34303094/
+    private static volatile ForkJoinPool scanIoForkJoinPool;
+
     static {
+        COMMON_IO_FORK_JOIN_POOL =
+                createForkJoinPool(
+                        "file-store-common-io", Runtime.getRuntime().availableProcessors());
+    }
+
+    private static ForkJoinPool createForkJoinPool(String namePrefix, int parallelism) {
+        // if we want to name threads in the fork join pool we need all these
+        // see https://stackoverflow.com/questions/34303094/
         ForkJoinPool.ForkJoinWorkerThreadFactory factory =
                 pool -> {
                     ForkJoinWorkerThread worker =
                             ForkJoinPool.defaultForkJoinWorkerThreadFactory.newThread(pool);
-                    worker.setName("file-store-common-io-" + worker.getPoolIndex());
+                    worker.setName(namePrefix + "-" + worker.getPoolIndex());
                     return worker;
                 };
-        COMMON_IO_FORK_JOIN_POOL =
-                new ForkJoinPool(Runtime.getRuntime().availableProcessors(), factory, null, false);
+        return new ForkJoinPool(parallelism, factory, null, false);
+    }
+
+    public static synchronized ForkJoinPool getScanIoForkJoinPool(int parallelism) {
+        if (scanIoForkJoinPool == null || parallelism > scanIoForkJoinPool.getParallelism()) {
+            if (scanIoForkJoinPool != null) {
+                scanIoForkJoinPool.shutdown();
+            }
+            scanIoForkJoinPool = createForkJoinPool("paimon-scan-io", parallelism);
+        }
+        return scanIoForkJoinPool;
     }
 
     public static <T> List<T> readListFromFile(

diff --git a/.../paimon/utils/ParallellyExecuteUtils.java → ...he/paimon/utils/ScanParallelExecutor.java b/.../paimon/utils/ParallellyExecuteUtils.java → ...he/paimon/utils/ScanParallelExecutor.java
@@ -28,25 +28,27 @@
 import java.util.NoSuchElementException;
 import java.util.Queue;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ForkJoinPool;
 import java.util.function.Function;
 
 /**
  * This class is a parallel execution util class, which mainly aim to process tasks parallelly with
  * memory control.
  */
-public class ParallellyExecuteUtils {
+public class ScanParallelExecutor {
 
     // reduce memory usage by batch iterable process, the cached result in memory will be queueSize
     public static <T, U> Iterable<T> parallelismBatchIterable(
             Function<List<U>, List<T>> processor, List<U> input, @Nullable Integer queueSize) {
+        ForkJoinPool poolCandidate = FileUtils.COMMON_IO_FORK_JOIN_POOL;
         if (queueSize == null) {
-            queueSize = FileUtils.COMMON_IO_FORK_JOIN_POOL.getParallelism();
+            queueSize = poolCandidate.getParallelism();
         } else if (queueSize <= 0) {
             throw new NegativeArraySizeException("queue size should not be negetive");
         }
 
         final Queue<List<U>> stack = new ArrayDeque<>(Lists.partition(input, queueSize));
-
+        final int settledQueueSize = queueSize;
         return () ->
                 new Iterator<T>() {
                     List<T> activeList = null;
@@ -76,7 +78,7 @@ private void advanceIfNeeded() {
                                 activeList =
                                         CompletableFuture.supplyAsync(
                                                         () -> processor.apply(stack.poll()),
-                                                        FileUtils.COMMON_IO_FORK_JOIN_POOL)
+                                                        getExecutePool(settledQueueSize))
                                                 .get();
                             } catch (Exception e) {
                                 throw new RuntimeException(e);
@@ -85,4 +87,10 @@ private void advanceIfNeeded() {
                     }
                 };
     }
+
+    private static ForkJoinPool getExecutePool(int queueSize) {
+        return queueSize > FileUtils.COMMON_IO_FORK_JOIN_POOL.getParallelism()
+                ? FileUtils.getScanIoForkJoinPool(queueSize)
+                : FileUtils.COMMON_IO_FORK_JOIN_POOL;
+    }
 }
diff --git a/...mon/utils/ParallellyExecuteUtilsTest.java → ...aimon/utils/ScanParallelExecutorTest.java b/...mon/utils/ParallellyExecuteUtilsTest.java → ...aimon/utils/ScanParallelExecutorTest.java
@@ -29,8 +29,8 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
 
-/** This test mainly test for the methods in {@link ParallellyExecuteUtils}. */
-public class ParallellyExecuteUtilsTest {
+/** This test mainly test for the methods in {@link ScanParallelExecutor}. */
+public class ScanParallelExecutorTest {
 
     @Test
     public void testParallelismBatchIterable() {
@@ -41,7 +41,7 @@ public void testParallelismBatchIterable() {
         }
 
         Iterable<Integer> re =
-                ParallellyExecuteUtils.parallelismBatchIterable(
+                ScanParallelExecutor.parallelismBatchIterable(
                         l -> l.parallelStream().map(i -> i + 1).collect(Collectors.toList()),
                         nums,
                         null);
@@ -62,7 +62,7 @@ public void testParallelismBatchIterable2() {
         }
 
         Iterable<Integer> re =
-                ParallellyExecuteUtils.parallelismBatchIterable(
+                ScanParallelExecutor.parallelismBatchIterable(
                         l -> l.parallelStream().map(i -> i + 1).collect(Collectors.toList()),
                         nums,
                         null);
@@ -83,7 +83,7 @@ public void testParallelismBatchIterable3() {
         }
 
         Iterable<Integer> re =
-                ParallellyExecuteUtils.parallelismBatchIterable(
+                ScanParallelExecutor.parallelismBatchIterable(
                         l -> l.parallelStream().map(i -> i + 1).collect(Collectors.toList()),
                         nums,
                         null);
@@ -109,7 +109,7 @@ public void testParallelismBatchIterable4() {
         }
 
         Iterable<Integer> re =
-                ParallellyExecuteUtils.parallelismBatchIterable(
+                ScanParallelExecutor.parallelismBatchIterable(
                         l -> l.parallelStream().map(i -> i + 1).collect(Collectors.toList()),
                         nums,
                         null);
@@ -129,7 +129,7 @@ public void testParallelismBatchIterable4() {
     @Test
     public void testForEmptyInput() {
         Iterable<Integer> re =
-                ParallellyExecuteUtils.parallelismBatchIterable(
+                ScanParallelExecutor.parallelismBatchIterable(
                         l -> l.parallelStream().map(i -> i + 1).collect(Collectors.toList()),
                         (List<Integer>) Collections.EMPTY_LIST,
                         null);
@@ -139,7 +139,7 @@ public void testForEmptyInput() {
     @Test
     public void testForSingletonInput() {
         Iterable<Integer> re =
-                ParallellyExecuteUtils.parallelismBatchIterable(
+                ScanParallelExecutor.parallelismBatchIterable(
                         l -> l.parallelStream().map(i -> i + 1).collect(Collectors.toList()),
                         Collections.singletonList(1),
                         null);
@@ -150,7 +150,7 @@ public void testForSingletonInput() {
     public void testDifferentQueueSizeWithFilterElement() {
         for (int queueSize = 1; queueSize < 20; queueSize++) {
             Iterable<Integer> re =
-                    ParallellyExecuteUtils.parallelismBatchIterable(
+                    ScanParallelExecutor.parallelismBatchIterable(
                             l -> l.parallelStream().filter(i -> i > 5).collect(Collectors.toList()),
                             Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10),
                             queueSize);