(incubator-paimon) branch master updated: [core] FieldIgnoreRetractAgg should implement the reset method by calling the wrapped aggregator (#2653)

[lz...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git


The following commit(s) were added to refs/heads/master by this push:
     new b767e8315 [core] FieldIgnoreRetractAgg should implement the reset method by calling the wrapped aggregator (#2653)
b767e8315 is described below

commit b767e83150363128de452821a901038aff1ae807
Author: tsreaper <ts...@gmail.com>
AuthorDate: Mon Jan 8 15:58:21 2024 +0800

    [core] FieldIgnoreRetractAgg should implement the reset method by calling the wrapped aggregator (#2653)
---
 .../compact/aggregate/FieldIgnoreRetractAgg.java   |  5 ++++
 .../apache/paimon/flink/PreAggregationITCase.java  | 33 +++++++++++++++++++++-
 2 files changed, 37 insertions(+), 1 deletion(-)

diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldIgnoreRetractAgg.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldIgnoreRetractAgg.java
index 449b55fba..7fe531378 100644
--- a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldIgnoreRetractAgg.java
+++ b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldIgnoreRetractAgg.java
@@ -38,6 +38,11 @@ public class FieldIgnoreRetractAgg extends FieldAggregator {
         return aggregator.agg(accumulator, inputField);
     }
 
+    @Override
+    public void reset() {
+        aggregator.reset();
+    }
+
     @Override
     public Object retract(Object accumulator, Object retractField) {
         return accumulator;
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PreAggregationITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PreAggregationITCase.java
index d2233a63e..33e0b4cab 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PreAggregationITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PreAggregationITCase.java
@@ -40,6 +40,7 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
+import java.util.stream.IntStream;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -1054,7 +1055,7 @@ public class PreAggregationITCase {
     public static class FirstValueAggregation extends CatalogITCaseBase {
         @Override
         protected List<String> ddl() {
-            return Collections.singletonList(
+            return Arrays.asList(
                     "CREATE TABLE T ("
                             + "k INT,"
                             + "a INT,"
@@ -1066,6 +1067,15 @@ public class PreAggregationITCase {
                             + "'fields.b.aggregate-function'='first_value',"
                             + "'fields.c.aggregate-function'='first_not_null_value',"
                             + "'sequence.field'='a'"
+                            + ");",
+                    "CREATE TABLE T2 ("
+                            + "k INT,"
+                            + "v STRING,"
+                            + "PRIMARY KEY (k) NOT ENFORCED)"
+                            + "WITH ("
+                            + "'merge-engine' = 'aggregation',"
+                            + "'fields.v.aggregate-function' = 'first_value',"
+                            + "'fields.v.ignore-retract' = 'true'"
                             + ");");
         }
 
@@ -1109,6 +1119,27 @@ public class PreAggregationITCase {
             assertThat(result)
                     .containsExactlyInAnyOrder(Row.of(1, 2, null, "1"), Row.of(2, 2, "2", "2"));
         }
+
+        @Test
+        public void testAggregatorResetWhenIgnoringRetract() {
+            int numRows = 100;
+            batchSql(
+                    "INSERT INTO T2 VALUES "
+                            + IntStream.range(0, numRows)
+                                    .mapToObj(i -> String.format("(%d, '%d')", i, i))
+                                    .collect(Collectors.joining(", ")));
+            batchSql(
+                    "INSERT INTO T2 VALUES "
+                            + IntStream.range(numRows / 2, numRows)
+                                    .mapToObj(i -> String.format("(%d, '%d')", i, i + numRows))
+                                    .collect(Collectors.joining(", ")));
+            List<Row> result = batchSql("SELECT * FROM T2");
+            assertThat(result)
+                    .containsExactlyInAnyOrder(
+                            IntStream.range(0, numRows)
+                                    .mapToObj(i -> Row.of(i, String.valueOf(i)))
+                                    .toArray(Row[]::new));
+        }
     }
 
     /** IT Test for aggregation merge engine. */