JingsongLi commented on code in PR #216:
URL: https://github.com/apache/flink-table-store/pull/216#discussion_r922940630
##########
flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/MergeTreeCompactManager.java:
##########
@@ -69,7 +75,17 @@ public void submitCompaction() {
throw new IllegalStateException(
"Please finish the previous compaction before submitting
new one.");
}
- strategy.pick(levels.numberOfLevels(), levels.levelSortedRuns())
+ List<LevelSortedRun> sortedRuns = levels.levelSortedRuns();
+ if (maxSortedRunNum != null && maxSortedRunNum < sortedRuns.size()) {
+ pickSortedRuns(sortedRuns.subList(0, maxSortedRunNum));
+ pickSortedRuns(sortedRuns.subList(maxSortedRunNum,
sortedRuns.size()));
+ } else {
+ pickSortedRuns(sortedRuns);
+ }
+ }
+
+ private void pickSortedRuns(List<LevelSortedRun> sortedRuns) {
+ strategy.pick(levels.numberOfLevels(), sortedRuns)
.ifPresent(
unit -> {
if (unit.files().size() < 2) {
Review Comment:
I think it is better to limit sorted runs in `CompactStrategy`. We can pass
`maxRuns` to `UniversalCompaction.this(...)`, and limit runs in `createUnit`.
The `strategy.pick(levels.numberOfLevels(), partial runs)` may lead to
incorrect runs, because strategy doesn't know the global information, it's not
sure if there are existing runs in the deep layers.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
