This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
The following commit(s) were added to refs/heads/master by this push:
new dbc06f77e [core] Ensure enough tasks for scan.manifest.parallelism
(#2722)
dbc06f77e is described below
commit dbc06f77e935a757aa0681219e4bf7df49873cee
Author: Jingsong Lee <[email protected]>
AuthorDate: Wed Jan 17 18:56:53 2024 +0800
[core] Ensure enough tasks for scan.manifest.parallelism (#2722)
---
.../paimon/operation/AbstractFileStoreScan.java | 4 ++--
.../java/org/apache/paimon/utils/FileUtils.java | 27 ++++++++++++++++++----
...ExecuteUtils.java => ScanParallelExecutor.java} | 16 +++++++++----
...tilsTest.java => ScanParallelExecutorTest.java} | 18 +++++++--------
4 files changed, 45 insertions(+), 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
index c15959b50..766e17417 100644
---
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.types.RowType;
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 @@ public abstract class AbstractFileStoreScan implements
FileStoreScan {
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
index faa27afa4..6f14d8a62 100644
--- 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-core/src/main/java/org/apache/paimon/utils/ParallellyExecuteUtils.java
b/paimon-core/src/main/java/org/apache/paimon/utils/ScanParallelExecutor.java
similarity index 84%
rename from
paimon-core/src/main/java/org/apache/paimon/utils/ParallellyExecuteUtils.java
rename to
paimon-core/src/main/java/org/apache/paimon/utils/ScanParallelExecutor.java
index 3d82f9923..1299b618b 100644
---
a/paimon-core/src/main/java/org/apache/paimon/utils/ParallellyExecuteUtils.java
+++
b/paimon-core/src/main/java/org/apache/paimon/utils/ScanParallelExecutor.java
@@ -28,25 +28,27 @@ import java.util.List;
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 @@ public class ParallellyExecuteUtils {
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 @@ public class ParallellyExecuteUtils {
}
};
}
+
+ 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/paimon-core/src/test/java/org/apache/paimon/utils/ParallellyExecuteUtilsTest.java
b/paimon-core/src/test/java/org/apache/paimon/utils/ScanParallelExecutorTest.java
similarity index 89%
rename from
paimon-core/src/test/java/org/apache/paimon/utils/ParallellyExecuteUtilsTest.java
rename to
paimon-core/src/test/java/org/apache/paimon/utils/ScanParallelExecutorTest.java
index 6d1fd1321..e312a418e 100644
---
a/paimon-core/src/test/java/org/apache/paimon/utils/ParallellyExecuteUtilsTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/utils/ScanParallelExecutorTest.java
@@ -29,8 +29,8 @@ import java.util.List;
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 class ParallellyExecuteUtilsTest {
}
Iterable<Integer> re =
- ParallellyExecuteUtils.parallelismBatchIterable(
+ ScanParallelExecutor.parallelismBatchIterable(
l -> l.parallelStream().map(i -> i +
1).collect(Collectors.toList()),
nums,
null);
@@ -62,7 +62,7 @@ public class ParallellyExecuteUtilsTest {
}
Iterable<Integer> re =
- ParallellyExecuteUtils.parallelismBatchIterable(
+ ScanParallelExecutor.parallelismBatchIterable(
l -> l.parallelStream().map(i -> i +
1).collect(Collectors.toList()),
nums,
null);
@@ -83,7 +83,7 @@ public class ParallellyExecuteUtilsTest {
}
Iterable<Integer> re =
- ParallellyExecuteUtils.parallelismBatchIterable(
+ ScanParallelExecutor.parallelismBatchIterable(
l -> l.parallelStream().map(i -> i +
1).collect(Collectors.toList()),
nums,
null);
@@ -109,7 +109,7 @@ public class ParallellyExecuteUtilsTest {
}
Iterable<Integer> re =
- ParallellyExecuteUtils.parallelismBatchIterable(
+ ScanParallelExecutor.parallelismBatchIterable(
l -> l.parallelStream().map(i -> i +
1).collect(Collectors.toList()),
nums,
null);
@@ -129,7 +129,7 @@ public class ParallellyExecuteUtilsTest {
@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 class ParallellyExecuteUtilsTest {
@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 class ParallellyExecuteUtilsTest {
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);
