[core] Support parallel reading of local orphan clean #4320
Conversation
| }); | ||
| } | ||
| usedFiles.addAll(dataFiles); | ||
| randomlyOnlyExecute( |
There was a problem hiding this comment.
I think if manifests size is very big, use randomlyOnlyExecute may oom?
Do you think so?
There was a problem hiding this comment.
I think there is indeed a risk of OOM.
But this can be avoided by reducing the parallelism (if it's equal to 1, then it's similar to serial execution before), and here it's just providing an ability to read in parallel.
If parallelism 1 is still OOM, then distributed mode is considered.
There was a problem hiding this comment.
What about change randomlyOnlyExecute to sequentialBatchedExecute?
There was a problem hiding this comment.
Yeah, I repleaced it with sequentialBatchedExecute since It has better control over memory
There was a problem hiding this comment.
In the end, you still need to put it in a set, so as long as there is no inflation during the reading process, there should be no risk of oom here.
There was a problem hiding this comment.
I test sequentialBatchedExecute randomlyOnlyExecute and randomlyExecute in 16 parallelism on table with 2PB of data , None of them happened OOM.
Besides sequentialBatchedExecute procedure cost 30 min , randomlyOnlyExecute and randomlyExecute cost about 20 min
So I prefer to use randomlyOnlyExecute because this method is more concise and no other variables are introduced, it can also reduce the gc process of temporary data files variable. The disadvantage of this method is that there will be waiting lock overhead while inserting useFiles, but this has no impact on the overall time consumption, because the overall time consumption mainly comes from IO.
What do you think? @JingsongLi @wwj6591812
| @@ -472,13 +476,18 @@ public List<Snapshot> safelyGetAllSnapshots() throws IOException { | |||
| .map(id -> snapshotPath(id)) | |||
| } catch (FileNotFoundException ignored) { | ||
| } | ||
| } | ||
| List<Snapshot> snapshots = Collections.synchronizedList(new ArrayList<>()); |
There was a problem hiding this comment.
Collections.synchronizedList(new ArrayList<>(paths.size()));
| @@ -489,18 +498,36 @@ public List<Changelog> safelyGetAllChangelogs() throws IOException { | |||
| .map(id -> longLivedChangelogPath(id)) | |||
There was a problem hiding this comment.
this::longLivedChangelogPath
| } catch (FileNotFoundException ignored) { | ||
| } | ||
| } | ||
| List<Changelog> changelogs = Collections.synchronizedList(new ArrayList<>()); |
There was a problem hiding this comment.
Collections.synchronizedList(new ArrayList<>(paths.size()));
|
|
||
| return changelogs; | ||
| } | ||
|
|
||
| private void collectSnapshots(Consumer<Path> pathConsumer, List<Path> paths) | ||
| throws IOException { | ||
| ExecutorService executor = |
There was a problem hiding this comment.
Why not use ThreadPoolUtils#createCachedThreadPool？
| .forEach(dataFiles::add); | ||
| }); | ||
| Iterable<String> dataFiles = | ||
| sequentialBatchedExecute( |
There was a problem hiding this comment.
Why using sequentialBatchedExecute? Maybe use randomlyExecute?
There was a problem hiding this comment.
replace sequentialBatchedExecute with randomlyOnlyExecute , detail in #4320 (comment)
This reverts commit 43350a6.
Purpose
Increase parallel reading of snapshot, manifest, and data file meta to improve execution efficiency
Tests
Test in table with 7TB size with local mode:
Before this patch: local orphan clean will cost 20 min
After this patcgh, use the same sql in parallelism 16, the procedure only cost 5 min
API and Format
Documentation