Re: [PR] [Flink]add sort compaction config [incubator-paimon]

["leaves12138 (via GitHub)" <gi...@apache.org>]

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: issues-unsubscribe@paimon.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org