[Bug] fix sort compaction failed on empty input (apache#2766)

tongcheng-elong · Jan 23, 2024 · 898c0ea · 898c0ea
1 parent 05c17b1
commit 898c0ea
Show file tree

Hide file tree

Showing 2 changed files with 20 additions and 2 deletions.
diff --git a/...flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/shuffle/RangeShuffle.java b/...flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/shuffle/RangeShuffle.java
@@ -54,6 +54,7 @@
 
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.Comparator;
 import java.util.Iterator;
 import java.util.List;
@@ -257,9 +258,10 @@ public void endInput() throws Exception {
                     T record = sampledData.get((int) (i * avgRange));
                     boundaries[i - 1] = record;
                 }
+                collector.collect(Arrays.asList(boundaries));
+            } else {
+                collector.collect(Collections.emptyList());
             }
-
-            collector.collect(Arrays.asList(boundaries));
         }
     }
 

diff --git a/...src/test/java/org/apache/paimon/flink/action/SortCompactActionForUnawareBucketITCase.java b/...src/test/java/org/apache/paimon/flink/action/SortCompactActionForUnawareBucketITCase.java
@@ -258,6 +258,22 @@ public void testRandomSuffixWorks() throws Exception {
         Assertions.assertThat(files.size()).isEqualTo(3);
     }
 
+    @Test
+    public void testSortCompactionOnEmptyData() throws Exception {
+        createTable();
+        SortCompactAction sortCompactAction =
+                new SortCompactAction(
+                                warehouse,
+                                database,
+                                tableName,
+                                Collections.emptyMap(),
+                                Collections.emptyMap())
+                        .withOrderStrategy("zorder")
+                        .withOrderColumns(Collections.singletonList("f0"));
+
+        sortCompactAction.run();
+    }
+
     private void zorder(List<String> columns) throws Exception {
         if (RANDOM.nextBoolean()) {
             createAction("zorder", columns).run();