leaves12138 commented on code in PR #2749:
URL: https://github.com/apache/incubator-paimon/pull/2749#discussion_r1464253705
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/shuffle/RangeShuffle.java:
##########
@@ -486,4 +548,45 @@ public int get() {
return list.get(RANDOM.nextInt(list.size()));
}
}
+
+ @VisibleForTesting
+ static <T> T[] allocateRangeBaseSize(Tuple2<T, Integer>[] elements, int
rangeNum) {
+ int elementsNum = elements.length;
+ long totalSize = Arrays.stream(elements).mapToLong(t -> (long)
t.f1).sum();
+ long restSize = totalSize;
+ double targetSize = (totalSize) / (double) (rangeNum);
+
+ @SuppressWarnings("unchecked")
+ T[] range = (T[]) new Object[rangeNum];
Review Comment:
why not rangeNum - 1? If we separate a array num to 1000 part， we should
set 999 point?
##########
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/SortCompactActionForUnawareBucketITCase.java:
##########
@@ -259,22 +261,26 @@ public void testRandomSuffixWorks() throws Exception {
}
private void zorder(List<String> columns) throws Exception {
+ String rangeStrategy = RANDOM.nextBoolean() ? "size" : "quantity";
if (RANDOM.nextBoolean()) {
- createAction("zorder", columns).run();
+ createAction("zorder", rangeStrategy, columns).run();
} else {
- callProcedure("zorder", columns);
+ callProcedure("zorder", rangeStrategy, columns);
Review Comment:
I think we need not to add rangStrategy in the parameters. We can find
anywhere in the sort process.
##########
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/SortCompactActionForUnawareBucketITCase.java:
##########
@@ -259,22 +261,26 @@ public void testRandomSuffixWorks() throws Exception {
}
private void zorder(List<String> columns) throws Exception {
+ String rangeStrategy = RANDOM.nextBoolean() ? "size" : "quantity";
if (RANDOM.nextBoolean()) {
- createAction("zorder", columns).run();
+ createAction("zorder", rangeStrategy, columns).run();
} else {
- callProcedure("zorder", columns);
+ callProcedure("zorder", rangeStrategy, columns);
Review Comment:
For example, we can find in SortUtils.sortStreamByKey, `Strategy strategy =
table.coreOptions().sortRangeStrategy`
##########
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/SortCompactActionForUnawareBucketITCase.java:
##########
@@ -287,14 +293,21 @@ private SortCompactAction createAction(String
orderStrategy, List<String> column
"--order_strategy",
orderStrategy,
"--order_by",
- String.join(",", columns));
+ String.join(",", columns),
+ "--sort_conf range_strategy=" + rangeStrategy,
Review Comment:
We can add this as table-config, when we do compaction, we just --table_conf
sort.range.strategy = xxx. By this way, we can simplify code and keep
compatibility with old code.
--
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]
