[core] limit parallelly read file memory usage, extract some methods #1072

leaves12138 · 2023-05-05T03:27:10Z

[core] limit parallelly read file memory usage, extract some methods(#1061)

Purpose

Tests

(List UT and IT cases to verify this change)

API and Format

(Does this change affect API or storage format)

Documentation

(Does this change introduce a new feature)

JingsongLi · 2023-05-05T04:16:53Z

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

+                            // however entry.bucket() was computed against the old numOfBuckets
+                            // and thus the filtered manifest entries might be empty
+                            // which renders the bucket check invalid
+                            if (filterByBucket(file)


'if' statement can be simplified

JingsongLi · 2023-05-05T04:18:17Z

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

+    }
+
+    private <T extends AbstractManifestEntry> Pair<Long, List<T>> doPlan(
+            Function<List<ManifestFileMeta>, List<T>> processor,


readManifestFile?

JingsongLi · 2023-05-05T04:18:38Z

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

+                doPlan(
+                        // how to process entry files
+                        entries ->
+                                entries.parallelStream()


Can we also put these logical into doPlan?
Only filterByStats is special, we can just instanceof ManifestEntry in filterByStats?

JingsongLi · 2023-05-05T04:20:16Z

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

+
+    private <T extends AbstractManifestEntry> Pair<Long, List<T>> doPlan(
+            Function<List<ManifestFileMeta>, List<T>> processor,
+            Function<T, Boolean> postFilterProcessor) {


Maybe we don't need this one, postFilter can be applied to SimpleManifestEntry too.
If there is level filter, throw exception.

leaves12138 · 2023-05-05T06:11:29Z

fixed comment

JingsongLi · 2023-05-05T06:48:04Z

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

+        if (entry instanceof ManifestEntry) {
+            return filterByStats((ManifestEntry) entry);
+        } else {
+            throw new RuntimeException("only complete manifest entry could be filter by stats");


return true;

JingsongLi · 2023-05-05T07:04:59Z

paimon-core/src/main/java/org/apache/paimon/utils/ParallellyExecuteUtils.java

+    // reduce memory usage by batch iterable process, the cached result in memory will be 2 *
+    // queueSize
+    public static <T, U> Iterable<T> parallelismBatchIterable(
+            Function<List<U>, List<T>> processor, List<U> input, int queueSize) {


default value of queueSize can be COMMON_IO_FORK_JOIN_POOL thread number * 2.

done this! set queueSize to FileUtils.COMMON_IO_FORK_JOIN_POOL.getParallelism() * 2

JingsongLi · 2023-05-05T07:05:39Z

paimon-core/src/main/java/org/apache/paimon/utils/ParallellyExecuteUtils.java

+                            if (index < activeList.size()) {
+                                next = activeList.get(index++);
+                                if (index == activeList.size()) {
+                                    activeList = null;


You implement a wrong iterator.
Test should cover multiple invoking for hasNext .

done this. added tests for multiple invoking hasNext method

JingsongLi · 2023-05-05T07:06:18Z

paimon-core/src/main/java/org/apache/paimon/utils/ParallellyExecuteUtils.java

+                                activeList = batch.get();
+                                if (stack.size() > 0) {
+                                    batch =
+                                            CompletableFuture.supplyAsync(


We don't need to produce more elements when the consumer not finish this batch.

JingsongLi · 2023-05-05T11:14:14Z

paimon-core/src/main/java/org/apache/paimon/utils/ParallellyExecuteUtils.java

+        int size = input.size();
+        int num = size / queueSize;
+
+        for (int i = 0; i < num; i++) {


Lists.partition?

JingsongLi · 2023-05-05T11:17:10Z

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

+        if (entry instanceof ManifestEntry) {
+            return filterByStats((ManifestEntry) entry);
+        }
+        return true;


Add comments here: filterByStats is an action that is completed as much as possible and does not have an impact if it is not done.

leaves12138 · 2023-05-05T11:35:28Z

fixed comment

wxplovecc · 2023-05-05T12:49:22Z

paimon-core/src/main/java/org/apache/paimon/utils/ParallellyExecuteUtils.java

+    public static <T, U> Iterable<T> parallelismBatchIterable(
+            Function<List<U>, List<T>> processor, List<U> input) {
+        // default queueSize
+        int queueSize = FileUtils.COMMON_IO_FORK_JOIN_POOL.getParallelism() * 2;


Maybe default queueSize should be smaller , After testing it in the outdoor environment ,it will oom and set smaller than 80 is Ok, how about not *2

Introduce an option: SCAN_MANIFEST_PARALLELISM, default is none (will be COMMON_IO_FORK_JOIN_POOL.getParallelism()).

leaves12138 · 2023-05-06T03:53:01Z

fixed comment

leaves12138 · 2023-05-06T06:38:55Z

test passed in my local environment?

JingsongLi · 2023-05-06T06:44:29Z

paimon-core/src/main/java/org/apache/paimon/utils/ParallellyExecuteUtils.java

+                                                        () -> processor.apply(stack.poll()),
+                                                        FileUtils.COMMON_IO_FORK_JOIN_POOL)
+                                                .get();
+                            } catch (InterruptedException e) {


Just

} catch (Exception e) { throw new RuntimeException(e); }

?

Swallow InterruptedException is not good.

Exception message is confused, "should never get here", IOException will go here.

…pache#1061)

…ck still has items

JingsongLi · 2023-05-08T02:03:16Z

paimon-core/src/main/java/org/apache/paimon/utils/ParallellyExecuteUtils.java

@@ -66,7 +67,7 @@ public T next() {
                    }

                    private void advanceIfNeeded() {
-                        if ((activeList == null || index >= activeList.size())
+                        while ((activeList == null || index >= activeList.size())


Add a unit test for this?

JingsongLi

+1

…methods (apache#1072)" This reverts commit 5d23c7d.

…everted code, we don't need AbstractEntry, but we still need memory control) (apache#1072)

leaves12138 changed the title ~~[core] limit parallelly read file memory usage, extract some methods(#1061)~~ [core] limit parallelly read file memory usage, extract some methods May 5, 2023

JingsongLi reviewed May 5, 2023

View reviewed changes

leaves12138 force-pushed the improve_memory-limit-read-file branch from 2f01378 to eed1bc3 Compare May 5, 2023 06:02

JingsongLi reviewed May 5, 2023

View reviewed changes

leaves12138 force-pushed the improve_memory-limit-read-file branch 3 times, most recently from ece57b0 to adf8b8e Compare May 5, 2023 08:25

JingsongLi reviewed May 5, 2023

View reviewed changes

leaves12138 force-pushed the improve_memory-limit-read-file branch from adf8b8e to 0faf3b7 Compare May 5, 2023 11:28

wxplovecc reviewed May 5, 2023

View reviewed changes

leaves12138 force-pushed the improve_memory-limit-read-file branch 2 times, most recently from 352903e to bb6f277 Compare May 6, 2023 03:45

JingsongLi reviewed May 6, 2023

View reviewed changes

[core] limit parallelly read file memory usage, extract some methods(a…

f0c4135

…pache#1061)

leaves12138 force-pushed the improve_memory-limit-read-file branch from bb6f277 to f0c4135 Compare May 6, 2023 07:12

yejunhao added 3 commits May 6, 2023 15:56

[fix] fix comment and repair bug -- hasNext may return false when sta…

ed4e581

…ck still has items

[fix] delete debug code

cbe8f5d

[fix] generate docs for scan.manifest.parallelism configuration

9470986

github-actions bot added documentation Improvements or additions to documentation core labels May 6, 2023

JingsongLi closed this May 6, 2023

JingsongLi reopened this May 6, 2023

JingsongLi reviewed May 8, 2023

View reviewed changes

yejunhao added 2 commits May 8, 2023 10:42

[test] add test for different queue size with filter

d0af636

[fix] fix checkstyle

6217834

JingsongLi approved these changes May 8, 2023

View reviewed changes

JingsongLi merged commit 5d23c7d into apache:master May 8, 2023

leaves12138 pushed a commit to leaves12138/incubator-paimon that referenced this pull request May 10, 2023

Revert "[core] limit parallelly read file memory usage, extract some …

9e769a9

…methods (apache#1072)" This reverts commit 5d23c7d.

leaves12138 pushed a commit to leaves12138/incubator-paimon that referenced this pull request May 10, 2023

[core] limit parallelly read file memory usage (cherry-pick part of r…

a0a53f4

…everted code, we don't need AbstractEntry, but we still need memory control) (apache#1072)

JingsongLi mentioned this pull request May 11, 2023

[Feature] Reduce memory use when scan manifests #907

Closed

2 tasks

leaves12138 deleted the improve_memory-limit-read-file branch May 11, 2023 09:50

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[core] limit parallelly read file memory usage, extract some methods #1072

[core] limit parallelly read file memory usage, extract some methods #1072

leaves12138 commented May 5, 2023

JingsongLi May 5, 2023

JingsongLi May 5, 2023

JingsongLi May 5, 2023

JingsongLi May 5, 2023

leaves12138 commented May 5, 2023

JingsongLi May 5, 2023

leaves12138 May 5, 2023

JingsongLi May 5, 2023

leaves12138 May 5, 2023

JingsongLi May 5, 2023

leaves12138 May 5, 2023

JingsongLi May 5, 2023

leaves12138 May 5, 2023

JingsongLi May 5, 2023

leaves12138 May 5, 2023

JingsongLi May 5, 2023

leaves12138 May 5, 2023

leaves12138 commented May 5, 2023

wxplovecc May 5, 2023 •

edited

Loading

JingsongLi May 6, 2023

leaves12138 May 6, 2023

leaves12138 commented May 6, 2023

leaves12138 commented May 6, 2023

JingsongLi May 6, 2023

JingsongLi May 8, 2023 •

edited

Loading

JingsongLi left a comment

[core] limit parallelly read file memory usage, extract some methods #1072

[core] limit parallelly read file memory usage, extract some methods #1072

Conversation

leaves12138 commented May 5, 2023

Purpose

Tests

API and Format

Documentation

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

leaves12138 commented May 5, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

leaves12138 commented May 5, 2023

wxplovecc May 5, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

leaves12138 commented May 6, 2023

leaves12138 commented May 6, 2023

Choose a reason for hiding this comment

JingsongLi May 8, 2023 • edited Loading

Choose a reason for hiding this comment

JingsongLi left a comment

Choose a reason for hiding this comment

wxplovecc May 5, 2023 •

edited

Loading

JingsongLi May 8, 2023 •

edited

Loading