fix

apache · Jan 31, 2024 · 0af443e · 0af443e
1 parent d877c44
commit 0af443e
Showing 1 changed file with 12 additions and 13 deletions.
diff --git a/...flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PreAggregationITCase.java b/...flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PreAggregationITCase.java
@@ -1444,6 +1444,12 @@ private boolean checkNestedTable(Row[] nestedTable, Row... subOrders) {
     /** ITCase for {@link FieldCollectAgg}. */
     public static class CollectAggregationITCase extends CatalogITCaseBase {
 
+        @Override
+        protected int defaultParallelism() {
+            // set parallelism to 1 so that the order of input data is determined
+            return 1;
+        }
+
         @Test
         public void testAggWithDistinct() {
             sql(
@@ -1525,11 +1531,8 @@ public void testRetractWithAggregation() throws Exception {
                             + "  'fields.f0.aggregate-function' = 'collect'"
                             + ")");
 
-            sql(
-                    "INSERT INTO INPUT VALUES "
-                            + "(1, CAST (NULL AS ARRAY<STRING>)), "
-                            + "(2, ARRAY['A', 'B'])");
-            sql("INSERT INTO INPUT VALUES (2, ARRAY['C', 'D'])");
+            sql("INSERT INTO INPUT VALUES (1, ARRAY['A', 'B'])");
+            sql("INSERT INTO INPUT VALUES (1, ARRAY['C', 'D'])");
 
             innerTestRetract();
         }
@@ -1548,11 +1551,8 @@ public void testRetractWithPartialUpdate() throws Exception {
                             + "  'fields.f1.sequence-group' = 'f0'"
                             + ")");
 
-            sql(
-                    "INSERT INTO INPUT VALUES "
-                            + "(1, CAST (NULL AS ARRAY<STRING>), 10), "
-                            + "(2, ARRAY['A', 'B'], 10)");
-            sql("INSERT INTO INPUT VALUES (2, ARRAY['C', 'D'], 20)");
+            sql("INSERT INTO INPUT VALUES (1, ARRAY['A', 'B'], 10)");
+            sql("INSERT INTO INPUT VALUES (1, ARRAY['C', 'D'], 20)");
 
             innerTestRetract();
         }
@@ -1563,14 +1563,13 @@ private void innerTestRetract() throws Exception {
                             "INSERT INTO test_collect SELECT * FROM INPUT /*+ OPTIONS('scan.snapshot-id'='1') */");
 
             CommonTestUtils.waitUtil(
-                    () -> sql("SELECT * FROM test_collect").size() == 2,
+                    () -> sql("SELECT * FROM test_collect").size() == 1,
                     Duration.ofMinutes(2),
                     Duration.ofMillis(100));
 
             List<Row> result = sql("SELECT * FROM test_collect");
-            checkOneRecord(result.get(0), 1);
             // if not retracted, the result would be ['A', 'B', 'C', 'D']
-            checkOneRecord(result.get(1), 2, "C", "D");
+            checkOneRecord(result.get(0), 1, "C", "D");
 
             insert.close();
         }