You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2025/07/07 07:36:51 UTC
(paimon) branch master updated: [core] Add aggregate function validation in SchemaValidation
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/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 6e83e04c4c [core] Add aggregate function validation in SchemaValidation
6e83e04c4c is described below
commit 6e83e04c4cb3b83acc730e4519bd86f7014cec62
Author: JingsongLi <ji...@gmail.com>
AuthorDate: Mon Jul 7 14:02:55 2025 +0800
[core] Add aggregate function validation in SchemaValidation
---
.../org/apache/paimon/schema/SchemaValidation.java | 30 +
.../apache/paimon/table/PrimaryKeyTableUtils.java | 18 -
.../paimon/flink/AbstractFlinkTableFactory.java | 2 -
.../apache/paimon/flink/PartialUpdateITCase.java | 19 -
.../apache/paimon/flink/PreAggregationITCase.java | 2170 --------------------
.../flink/aggregation/BasicAggregateITCase.java | 85 +
.../aggregation/BoolOrAndAggregationITCase.java | 97 +
.../aggregation/CollectAggregationITCase.java | 298 +++
.../FieldsDefaultAggregationITCase.java | 117 ++
.../aggregation/FirstValueAggregationITCase.java | 129 ++
.../LastNonNullValueAggregationITCase.java | 112 +
.../aggregation/LastValueAggregationITCase.java | 109 +
.../aggregation/ListAggAggregationITCase.java | 106 +
.../flink/aggregation/MaxAggregationITCase.java | 209 ++
.../aggregation/MergeMapAggregationITCase.java | 134 ++
.../flink/aggregation/MinAggregationITCase.java | 206 ++
.../aggregation/NestedUpdateAggregationITCase.java | 332 +++
.../NotExistAggregationFunctionITCase.java | 45 +
.../aggregation/ProductAggregationITCase.java | 184 ++
.../RoaringBitmapAggAggregationITCase.java | 156 ++
.../flink/aggregation/SumAggregationITCase.java | 180 ++
.../SumRetractionAggregationITCase.java | 74 +
.../ThetaSketchAggAggregationITCase.java | 75 +
23 files changed, 2678 insertions(+), 2209 deletions(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
index cf6b919499..ddde77cbb1 100644
--- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
+++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
@@ -21,7 +21,10 @@ package org.apache.paimon.schema;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.CoreOptions.ChangelogProducer;
import org.apache.paimon.CoreOptions.MergeEngine;
+import org.apache.paimon.factories.FactoryUtil;
import org.apache.paimon.format.FileFormat;
+import org.apache.paimon.mergetree.compact.aggregate.FieldAggregator;
+import org.apache.paimon.mergetree.compact.aggregate.factory.FieldAggregatorFactory;
import org.apache.paimon.options.ConfigOption;
import org.apache.paimon.options.Options;
import org.apache.paimon.table.BucketMode;
@@ -223,6 +226,8 @@ public class SchemaValidation {
if (options.deletionVectorsEnabled()) {
validateForDeletionVectors(options);
}
+
+ validateMergeFunctionFactory(schema);
}
public static void validateFallbackBranch(SchemaManager schemaManager, TableSchema schema) {
@@ -589,4 +594,29 @@ public class SchemaValidation {
private static boolean isPostponeBucketTable(TableSchema schema, int bucket) {
return !schema.primaryKeys().isEmpty() && bucket == BucketMode.POSTPONE_BUCKET;
}
+
+ private static void validateMergeFunctionFactory(TableSchema schema) {
+ if (schema.primaryKeys().isEmpty()) {
+ return;
+ }
+ CoreOptions options = new CoreOptions(schema.options());
+ switch (options.mergeEngine()) {
+ case DEDUPLICATE:
+ case FIRST_ROW:
+ return;
+ default:
+ }
+
+ for (int i = 0; i < schema.logicalRowType().getFieldNames().size(); i++) {
+ String fieldName = schema.logicalRowType().getFieldNames().get(i);
+ String aggFuncName = options.fieldAggFunc(fieldName);
+ aggFuncName = aggFuncName == null ? options.fieldsDefaultFunc() : aggFuncName;
+ if (aggFuncName != null) {
+ FactoryUtil.discoverFactory(
+ FieldAggregator.class.getClassLoader(),
+ FieldAggregatorFactory.class,
+ aggFuncName);
+ }
+ }
+ }
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/PrimaryKeyTableUtils.java b/paimon-core/src/main/java/org/apache/paimon/table/PrimaryKeyTableUtils.java
index 2d152e958e..a47e44718e 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/PrimaryKeyTableUtils.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/PrimaryKeyTableUtils.java
@@ -20,14 +20,11 @@ package org.apache.paimon.table;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.KeyValue;
-import org.apache.paimon.factories.FactoryUtil;
import org.apache.paimon.mergetree.compact.DeduplicateMergeFunction;
import org.apache.paimon.mergetree.compact.FirstRowMergeFunction;
import org.apache.paimon.mergetree.compact.MergeFunctionFactory;
import org.apache.paimon.mergetree.compact.PartialUpdateMergeFunction;
import org.apache.paimon.mergetree.compact.aggregate.AggregateMergeFunction;
-import org.apache.paimon.mergetree.compact.aggregate.FieldAggregator;
-import org.apache.paimon.mergetree.compact.aggregate.factory.FieldAggregatorFactory;
import org.apache.paimon.options.Options;
import org.apache.paimon.schema.KeyValueFieldsExtractor;
import org.apache.paimon.schema.TableSchema;
@@ -78,21 +75,6 @@ public class PrimaryKeyTableUtils {
}
}
- public static void validateMergeFunctionFactory(TableSchema tableSchema) {
- CoreOptions options = new CoreOptions(tableSchema.options());
- for (int i = 0; i < tableSchema.logicalRowType().getFieldNames().size(); i++) {
- String fieldName = tableSchema.logicalRowType().getFieldNames().get(i);
- String aggFuncName = options.fieldAggFunc(fieldName);
- aggFuncName = aggFuncName == null ? options.fieldsDefaultFunc() : aggFuncName;
- if (aggFuncName != null) {
- FactoryUtil.discoverFactory(
- FieldAggregator.class.getClassLoader(),
- FieldAggregatorFactory.class,
- aggFuncName);
- }
- }
- }
-
/** Primary key fields extractor. */
public static class PrimaryKeyFieldsExtractor implements KeyValueFieldsExtractor {
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/AbstractFlinkTableFactory.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/AbstractFlinkTableFactory.java
index 36d8525f04..a2e78d4b34 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/AbstractFlinkTableFactory.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/AbstractFlinkTableFactory.java
@@ -35,7 +35,6 @@ import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.FileStoreTableFactory;
-import org.apache.paimon.table.PrimaryKeyTableUtils;
import org.apache.paimon.table.Table;
import org.apache.paimon.table.object.ObjectTable;
import org.apache.paimon.utils.Preconditions;
@@ -227,7 +226,6 @@ public abstract class AbstractFlinkTableFactory
throw new RuntimeException(e);
}
}
- PrimaryKeyTableUtils.validateMergeFunctionFactory(fileStoreTable.schema());
FileStoreTable table = fileStoreTable.copyWithoutTimeTravel(newOptions);
if (Options.fromMap(table.options()).get(FILESYSTEM_JOB_LEVEL_SETTINGS_ENABLED)) {
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PartialUpdateITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PartialUpdateITCase.java
index 59ecc65a36..2da837d2f4 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PartialUpdateITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PartialUpdateITCase.java
@@ -22,7 +22,6 @@ import org.apache.paimon.utils.BlockingIterator;
import org.apache.paimon.utils.CommonTestUtils;
import org.apache.flink.configuration.RestartStrategyOptions;
-import org.apache.flink.table.api.ValidationException;
import org.apache.flink.table.api.config.ExecutionConfigOptions;
import org.apache.flink.table.planner.factories.TestValuesTableFactory;
import org.apache.flink.types.Row;
@@ -773,22 +772,4 @@ public class PartialUpdateITCase extends CatalogITCaseBase {
assertThat(sql("SELECT * FROM seq_default_agg")).containsExactly(Row.of(0, 2, 3));
}
-
- @Test
- public void testSequenceGroupWithNotExistAgg() {
- sql(
- "CREATE TABLE seq_not_exist_agg ("
- + " pk INT PRIMARY KEY NOT ENFORCED,"
- + " seq INT,"
- + " v INT) WITH ("
- + " 'merge-engine'='partial-update',"
- + " 'fields.seq.sequence-group'='v',"
- + " 'fields.default-aggregate-function'='not_exist'"
- + ")");
-
- assertThatThrownBy(() -> sql("INSERT INTO seq_not_exist_agg VALUES (0, 1, 1)"))
- .hasCauseInstanceOf(ValidationException.class)
- .hasMessageContaining(
- "Unable to create a sink for writing table 'PAIMON.default.seq_not_exist_agg'.");
- }
}
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
deleted file mode 100644
index e37d18ddb8..0000000000
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PreAggregationITCase.java
+++ /dev/null
@@ -1,2170 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.paimon.flink;
-
-import org.apache.paimon.mergetree.compact.aggregate.FieldCollectAgg;
-import org.apache.paimon.mergetree.compact.aggregate.FieldMergeMapAgg;
-import org.apache.paimon.mergetree.compact.aggregate.FieldNestedUpdateAgg;
-import org.apache.paimon.utils.BlockingIterator;
-import org.apache.paimon.utils.RoaringBitmap32;
-import org.apache.paimon.utils.RoaringBitmap64;
-
-import org.apache.commons.codec.binary.Hex;
-import org.apache.flink.table.api.ValidationException;
-import org.apache.flink.table.api.config.ExecutionConfigOptions;
-import org.apache.flink.table.planner.factories.TestValuesTableFactory;
-import org.apache.flink.types.Row;
-import org.apache.flink.types.RowKind;
-import org.apache.flink.util.CloseableIterator;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.Timeout;
-import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.Arguments;
-import org.junit.jupiter.params.provider.MethodSource;
-
-import java.io.IOException;
-import java.math.BigDecimal;
-import java.time.LocalDate;
-import java.time.LocalDateTime;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.Comparator;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Objects;
-import java.util.stream.Collectors;
-import java.util.stream.IntStream;
-
-import static org.apache.paimon.utils.ThetaSketch.sketchOf;
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.assertThatThrownBy;
-
-/** ITCase for pre-aggregation. */
-public class PreAggregationITCase {
- /** ITCase for bool_or/bool_and aggregate function. */
- public static class BoolOrAndAggregation extends CatalogITCaseBase {
- @Override
- protected List<String> ddl() {
- return Collections.singletonList(
- "CREATE TABLE IF NOT EXISTS T7 ("
- + "j INT, k INT, "
- + "a BOOLEAN, "
- + "b BOOLEAN,"
- + "PRIMARY KEY (j,k) NOT ENFORCED)"
- + " WITH ('merge-engine'='aggregation', "
- + "'fields.a.aggregate-function'='bool_or',"
- + "'fields.b.aggregate-function'='bool_and'"
- + ");");
- }
-
- @Test
- public void testMergeInMemory() {
- batchSql(
- "INSERT INTO T7 VALUES "
- + "(1, 2, CAST('TRUE' AS BOOLEAN), CAST('TRUE' AS BOOLEAN)),"
- + "(1, 2, CAST(NULL AS BOOLEAN), CAST(NULL AS BOOLEAN)), "
- + "(1, 2, CAST('FALSE' AS BOOLEAN), CAST('FALSE' AS BOOLEAN))");
- List<Row> result = batchSql("SELECT * FROM T7");
- assertThat(result).containsExactlyInAnyOrder(Row.of(1, 2, true, false));
- }
-
- @Test
- public void testMergeRead() {
- batchSql(
- "INSERT INTO T7 VALUES (1, 2, CAST('TRUE' AS BOOLEAN), CAST('TRUE' AS BOOLEAN))");
- batchSql("INSERT INTO T7 VALUES (1, 2, CAST(NULL AS BOOLEAN), CAST(NULL AS BOOLEAN))");
- batchSql(
- "INSERT INTO T7 VALUES (1, 2, CAST('FALSE' AS BOOLEAN), CAST('FALSE' AS BOOLEAN))");
-
- List<Row> result = batchSql("SELECT * FROM T7");
- assertThat(result).containsExactlyInAnyOrder(Row.of(1, 2, true, false));
- }
-
- @Test
- public void testMergeCompaction() {
- // Wait compaction
- batchSql("ALTER TABLE T7 SET ('commit.force-compact'='true')");
-
- // key 1 2
- batchSql(
- "INSERT INTO T7 VALUES (1, 2, CAST('TRUE' AS BOOLEAN), CAST('TRUE' AS BOOLEAN))");
- batchSql("INSERT INTO T7 VALUES (1, 2, CAST(NULL AS BOOLEAN), CAST(NULL AS BOOLEAN))");
- batchSql(
- "INSERT INTO T7 VALUES (1, 2, CAST('FALSE' AS BOOLEAN), CAST('FALSE' AS BOOLEAN))");
-
- // key 1 3
- batchSql(
- "INSERT INTO T7 VALUES (1, 3, CAST('FALSE' AS BOOLEAN), CAST('TRUE' AS BOOLEAN))");
- batchSql("INSERT INTO T7 VALUES (1, 3, CAST(NULL AS BOOLEAN), CAST(NULL AS BOOLEAN))");
- batchSql(
- "INSERT INTO T7 VALUES (1, 3, CAST('FALSE' AS BOOLEAN), CAST('TRUE' AS BOOLEAN))");
-
- assertThat(batchSql("SELECT * FROM T7"))
- .containsExactlyInAnyOrder(
- Row.of(1, 2, true, false), Row.of(1, 3, false, true));
- }
-
- @Test
- public void testStreamingRead() {
- assertThatThrownBy(
- () -> sEnv.from("T7").execute().print(),
- "Pre-aggregate continuous reading is not supported");
- }
- }
-
- /** ITCase for listagg aggregate function. */
- public static class ListAggAggregation extends CatalogITCaseBase {
- @Override
- protected int defaultParallelism() {
- // set parallelism to 1 so that the order of input data is determined
- return 1;
- }
-
- @Override
- protected List<String> ddl() {
- return Collections.singletonList(
- "CREATE TABLE IF NOT EXISTS T6 ("
- + "j INT, k INT, "
- + "a STRING, "
- + "PRIMARY KEY (j,k) NOT ENFORCED)"
- + " WITH ('merge-engine'='aggregation', "
- + "'fields.a.aggregate-function'='listagg'"
- + ");");
- }
-
- @Test
- public void testMergeInMemory() {
- // VALUES does not guarantee order, but order is important for list aggregations.
- // So we need to sort the input data.
- batchSql(
- "CREATE TABLE myTable AS "
- + "SELECT b, c, d FROM "
- + "(VALUES "
- + " (1, 1, 2, 'first line'),"
- + " (2, 1, 2, CAST(NULL AS STRING)),"
- + " (3, 1, 2, 'second line')"
- + ") AS V(a, b, c, d) "
- + "ORDER BY a");
- batchSql("INSERT INTO T6 SELECT * FROM myTable");
- List<Row> result = batchSql("SELECT * FROM T6");
- assertThat(result).containsExactlyInAnyOrder(Row.of(1, 2, "first line,second line"));
- }
-
- @Test
- public void testMergeRead() {
- batchSql("INSERT INTO T6 VALUES (1, 2, 'first line')");
- batchSql("INSERT INTO T6 VALUES (1, 2, CAST(NULL AS STRING))");
- batchSql("INSERT INTO T6 VALUES (1, 2, 'second line')");
-
- List<Row> result = batchSql("SELECT * FROM T6");
- assertThat(result).containsExactlyInAnyOrder(Row.of(1, 2, "first line,second line"));
- }
-
- @Test
- public void testMergeCompaction() {
- // Wait compaction
- batchSql("ALTER TABLE T6 SET ('commit.force-compact'='true')");
-
- // key 1 2
- batchSql("INSERT INTO T6 VALUES (1, 2, 'first line')");
- batchSql("INSERT INTO T6 VALUES (1, 2, CAST(NULL AS STRING))");
- batchSql("INSERT INTO T6 VALUES (1, 2, 'second line')");
-
- // key 1 3
- batchSql("INSERT INTO T6 VALUES (1, 3, CAST(NULL AS STRING))");
- batchSql("INSERT INTO T6 VALUES (1, 3, CAST(NULL AS STRING))");
- batchSql("INSERT INTO T6 VALUES (1, 3, CAST(NULL AS STRING))");
-
- assertThat(batchSql("SELECT * FROM T6"))
- .containsExactlyInAnyOrder(
- Row.of(1, 2, "first line,second line"), Row.of(1, 3, null));
- }
-
- @Test
- public void testStreamingRead() {
- assertThatThrownBy(
- () -> sEnv.from("T6").execute().print(),
- "Pre-aggregate continuous reading is not supported");
- }
- }
-
- /** ITCase for last value aggregate function. */
- public static class LastValueAggregation extends CatalogITCaseBase {
- @Override
- protected int defaultParallelism() {
- // set parallelism to 1 so that the order of input data is determined
- return 1;
- }
-
- @Override
- protected List<String> ddl() {
- return Collections.singletonList(
- "CREATE TABLE IF NOT EXISTS T5 ("
- + "j INT, k INT, "
- + "a INT, "
- + "i DATE,"
- + "PRIMARY KEY (j,k) NOT ENFORCED)"
- + " WITH ('merge-engine'='aggregation', "
- + "'fields.a.aggregate-function'='last_value', "
- + "'fields.i.aggregate-function'='last_value'"
- + ");");
- }
-
- @Test
- public void testMergeInMemory() {
- // VALUES does not guarantee order, but order is important for last value aggregations.
- // So we need to sort the input data.
- batchSql(
- "CREATE TABLE myTable AS "
- + "SELECT b, c, d, e FROM "
- + "(VALUES "
- + " (1, 1, 2, CAST(NULL AS INT), CAST('2020-01-01' AS DATE)),"
- + " (2, 1, 2, 2, CAST('2020-01-02' AS DATE)),"
- + " (3, 1, 2, 3, CAST(NULL AS DATE))"
- + ") AS V(a, b, c, d, e) "
- + "ORDER BY a");
- batchSql("INSERT INTO T5 SELECT * FROM myTable");
- List<Row> result = batchSql("SELECT * FROM T5");
- assertThat(result).containsExactlyInAnyOrder(Row.of(1, 2, 3, null));
- }
-
- @Test
- public void testMergeRead() {
- batchSql("INSERT INTO T5 VALUES (1, 2, CAST(NULL AS INT), CAST('2020-01-01' AS DATE))");
- batchSql("INSERT INTO T5 VALUES (1, 2, 2, CAST('2020-01-02' AS DATE))");
- batchSql("INSERT INTO T5 VALUES (1, 2, 3, CAST(NULL AS DATE))");
-
- List<Row> result = batchSql("SELECT * FROM T5");
- assertThat(result).containsExactlyInAnyOrder(Row.of(1, 2, 3, null));
- }
-
- @Test
- public void testMergeCompaction() {
- // Wait compaction
- batchSql("ALTER TABLE T5 SET ('commit.force-compact'='true')");
-
- // key 1 2
- batchSql("INSERT INTO T5 VALUES (1, 2, CAST(NULL AS INT), CAST('2020-01-01' AS DATE))");
- batchSql("INSERT INTO T5 VALUES (1, 2, 2, CAST('2020-01-02' AS DATE))");
- batchSql("INSERT INTO T5 VALUES (1, 2, 3, CAST(NULL AS DATE))");
-
- // key 1 3
- batchSql("INSERT INTO T5 VALUES (1, 3, 3, CAST('2020-01-01' AS DATE))");
- batchSql("INSERT INTO T5 VALUES (1, 3, 2, CAST(NULL AS DATE))");
- batchSql("INSERT INTO T5 VALUES (1, 3, CAST(NULL AS INT), CAST('2022-01-02' AS DATE))");
-
- assertThat(batchSql("SELECT * FROM T5"))
- .containsExactlyInAnyOrder(
- Row.of(1, 2, 3, null), Row.of(1, 3, null, LocalDate.of(2022, 1, 2)));
- }
-
- @Test
- public void testStreamingRead() {
- assertThatThrownBy(
- () -> sEnv.from("T5").execute().print(),
- "Pre-aggregate continuous reading is not supported");
- }
- }
-
- /** ITCase for last non-null value aggregate function. */
- public static class LastNonNullValueAggregation extends CatalogITCaseBase {
- @Override
- protected int defaultParallelism() {
- // set parallelism to 1 so that the order of input data is determined
- return 1;
- }
-
- @Override
- protected List<String> ddl() {
- return Collections.singletonList(
- "CREATE TABLE IF NOT EXISTS T4 ("
- + "j INT, k INT, "
- + "a INT, "
- + "b INT, "
- + "i DATE,"
- + "PRIMARY KEY (j,k) NOT ENFORCED)"
- + " WITH ('merge-engine'='aggregation', "
- + "'fields.a.aggregate-function'='last_non_null_value', "
- + "'fields.i.aggregate-function'='last_non_null_value'"
- + ");");
- }
-
- @Test
- public void testMergeInMemory() {
- // VALUES does not guarantee order, but order is important for last value aggregations.
- // So we need to sort the input data.
- batchSql(
- "CREATE TABLE myTable AS "
- + "SELECT b, c, d, e, f FROM "
- + "(VALUES "
- + " (1, 1, 2, CAST(NULL AS INT), 4, CAST('2020-01-01' AS DATE)),"
- + " (2, 1, 2, 2, CAST(NULL as INT), CAST('2020-01-02' AS DATE)),"
- + " (3, 1, 2, 3, 5, CAST(NULL AS DATE))"
- + ") AS V(a, b, c, d, e, f) "
- + "ORDER BY a");
- batchSql("INSERT INTO T4 SELECT * FROM myTable");
- List<Row> result = batchSql("SELECT * FROM T4");
- assertThat(result)
- .containsExactlyInAnyOrder(Row.of(1, 2, 3, 5, LocalDate.of(2020, 1, 2)));
- }
-
- @Test
- public void testMergeRead() {
- batchSql(
- "INSERT INTO T4 VALUES (1, 2, CAST(NULL AS INT), 3, CAST('2020-01-01' AS DATE))");
- batchSql(
- "INSERT INTO T4 VALUES (1, 2, 2, CAST(NULL AS INT), CAST('2020-01-02' AS DATE))");
- batchSql("INSERT INTO T4 VALUES (1, 2, 3, 5, CAST(NULL AS DATE))");
-
- List<Row> result = batchSql("SELECT * FROM T4");
- assertThat(result)
- .containsExactlyInAnyOrder(Row.of(1, 2, 3, 5, LocalDate.of(2020, 1, 2)));
- }
-
- @Test
- public void testMergeCompaction() {
- // Wait compaction
- batchSql("ALTER TABLE T4 SET ('commit.force-compact'='true')");
-
- // key 1 2
- batchSql(
- "INSERT INTO T4 VALUES (1, 2, CAST(NULL AS INT), 3, CAST('2020-01-01' AS DATE))");
- batchSql(
- "INSERT INTO T4 VALUES (1, 2, 2, CAST(NULL AS INT), CAST('2020-01-02' AS DATE))");
- batchSql("INSERT INTO T4 VALUES (1, 2, 3, 5, CAST(NULL AS DATE))");
-
- // key 1 3
- batchSql("INSERT INTO T4 VALUES (1, 3, 3, 4, CAST('2020-01-01' AS DATE))");
- batchSql("INSERT INTO T4 VALUES (1, 3, 2, 6, CAST(NULL AS DATE))");
- batchSql(
- "INSERT INTO T4 VALUES (1, 3, CAST(NULL AS INT), CAST(NULL AS INT), CAST('2022-01-02' AS DATE))");
-
- assertThat(batchSql("SELECT * FROM T4"))
- .containsExactlyInAnyOrder(
- Row.of(1, 2, 3, 5, LocalDate.of(2020, 1, 2)),
- Row.of(1, 3, 2, 6, LocalDate.of(2022, 1, 2)));
- }
-
- @Test
- public void testStreamingRead() {
- assertThatThrownBy(
- () -> sEnv.from("T4").execute().print(),
- "Pre-aggregate continuous reading is not supported");
- }
- }
-
- /** ITCase for min aggregate function. */
- public static class MinAggregation extends CatalogITCaseBase {
- @Override
- protected List<String> ddl() {
- return Collections.singletonList(
- "CREATE TABLE IF NOT EXISTS T3 ("
- + "j INT, k INT, "
- + "a INT, "
- + "b Decimal(4,2), "
- + "c TINYINT,"
- + "d SMALLINT,"
- + "e BIGINT,"
- + "f FLOAT,"
- + "h DOUBLE,"
- + "i DATE,"
- + "l TIMESTAMP,"
- + "m CHAR(1),"
- + "n VARCHAR,"
- + "PRIMARY KEY (j,k) NOT ENFORCED)"
- + " WITH ('merge-engine'='aggregation', "
- + "'fields.a.aggregate-function'='min', "
- + "'fields.b.aggregate-function'='min', "
- + "'fields.c.aggregate-function'='min', "
- + "'fields.d.aggregate-function'='min', "
- + "'fields.e.aggregate-function'='min', "
- + "'fields.f.aggregate-function'='min',"
- + "'fields.h.aggregate-function'='min',"
- + "'fields.i.aggregate-function'='min',"
- + "'fields.l.aggregate-function'='min',"
- + "'fields.m.aggregate-function'='min',"
- + "'fields.n.aggregate-function'='min'"
- + ");");
- }
-
- @Test
- public void testMergeInMemory() {
- batchSql(
- "INSERT INTO T3 VALUES "
- + "(1, 2, CAST(NULL AS INT), 1.01, CAST(-1 AS TINYINT), CAST(-1 AS SMALLINT), "
- + "CAST(1000 AS BIGINT), 1.11, CAST(1.11 AS DOUBLE), CAST('2020-01-01' AS DATE), "
- + "CAST('2021-01-01 01:01:01' AS TIMESTAMP), 'a', 'aaa'),"
- + "(1, 2, 2, 1.10, CAST(2 AS TINYINT), CAST(2 AS SMALLINT), "
- + "CAST(100000 AS BIGINT), -1.11, CAST(1.21 AS DOUBLE), CAST('2020-01-02' AS DATE), "
- + "CAST('2022-01-01 01:01:01' AS TIMESTAMP), 'b', 'bbb'), "
- + "(1, 2, 3, 10.00, CAST(1 AS TINYINT), CAST(1 AS SMALLINT), "
- + "CAST(10000000 AS BIGINT), 0, CAST(-1.11 AS DOUBLE), CAST('2022-01-02' AS DATE), "
- + "CAST('2022-01-01 02:00:00' AS TIMESTAMP), 'c', 'ccc')");
- List<Row> result = batchSql("SELECT * FROM T3");
- assertThat(result)
- .containsExactlyInAnyOrder(
- Row.of(
- 1,
- 2,
- 2,
- new BigDecimal("1.01"),
- (byte) -1,
- (short) -1,
- (long) 1000,
- (float) -1.11,
- -1.11,
- LocalDate.of(2020, 1, 1),
- LocalDateTime.of(2021, 1, 1, 1, 1, 1),
- "a",
- "aaa"));
- }
-
- @Test
- public void testMergeRead() {
- batchSql(
- "INSERT INTO T3 VALUES "
- + "(1, 2, CAST(NULL AS INT), 1.01, CAST(1 AS TINYINT), CAST(-1 AS SMALLINT), CAST(1000 AS BIGINT), "
- + "1.11, CAST(1.11 AS DOUBLE), CAST('2020-01-01' AS DATE), CAST('2021-01-01 01:01:01' AS TIMESTAMP), 'a', 'aaa')");
- batchSql(
- "INSERT INTO T3 VALUES "
- + "(1, 2, 2, 1.10, CAST(2 AS TINYINT), CAST(2 AS SMALLINT), CAST(100000 AS BIGINT), "
- + "-1.11, CAST(1.21 AS DOUBLE), CAST('2020-01-02' AS DATE), CAST('2022-01-01 01:01:01' AS TIMESTAMP), 'b', 'bbb')");
- batchSql(
- "INSERT INTO T3 VALUES "
- + "(1, 2, 3, 10.00, CAST(-1 AS TINYINT), CAST(1 AS SMALLINT), CAST(10000000 AS BIGINT), "
- + "0, CAST(-1.11 AS DOUBLE), CAST('2022-01-02' AS DATE), CAST('2022-01-01 02:00:00' AS TIMESTAMP), 'c', 'ccc')");
-
- List<Row> result = batchSql("SELECT * FROM T3");
- assertThat(result)
- .containsExactlyInAnyOrder(
- Row.of(
- 1,
- 2,
- 2,
- new BigDecimal("1.01"),
- (byte) -1,
- (short) -1,
- (long) 1000,
- (float) -1.11,
- -1.11,
- LocalDate.of(2020, 1, 1),
- LocalDateTime.of(2021, 1, 1, 1, 1, 1),
- "a",
- "aaa"));
- }
-
- @Test
- public void testMergeCompaction() {
- // Wait compaction
- batchSql("ALTER TABLE T3 SET ('commit.force-compact'='true')");
-
- // key 1 2
- batchSql(
- "INSERT INTO T3 VALUES "
- + "(1, 2, CAST(NULL AS INT), 1.01, CAST(1 AS TINYINT), CAST(-1 AS SMALLINT), CAST(1000 AS BIGINT), "
- + "1.11, CAST(1.11 AS DOUBLE), CAST('2020-01-01' AS DATE), CAST('2021-01-01 01:01:01' AS TIMESTAMP), 'a', 'aaa')");
- batchSql(
- "INSERT INTO T3 VALUES "
- + "(1, 2, 2, 1.10, CAST(2 AS TINYINT), CAST(2 AS SMALLINT), CAST(100000 AS BIGINT), "
- + "-1.11, CAST(1.21 AS DOUBLE), CAST('2020-01-02' AS DATE), CAST('2022-01-01 01:01:01' AS TIMESTAMP), 'b', 'bbb')");
- batchSql(
- "INSERT INTO T3 VALUES "
- + "(1, 2, 3, 10.00, CAST(-1 AS TINYINT), CAST(1 AS SMALLINT), CAST(10000000 AS BIGINT), "
- + "0, CAST(-1.11 AS DOUBLE), CAST('2022-01-02' AS DATE), CAST('2022-01-01 02:00:00' AS TIMESTAMP), 'c', 'ccc')");
-
- // key 1 3
- batchSql(
- "INSERT INTO T3 VALUES "
- + "(1, 3, CAST(NULL AS INT), 1.01, CAST(1 AS TINYINT), CAST(-1 AS SMALLINT), CAST(1000 AS BIGINT), "
- + "1.11, CAST(1.11 AS DOUBLE), CAST('2020-01-01' AS DATE), CAST('2021-01-01 01:01:01' AS TIMESTAMP), 'a', 'aaa')");
- batchSql(
- "INSERT INTO T3 VALUES "
- + "(1, 3, 6, 1.10, CAST(2 AS TINYINT), CAST(2 AS SMALLINT), CAST(100000 AS BIGINT), "
- + "-1.11, CAST(1.21 AS DOUBLE), CAST('2020-01-02' AS DATE), CAST('2022-01-01 01:01:01' AS TIMESTAMP), 'b', 'bbb')");
- batchSql(
- "INSERT INTO T3 VALUES "
- + "(1, 3, 3, 10.00, CAST(-1 AS TINYINT), CAST(1 AS SMALLINT), CAST(10000000 AS BIGINT), "
- + "0, CAST(-1.11 AS DOUBLE), CAST('2022-01-02' AS DATE), CAST('2022-01-01 02:00:00' AS TIMESTAMP), 'c', 'ccc')");
-
- assertThat(batchSql("SELECT * FROM T3"))
- .containsExactlyInAnyOrder(
- Row.of(
- 1,
- 2,
- 2,
- new BigDecimal("1.01"),
- (byte) -1,
- (short) -1,
- (long) 1000,
- (float) -1.11,
- -1.11,
- LocalDate.of(2020, 1, 1),
- LocalDateTime.of(2021, 1, 1, 1, 1, 1),
- "a",
- "aaa"),
- Row.of(
- 1,
- 3,
- 3,
- new BigDecimal("1.01"),
- (byte) -1,
- (short) -1,
- (long) 1000,
- (float) -1.11,
- -1.11,
- LocalDate.of(2020, 1, 1),
- LocalDateTime.of(2021, 1, 1, 1, 1, 1),
- "a",
- "aaa"));
- }
-
- @Test
- public void testStreamingRead() {
- assertThatThrownBy(
- () -> sEnv.from("T3").execute().print(),
- "Pre-aggregate continuous reading is not supported");
- }
- }
-
- /** ITCase for max aggregate function. */
- public static class MaxAggregation extends CatalogITCaseBase {
- @Override
- protected List<String> ddl() {
- return Collections.singletonList(
- "CREATE TABLE IF NOT EXISTS T2 ("
- + "j INT, k INT, "
- + "a INT, "
- + "b Decimal(4,2), "
- + "c TINYINT,"
- + "d SMALLINT,"
- + "e BIGINT,"
- + "f FLOAT,"
- + "h DOUBLE,"
- + "i DATE,"
- + "l TIMESTAMP,"
- + "m CHAR,"
- + "n VARCHAR,"
- + "PRIMARY KEY (j,k) NOT ENFORCED)"
- + " WITH ('merge-engine'='aggregation', "
- + "'fields.a.aggregate-function'='max', "
- + "'fields.b.aggregate-function'='max', "
- + "'fields.c.aggregate-function'='max', "
- + "'fields.d.aggregate-function'='max', "
- + "'fields.e.aggregate-function'='max', "
- + "'fields.f.aggregate-function'='max',"
- + "'fields.h.aggregate-function'='max',"
- + "'fields.i.aggregate-function'='max',"
- + "'fields.l.aggregate-function'='max',"
- + "'fields.m.aggregate-function'='max',"
- + "'fields.n.aggregate-function'='max'"
- + ");");
- }
-
- @Test
- public void testMergeInMemory() {
- batchSql(
- "INSERT INTO T2 VALUES "
- + "(1, 2, CAST(NULL AS INT), 1.01, CAST(1 AS TINYINT), CAST(-1 AS SMALLINT), "
- + "CAST(1000 AS BIGINT), 1.11, CAST(1.11 AS DOUBLE), CAST('2020-01-01' AS DATE), "
- + "CAST('2021-01-01 01:01:01' AS TIMESTAMP), 'a', 'aaa'),"
- + "(1, 2, 2, 1.10, CAST(2 AS TINYINT), CAST(2 AS SMALLINT), CAST(100000 AS BIGINT), "
- + "-1.11, CAST(1.21 AS DOUBLE), CAST('2020-01-02' AS DATE), "
- + "CAST('2022-01-01 01:01:01' AS TIMESTAMP), 'b', 'bbb'), "
- + "(1, 2, 3, 10.00, CAST(1 AS TINYINT), CAST(1 AS SMALLINT), CAST(10000000 AS BIGINT), "
- + "0, CAST(-1.11 AS DOUBLE), CAST('2022-01-02' AS DATE), "
- + "CAST('2022-01-01 02:00:00' AS TIMESTAMP), 'c', 'ccc')");
- List<Row> result = batchSql("SELECT * FROM T2");
- assertThat(result)
- .containsExactlyInAnyOrder(
- Row.of(
- 1,
- 2,
- 3,
- new BigDecimal("10.00"),
- (byte) 2,
- (short) 2,
- (long) 10000000,
- (float) 1.11,
- 1.21,
- LocalDate.of(2022, 1, 2),
- LocalDateTime.of(2022, 1, 1, 2, 0, 0),
- "c",
- "ccc"));
- }
-
- @Test
- public void testMergeRead() {
- batchSql(
- "INSERT INTO T2 VALUES "
- + "(1, 2, CAST(NULL AS INT), 1.01, CAST(1 AS TINYINT), CAST(-1 AS SMALLINT), CAST(1000 AS BIGINT), "
- + "1.11, CAST(1.11 AS DOUBLE), CAST('2020-01-01' AS DATE), "
- + "CAST('2021-01-01 01:01:01' AS TIMESTAMP), 'a', 'aaa')");
- batchSql(
- "INSERT INTO T2 VALUES "
- + "(1, 2, 2, 1.10, CAST(2 AS TINYINT), CAST(2 AS SMALLINT), CAST(100000 AS BIGINT), -1.11, "
- + "CAST(1.21 AS DOUBLE), CAST('2020-01-02' AS DATE), "
- + "CAST('2022-01-01 01:01:01' AS TIMESTAMP), 'b', 'bbb')");
- batchSql(
- "INSERT INTO T2 VALUES "
- + "(1, 2, 3, 10.00, CAST(1 AS TINYINT), CAST(1 AS SMALLINT), CAST(10000000 AS BIGINT), 0, "
- + "CAST(-1.11 AS DOUBLE), CAST('2022-01-02' AS DATE), "
- + "CAST('2022-01-01 02:00:00' AS TIMESTAMP), 'c', 'ccc')");
-
- List<Row> result = batchSql("SELECT * FROM T2");
- assertThat(result)
- .containsExactlyInAnyOrder(
- Row.of(
- 1,
- 2,
- 3,
- new BigDecimal("10.00"),
- (byte) 2,
- (short) 2,
- (long) 10000000,
- (float) 1.11,
- 1.21,
- LocalDate.of(2022, 1, 2),
- LocalDateTime.of(2022, 1, 1, 2, 0, 0),
- "c",
- "ccc"));
- }
-
- @Test
- public void testMergeCompaction() {
- // Wait compaction
- batchSql("ALTER TABLE T2 SET ('commit.force-compact'='true')");
-
- // key 1 2
- batchSql(
- "INSERT INTO T2 VALUES "
- + "(1, 2, CAST(NULL AS INT), 1.01, CAST(1 AS TINYINT), CAST(-1 AS SMALLINT), CAST(1000 AS BIGINT), "
- + "1.11, CAST(1.11 AS DOUBLE), CAST('2020-01-01' AS DATE), CAST('2021-01-01 01:01:01' AS TIMESTAMP), 'a', 'aaa')");
- batchSql(
- "INSERT INTO T2 VALUES "
- + "(1, 2, 2, 1.10, CAST(2 AS TINYINT), CAST(2 AS SMALLINT), CAST(100000 AS BIGINT), -1.11, "
- + "CAST(1.21 AS DOUBLE), CAST('2020-01-02' AS DATE), CAST('2022-01-01 01:01:01' AS TIMESTAMP), 'c', 'ccc')");
- batchSql(
- "INSERT INTO T2 VALUES "
- + "(1, 2, 3, 10.00, CAST(1 AS TINYINT), CAST(1 AS SMALLINT), CAST(10000000 AS BIGINT), 0, "
- + "CAST(-1.11 AS DOUBLE), CAST('2022-01-02' AS DATE), CAST('2022-01-01 02:00:00' AS TIMESTAMP), 'b', 'bbb')");
-
- // key 1 3
- batchSql(
- "INSERT INTO T2 VALUES "
- + "(1, 3, CAST(NULL AS INT), 1.01, CAST(1 AS TINYINT), CAST(-1 AS SMALLINT), CAST(1000 AS BIGINT), "
- + "1.11, CAST(1.11 AS DOUBLE), CAST('2020-01-01' AS DATE), CAST('2021-01-01 01:01:01' AS TIMESTAMP), 'a', 'aaa')");
- batchSql(
- "INSERT INTO T2 VALUES "
- + "(1, 3, 6, 1.10, CAST(2 AS TINYINT), CAST(2 AS SMALLINT), CAST(100000 AS BIGINT), -1.11, "
- + "CAST(1.21 AS DOUBLE), CAST('2020-01-02' AS DATE), CAST('2022-01-01 01:01:01' AS TIMESTAMP), 'c', 'ccc')");
- batchSql(
- "INSERT INTO T2 VALUES "
- + "(1, 3, 3, 10.00, CAST(1 AS TINYINT), CAST(1 AS SMALLINT), CAST(10000000 AS BIGINT), 0, "
- + "CAST(-1.11 AS DOUBLE), CAST('2022-01-02' AS DATE), CAST('2022-01-01 02:00:00' AS TIMESTAMP), 'b', 'bbb')");
-
- assertThat(batchSql("SELECT * FROM T2"))
- .containsExactlyInAnyOrder(
- Row.of(
- 1,
- 2,
- 3,
- new BigDecimal("10.00"),
- (byte) 2,
- (short) 2,
- (long) 10000000,
- (float) 1.11,
- 1.21,
- LocalDate.of(2022, 1, 2),
- LocalDateTime.of(2022, 1, 1, 2, 0, 0),
- "c",
- "ccc"),
- Row.of(
- 1,
- 3,
- 6,
- new BigDecimal("10.00"),
- (byte) 2,
- (short) 2,
- (long) 10000000,
- (float) 1.11,
- 1.21,
- LocalDate.of(2022, 1, 2),
- LocalDateTime.of(2022, 1, 1, 2, 0, 0),
- "c",
- "ccc" + ""));
- }
-
- @Test
- public void testStreamingRead() {
- assertThatThrownBy(
- () -> sEnv.from("T2").execute().print(),
- "Pre-aggregate continuous reading is not supported");
- }
- }
-
- /** ITCase for sum aggregate function. */
- public static class SumAggregation extends CatalogITCaseBase {
- @Override
- protected List<String> ddl() {
- return Collections.singletonList(
- "CREATE TABLE IF NOT EXISTS T1 ("
- + "j INT, k INT, "
- + "a INT, "
- + "b Decimal(4,2), "
- + "c TINYINT,"
- + "d SMALLINT,"
- + "e BIGINT,"
- + "f FLOAT,"
- + "h DOUBLE,"
- + "PRIMARY KEY (j,k) NOT ENFORCED)"
- + " WITH ('merge-engine'='aggregation', "
- + "'fields.a.aggregate-function'='sum', "
- + "'fields.b.aggregate-function'='sum', "
- + "'fields.c.aggregate-function'='sum', "
- + "'fields.d.aggregate-function'='sum', "
- + "'fields.e.aggregate-function'='sum', "
- + "'fields.f.aggregate-function'='sum',"
- + "'fields.h.aggregate-function'='sum'"
- + ");");
- }
-
- @Test
- public void testMergeInMemory() {
- batchSql(
- "INSERT INTO T1 VALUES "
- + "(1, 2, CAST(NULL AS INT), 1.01, CAST(1 AS TINYINT), CAST(-1 AS SMALLINT), "
- + "CAST(1000 AS BIGINT), 1.11, CAST(1.11 AS DOUBLE)),"
- + "(1, 2, 2, 1.10, CAST(2 AS TINYINT), CAST(2 AS SMALLINT), "
- + "CAST(100000 AS BIGINT), -1.11, CAST(1.11 AS DOUBLE)), "
- + "(1, 2, 3, 10.00, CAST(1 AS TINYINT), CAST(1 AS SMALLINT), "
- + "CAST(10000000 AS BIGINT), 0, CAST(-1.11 AS DOUBLE))");
- assertThat(batchSql("SELECT * FROM T1"))
- .containsExactlyInAnyOrder(
- Row.of(
- 1,
- 2,
- 5,
- new BigDecimal("12.11"),
- (byte) 4,
- (short) 2,
- (long) 10101000,
- (float) 0,
- 1.11));
-
- // projection
- assertThat(batchSql("SELECT f,e FROM T1"))
- .containsExactlyInAnyOrder(Row.of((float) 0, (long) 10101000));
- }
-
- @Test
- public void testMergeRead() {
- batchSql(
- "INSERT INTO T1 VALUES "
- + "(1, 2, 1, 1.01, CAST(1 AS TINYINT), CAST(-1 AS SMALLINT), CAST(1000 AS BIGINT), "
- + "1.11, CAST(1.11 AS DOUBLE))");
- batchSql(
- "INSERT INTO T1 VALUES "
- + "(1, 2, 2, 1.10, CAST(2 AS TINYINT), CAST(2 AS SMALLINT), CAST(100000 AS BIGINT), "
- + "CAST(NULL AS FLOAT), CAST(1.11 AS DOUBLE))");
- batchSql(
- "INSERT INTO T1 VALUES "
- + "(1, 2, 3, 10.00, CAST(1 AS TINYINT), CAST(1 AS SMALLINT), CAST(10000000 AS BIGINT), "
- + "-1.11, CAST(-1.11 AS DOUBLE))");
-
- List<Row> result = batchSql("SELECT * FROM T1");
- assertThat(result)
- .containsExactlyInAnyOrder(
- Row.of(
- 1,
- 2,
- 6,
- new BigDecimal("12.11"),
- (byte) 4,
- (short) 2,
- (long) 10101000,
- (float) 0,
- 1.11));
- }
-
- @Test
- public void testMergeCompaction() {
- // Wait compaction
- batchSql("ALTER TABLE T1 SET ('commit.force-compact'='true')");
-
- // key 1 2
- batchSql(
- "INSERT INTO T1 VALUES "
- + "(1, 2, 1, 1.01, CAST(1 AS TINYINT), CAST(-1 AS SMALLINT), CAST(1000 AS BIGINT), "
- + "1.11, CAST(1.11 AS DOUBLE))");
- batchSql(
- "INSERT INTO T1 VALUES "
- + "(1, 2, 2, 1.10, CAST(2 AS TINYINT), CAST(2 AS SMALLINT), CAST(100000 AS BIGINT), "
- + "CAST(NULL AS FLOAT), CAST(1.11 AS DOUBLE))");
- batchSql(
- "INSERT INTO T1 VALUES "
- + "(1, 2, 3, 10.00, CAST(1 AS TINYINT), CAST(1 AS SMALLINT), CAST(10000000 AS BIGINT), "
- + "-1.11, CAST(-1.11 AS DOUBLE))");
-
- // key 1 3
- batchSql(
- "INSERT INTO T1 VALUES "
- + "(1, 3, 2, 1.01, CAST(1 AS TINYINT), CAST(-1 AS SMALLINT), CAST(1000 AS BIGINT), "
- + "1.11, CAST(1.11 AS DOUBLE))");
- batchSql(
- "INSERT INTO T1 VALUES "
- + "(1, 3, 2, 1.10, CAST(2 AS TINYINT), CAST(2 AS SMALLINT), CAST(100000 AS BIGINT), "
- + "CAST(NULL AS FLOAT), CAST(1.11 AS DOUBLE))");
- batchSql(
- "INSERT INTO T1 VALUES "
- + "(1, 3, 3, 10.00, CAST(1 AS TINYINT), CAST(1 AS SMALLINT), CAST(10000000 AS BIGINT), "
- + "-1.11, CAST(-1.11 AS DOUBLE))");
-
- assertThat(batchSql("SELECT * FROM T1"))
- .containsExactlyInAnyOrder(
- Row.of(
- 1,
- 2,
- 6,
- new BigDecimal("12.11"),
- (byte) 4,
- (short) 2,
- (long) 10101000,
- (float) 0,
- 1.11),
- Row.of(
- 1,
- 3,
- 7,
- new BigDecimal("12.11"),
- (byte) 4,
- (short) 2,
- (long) 10101000,
- (float) 0,
- 1.11));
- }
-
- @Test
- public void testStreamingRead() {
- assertThatThrownBy(
- () -> sEnv.from("T1").execute().print(),
- "Pre-aggregate continuous reading is not supported");
- }
- }
-
- /** ITCase for product aggregate function. */
- public static class ProductAggregation extends CatalogITCaseBase {
- @Override
- protected List<String> ddl() {
- return Collections.singletonList(
- "CREATE TABLE IF NOT EXISTS T1 ("
- + "j INT, k INT, "
- + "a INT, "
- + "b Decimal(4,2), "
- + "c TINYINT,"
- + "d SMALLINT,"
- + "e BIGINT,"
- + "f FLOAT,"
- + "h DOUBLE,"
- + "PRIMARY KEY (j,k) NOT ENFORCED)"
- + " WITH ('merge-engine'='aggregation', "
- + "'fields.a.aggregate-function'='product', "
- + "'fields.b.aggregate-function'='product', "
- + "'fields.c.aggregate-function'='product', "
- + "'fields.d.aggregate-function'='product', "
- + "'fields.e.aggregate-function'='product', "
- + "'fields.f.aggregate-function'='product',"
- + "'fields.h.aggregate-function'='product'"
- + ");");
- }
-
- @Test
- public void testMergeInMemory() {
- batchSql("ALTER TABLE T1 MODIFY b DECIMAL(5, 3)");
-
- batchSql(
- "INSERT INTO T1 VALUES "
- + "(1, 2, CAST(NULL AS INT), 1.01, CAST(1 AS TINYINT), CAST(-1 AS SMALLINT), "
- + "CAST(1000 AS BIGINT), 1.11, CAST(1.11 AS DOUBLE)),"
- + "(1, 2, 2, 1.10, CAST(2 AS TINYINT), CAST(2 AS SMALLINT), "
- + "CAST(100000 AS BIGINT), -1.11, CAST(1.11 AS DOUBLE)), "
- + "(1, 2, 3, 10.00, CAST(1 AS TINYINT), CAST(1 AS SMALLINT), "
- + "CAST(10000000 AS BIGINT), 0, CAST(-1.11 AS DOUBLE))");
- List<Row> result = batchSql("SELECT * FROM T1");
- assertThat(result)
- .containsExactlyInAnyOrder(
- Row.of(
- 1,
- 2,
- 6,
- new BigDecimal("11.110"),
- (byte) 2,
- (short) -2,
- 1000000000000000L,
- (float) -0.0,
- -1.3676310000000003));
-
- // projection
- assertThat(batchSql("SELECT f,e FROM T1"))
- .containsExactlyInAnyOrder(Row.of((float) -0.0, 1000000000000000L));
- }
-
- @Test
- public void testMergeRead() {
- batchSql(
- "INSERT INTO T1 VALUES "
- + "(1, 2, 1, 1.01, CAST(1 AS TINYINT), CAST(-1 AS SMALLINT), CAST(1000 AS BIGINT), "
- + "1.11, CAST(1.11 AS DOUBLE))");
- batchSql(
- "INSERT INTO T1 VALUES "
- + "(1, 2, 2, 1.10, CAST(2 AS TINYINT), CAST(2 AS SMALLINT), CAST(100000 AS BIGINT), "
- + "CAST(NULL AS FLOAT), CAST(1.11 AS DOUBLE))");
- batchSql(
- "INSERT INTO T1 VALUES "
- + "(1, 2, 3, 10.00, CAST(1 AS TINYINT), CAST(1 AS SMALLINT), CAST(10000000 AS BIGINT), "
- + "-1.11, CAST(-1.11 AS DOUBLE))");
-
- List<Row> result = batchSql("SELECT * FROM T1");
- assertThat(result)
- .containsExactlyInAnyOrder(
- Row.of(
- 1,
- 2,
- 6,
- new BigDecimal("11.10"),
- (byte) 2,
- (short) -2,
- 1000000000000000L,
- (float) -1.2321,
- -1.3676310000000003));
- }
-
- @Test
- public void testMergeCompaction() {
- // Wait compaction
- batchSql("ALTER TABLE T1 SET ('commit.force-compact'='true')");
-
- // key 1 2
- batchSql(
- "INSERT INTO T1 VALUES "
- + "(1, 2, 1, 1.01, CAST(1 AS TINYINT), CAST(-1 AS SMALLINT), CAST(1000 AS BIGINT), "
- + "1.11, CAST(1.11 AS DOUBLE))");
- batchSql(
- "INSERT INTO T1 VALUES "
- + "(1, 2, 2, 1.10, CAST(2 AS TINYINT), CAST(2 AS SMALLINT), CAST(100000 AS BIGINT), "
- + "CAST(NULL AS FLOAT), CAST(1.11 AS DOUBLE))");
- batchSql(
- "INSERT INTO T1 VALUES "
- + "(1, 2, 3, 10.00, CAST(1 AS TINYINT), CAST(1 AS SMALLINT), CAST(10000000 AS BIGINT), "
- + "-1.11, CAST(-1.11 AS DOUBLE))");
-
- // key 1 3
- batchSql(
- "INSERT INTO T1 VALUES "
- + "(1, 3, 2, 1.01, CAST(1 AS TINYINT), CAST(-1 AS SMALLINT), CAST(1000 AS BIGINT), "
- + "1.11, CAST(1.11 AS DOUBLE))");
- batchSql(
- "INSERT INTO T1 VALUES "
- + "(1, 3, 2, 1.10, CAST(2 AS TINYINT), CAST(2 AS SMALLINT), CAST(100000 AS BIGINT), "
- + "CAST(NULL AS FLOAT), CAST(1.11 AS DOUBLE))");
- batchSql(
- "INSERT INTO T1 VALUES "
- + "(1, 3, 3, 10.00, CAST(1 AS TINYINT), CAST(1 AS SMALLINT), CAST(10000000 AS BIGINT), "
- + "-1.11, CAST(-1.11 AS DOUBLE))");
-
- List<Row> result = batchSql("SELECT * FROM T1");
- assertThat(result)
- .containsExactlyInAnyOrder(
- Row.of(
- 1,
- 2,
- 6,
- new BigDecimal("11.10"),
- (byte) 2,
- (short) -2,
- 1000000000000000L,
- (float) -1.2321,
- -1.3676310000000003),
- Row.of(
- 1,
- 3,
- 12,
- new BigDecimal("11.10"),
- (byte) 2,
- (short) -2,
- 1000000000000000L,
- (float) -1.2321,
- -1.3676310000000003));
- }
-
- @Test
- public void testStreamingRead() {
- assertThatThrownBy(
- () -> sEnv.from("T1").execute().print(),
- "Pre-aggregate continuous reading is not supported");
- }
- }
-
- /** ITCase for sum aggregate function retraction. */
- public static class SumRetractionAggregation extends CatalogITCaseBase {
- @Override
- protected List<String> ddl() {
- return Collections.singletonList(
- "CREATE TABLE T ("
- + "k INT,"
- + "b Decimal(12, 2),"
- + "PRIMARY KEY (k) NOT ENFORCED)"
- + " WITH ('merge-engine'='aggregation', "
- + "'changelog-producer' = 'full-compaction',"
- + "'fields.b.aggregate-function'='sum'"
- + ");");
- }
-
- @Test
- public void testRetraction() throws Exception {
- sql("CREATE TABLE INPUT (" + "k INT," + "b INT," + "PRIMARY KEY (k) NOT ENFORCED);");
- CloseableIterator<Row> insert =
- streamSqlIter("INSERT INTO T SELECT k, SUM(b) FROM INPUT GROUP BY k;");
- BlockingIterator<Row, Row> select = streamSqlBlockIter("SELECT * FROM T");
-
- sql("INSERT INTO INPUT VALUES (1, 1), (2, 2)");
- assertThat(select.collect(2))
- .containsExactlyInAnyOrder(
- Row.of(1, BigDecimal.valueOf(100, 2)),
- Row.of(2, BigDecimal.valueOf(200, 2)));
-
- sql("INSERT INTO INPUT VALUES (1, 3), (2, 4)");
- assertThat(select.collect(4))
- .containsExactlyInAnyOrder(
- Row.ofKind(RowKind.UPDATE_BEFORE, 1, BigDecimal.valueOf(100, 2)),
- Row.ofKind(RowKind.UPDATE_AFTER, 1, BigDecimal.valueOf(300, 2)),
- Row.ofKind(RowKind.UPDATE_BEFORE, 2, BigDecimal.valueOf(200, 2)),
- Row.ofKind(RowKind.UPDATE_AFTER, 2, BigDecimal.valueOf(400, 2)));
-
- select.close();
- insert.close();
- }
- }
-
- /** ITCase for first_value aggregate function. */
- public static class FirstValueAggregation extends CatalogITCaseBase {
- @Override
- protected List<String> ddl() {
- return Arrays.asList(
- "CREATE TABLE T ("
- + "k INT,"
- + "a INT,"
- + "b VARCHAR,"
- + "c VARCHAR,"
- + "d VARCHAR,"
- + "PRIMARY KEY (k) NOT ENFORCED)"
- + " WITH ('merge-engine'='aggregation', "
- + "'changelog-producer' = 'full-compaction',"
- + "'fields.b.aggregate-function'='first_value',"
- + "'fields.c.aggregate-function'='first_non_null_value',"
- + "'fields.d.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'"
- + ");");
- }
-
- @Test
- public void tesInMemoryMerge() {
- batchSql(
- "INSERT INTO T VALUES "
- + "(1, 0, CAST(NULL AS VARCHAR), CAST(NULL AS VARCHAR), CAST(NULL AS VARCHAR)),"
- + "(1, 1, '1', '1', '1'), "
- + "(2, 2, '2', '2', '2'),"
- + "(2, 3, '22', '22', '22')");
- List<Row> result = batchSql("SELECT * FROM T");
- assertThat(result)
- .containsExactlyInAnyOrder(
- Row.of(1, 1, null, "1", "1"), Row.of(2, 3, "2", "2", "2"));
- }
-
- @Test
- public void tesUnOrderInput() {
- batchSql(
- "INSERT INTO T VALUES "
- + "(1, 0, CAST(NULL AS VARCHAR), CAST(NULL AS VARCHAR), CAST(NULL AS VARCHAR)),"
- + "(1, 1, '1', '1', '1'), "
- + "(2, 3, '2', '2', '2'),"
- + "(2, 2, '22', '22', '22')");
- List<Row> result = batchSql("SELECT * FROM T");
- assertThat(result)
- .containsExactlyInAnyOrder(
- Row.of(1, 1, null, "1", "1"), Row.of(2, 3, "22", "22", "22"));
- batchSql("INSERT INTO T VALUES (2, 1, '1', '1', '1')");
- result = batchSql("SELECT * FROM T");
- assertThat(result)
- .containsExactlyInAnyOrder(
- Row.of(1, 1, null, "1", "1"), Row.of(2, 3, "1", "1", "1"));
- }
-
- @Test
- public void testMergeRead() {
- batchSql(
- "INSERT INTO T VALUES (1, 1, CAST(NULL AS VARCHAR), CAST(NULL AS VARCHAR), CAST(NULL AS VARCHAR))");
- batchSql("INSERT INTO T VALUES (1, 2, '1', '1', '1')");
- batchSql("INSERT INTO T VALUES (2, 1, '2', '2', '2')");
- batchSql("INSERT INTO T VALUES (2, 2, '22', '22', '22')");
- List<Row> result = batchSql("SELECT * FROM T");
- assertThat(result)
- .containsExactlyInAnyOrder(
- Row.of(1, 2, null, "1", "1"), Row.of(2, 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. */
- public static class BasicAggregateITCase extends CatalogITCaseBase {
-
- @Override
- protected List<String> ddl() {
- return Collections.singletonList(
- "CREATE TABLE T ("
- + "k INT,"
- + "v INT,"
- + "d INT,"
- + "PRIMARY KEY (k, d) NOT ENFORCED) PARTITIONED BY (d) "
- + " WITH ('merge-engine'='aggregation', "
- + "'fields.v.aggregate-function'='sum',"
- + "'local-merge-buffer-size'='5m'"
- + ");");
- }
-
- @Test
- public void testLocalMerge() {
- sql("INSERT INTO T VALUES(1, 1, 1), (2, 1, 1), (1, 2, 1)");
- assertThat(batchSql("SELECT * FROM T"))
- .containsExactlyInAnyOrder(Row.of(1, 3, 1), Row.of(2, 1, 1));
- }
-
- @Test
- public void testMergeRead() {
- sql("INSERT INTO T VALUES(1, 1, 1), (2, 1, 1)");
- sql("INSERT INTO T VALUES(1, 2, 1)");
- assertThat(batchSql("SELECT * FROM T"))
- .containsExactlyInAnyOrder(Row.of(1, 3, 1), Row.of(2, 1, 1));
- // filter
- assertThat(batchSql("SELECT * FROM T where v = 3"))
- .containsExactlyInAnyOrder(Row.of(1, 3, 1));
- assertThat(batchSql("SELECT * FROM T where v = 1"))
- .containsExactlyInAnyOrder(Row.of(2, 1, 1));
- }
-
- @Test
- public void testSequenceFieldWithDefaultAgg() {
- sql(
- "CREATE TABLE seq_default_agg ("
- + " pk INT PRIMARY KEY NOT ENFORCED,"
- + " seq INT,"
- + " v INT) WITH ("
- + " 'merge-engine'='aggregation',"
- + " 'sequence.field'='seq',"
- + " 'fields.default-aggregate-function'='sum'"
- + ")");
-
- sql("INSERT INTO seq_default_agg VALUES (0, 1, 1)");
- sql("INSERT INTO seq_default_agg VALUES (0, 2, 2)");
-
- assertThat(sql("SELECT * FROM seq_default_agg")).containsExactly(Row.of(0, 2, 3));
- }
- }
-
- /** ITCase for {@link FieldNestedUpdateAgg}. */
- public static class NestedUpdateAggregationITCase extends CatalogITCaseBase {
-
- @Override
- protected List<String> ddl() {
- String ordersTable =
- "CREATE TABLE orders (\n"
- + " order_id INT PRIMARY KEY NOT ENFORCED,\n"
- + " user_name STRING,\n"
- + " address STRING\n"
- + ");";
-
- String subordersTable =
- "CREATE TABLE sub_orders (\n"
- + " order_id INT,\n"
- + " daily_id INT,\n"
- + " today STRING,\n"
- + " product_name STRING,\n"
- + " price BIGINT,\n"
- + " PRIMARY KEY (order_id, daily_id, today) NOT ENFORCED\n"
- + ");";
-
- String wideTable =
- "CREATE TABLE order_wide (\n"
- + " order_id INT PRIMARY KEY NOT ENFORCED,\n"
- + " user_name STRING,\n"
- + " address STRING,\n"
- + " sub_orders ARRAY<ROW<daily_id INT, today STRING, product_name STRING, price BIGINT>>\n"
- + ") WITH (\n"
- + " 'merge-engine' = 'aggregation',\n"
- + " 'fields.sub_orders.aggregate-function' = 'nested_update',\n"
- + " 'fields.sub_orders.nested-key' = 'daily_id,today',\n"
- + " 'fields.sub_orders.ignore-retract' = 'true',"
- + " 'fields.user_name.ignore-retract' = 'true',"
- + " 'fields.address.ignore-retract' = 'true'"
- + ")";
-
- String wideAppendTable =
- "CREATE TABLE order_append_wide (\n"
- + " order_id INT PRIMARY KEY NOT ENFORCED,\n"
- + " user_name STRING,\n"
- + " address STRING,\n"
- + " sub_orders ARRAY<ROW<daily_id INT, today STRING, product_name STRING, price BIGINT>>\n"
- + ") WITH (\n"
- + " 'merge-engine' = 'aggregation',\n"
- + " 'fields.sub_orders.aggregate-function' = 'nested_update',\n"
- + " 'fields.sub_orders.ignore-retract' = 'true',"
- + " 'fields.user_name.ignore-retract' = 'true',"
- + " 'fields.address.ignore-retract' = 'true'"
- + ")";
-
- return Arrays.asList(ordersTable, subordersTable, wideTable, wideAppendTable);
- }
-
- @Test
- public void testUseCase() {
- sql(
- "INSERT INTO orders VALUES "
- + "(1, 'Wang', 'HangZhou'),"
- + "(2, 'Zhao', 'ChengDu'),"
- + "(3, 'Liu', 'NanJing')");
-
- sql(
- "INSERT INTO sub_orders VALUES "
- + "(1, 1, '12-20', 'Apple', 8000),"
- + "(1, 2, '12-20', 'Tesla', 400000),"
- + "(1, 1, '12-21', 'Sangsung', 5000),"
- + "(2, 1, '12-20', 'Tea', 40),"
- + "(2, 2, '12-20', 'Pot', 60),"
- + "(3, 1, '12-25', 'Bat', 15),"
- + "(3, 1, '12-26', 'Cup', 30)");
-
- sql(widenSql());
-
- List<Row> result =
- sql("SELECT * FROM order_wide").stream()
- .sorted(Comparator.comparingInt(r -> r.getFieldAs(0)))
- .collect(Collectors.toList());
-
- assertThat(
- checkOneRecord(
- result.get(0),
- 1,
- "Wang",
- "HangZhou",
- Row.of(1, "12-20", "Apple", 8000L),
- Row.of(1, "12-21", "Sangsung", 5000L),
- Row.of(2, "12-20", "Tesla", 400000L)))
- .isTrue();
- assertThat(
- checkOneRecord(
- result.get(1),
- 2,
- "Zhao",
- "ChengDu",
- Row.of(1, "12-20", "Tea", 40L),
- Row.of(2, "12-20", "Pot", 60L)))
- .isTrue();
- assertThat(
- checkOneRecord(
- result.get(2),
- 3,
- "Liu",
- "NanJing",
- Row.of(1, "12-25", "Bat", 15L),
- Row.of(1, "12-26", "Cup", 30L)))
- .isTrue();
-
- // query using UNNEST
- List<Row> unnested =
- sql(
- "SELECT order_id, user_name, address, daily_id, today, product_name, price "
- + "FROM order_wide, UNNEST(sub_orders) AS so(daily_id, today, product_name, price)");
-
- assertThat(unnested)
- .containsExactlyInAnyOrder(
- Row.of(1, "Wang", "HangZhou", 1, "12-20", "Apple", 8000L),
- Row.of(1, "Wang", "HangZhou", 2, "12-20", "Tesla", 400000L),
- Row.of(1, "Wang", "HangZhou", 1, "12-21", "Sangsung", 5000L),
- Row.of(2, "Zhao", "ChengDu", 1, "12-20", "Tea", 40L),
- Row.of(2, "Zhao", "ChengDu", 2, "12-20", "Pot", 60L),
- Row.of(3, "Liu", "NanJing", 1, "12-25", "Bat", 15L),
- Row.of(3, "Liu", "NanJing", 1, "12-26", "Cup", 30L));
- }
-
- @Test
- public void testUseCaseWithNullValue() {
- sql(
- "INSERT INTO order_wide\n"
- + "SELECT 6, CAST (NULL AS STRING), CAST (NULL AS STRING), "
- + "ARRAY[cast(null as ROW<daily_id INT, today STRING, product_name STRING, price BIGINT>)]");
-
- List<Row> result =
- sql("SELECT * FROM order_wide").stream()
- .sorted(Comparator.comparingInt(r -> r.getFieldAs(0)))
- .collect(Collectors.toList());
-
- assertThat(checkOneRecord(result.get(0), 6, null, null, (Row) null)).isTrue();
-
- sql(
- "INSERT INTO order_wide\n"
- + "SELECT 6, 'Sun', CAST (NULL AS STRING), "
- + "ARRAY[ROW(1, '01-01','Apple', 6999)]");
-
- result =
- sql("SELECT * FROM order_wide").stream()
- .sorted(Comparator.comparingInt(r -> r.getFieldAs(0)))
- .collect(Collectors.toList());
- assertThat(
- checkOneRecord(
- result.get(0),
- 6,
- "Sun",
- null,
- Row.of(1, "01-01", "Apple", 6999L)))
- .isTrue();
- }
-
- @Test
- public void testUseCaseAppend() {
- sql(
- "INSERT INTO orders VALUES "
- + "(1, 'Wang', 'HangZhou'),"
- + "(2, 'Zhao', 'ChengDu'),"
- + "(3, 'Liu', 'NanJing')");
-
- sql(
- "INSERT INTO sub_orders VALUES "
- + "(1, 1, '12-20', 'Apple', 8000),"
- + "(2, 1, '12-20', 'Tesla', 400000),"
- + "(3, 1, '12-25', 'Bat', 15),"
- + "(3, 1, '12-26', 'Cup', 30)");
-
- sql(widenAppendSql());
-
- // query using UNNEST
- List<Row> unnested =
- sql(
- "SELECT order_id, user_name, address, daily_id, today, product_name, price "
- + "FROM order_append_wide, UNNEST(sub_orders) AS so(daily_id, today, product_name, price)");
-
- assertThat(unnested)
- .containsExactlyInAnyOrder(
- Row.of(1, "Wang", "HangZhou", 1, "12-20", "Apple", 8000L),
- Row.of(2, "Zhao", "ChengDu", 1, "12-20", "Tesla", 400000L),
- Row.of(3, "Liu", "NanJing", 1, "12-25", "Bat", 15L),
- Row.of(3, "Liu", "NanJing", 1, "12-26", "Cup", 30L));
- }
-
- @Test
- @Timeout(60)
- public void testUpdateWithIgnoreRetract() throws Exception {
- sEnv.getConfig()
- .set(
- ExecutionConfigOptions.TABLE_EXEC_SINK_UPSERT_MATERIALIZE,
- ExecutionConfigOptions.UpsertMaterialize.NONE);
-
- sql("INSERT INTO orders VALUES (1, 'Wang', 'HangZhou')");
-
- sql(
- "INSERT INTO sub_orders VALUES "
- + "(1, 1, '12-20', 'Apple', 8000),"
- + "(1, 2, '12-20', 'Tesla', 400000),"
- + "(1, 1, '12-21', 'Sangsung', 5000)");
-
- sEnv.executeSql(widenSql());
-
- boolean checkResult;
- List<Row> result;
-
- do {
- Thread.sleep(500);
- result = sql("SELECT * FROM order_wide");
- checkResult =
- !result.isEmpty()
- && checkOneRecord(
- result.get(0),
- 1,
- "Wang",
- "HangZhou",
- Row.of(1, "12-20", "Apple", 8000L),
- Row.of(1, "12-21", "Sangsung", 5000L),
- Row.of(2, "12-20", "Tesla", 400000L));
- } while (!checkResult);
-
- sql("INSERT INTO sub_orders VALUES (1, 2, '12-20', 'Benz', 380000)");
-
- do {
- Thread.sleep(500);
- result = sql("SELECT * FROM order_wide");
- checkResult =
- !result.isEmpty()
- && checkOneRecord(
- result.get(0),
- 1,
- "Wang",
- "HangZhou",
- Row.of(1, "12-20", "Apple", 8000L),
- Row.of(1, "12-21", "Sangsung", 5000L),
- Row.of(2, "12-20", "Benz", 380000L));
- } while (!checkResult);
- }
-
- private String widenSql() {
- return "INSERT INTO order_wide\n"
- + "SELECT order_id, user_name, address, "
- + "CAST (NULL AS ARRAY<ROW<daily_id INT, today STRING, product_name STRING, price BIGINT>>) FROM orders\n"
- + "UNION ALL\n"
- + "SELECT order_id, CAST (NULL AS STRING), CAST (NULL AS STRING), "
- + "ARRAY[ROW(daily_id, today, product_name, price)] FROM sub_orders";
- }
-
- private String widenAppendSql() {
- return "INSERT INTO order_append_wide\n"
- + "SELECT order_id, user_name, address, "
- + "CAST (NULL AS ARRAY<ROW<daily_id INT, today STRING, product_name STRING, price BIGINT>>) FROM orders\n"
- + "UNION ALL\n"
- + "SELECT order_id, CAST (NULL AS STRING), CAST (NULL AS STRING), "
- + "ARRAY[ROW(daily_id, today, product_name, price)] FROM sub_orders";
- }
-
- private boolean checkOneRecord(
- Row record, int orderId, String userName, String address, Row... subOrders) {
- if ((int) record.getField(0) != orderId) {
- return false;
- }
- if (!Objects.equals(record.getFieldAs(1), userName)) {
- return false;
- }
- if (!Objects.equals(record.getFieldAs(2), address)) {
- return false;
- }
-
- return checkNestedTable(record.getFieldAs(3), subOrders);
- }
-
- private boolean checkNestedTable(Row[] nestedTable, Row... subOrders) {
- if (nestedTable.length != subOrders.length) {
- return false;
- }
-
- Comparator<Row> comparator =
- (Comparator)
- Comparator.comparingInt(r -> ((Row) r).getFieldAs(0))
- .thenComparing(r -> (String) ((Row) r).getField(1));
-
- List<Row> sortedActual =
- Arrays.stream(nestedTable).sorted(comparator).collect(Collectors.toList());
- List<Row> sortedExpected =
- Arrays.stream(subOrders).sorted(comparator).collect(Collectors.toList());
-
- for (int i = 0; i < sortedActual.size(); i++) {
- if (!Objects.equals(sortedActual.get(i), sortedExpected.get(i))) {
- return false;
- }
- }
-
- return true;
- }
- }
-
- /** 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(
- "CREATE TABLE test_collect("
- + " id INT PRIMARY KEY NOT ENFORCED,"
- + " f0 ARRAY<STRING>"
- + ") WITH ("
- + " 'merge-engine' = 'aggregation',"
- + " 'fields.f0.aggregate-function' = 'collect',"
- + " 'fields.f0.distinct' = 'true'"
- + ")");
-
- sql(
- "INSERT INTO test_collect VALUES "
- + "(1, CAST (NULL AS ARRAY<STRING>)), "
- + "(2, ARRAY['A', 'B']), "
- + "(3, ARRAY['car', 'watch'])");
-
- List<Row> result = queryAndSort("SELECT * FROM test_collect");
- checkOneRecord(result.get(0), 1);
- checkOneRecord(result.get(1), 2, "A", "B");
- checkOneRecord(result.get(2), 3, "car", "watch");
-
- sql(
- "INSERT INTO test_collect VALUES "
- + "(1, ARRAY['paimon', 'paimon']), "
- + "(2, ARRAY['A', 'B', 'C']), "
- + "(3, CAST (NULL AS ARRAY<STRING>))");
-
- result = queryAndSort("SELECT * FROM test_collect");
- checkOneRecord(result.get(0), 1, "paimon");
- checkOneRecord(result.get(1), 2, "A", "B", "C");
- checkOneRecord(result.get(2), 3, "car", "watch");
- }
-
- @Test
- public void testAggWithoutDistinct() {
- sql(
- "CREATE TABLE test_collect("
- + " id INT PRIMARY KEY NOT ENFORCED,"
- + " f0 ARRAY<STRING>"
- + ") WITH ("
- + " 'merge-engine' = 'aggregation',"
- + " 'fields.f0.aggregate-function' = 'collect'"
- + ")");
-
- sql(
- "INSERT INTO test_collect VALUES "
- + "(1, CAST (NULL AS ARRAY<STRING>)), "
- + "(2, ARRAY['A', 'B', 'B']), "
- + "(3, ARRAY['car', 'watch'])");
-
- List<Row> result = queryAndSort("SELECT * FROM test_collect");
- checkOneRecord(result.get(0), 1);
- checkOneRecord(result.get(1), 2, "A", "B", "B");
- checkOneRecord(result.get(2), 3, "car", "watch");
-
- sql(
- "INSERT INTO test_collect VALUES "
- + "(1, ARRAY['paimon', 'paimon']), "
- + "(2, ARRAY['A', 'B', 'C']), "
- + "(3, CAST (NULL AS ARRAY<STRING>))");
-
- result = queryAndSort("SELECT * FROM test_collect");
- checkOneRecord(result.get(0), 1, "paimon", "paimon");
- checkOneRecord(result.get(1), 2, "A", "A", "B", "B", "B", "C");
- checkOneRecord(result.get(2), 3, "car", "watch");
- }
-
- private static List<Arguments> retractArguments() {
- return Arrays.asList(
- Arguments.arguments("lookup", "aggregation"),
- Arguments.arguments("lookup", "partial-update"),
- Arguments.arguments("full-compaction", "aggregation"),
- Arguments.arguments("full-compaction", "partial-update"));
- }
-
- @ParameterizedTest(name = "changelog-producer = {0}, merge-engine = {1}")
- @MethodSource("retractArguments")
- public void testRetract(String changelogProducer, String mergeEngine) throws Exception {
- String sequenceGroup = "";
- if (mergeEngine.equals("partial-update")) {
- sequenceGroup = ", 'fields.f1.sequence-group' = 'f0'";
- }
- sql(
- "CREATE TABLE test_collect("
- + " id INT PRIMARY KEY NOT ENFORCED,"
- + " f0 ARRAY<STRING>,"
- + " f1 INT"
- + ") WITH ("
- + " 'changelog-producer' = '%s',"
- + " 'merge-engine' = '%s',"
- + " 'fields.f0.aggregate-function' = 'collect'"
- + " %s"
- + ")",
- changelogProducer, mergeEngine, sequenceGroup);
-
- BlockingIterator<Row, Row> select = streamSqlBlockIter("SELECT * FROM test_collect");
-
- String temporaryTableTemplate =
- "CREATE TEMPORARY TABLE %s ("
- + " id INT PRIMARY KEY NOT ENFORCED,"
- + " f0 ARRAY<STRING>,"
- + " f1 INT"
- + ") WITH ("
- + " 'connector' = 'values',"
- + " 'data-id' = '%s',"
- + " 'bounded' = 'true',"
- + " 'changelog-mode' = '%s'"
- + ")";
-
- // 1 only exists single key record
- sql("INSERT INTO test_collect VALUES (1, ARRAY['A', 'B'], 1)");
- List<Row> result = select.collect(1);
- checkOneRecord(result.get(0), 1, "A", "B");
-
- // 1.1 UB + UA (CANNOT handle)
- List<Row> inputRecords =
- Arrays.asList(
- Row.ofKind(RowKind.UPDATE_BEFORE, 1, new String[] {"A", "B"}, 2),
- Row.ofKind(RowKind.UPDATE_AFTER, 1, new String[] {"C", "D"}, 3));
- sEnv.executeSql(
- String.format(
- temporaryTableTemplate,
- "INPUT11",
- TestValuesTableFactory.registerData(inputRecords),
- "UB,UA"))
- .await();
- sEnv.executeSql("INSERT INTO test_collect SELECT * FROM INPUT11").await();
-
- result = select.collect(2);
- assertThat(result.get(0).getKind()).isEqualTo(RowKind.UPDATE_BEFORE);
- checkOneRecord(result.get(0), 1, "A", "B");
- assertThat(result.get(1).getKind()).isEqualTo(RowKind.UPDATE_AFTER);
- checkOneRecord(result.get(1), 1, "A", "B", "C", "D");
-
- // 1.2 -D
- inputRecords =
- Collections.singletonList(
- Row.ofKind(RowKind.DELETE, 1, new String[] {"C", "D"}, 4));
- sEnv.executeSql(
- String.format(
- temporaryTableTemplate,
- "INPUT12",
- TestValuesTableFactory.registerData(inputRecords),
- "D"))
- .await();
- sEnv.executeSql("INSERT INTO test_collect SELECT * FROM INPUT12").await();
-
- result = select.collect(2);
- assertThat(result.get(0).getKind()).isEqualTo(RowKind.UPDATE_BEFORE);
- checkOneRecord(result.get(0), 1, "A", "B", "C", "D");
- assertThat(result.get(1).getKind()).isEqualTo(RowKind.UPDATE_AFTER);
- checkOneRecord(result.get(1), 1, "A", "B");
-
- // 2 exists multiple key records
- sql("INSERT INTO test_collect VALUES (2, ARRAY['A', 'B'], 5), (3, ARRAY['A', 'B'], 6)");
- result = select.collect(2);
- checkOneRecord(result.get(0), 2, "A", "B");
- checkOneRecord(result.get(1), 3, "A", "B");
-
- // 2.1 UB + UA (CANNOT handle)
- inputRecords =
- Arrays.asList(
- Row.ofKind(RowKind.UPDATE_BEFORE, 2, new String[] {"A", "B"}, 7),
- Row.ofKind(RowKind.UPDATE_AFTER, 2, new String[] {"C", "D"}, 8));
- sEnv.executeSql(
- String.format(
- temporaryTableTemplate,
- "INPUT21",
- TestValuesTableFactory.registerData(inputRecords),
- "UB,UA"))
- .await();
- sEnv.executeSql("INSERT INTO test_collect SELECT * FROM INPUT21").await();
-
- result = select.collect(2);
- assertThat(result.get(0).getKind()).isEqualTo(RowKind.UPDATE_BEFORE);
- checkOneRecord(result.get(0), 2, "A", "B");
- assertThat(result.get(1).getKind()).isEqualTo(RowKind.UPDATE_AFTER);
- checkOneRecord(result.get(1), 2, "A", "B", "C", "D");
-
- // 2.2 -D
- inputRecords =
- Collections.singletonList(Row.ofKind(RowKind.DELETE, 3, new String[] {"A"}, 9));
- sEnv.executeSql(
- String.format(
- temporaryTableTemplate,
- "INPUT22",
- TestValuesTableFactory.registerData(inputRecords),
- "D"))
- .await();
- sEnv.executeSql("INSERT INTO test_collect SELECT * FROM INPUT22").await();
-
- result = select.collect(2);
- assertThat(result.get(0).getKind()).isEqualTo(RowKind.UPDATE_BEFORE);
- checkOneRecord(result.get(0), 3, "A", "B");
- assertThat(result.get(1).getKind()).isEqualTo(RowKind.UPDATE_AFTER);
- checkOneRecord(result.get(1), 3, "B");
-
- select.close();
- }
-
- @Test
- public void testRetractInputNull() throws Exception {
- sql(
- "CREATE TABLE test_collect ("
- + " id INT PRIMARY KEY NOT ENFORCED,"
- + " f0 ARRAY<STRING>,"
- + " f1 INT"
- + ") WITH ("
- + " 'changelog-producer' = 'lookup',"
- + " 'merge-engine' = 'partial-update',"
- + " 'fields.f0.aggregate-function' = 'collect',"
- + " 'fields.f1.sequence-group' = 'f0'"
- + ")");
-
- List<Row> input =
- Arrays.asList(
- Row.ofKind(RowKind.INSERT, 1, null, 1),
- Row.ofKind(RowKind.INSERT, 1, new String[] {"A"}, 2),
- Row.ofKind(RowKind.UPDATE_BEFORE, 1, null, 1),
- Row.ofKind(RowKind.UPDATE_AFTER, 1, new String[] {"B"}, 3));
- sEnv.executeSql(
- String.format(
- "CREATE TEMPORARY TABLE input ("
- + " id INT PRIMARY KEY NOT ENFORCED,"
- + " f0 ARRAY<STRING>,"
- + " f1 INT"
- + ") WITH ("
- + " 'connector' = 'values',"
- + " 'data-id' = '%s',"
- + " 'bounded' = 'true',"
- + " 'changelog-mode' = 'UB,UA'"
- + ")",
- TestValuesTableFactory.registerData(input)))
- .await();
- sEnv.executeSql("INSERT INTO test_collect SELECT * FROM input").await();
-
- assertThat(sql("SELECT * FROM test_collect"))
- .containsExactly(Row.of(1, new String[] {"A", "B"}, 3));
- }
-
- private void checkOneRecord(Row row, int id, String... elements) {
- assertThat(row.getField(0)).isEqualTo(id);
- if (elements == null || elements.length == 0) {
- assertThat(row.getField(1)).isNull();
- } else {
- assertThat((String[]) row.getField(1)).containsExactlyInAnyOrder(elements);
- }
- }
- }
-
- /** ITCase for {@link FieldMergeMapAgg}. */
- public static class MergeMapAggregationITCase extends CatalogITCaseBase {
-
- @Override
- protected List<String> ddl() {
- return Collections.singletonList(
- "CREATE TABLE test_merge_map("
- + " id INT PRIMARY KEY NOT ENFORCED,"
- + " f0 MAP<INT, STRING>"
- + ") WITH ("
- + " 'merge-engine' = 'aggregation',"
- + " 'fields.f0.aggregate-function' = 'merge_map'"
- + ")");
- }
-
- @Test
- public void testMergeMap() {
- sql(
- "INSERT INTO test_merge_map VALUES "
- + "(1, CAST (NULL AS MAP<INT, STRING>)), "
- + "(2, MAP[1, 'A']), "
- + "(3, MAP[1, 'A', 2, 'B'])");
-
- List<Row> result = queryAndSort("SELECT * FROM test_merge_map");
- checkOneRecord(result.get(0), 1, null);
- checkOneRecord(result.get(1), 2, toMap(1, "A"));
- checkOneRecord(result.get(2), 3, toMap(1, "A", 2, "B"));
-
- sql(
- "INSERT INTO test_merge_map VALUES "
- + "(1, MAP[1, 'A']), "
- + "(2, MAP[1, 'B']), "
- + "(3, MAP[1, 'a', 2, 'b', 3, 'c'])");
-
- result = queryAndSort("SELECT * FROM test_merge_map");
- checkOneRecord(result.get(0), 1, toMap(1, "A"));
- checkOneRecord(result.get(1), 2, toMap(1, "B"));
- checkOneRecord(result.get(2), 3, toMap(1, "a", 2, "b", 3, "c"));
- }
-
- @Test
- public void testRetractInputNull() throws Exception {
- sql(
- "CREATE TABLE test_merge_map1 ("
- + " id INT PRIMARY KEY NOT ENFORCED,"
- + " f0 MAP<INT, STRING>,"
- + " f1 INT"
- + ") WITH ("
- + " 'changelog-producer' = 'lookup',"
- + " 'merge-engine' = 'partial-update',"
- + " 'fields.f0.aggregate-function' = 'merge_map',"
- + " 'fields.f1.sequence-group' = 'f0'"
- + ")");
-
- List<Row> input =
- Arrays.asList(
- Row.ofKind(RowKind.INSERT, 1, null, 1),
- Row.ofKind(RowKind.INSERT, 1, Collections.singletonMap(1, "A"), 2),
- Row.ofKind(RowKind.UPDATE_BEFORE, 1, null, 1),
- Row.ofKind(
- RowKind.UPDATE_AFTER, 1, Collections.singletonMap(2, "B"), 3));
- sEnv.executeSql(
- String.format(
- "CREATE TEMPORARY TABLE input ("
- + " id INT PRIMARY KEY NOT ENFORCED,"
- + " f0 MAP<INT, STRING>,"
- + " f1 INT"
- + ") WITH ("
- + " 'connector' = 'values',"
- + " 'data-id' = '%s',"
- + " 'bounded' = 'true',"
- + " 'changelog-mode' = 'UB,UA'"
- + ")",
- TestValuesTableFactory.registerData(input)))
- .await();
- sEnv.executeSql("INSERT INTO test_merge_map1 SELECT * FROM input").await();
-
- assertThat(sql("SELECT * FROM test_merge_map1"))
- .containsExactly(Row.of(1, toMap(1, "A", 2, "B"), 3));
- }
-
- private Map<Object, Object> toMap(Object... kvs) {
- Map<Object, Object> result = new HashMap<>();
- for (int i = 0; i < kvs.length; i += 2) {
- result.put(kvs[i], kvs[i + 1]);
- }
- return result;
- }
-
- private void checkOneRecord(Row row, int id, Map<Object, Object> map) {
- assertThat(row.getField(0)).isEqualTo(id);
- if (map == null || map.isEmpty()) {
- assertThat(row.getField(1)).isNull();
- } else {
- assertThat((Map<Object, Object>) row.getField(1))
- .containsExactlyInAnyOrderEntriesOf(map);
- }
- }
- }
-
- /** ITCase for last non-null value aggregate function. */
- public static class FieldsDefaultAggregationITCase extends CatalogITCaseBase {
- @Override
- protected int defaultParallelism() {
- // set parallelism to 1 so that the order of input data is determined
- return 1;
- }
-
- @Override
- protected List<String> ddl() {
- return Collections.singletonList(
- "CREATE TABLE IF NOT EXISTS test_default_agg_func ("
- + "j INT, k INT, "
- + "a INT, "
- + "b INT, "
- + "i DATE,"
- + "PRIMARY KEY (j,k) NOT ENFORCED)"
- + " WITH ('merge-engine'='aggregation', "
- + "'fields.default-aggregate-function'='first_non_null_value', "
- + "'fields.i.aggregate-function'='last_non_null_value'"
- + ");");
- }
-
- @Test
- public void testMergeInMemory() {
- // VALUES does not guarantee order, but order is important for last value aggregations.
- // So we need to sort the input data.
- batchSql(
- "CREATE TABLE myTable AS "
- + "SELECT b, c, d, e, f FROM "
- + "(VALUES "
- + " (1, 1, 2, CAST(NULL AS INT), 4, CAST('2020-01-01' AS DATE)),"
- + " (2, 1, 2, 2, CAST(NULL as INT), CAST('2020-01-02' AS DATE)),"
- + " (3, 1, 2, 3, 5, CAST(NULL AS DATE))"
- + ") AS V(a, b, c, d, e, f) "
- + "ORDER BY a");
- batchSql("INSERT INTO test_default_agg_func SELECT * FROM myTable");
- List<Row> result = batchSql("SELECT * FROM test_default_agg_func");
- assertThat(result)
- .containsExactlyInAnyOrder(Row.of(1, 2, 2, 4, LocalDate.of(2020, 1, 2)));
- }
-
- @Test
- public void testMergeRead() {
- batchSql(
- "INSERT INTO test_default_agg_func VALUES (1, 2, CAST(NULL AS INT), 3, CAST('2020-01-01' AS DATE))");
- batchSql(
- "INSERT INTO test_default_agg_func VALUES (1, 2, 2, CAST(NULL AS INT), CAST('2020-01-02' AS DATE))");
- batchSql("INSERT INTO test_default_agg_func VALUES (1, 2, 3, 5, CAST(NULL AS DATE))");
-
- List<Row> result = batchSql("SELECT * FROM test_default_agg_func");
- assertThat(result)
- .containsExactlyInAnyOrder(Row.of(1, 2, 2, 3, LocalDate.of(2020, 1, 2)));
- }
-
- @Test
- public void testMergeCompaction() {
- // Wait compaction
- batchSql("ALTER TABLE test_default_agg_func SET ('commit.force-compact'='true')");
-
- // key 1 2
- batchSql(
- "INSERT INTO test_default_agg_func VALUES (1, 2, CAST(NULL AS INT), 3, CAST('2020-01-01' AS DATE))");
- batchSql(
- "INSERT INTO test_default_agg_func VALUES (1, 2, 2, CAST(NULL AS INT), CAST('2020-01-02' AS DATE))");
- batchSql("INSERT INTO test_default_agg_func VALUES (1, 2, 3, 5, CAST(NULL AS DATE))");
-
- // key 1 3
- batchSql(
- "INSERT INTO test_default_agg_func VALUES (1, 3, 3, 4, CAST('2020-01-01' AS DATE))");
- batchSql("INSERT INTO test_default_agg_func VALUES (1, 3, 2, 6, CAST(NULL AS DATE))");
- batchSql(
- "INSERT INTO test_default_agg_func VALUES (1, 3, CAST(NULL AS INT), CAST(NULL AS INT), CAST('2022-01-02' AS DATE))");
-
- assertThat(batchSql("SELECT * FROM test_default_agg_func"))
- .containsExactlyInAnyOrder(
- Row.of(1, 2, 2, 3, LocalDate.of(2020, 1, 2)),
- Row.of(1, 3, 3, 4, LocalDate.of(2022, 1, 2)));
- }
-
- @Test
- public void testStreamingRead() {
- assertThatThrownBy(
- () -> sEnv.from("test_default_agg_func").execute().print(),
- "Pre-aggregate continuous reading is not supported");
- }
- }
-
- /** ITCase for {@link org.apache.paimon.mergetree.compact.aggregate.FieldThetaSketchAgg}. */
- public static class ThetaSketchAggAggregationITCase extends CatalogITCaseBase {
-
- @Test
- public void testThetaSketchAgg() {
- sql(
- "CREATE TABLE test_collect("
- + " id INT PRIMARY KEY NOT ENFORCED,"
- + " f0 VARBINARY"
- + ") WITH ("
- + " 'merge-engine' = 'aggregation',"
- + " 'fields.f0.aggregate-function' = 'theta_sketch'"
- + ")");
-
- String str1 = Hex.encodeHexString(sketchOf(1)).toUpperCase();
- String str2 = Hex.encodeHexString(sketchOf(2)).toUpperCase();
- String str3 = Hex.encodeHexString(sketchOf(3)).toUpperCase();
-
- sql(
- String.format(
- "INSERT INTO test_collect VALUES (1, CAST (NULL AS VARBINARY)),(2, CAST(x'%s' AS VARBINARY)), (3, CAST(x'%s' AS VARBINARY))",
- str1, str2));
-
- List<Row> result = queryAndSort("SELECT * FROM test_collect");
- checkOneRecord(result.get(0), 1, null);
- checkOneRecord(result.get(1), 2, sketchOf(1));
- checkOneRecord(result.get(2), 3, sketchOf(2));
-
- sql(
- String.format(
- "INSERT INTO test_collect VALUES (1, CAST (x'%s' AS VARBINARY)),(2, CAST(x'%s' AS VARBINARY)), (2, CAST(x'%s' AS VARBINARY)), (3, CAST(x'%s' AS VARBINARY))",
- str1, str2, str2, str3));
-
- result = queryAndSort("SELECT * FROM test_collect");
- checkOneRecord(result.get(0), 1, sketchOf(1));
- checkOneRecord(result.get(1), 2, sketchOf(1, 2));
- checkOneRecord(result.get(2), 3, sketchOf(2, 3));
- }
-
- private void checkOneRecord(Row row, int id, byte[] expected) {
- assertThat(row.getField(0)).isEqualTo(id);
- assertThat(row.getField(1)).isEqualTo(expected);
- }
- }
-
- /**
- * ITCase for {@link org.apache.paimon.mergetree.compact.aggregate.FieldRoaringBitmap32Agg} &
- * {@link org.apache.paimon.mergetree.compact.aggregate.FieldRoaringBitmap64Agg}.
- */
- public static class RoaringBitmapAggAggregationITCase extends CatalogITCaseBase {
-
- @Test
- public void testRoaring32BitmapAgg() throws IOException {
- sql(
- "CREATE TABLE test_rbm64("
- + " id INT PRIMARY KEY NOT ENFORCED,"
- + " f0 VARBINARY"
- + ") WITH ("
- + " 'merge-engine' = 'aggregation',"
- + " 'fields.f0.aggregate-function' = 'rbm32'"
- + ")");
-
- byte[] v1Bytes = RoaringBitmap32.bitmapOf(1).serialize();
- byte[] v2Bytes = RoaringBitmap32.bitmapOf(2).serialize();
- byte[] v3Bytes = RoaringBitmap32.bitmapOf(3).serialize();
- byte[] v4Bytes = RoaringBitmap32.bitmapOf(1, 2).serialize();
- byte[] v5Bytes = RoaringBitmap32.bitmapOf(2, 3).serialize();
- String v1 = Hex.encodeHexString(v1Bytes).toUpperCase();
- String v2 = Hex.encodeHexString(v2Bytes).toUpperCase();
- String v3 = Hex.encodeHexString(v3Bytes).toUpperCase();
-
- sql(
- "INSERT INTO test_rbm64 VALUES "
- + "(1, CAST (NULL AS VARBINARY)), "
- + "(2, CAST (x'"
- + v1
- + "' AS VARBINARY)), "
- + "(3, CAST (x'"
- + v2
- + "' AS VARBINARY))");
-
- List<Row> result = queryAndSort("SELECT * FROM test_rbm64");
- checkOneRecord(result.get(0), 1, null);
- checkOneRecord(result.get(1), 2, v1Bytes);
- checkOneRecord(result.get(2), 3, v2Bytes);
-
- sql(
- "INSERT INTO test_rbm64 VALUES "
- + "(1, CAST (x'"
- + v1
- + "' AS VARBINARY)), "
- + "(2, CAST (x'"
- + v2
- + "' AS VARBINARY)), "
- + "(2, CAST (x'"
- + v2
- + "' AS VARBINARY)), "
- + "(3, CAST (x'"
- + v3
- + "' AS VARBINARY))");
-
- result = queryAndSort("SELECT * FROM test_rbm64");
- checkOneRecord(result.get(0), 1, v1Bytes);
- checkOneRecord(result.get(1), 2, v4Bytes);
- checkOneRecord(result.get(2), 3, v5Bytes);
- }
-
- @Test
- public void testRoaring64BitmapAgg() throws IOException {
- sql(
- "CREATE TABLE test_rbm64("
- + " id INT PRIMARY KEY NOT ENFORCED,"
- + " f0 VARBINARY"
- + ") WITH ("
- + " 'merge-engine' = 'aggregation',"
- + " 'fields.f0.aggregate-function' = 'rbm64'"
- + ")");
-
- byte[] v1Bytes = RoaringBitmap64.bitmapOf(1L).serialize();
- byte[] v2Bytes = RoaringBitmap64.bitmapOf(2L).serialize();
- byte[] v3Bytes = RoaringBitmap64.bitmapOf(3L).serialize();
- byte[] v4Bytes = RoaringBitmap64.bitmapOf(1L, 2L).serialize();
- byte[] v5Bytes = RoaringBitmap64.bitmapOf(2L, 3L).serialize();
- String v1 = Hex.encodeHexString(v1Bytes).toUpperCase();
- String v2 = Hex.encodeHexString(v2Bytes).toUpperCase();
- String v3 = Hex.encodeHexString(v3Bytes).toUpperCase();
-
- sql(
- "INSERT INTO test_rbm64 VALUES "
- + "(1, CAST (NULL AS VARBINARY)), "
- + "(2, CAST (x'"
- + v1
- + "' AS VARBINARY)), "
- + "(3, CAST (x'"
- + v2
- + "' AS VARBINARY))");
-
- List<Row> result = queryAndSort("SELECT * FROM test_rbm64");
- checkOneRecord(result.get(0), 1, null);
- checkOneRecord(result.get(1), 2, v1Bytes);
- checkOneRecord(result.get(2), 3, v2Bytes);
-
- sql(
- "INSERT INTO test_rbm64 VALUES "
- + "(1, CAST (x'"
- + v1
- + "' AS VARBINARY)), "
- + "(2, CAST (x'"
- + v2
- + "' AS VARBINARY)), "
- + "(2, CAST (x'"
- + v2
- + "' AS VARBINARY)), "
- + "(3, CAST (x'"
- + v3
- + "' AS VARBINARY))");
-
- result = queryAndSort("SELECT * FROM test_rbm64");
- checkOneRecord(result.get(0), 1, v1Bytes);
- checkOneRecord(result.get(1), 2, v4Bytes);
- checkOneRecord(result.get(2), 3, v5Bytes);
- }
-
- private void checkOneRecord(Row row, int id, byte[] expected) {
- assertThat(row.getField(0)).isEqualTo(id);
- assertThat(row.getField(1)).isEqualTo(expected);
- }
- }
-
- /** ITCase for testing the aggregation merge engine with not exist aggregation function. */
- public static class NotExistAggregationFunctionITCase extends CatalogITCaseBase {
-
- @Override
- protected List<String> ddl() {
- return Collections.singletonList(
- "CREATE TABLE test_not_exist("
- + " id INT PRIMARY KEY NOT ENFORCED,"
- + " f0 INT"
- + ") WITH ("
- + " 'merge-engine' = 'aggregation',"
- + " 'fields.f0.aggregate-function' = 'not_exist'"
- + ")");
- }
-
- @Test
- public void testInsert() {
- assertThatThrownBy(
- () -> sql("INSERT INTO test_not_exist VALUES (1, 1), (2, 2), (3, 3)"))
- .hasCauseInstanceOf(ValidationException.class)
- .hasMessageContaining(
- "Unable to create a sink for writing table 'PAIMON.default.test_not_exist'.");
- }
-
- @Test
- public void testSelect() {
- assertThatThrownBy(() -> sql("select * from test_not_exist"))
- .hasCauseInstanceOf(ValidationException.class)
- .hasMessageContaining(
- "Unable to create a source for reading table 'PAIMON.default.test_not_exist'.");
- }
- }
-}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/BasicAggregateITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/BasicAggregateITCase.java
new file mode 100644
index 0000000000..612f456805
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/BasicAggregateITCase.java
@@ -0,0 +1,85 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.aggregation;
+
+import org.apache.paimon.flink.CatalogITCaseBase;
+
+import org.apache.flink.types.Row;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** IT Test for aggregation merge engine. */
+public class BasicAggregateITCase extends CatalogITCaseBase {
+
+ @Override
+ protected List<String> ddl() {
+ return Collections.singletonList(
+ "CREATE TABLE T ("
+ + "k INT,"
+ + "v INT,"
+ + "d INT,"
+ + "PRIMARY KEY (k, d) NOT ENFORCED) PARTITIONED BY (d) "
+ + " WITH ('merge-engine'='aggregation', "
+ + "'fields.v.aggregate-function'='sum',"
+ + "'local-merge-buffer-size'='5m'"
+ + ");");
+ }
+
+ @Test
+ public void testLocalMerge() {
+ sql("INSERT INTO T VALUES(1, 1, 1), (2, 1, 1), (1, 2, 1)");
+ assertThat(batchSql("SELECT * FROM T"))
+ .containsExactlyInAnyOrder(Row.of(1, 3, 1), Row.of(2, 1, 1));
+ }
+
+ @Test
+ public void testMergeRead() {
+ sql("INSERT INTO T VALUES(1, 1, 1), (2, 1, 1)");
+ sql("INSERT INTO T VALUES(1, 2, 1)");
+ assertThat(batchSql("SELECT * FROM T"))
+ .containsExactlyInAnyOrder(Row.of(1, 3, 1), Row.of(2, 1, 1));
+ // filter
+ assertThat(batchSql("SELECT * FROM T where v = 3"))
+ .containsExactlyInAnyOrder(Row.of(1, 3, 1));
+ assertThat(batchSql("SELECT * FROM T where v = 1"))
+ .containsExactlyInAnyOrder(Row.of(2, 1, 1));
+ }
+
+ @Test
+ public void testSequenceFieldWithDefaultAgg() {
+ sql(
+ "CREATE TABLE seq_default_agg ("
+ + " pk INT PRIMARY KEY NOT ENFORCED,"
+ + " seq INT,"
+ + " v INT) WITH ("
+ + " 'merge-engine'='aggregation',"
+ + " 'sequence.field'='seq',"
+ + " 'fields.default-aggregate-function'='sum'"
+ + ")");
+
+ sql("INSERT INTO seq_default_agg VALUES (0, 1, 1)");
+ sql("INSERT INTO seq_default_agg VALUES (0, 2, 2)");
+
+ assertThat(sql("SELECT * FROM seq_default_agg")).containsExactly(Row.of(0, 2, 3));
+ }
+}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/BoolOrAndAggregationITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/BoolOrAndAggregationITCase.java
new file mode 100644
index 0000000000..c2e7d86469
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/BoolOrAndAggregationITCase.java
@@ -0,0 +1,97 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.aggregation;
+
+import org.apache.paimon.flink.CatalogITCaseBase;
+
+import org.apache.flink.types.Row;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** ITCase for bool_or/bool_and aggregate function. */
+public class BoolOrAndAggregationITCase extends CatalogITCaseBase {
+ @Override
+ protected List<String> ddl() {
+ return Collections.singletonList(
+ "CREATE TABLE IF NOT EXISTS T7 ("
+ + "j INT, k INT, "
+ + "a BOOLEAN, "
+ + "b BOOLEAN,"
+ + "PRIMARY KEY (j,k) NOT ENFORCED)"
+ + " WITH ('merge-engine'='aggregation', "
+ + "'fields.a.aggregate-function'='bool_or',"
+ + "'fields.b.aggregate-function'='bool_and'"
+ + ");");
+ }
+
+ @Test
+ public void testMergeInMemory() {
+ batchSql(
+ "INSERT INTO T7 VALUES "
+ + "(1, 2, CAST('TRUE' AS BOOLEAN), CAST('TRUE' AS BOOLEAN)),"
+ + "(1, 2, CAST(NULL AS BOOLEAN), CAST(NULL AS BOOLEAN)), "
+ + "(1, 2, CAST('FALSE' AS BOOLEAN), CAST('FALSE' AS BOOLEAN))");
+ List<Row> result = batchSql("SELECT * FROM T7");
+ assertThat(result).containsExactlyInAnyOrder(Row.of(1, 2, true, false));
+ }
+
+ @Test
+ public void testMergeRead() {
+ batchSql("INSERT INTO T7 VALUES (1, 2, CAST('TRUE' AS BOOLEAN), CAST('TRUE' AS BOOLEAN))");
+ batchSql("INSERT INTO T7 VALUES (1, 2, CAST(NULL AS BOOLEAN), CAST(NULL AS BOOLEAN))");
+ batchSql(
+ "INSERT INTO T7 VALUES (1, 2, CAST('FALSE' AS BOOLEAN), CAST('FALSE' AS BOOLEAN))");
+
+ List<Row> result = batchSql("SELECT * FROM T7");
+ assertThat(result).containsExactlyInAnyOrder(Row.of(1, 2, true, false));
+ }
+
+ @Test
+ public void testMergeCompaction() {
+ // Wait compaction
+ batchSql("ALTER TABLE T7 SET ('commit.force-compact'='true')");
+
+ // key 1 2
+ batchSql("INSERT INTO T7 VALUES (1, 2, CAST('TRUE' AS BOOLEAN), CAST('TRUE' AS BOOLEAN))");
+ batchSql("INSERT INTO T7 VALUES (1, 2, CAST(NULL AS BOOLEAN), CAST(NULL AS BOOLEAN))");
+ batchSql(
+ "INSERT INTO T7 VALUES (1, 2, CAST('FALSE' AS BOOLEAN), CAST('FALSE' AS BOOLEAN))");
+
+ // key 1 3
+ batchSql(
+ "INSERT INTO T7 VALUES (1, 3, CAST('FALSE' AS BOOLEAN), CAST('TRUE' AS BOOLEAN))");
+ batchSql("INSERT INTO T7 VALUES (1, 3, CAST(NULL AS BOOLEAN), CAST(NULL AS BOOLEAN))");
+ batchSql("INSERT INTO T7 VALUES (1, 3, CAST('FALSE' AS BOOLEAN), CAST('TRUE' AS BOOLEAN))");
+
+ assertThat(batchSql("SELECT * FROM T7"))
+ .containsExactlyInAnyOrder(Row.of(1, 2, true, false), Row.of(1, 3, false, true));
+ }
+
+ @Test
+ public void testStreamingRead() {
+ assertThatThrownBy(
+ () -> sEnv.from("T7").execute().print(),
+ "Pre-aggregate continuous reading is not supported");
+ }
+}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/CollectAggregationITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/CollectAggregationITCase.java
new file mode 100644
index 0000000000..518dff4c25
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/CollectAggregationITCase.java
@@ -0,0 +1,298 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.aggregation;
+
+import org.apache.paimon.flink.CatalogITCaseBase;
+import org.apache.paimon.mergetree.compact.aggregate.FieldCollectAgg;
+import org.apache.paimon.utils.BlockingIterator;
+
+import org.apache.flink.table.planner.factories.TestValuesTableFactory;
+import org.apache.flink.types.Row;
+import org.apache.flink.types.RowKind;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** ITCase for {@link FieldCollectAgg}. */
+public 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(
+ "CREATE TABLE test_collect("
+ + " id INT PRIMARY KEY NOT ENFORCED,"
+ + " f0 ARRAY<STRING>"
+ + ") WITH ("
+ + " 'merge-engine' = 'aggregation',"
+ + " 'fields.f0.aggregate-function' = 'collect',"
+ + " 'fields.f0.distinct' = 'true'"
+ + ")");
+
+ sql(
+ "INSERT INTO test_collect VALUES "
+ + "(1, CAST (NULL AS ARRAY<STRING>)), "
+ + "(2, ARRAY['A', 'B']), "
+ + "(3, ARRAY['car', 'watch'])");
+
+ List<Row> result = queryAndSort("SELECT * FROM test_collect");
+ checkOneRecord(result.get(0), 1);
+ checkOneRecord(result.get(1), 2, "A", "B");
+ checkOneRecord(result.get(2), 3, "car", "watch");
+
+ sql(
+ "INSERT INTO test_collect VALUES "
+ + "(1, ARRAY['paimon', 'paimon']), "
+ + "(2, ARRAY['A', 'B', 'C']), "
+ + "(3, CAST (NULL AS ARRAY<STRING>))");
+
+ result = queryAndSort("SELECT * FROM test_collect");
+ checkOneRecord(result.get(0), 1, "paimon");
+ checkOneRecord(result.get(1), 2, "A", "B", "C");
+ checkOneRecord(result.get(2), 3, "car", "watch");
+ }
+
+ @Test
+ public void testAggWithoutDistinct() {
+ sql(
+ "CREATE TABLE test_collect("
+ + " id INT PRIMARY KEY NOT ENFORCED,"
+ + " f0 ARRAY<STRING>"
+ + ") WITH ("
+ + " 'merge-engine' = 'aggregation',"
+ + " 'fields.f0.aggregate-function' = 'collect'"
+ + ")");
+
+ sql(
+ "INSERT INTO test_collect VALUES "
+ + "(1, CAST (NULL AS ARRAY<STRING>)), "
+ + "(2, ARRAY['A', 'B', 'B']), "
+ + "(3, ARRAY['car', 'watch'])");
+
+ List<Row> result = queryAndSort("SELECT * FROM test_collect");
+ checkOneRecord(result.get(0), 1);
+ checkOneRecord(result.get(1), 2, "A", "B", "B");
+ checkOneRecord(result.get(2), 3, "car", "watch");
+
+ sql(
+ "INSERT INTO test_collect VALUES "
+ + "(1, ARRAY['paimon', 'paimon']), "
+ + "(2, ARRAY['A', 'B', 'C']), "
+ + "(3, CAST (NULL AS ARRAY<STRING>))");
+
+ result = queryAndSort("SELECT * FROM test_collect");
+ checkOneRecord(result.get(0), 1, "paimon", "paimon");
+ checkOneRecord(result.get(1), 2, "A", "A", "B", "B", "B", "C");
+ checkOneRecord(result.get(2), 3, "car", "watch");
+ }
+
+ private static List<Arguments> retractArguments() {
+ return Arrays.asList(
+ Arguments.arguments("lookup", "aggregation"),
+ Arguments.arguments("lookup", "partial-update"),
+ Arguments.arguments("full-compaction", "aggregation"),
+ Arguments.arguments("full-compaction", "partial-update"));
+ }
+
+ @ParameterizedTest(name = "changelog-producer = {0}, merge-engine = {1}")
+ @MethodSource("retractArguments")
+ public void testRetract(String changelogProducer, String mergeEngine) throws Exception {
+ String sequenceGroup = "";
+ if (mergeEngine.equals("partial-update")) {
+ sequenceGroup = ", 'fields.f1.sequence-group' = 'f0'";
+ }
+ sql(
+ "CREATE TABLE test_collect("
+ + " id INT PRIMARY KEY NOT ENFORCED,"
+ + " f0 ARRAY<STRING>,"
+ + " f1 INT"
+ + ") WITH ("
+ + " 'changelog-producer' = '%s',"
+ + " 'merge-engine' = '%s',"
+ + " 'fields.f0.aggregate-function' = 'collect'"
+ + " %s"
+ + ")",
+ changelogProducer, mergeEngine, sequenceGroup);
+
+ BlockingIterator<Row, Row> select = streamSqlBlockIter("SELECT * FROM test_collect");
+
+ String temporaryTableTemplate =
+ "CREATE TEMPORARY TABLE %s ("
+ + " id INT PRIMARY KEY NOT ENFORCED,"
+ + " f0 ARRAY<STRING>,"
+ + " f1 INT"
+ + ") WITH ("
+ + " 'connector' = 'values',"
+ + " 'data-id' = '%s',"
+ + " 'bounded' = 'true',"
+ + " 'changelog-mode' = '%s'"
+ + ")";
+
+ // 1 only exists single key record
+ sql("INSERT INTO test_collect VALUES (1, ARRAY['A', 'B'], 1)");
+ List<Row> result = select.collect(1);
+ checkOneRecord(result.get(0), 1, "A", "B");
+
+ // 1.1 UB + UA (CANNOT handle)
+ List<Row> inputRecords =
+ Arrays.asList(
+ Row.ofKind(RowKind.UPDATE_BEFORE, 1, new String[] {"A", "B"}, 2),
+ Row.ofKind(RowKind.UPDATE_AFTER, 1, new String[] {"C", "D"}, 3));
+ sEnv.executeSql(
+ String.format(
+ temporaryTableTemplate,
+ "INPUT11",
+ TestValuesTableFactory.registerData(inputRecords),
+ "UB,UA"))
+ .await();
+ sEnv.executeSql("INSERT INTO test_collect SELECT * FROM INPUT11").await();
+
+ result = select.collect(2);
+ assertThat(result.get(0).getKind()).isEqualTo(RowKind.UPDATE_BEFORE);
+ checkOneRecord(result.get(0), 1, "A", "B");
+ assertThat(result.get(1).getKind()).isEqualTo(RowKind.UPDATE_AFTER);
+ checkOneRecord(result.get(1), 1, "A", "B", "C", "D");
+
+ // 1.2 -D
+ inputRecords =
+ Collections.singletonList(
+ Row.ofKind(RowKind.DELETE, 1, new String[] {"C", "D"}, 4));
+ sEnv.executeSql(
+ String.format(
+ temporaryTableTemplate,
+ "INPUT12",
+ TestValuesTableFactory.registerData(inputRecords),
+ "D"))
+ .await();
+ sEnv.executeSql("INSERT INTO test_collect SELECT * FROM INPUT12").await();
+
+ result = select.collect(2);
+ assertThat(result.get(0).getKind()).isEqualTo(RowKind.UPDATE_BEFORE);
+ checkOneRecord(result.get(0), 1, "A", "B", "C", "D");
+ assertThat(result.get(1).getKind()).isEqualTo(RowKind.UPDATE_AFTER);
+ checkOneRecord(result.get(1), 1, "A", "B");
+
+ // 2 exists multiple key records
+ sql("INSERT INTO test_collect VALUES (2, ARRAY['A', 'B'], 5), (3, ARRAY['A', 'B'], 6)");
+ result = select.collect(2);
+ checkOneRecord(result.get(0), 2, "A", "B");
+ checkOneRecord(result.get(1), 3, "A", "B");
+
+ // 2.1 UB + UA (CANNOT handle)
+ inputRecords =
+ Arrays.asList(
+ Row.ofKind(RowKind.UPDATE_BEFORE, 2, new String[] {"A", "B"}, 7),
+ Row.ofKind(RowKind.UPDATE_AFTER, 2, new String[] {"C", "D"}, 8));
+ sEnv.executeSql(
+ String.format(
+ temporaryTableTemplate,
+ "INPUT21",
+ TestValuesTableFactory.registerData(inputRecords),
+ "UB,UA"))
+ .await();
+ sEnv.executeSql("INSERT INTO test_collect SELECT * FROM INPUT21").await();
+
+ result = select.collect(2);
+ assertThat(result.get(0).getKind()).isEqualTo(RowKind.UPDATE_BEFORE);
+ checkOneRecord(result.get(0), 2, "A", "B");
+ assertThat(result.get(1).getKind()).isEqualTo(RowKind.UPDATE_AFTER);
+ checkOneRecord(result.get(1), 2, "A", "B", "C", "D");
+
+ // 2.2 -D
+ inputRecords =
+ Collections.singletonList(Row.ofKind(RowKind.DELETE, 3, new String[] {"A"}, 9));
+ sEnv.executeSql(
+ String.format(
+ temporaryTableTemplate,
+ "INPUT22",
+ TestValuesTableFactory.registerData(inputRecords),
+ "D"))
+ .await();
+ sEnv.executeSql("INSERT INTO test_collect SELECT * FROM INPUT22").await();
+
+ result = select.collect(2);
+ assertThat(result.get(0).getKind()).isEqualTo(RowKind.UPDATE_BEFORE);
+ checkOneRecord(result.get(0), 3, "A", "B");
+ assertThat(result.get(1).getKind()).isEqualTo(RowKind.UPDATE_AFTER);
+ checkOneRecord(result.get(1), 3, "B");
+
+ select.close();
+ }
+
+ @Test
+ public void testRetractInputNull() throws Exception {
+ sql(
+ "CREATE TABLE test_collect ("
+ + " id INT PRIMARY KEY NOT ENFORCED,"
+ + " f0 ARRAY<STRING>,"
+ + " f1 INT"
+ + ") WITH ("
+ + " 'changelog-producer' = 'lookup',"
+ + " 'merge-engine' = 'partial-update',"
+ + " 'fields.f0.aggregate-function' = 'collect',"
+ + " 'fields.f1.sequence-group' = 'f0'"
+ + ")");
+
+ List<Row> input =
+ Arrays.asList(
+ Row.ofKind(RowKind.INSERT, 1, null, 1),
+ Row.ofKind(RowKind.INSERT, 1, new String[] {"A"}, 2),
+ Row.ofKind(RowKind.UPDATE_BEFORE, 1, null, 1),
+ Row.ofKind(RowKind.UPDATE_AFTER, 1, new String[] {"B"}, 3));
+ sEnv.executeSql(
+ String.format(
+ "CREATE TEMPORARY TABLE input ("
+ + " id INT PRIMARY KEY NOT ENFORCED,"
+ + " f0 ARRAY<STRING>,"
+ + " f1 INT"
+ + ") WITH ("
+ + " 'connector' = 'values',"
+ + " 'data-id' = '%s',"
+ + " 'bounded' = 'true',"
+ + " 'changelog-mode' = 'UB,UA'"
+ + ")",
+ TestValuesTableFactory.registerData(input)))
+ .await();
+ sEnv.executeSql("INSERT INTO test_collect SELECT * FROM input").await();
+
+ assertThat(sql("SELECT * FROM test_collect"))
+ .containsExactly(Row.of(1, new String[] {"A", "B"}, 3));
+ }
+
+ private void checkOneRecord(Row row, int id, String... elements) {
+ assertThat(row.getField(0)).isEqualTo(id);
+ if (elements == null || elements.length == 0) {
+ assertThat(row.getField(1)).isNull();
+ } else {
+ assertThat((String[]) row.getField(1)).containsExactlyInAnyOrder(elements);
+ }
+ }
+}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/FieldsDefaultAggregationITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/FieldsDefaultAggregationITCase.java
new file mode 100644
index 0000000000..a0fab43f01
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/FieldsDefaultAggregationITCase.java
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.aggregation;
+
+import org.apache.paimon.flink.CatalogITCaseBase;
+
+import org.apache.flink.types.Row;
+import org.junit.jupiter.api.Test;
+
+import java.time.LocalDate;
+import java.util.Collections;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** ITCase for last non-null value aggregate function. */
+public class FieldsDefaultAggregationITCase extends CatalogITCaseBase {
+ @Override
+ protected int defaultParallelism() {
+ // set parallelism to 1 so that the order of input data is determined
+ return 1;
+ }
+
+ @Override
+ protected List<String> ddl() {
+ return Collections.singletonList(
+ "CREATE TABLE IF NOT EXISTS test_default_agg_func ("
+ + "j INT, k INT, "
+ + "a INT, "
+ + "b INT, "
+ + "i DATE,"
+ + "PRIMARY KEY (j,k) NOT ENFORCED)"
+ + " WITH ('merge-engine'='aggregation', "
+ + "'fields.default-aggregate-function'='first_non_null_value', "
+ + "'fields.i.aggregate-function'='last_non_null_value'"
+ + ");");
+ }
+
+ @Test
+ public void testMergeInMemory() {
+ // VALUES does not guarantee order, but order is important for last value aggregations.
+ // So we need to sort the input data.
+ batchSql(
+ "CREATE TABLE myTable AS "
+ + "SELECT b, c, d, e, f FROM "
+ + "(VALUES "
+ + " (1, 1, 2, CAST(NULL AS INT), 4, CAST('2020-01-01' AS DATE)),"
+ + " (2, 1, 2, 2, CAST(NULL as INT), CAST('2020-01-02' AS DATE)),"
+ + " (3, 1, 2, 3, 5, CAST(NULL AS DATE))"
+ + ") AS V(a, b, c, d, e, f) "
+ + "ORDER BY a");
+ batchSql("INSERT INTO test_default_agg_func SELECT * FROM myTable");
+ List<Row> result = batchSql("SELECT * FROM test_default_agg_func");
+ assertThat(result).containsExactlyInAnyOrder(Row.of(1, 2, 2, 4, LocalDate.of(2020, 1, 2)));
+ }
+
+ @Test
+ public void testMergeRead() {
+ batchSql(
+ "INSERT INTO test_default_agg_func VALUES (1, 2, CAST(NULL AS INT), 3, CAST('2020-01-01' AS DATE))");
+ batchSql(
+ "INSERT INTO test_default_agg_func VALUES (1, 2, 2, CAST(NULL AS INT), CAST('2020-01-02' AS DATE))");
+ batchSql("INSERT INTO test_default_agg_func VALUES (1, 2, 3, 5, CAST(NULL AS DATE))");
+
+ List<Row> result = batchSql("SELECT * FROM test_default_agg_func");
+ assertThat(result).containsExactlyInAnyOrder(Row.of(1, 2, 2, 3, LocalDate.of(2020, 1, 2)));
+ }
+
+ @Test
+ public void testMergeCompaction() {
+ // Wait compaction
+ batchSql("ALTER TABLE test_default_agg_func SET ('commit.force-compact'='true')");
+
+ // key 1 2
+ batchSql(
+ "INSERT INTO test_default_agg_func VALUES (1, 2, CAST(NULL AS INT), 3, CAST('2020-01-01' AS DATE))");
+ batchSql(
+ "INSERT INTO test_default_agg_func VALUES (1, 2, 2, CAST(NULL AS INT), CAST('2020-01-02' AS DATE))");
+ batchSql("INSERT INTO test_default_agg_func VALUES (1, 2, 3, 5, CAST(NULL AS DATE))");
+
+ // key 1 3
+ batchSql(
+ "INSERT INTO test_default_agg_func VALUES (1, 3, 3, 4, CAST('2020-01-01' AS DATE))");
+ batchSql("INSERT INTO test_default_agg_func VALUES (1, 3, 2, 6, CAST(NULL AS DATE))");
+ batchSql(
+ "INSERT INTO test_default_agg_func VALUES (1, 3, CAST(NULL AS INT), CAST(NULL AS INT), CAST('2022-01-02' AS DATE))");
+
+ assertThat(batchSql("SELECT * FROM test_default_agg_func"))
+ .containsExactlyInAnyOrder(
+ Row.of(1, 2, 2, 3, LocalDate.of(2020, 1, 2)),
+ Row.of(1, 3, 3, 4, LocalDate.of(2022, 1, 2)));
+ }
+
+ @Test
+ public void testStreamingRead() {
+ assertThatThrownBy(
+ () -> sEnv.from("test_default_agg_func").execute().print(),
+ "Pre-aggregate continuous reading is not supported");
+ }
+}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/FirstValueAggregationITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/FirstValueAggregationITCase.java
new file mode 100644
index 0000000000..4a7db61ef2
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/FirstValueAggregationITCase.java
@@ -0,0 +1,129 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.aggregation;
+
+import org.apache.paimon.flink.CatalogITCaseBase;
+
+import org.apache.flink.types.Row;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** ITCase for first_value aggregate function. */
+public class FirstValueAggregationITCase extends CatalogITCaseBase {
+ @Override
+ protected List<String> ddl() {
+ return Arrays.asList(
+ "CREATE TABLE T ("
+ + "k INT,"
+ + "a INT,"
+ + "b VARCHAR,"
+ + "c VARCHAR,"
+ + "d VARCHAR,"
+ + "PRIMARY KEY (k) NOT ENFORCED)"
+ + " WITH ('merge-engine'='aggregation', "
+ + "'changelog-producer' = 'full-compaction',"
+ + "'fields.b.aggregate-function'='first_value',"
+ + "'fields.c.aggregate-function'='first_non_null_value',"
+ + "'fields.d.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'"
+ + ");");
+ }
+
+ @Test
+ public void tesInMemoryMerge() {
+ batchSql(
+ "INSERT INTO T VALUES "
+ + "(1, 0, CAST(NULL AS VARCHAR), CAST(NULL AS VARCHAR), CAST(NULL AS VARCHAR)),"
+ + "(1, 1, '1', '1', '1'), "
+ + "(2, 2, '2', '2', '2'),"
+ + "(2, 3, '22', '22', '22')");
+ List<Row> result = batchSql("SELECT * FROM T");
+ assertThat(result)
+ .containsExactlyInAnyOrder(
+ Row.of(1, 1, null, "1", "1"), Row.of(2, 3, "2", "2", "2"));
+ }
+
+ @Test
+ public void tesUnOrderInput() {
+ batchSql(
+ "INSERT INTO T VALUES "
+ + "(1, 0, CAST(NULL AS VARCHAR), CAST(NULL AS VARCHAR), CAST(NULL AS VARCHAR)),"
+ + "(1, 1, '1', '1', '1'), "
+ + "(2, 3, '2', '2', '2'),"
+ + "(2, 2, '22', '22', '22')");
+ List<Row> result = batchSql("SELECT * FROM T");
+ assertThat(result)
+ .containsExactlyInAnyOrder(
+ Row.of(1, 1, null, "1", "1"), Row.of(2, 3, "22", "22", "22"));
+ batchSql("INSERT INTO T VALUES (2, 1, '1', '1', '1')");
+ result = batchSql("SELECT * FROM T");
+ assertThat(result)
+ .containsExactlyInAnyOrder(
+ Row.of(1, 1, null, "1", "1"), Row.of(2, 3, "1", "1", "1"));
+ }
+
+ @Test
+ public void testMergeRead() {
+ batchSql(
+ "INSERT INTO T VALUES (1, 1, CAST(NULL AS VARCHAR), CAST(NULL AS VARCHAR), CAST(NULL AS VARCHAR))");
+ batchSql("INSERT INTO T VALUES (1, 2, '1', '1', '1')");
+ batchSql("INSERT INTO T VALUES (2, 1, '2', '2', '2')");
+ batchSql("INSERT INTO T VALUES (2, 2, '22', '22', '22')");
+ List<Row> result = batchSql("SELECT * FROM T");
+ assertThat(result)
+ .containsExactlyInAnyOrder(
+ Row.of(1, 2, null, "1", "1"), Row.of(2, 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));
+ }
+}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/LastNonNullValueAggregationITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/LastNonNullValueAggregationITCase.java
new file mode 100644
index 0000000000..8896f0a762
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/LastNonNullValueAggregationITCase.java
@@ -0,0 +1,112 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.aggregation;
+
+import org.apache.paimon.flink.CatalogITCaseBase;
+
+import org.apache.flink.types.Row;
+import org.junit.jupiter.api.Test;
+
+import java.time.LocalDate;
+import java.util.Collections;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** ITCase for last non-null value aggregate function. */
+public class LastNonNullValueAggregationITCase extends CatalogITCaseBase {
+ @Override
+ protected int defaultParallelism() {
+ // set parallelism to 1 so that the order of input data is determined
+ return 1;
+ }
+
+ @Override
+ protected List<String> ddl() {
+ return Collections.singletonList(
+ "CREATE TABLE IF NOT EXISTS T4 ("
+ + "j INT, k INT, "
+ + "a INT, "
+ + "b INT, "
+ + "i DATE,"
+ + "PRIMARY KEY (j,k) NOT ENFORCED)"
+ + " WITH ('merge-engine'='aggregation', "
+ + "'fields.a.aggregate-function'='last_non_null_value', "
+ + "'fields.i.aggregate-function'='last_non_null_value'"
+ + ");");
+ }
+
+ @Test
+ public void testMergeInMemory() {
+ // VALUES does not guarantee order, but order is important for last value aggregations.
+ // So we need to sort the input data.
+ batchSql(
+ "CREATE TABLE myTable AS "
+ + "SELECT b, c, d, e, f FROM "
+ + "(VALUES "
+ + " (1, 1, 2, CAST(NULL AS INT), 4, CAST('2020-01-01' AS DATE)),"
+ + " (2, 1, 2, 2, CAST(NULL as INT), CAST('2020-01-02' AS DATE)),"
+ + " (3, 1, 2, 3, 5, CAST(NULL AS DATE))"
+ + ") AS V(a, b, c, d, e, f) "
+ + "ORDER BY a");
+ batchSql("INSERT INTO T4 SELECT * FROM myTable");
+ List<Row> result = batchSql("SELECT * FROM T4");
+ assertThat(result).containsExactlyInAnyOrder(Row.of(1, 2, 3, 5, LocalDate.of(2020, 1, 2)));
+ }
+
+ @Test
+ public void testMergeRead() {
+ batchSql("INSERT INTO T4 VALUES (1, 2, CAST(NULL AS INT), 3, CAST('2020-01-01' AS DATE))");
+ batchSql("INSERT INTO T4 VALUES (1, 2, 2, CAST(NULL AS INT), CAST('2020-01-02' AS DATE))");
+ batchSql("INSERT INTO T4 VALUES (1, 2, 3, 5, CAST(NULL AS DATE))");
+
+ List<Row> result = batchSql("SELECT * FROM T4");
+ assertThat(result).containsExactlyInAnyOrder(Row.of(1, 2, 3, 5, LocalDate.of(2020, 1, 2)));
+ }
+
+ @Test
+ public void testMergeCompaction() {
+ // Wait compaction
+ batchSql("ALTER TABLE T4 SET ('commit.force-compact'='true')");
+
+ // key 1 2
+ batchSql("INSERT INTO T4 VALUES (1, 2, CAST(NULL AS INT), 3, CAST('2020-01-01' AS DATE))");
+ batchSql("INSERT INTO T4 VALUES (1, 2, 2, CAST(NULL AS INT), CAST('2020-01-02' AS DATE))");
+ batchSql("INSERT INTO T4 VALUES (1, 2, 3, 5, CAST(NULL AS DATE))");
+
+ // key 1 3
+ batchSql("INSERT INTO T4 VALUES (1, 3, 3, 4, CAST('2020-01-01' AS DATE))");
+ batchSql("INSERT INTO T4 VALUES (1, 3, 2, 6, CAST(NULL AS DATE))");
+ batchSql(
+ "INSERT INTO T4 VALUES (1, 3, CAST(NULL AS INT), CAST(NULL AS INT), CAST('2022-01-02' AS DATE))");
+
+ assertThat(batchSql("SELECT * FROM T4"))
+ .containsExactlyInAnyOrder(
+ Row.of(1, 2, 3, 5, LocalDate.of(2020, 1, 2)),
+ Row.of(1, 3, 2, 6, LocalDate.of(2022, 1, 2)));
+ }
+
+ @Test
+ public void testStreamingRead() {
+ assertThatThrownBy(
+ () -> sEnv.from("T4").execute().print(),
+ "Pre-aggregate continuous reading is not supported");
+ }
+}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/LastValueAggregationITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/LastValueAggregationITCase.java
new file mode 100644
index 0000000000..55f98062d7
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/LastValueAggregationITCase.java
@@ -0,0 +1,109 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.aggregation;
+
+import org.apache.paimon.flink.CatalogITCaseBase;
+
+import org.apache.flink.types.Row;
+import org.junit.jupiter.api.Test;
+
+import java.time.LocalDate;
+import java.util.Collections;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** ITCase for last value aggregate function. */
+public class LastValueAggregationITCase extends CatalogITCaseBase {
+ @Override
+ protected int defaultParallelism() {
+ // set parallelism to 1 so that the order of input data is determined
+ return 1;
+ }
+
+ @Override
+ protected List<String> ddl() {
+ return Collections.singletonList(
+ "CREATE TABLE IF NOT EXISTS T5 ("
+ + "j INT, k INT, "
+ + "a INT, "
+ + "i DATE,"
+ + "PRIMARY KEY (j,k) NOT ENFORCED)"
+ + " WITH ('merge-engine'='aggregation', "
+ + "'fields.a.aggregate-function'='last_value', "
+ + "'fields.i.aggregate-function'='last_value'"
+ + ");");
+ }
+
+ @Test
+ public void testMergeInMemory() {
+ // VALUES does not guarantee order, but order is important for last value aggregations.
+ // So we need to sort the input data.
+ batchSql(
+ "CREATE TABLE myTable AS "
+ + "SELECT b, c, d, e FROM "
+ + "(VALUES "
+ + " (1, 1, 2, CAST(NULL AS INT), CAST('2020-01-01' AS DATE)),"
+ + " (2, 1, 2, 2, CAST('2020-01-02' AS DATE)),"
+ + " (3, 1, 2, 3, CAST(NULL AS DATE))"
+ + ") AS V(a, b, c, d, e) "
+ + "ORDER BY a");
+ batchSql("INSERT INTO T5 SELECT * FROM myTable");
+ List<Row> result = batchSql("SELECT * FROM T5");
+ assertThat(result).containsExactlyInAnyOrder(Row.of(1, 2, 3, null));
+ }
+
+ @Test
+ public void testMergeRead() {
+ batchSql("INSERT INTO T5 VALUES (1, 2, CAST(NULL AS INT), CAST('2020-01-01' AS DATE))");
+ batchSql("INSERT INTO T5 VALUES (1, 2, 2, CAST('2020-01-02' AS DATE))");
+ batchSql("INSERT INTO T5 VALUES (1, 2, 3, CAST(NULL AS DATE))");
+
+ List<Row> result = batchSql("SELECT * FROM T5");
+ assertThat(result).containsExactlyInAnyOrder(Row.of(1, 2, 3, null));
+ }
+
+ @Test
+ public void testMergeCompaction() {
+ // Wait compaction
+ batchSql("ALTER TABLE T5 SET ('commit.force-compact'='true')");
+
+ // key 1 2
+ batchSql("INSERT INTO T5 VALUES (1, 2, CAST(NULL AS INT), CAST('2020-01-01' AS DATE))");
+ batchSql("INSERT INTO T5 VALUES (1, 2, 2, CAST('2020-01-02' AS DATE))");
+ batchSql("INSERT INTO T5 VALUES (1, 2, 3, CAST(NULL AS DATE))");
+
+ // key 1 3
+ batchSql("INSERT INTO T5 VALUES (1, 3, 3, CAST('2020-01-01' AS DATE))");
+ batchSql("INSERT INTO T5 VALUES (1, 3, 2, CAST(NULL AS DATE))");
+ batchSql("INSERT INTO T5 VALUES (1, 3, CAST(NULL AS INT), CAST('2022-01-02' AS DATE))");
+
+ assertThat(batchSql("SELECT * FROM T5"))
+ .containsExactlyInAnyOrder(
+ Row.of(1, 2, 3, null), Row.of(1, 3, null, LocalDate.of(2022, 1, 2)));
+ }
+
+ @Test
+ public void testStreamingRead() {
+ assertThatThrownBy(
+ () -> sEnv.from("T5").execute().print(),
+ "Pre-aggregate continuous reading is not supported");
+ }
+}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/ListAggAggregationITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/ListAggAggregationITCase.java
new file mode 100644
index 0000000000..0e76408c93
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/ListAggAggregationITCase.java
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.aggregation;
+
+import org.apache.paimon.flink.CatalogITCaseBase;
+
+import org.apache.flink.types.Row;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** ITCase for listagg aggregate function. */
+public class ListAggAggregationITCase extends CatalogITCaseBase {
+ @Override
+ protected int defaultParallelism() {
+ // set parallelism to 1 so that the order of input data is determined
+ return 1;
+ }
+
+ @Override
+ protected List<String> ddl() {
+ return Collections.singletonList(
+ "CREATE TABLE IF NOT EXISTS T6 ("
+ + "j INT, k INT, "
+ + "a STRING, "
+ + "PRIMARY KEY (j,k) NOT ENFORCED)"
+ + " WITH ('merge-engine'='aggregation', "
+ + "'fields.a.aggregate-function'='listagg'"
+ + ");");
+ }
+
+ @Test
+ public void testMergeInMemory() {
+ // VALUES does not guarantee order, but order is important for list aggregations.
+ // So we need to sort the input data.
+ batchSql(
+ "CREATE TABLE myTable AS "
+ + "SELECT b, c, d FROM "
+ + "(VALUES "
+ + " (1, 1, 2, 'first line'),"
+ + " (2, 1, 2, CAST(NULL AS STRING)),"
+ + " (3, 1, 2, 'second line')"
+ + ") AS V(a, b, c, d) "
+ + "ORDER BY a");
+ batchSql("INSERT INTO T6 SELECT * FROM myTable");
+ List<Row> result = batchSql("SELECT * FROM T6");
+ assertThat(result).containsExactlyInAnyOrder(Row.of(1, 2, "first line,second line"));
+ }
+
+ @Test
+ public void testMergeRead() {
+ batchSql("INSERT INTO T6 VALUES (1, 2, 'first line')");
+ batchSql("INSERT INTO T6 VALUES (1, 2, CAST(NULL AS STRING))");
+ batchSql("INSERT INTO T6 VALUES (1, 2, 'second line')");
+
+ List<Row> result = batchSql("SELECT * FROM T6");
+ assertThat(result).containsExactlyInAnyOrder(Row.of(1, 2, "first line,second line"));
+ }
+
+ @Test
+ public void testMergeCompaction() {
+ // Wait compaction
+ batchSql("ALTER TABLE T6 SET ('commit.force-compact'='true')");
+
+ // key 1 2
+ batchSql("INSERT INTO T6 VALUES (1, 2, 'first line')");
+ batchSql("INSERT INTO T6 VALUES (1, 2, CAST(NULL AS STRING))");
+ batchSql("INSERT INTO T6 VALUES (1, 2, 'second line')");
+
+ // key 1 3
+ batchSql("INSERT INTO T6 VALUES (1, 3, CAST(NULL AS STRING))");
+ batchSql("INSERT INTO T6 VALUES (1, 3, CAST(NULL AS STRING))");
+ batchSql("INSERT INTO T6 VALUES (1, 3, CAST(NULL AS STRING))");
+
+ assertThat(batchSql("SELECT * FROM T6"))
+ .containsExactlyInAnyOrder(
+ Row.of(1, 2, "first line,second line"), Row.of(1, 3, null));
+ }
+
+ @Test
+ public void testStreamingRead() {
+ assertThatThrownBy(
+ () -> sEnv.from("T6").execute().print(),
+ "Pre-aggregate continuous reading is not supported");
+ }
+}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/MaxAggregationITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/MaxAggregationITCase.java
new file mode 100644
index 0000000000..098580ef97
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/MaxAggregationITCase.java
@@ -0,0 +1,209 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.aggregation;
+
+import org.apache.paimon.flink.CatalogITCaseBase;
+
+import org.apache.flink.types.Row;
+import org.junit.jupiter.api.Test;
+
+import java.math.BigDecimal;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.util.Collections;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** ITCase for max aggregate function. */
+public class MaxAggregationITCase extends CatalogITCaseBase {
+ @Override
+ protected List<String> ddl() {
+ return Collections.singletonList(
+ "CREATE TABLE IF NOT EXISTS T2 ("
+ + "j INT, k INT, "
+ + "a INT, "
+ + "b Decimal(4,2), "
+ + "c TINYINT,"
+ + "d SMALLINT,"
+ + "e BIGINT,"
+ + "f FLOAT,"
+ + "h DOUBLE,"
+ + "i DATE,"
+ + "l TIMESTAMP,"
+ + "m CHAR,"
+ + "n VARCHAR,"
+ + "PRIMARY KEY (j,k) NOT ENFORCED)"
+ + " WITH ('merge-engine'='aggregation', "
+ + "'fields.a.aggregate-function'='max', "
+ + "'fields.b.aggregate-function'='max', "
+ + "'fields.c.aggregate-function'='max', "
+ + "'fields.d.aggregate-function'='max', "
+ + "'fields.e.aggregate-function'='max', "
+ + "'fields.f.aggregate-function'='max',"
+ + "'fields.h.aggregate-function'='max',"
+ + "'fields.i.aggregate-function'='max',"
+ + "'fields.l.aggregate-function'='max',"
+ + "'fields.m.aggregate-function'='max',"
+ + "'fields.n.aggregate-function'='max'"
+ + ");");
+ }
+
+ @Test
+ public void testMergeInMemory() {
+ batchSql(
+ "INSERT INTO T2 VALUES "
+ + "(1, 2, CAST(NULL AS INT), 1.01, CAST(1 AS TINYINT), CAST(-1 AS SMALLINT), "
+ + "CAST(1000 AS BIGINT), 1.11, CAST(1.11 AS DOUBLE), CAST('2020-01-01' AS DATE), "
+ + "CAST('2021-01-01 01:01:01' AS TIMESTAMP), 'a', 'aaa'),"
+ + "(1, 2, 2, 1.10, CAST(2 AS TINYINT), CAST(2 AS SMALLINT), CAST(100000 AS BIGINT), "
+ + "-1.11, CAST(1.21 AS DOUBLE), CAST('2020-01-02' AS DATE), "
+ + "CAST('2022-01-01 01:01:01' AS TIMESTAMP), 'b', 'bbb'), "
+ + "(1, 2, 3, 10.00, CAST(1 AS TINYINT), CAST(1 AS SMALLINT), CAST(10000000 AS BIGINT), "
+ + "0, CAST(-1.11 AS DOUBLE), CAST('2022-01-02' AS DATE), "
+ + "CAST('2022-01-01 02:00:00' AS TIMESTAMP), 'c', 'ccc')");
+ List<Row> result = batchSql("SELECT * FROM T2");
+ assertThat(result)
+ .containsExactlyInAnyOrder(
+ Row.of(
+ 1,
+ 2,
+ 3,
+ new BigDecimal("10.00"),
+ (byte) 2,
+ (short) 2,
+ (long) 10000000,
+ (float) 1.11,
+ 1.21,
+ LocalDate.of(2022, 1, 2),
+ LocalDateTime.of(2022, 1, 1, 2, 0, 0),
+ "c",
+ "ccc"));
+ }
+
+ @Test
+ public void testMergeRead() {
+ batchSql(
+ "INSERT INTO T2 VALUES "
+ + "(1, 2, CAST(NULL AS INT), 1.01, CAST(1 AS TINYINT), CAST(-1 AS SMALLINT), CAST(1000 AS BIGINT), "
+ + "1.11, CAST(1.11 AS DOUBLE), CAST('2020-01-01' AS DATE), "
+ + "CAST('2021-01-01 01:01:01' AS TIMESTAMP), 'a', 'aaa')");
+ batchSql(
+ "INSERT INTO T2 VALUES "
+ + "(1, 2, 2, 1.10, CAST(2 AS TINYINT), CAST(2 AS SMALLINT), CAST(100000 AS BIGINT), -1.11, "
+ + "CAST(1.21 AS DOUBLE), CAST('2020-01-02' AS DATE), "
+ + "CAST('2022-01-01 01:01:01' AS TIMESTAMP), 'b', 'bbb')");
+ batchSql(
+ "INSERT INTO T2 VALUES "
+ + "(1, 2, 3, 10.00, CAST(1 AS TINYINT), CAST(1 AS SMALLINT), CAST(10000000 AS BIGINT), 0, "
+ + "CAST(-1.11 AS DOUBLE), CAST('2022-01-02' AS DATE), "
+ + "CAST('2022-01-01 02:00:00' AS TIMESTAMP), 'c', 'ccc')");
+
+ List<Row> result = batchSql("SELECT * FROM T2");
+ assertThat(result)
+ .containsExactlyInAnyOrder(
+ Row.of(
+ 1,
+ 2,
+ 3,
+ new BigDecimal("10.00"),
+ (byte) 2,
+ (short) 2,
+ (long) 10000000,
+ (float) 1.11,
+ 1.21,
+ LocalDate.of(2022, 1, 2),
+ LocalDateTime.of(2022, 1, 1, 2, 0, 0),
+ "c",
+ "ccc"));
+ }
+
+ @Test
+ public void testMergeCompaction() {
+ // Wait compaction
+ batchSql("ALTER TABLE T2 SET ('commit.force-compact'='true')");
+
+ // key 1 2
+ batchSql(
+ "INSERT INTO T2 VALUES "
+ + "(1, 2, CAST(NULL AS INT), 1.01, CAST(1 AS TINYINT), CAST(-1 AS SMALLINT), CAST(1000 AS BIGINT), "
+ + "1.11, CAST(1.11 AS DOUBLE), CAST('2020-01-01' AS DATE), CAST('2021-01-01 01:01:01' AS TIMESTAMP), 'a', 'aaa')");
+ batchSql(
+ "INSERT INTO T2 VALUES "
+ + "(1, 2, 2, 1.10, CAST(2 AS TINYINT), CAST(2 AS SMALLINT), CAST(100000 AS BIGINT), -1.11, "
+ + "CAST(1.21 AS DOUBLE), CAST('2020-01-02' AS DATE), CAST('2022-01-01 01:01:01' AS TIMESTAMP), 'c', 'ccc')");
+ batchSql(
+ "INSERT INTO T2 VALUES "
+ + "(1, 2, 3, 10.00, CAST(1 AS TINYINT), CAST(1 AS SMALLINT), CAST(10000000 AS BIGINT), 0, "
+ + "CAST(-1.11 AS DOUBLE), CAST('2022-01-02' AS DATE), CAST('2022-01-01 02:00:00' AS TIMESTAMP), 'b', 'bbb')");
+
+ // key 1 3
+ batchSql(
+ "INSERT INTO T2 VALUES "
+ + "(1, 3, CAST(NULL AS INT), 1.01, CAST(1 AS TINYINT), CAST(-1 AS SMALLINT), CAST(1000 AS BIGINT), "
+ + "1.11, CAST(1.11 AS DOUBLE), CAST('2020-01-01' AS DATE), CAST('2021-01-01 01:01:01' AS TIMESTAMP), 'a', 'aaa')");
+ batchSql(
+ "INSERT INTO T2 VALUES "
+ + "(1, 3, 6, 1.10, CAST(2 AS TINYINT), CAST(2 AS SMALLINT), CAST(100000 AS BIGINT), -1.11, "
+ + "CAST(1.21 AS DOUBLE), CAST('2020-01-02' AS DATE), CAST('2022-01-01 01:01:01' AS TIMESTAMP), 'c', 'ccc')");
+ batchSql(
+ "INSERT INTO T2 VALUES "
+ + "(1, 3, 3, 10.00, CAST(1 AS TINYINT), CAST(1 AS SMALLINT), CAST(10000000 AS BIGINT), 0, "
+ + "CAST(-1.11 AS DOUBLE), CAST('2022-01-02' AS DATE), CAST('2022-01-01 02:00:00' AS TIMESTAMP), 'b', 'bbb')");
+
+ assertThat(batchSql("SELECT * FROM T2"))
+ .containsExactlyInAnyOrder(
+ Row.of(
+ 1,
+ 2,
+ 3,
+ new BigDecimal("10.00"),
+ (byte) 2,
+ (short) 2,
+ (long) 10000000,
+ (float) 1.11,
+ 1.21,
+ LocalDate.of(2022, 1, 2),
+ LocalDateTime.of(2022, 1, 1, 2, 0, 0),
+ "c",
+ "ccc"),
+ Row.of(
+ 1,
+ 3,
+ 6,
+ new BigDecimal("10.00"),
+ (byte) 2,
+ (short) 2,
+ (long) 10000000,
+ (float) 1.11,
+ 1.21,
+ LocalDate.of(2022, 1, 2),
+ LocalDateTime.of(2022, 1, 1, 2, 0, 0),
+ "c",
+ "ccc" + ""));
+ }
+
+ @Test
+ public void testStreamingRead() {
+ assertThatThrownBy(
+ () -> sEnv.from("T2").execute().print(),
+ "Pre-aggregate continuous reading is not supported");
+ }
+}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/MergeMapAggregationITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/MergeMapAggregationITCase.java
new file mode 100644
index 0000000000..aaf910b85d
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/MergeMapAggregationITCase.java
@@ -0,0 +1,134 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.aggregation;
+
+import org.apache.paimon.flink.CatalogITCaseBase;
+import org.apache.paimon.mergetree.compact.aggregate.FieldMergeMapAgg;
+
+import org.apache.flink.table.planner.factories.TestValuesTableFactory;
+import org.apache.flink.types.Row;
+import org.apache.flink.types.RowKind;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** ITCase for {@link FieldMergeMapAgg}. */
+public class MergeMapAggregationITCase extends CatalogITCaseBase {
+
+ @Override
+ protected List<String> ddl() {
+ return Collections.singletonList(
+ "CREATE TABLE test_merge_map("
+ + " id INT PRIMARY KEY NOT ENFORCED,"
+ + " f0 MAP<INT, STRING>"
+ + ") WITH ("
+ + " 'merge-engine' = 'aggregation',"
+ + " 'fields.f0.aggregate-function' = 'merge_map'"
+ + ")");
+ }
+
+ @Test
+ public void testMergeMap() {
+ sql(
+ "INSERT INTO test_merge_map VALUES "
+ + "(1, CAST (NULL AS MAP<INT, STRING>)), "
+ + "(2, MAP[1, 'A']), "
+ + "(3, MAP[1, 'A', 2, 'B'])");
+
+ List<Row> result = queryAndSort("SELECT * FROM test_merge_map");
+ checkOneRecord(result.get(0), 1, null);
+ checkOneRecord(result.get(1), 2, toMap(1, "A"));
+ checkOneRecord(result.get(2), 3, toMap(1, "A", 2, "B"));
+
+ sql(
+ "INSERT INTO test_merge_map VALUES "
+ + "(1, MAP[1, 'A']), "
+ + "(2, MAP[1, 'B']), "
+ + "(3, MAP[1, 'a', 2, 'b', 3, 'c'])");
+
+ result = queryAndSort("SELECT * FROM test_merge_map");
+ checkOneRecord(result.get(0), 1, toMap(1, "A"));
+ checkOneRecord(result.get(1), 2, toMap(1, "B"));
+ checkOneRecord(result.get(2), 3, toMap(1, "a", 2, "b", 3, "c"));
+ }
+
+ @Test
+ public void testRetractInputNull() throws Exception {
+ sql(
+ "CREATE TABLE test_merge_map1 ("
+ + " id INT PRIMARY KEY NOT ENFORCED,"
+ + " f0 MAP<INT, STRING>,"
+ + " f1 INT"
+ + ") WITH ("
+ + " 'changelog-producer' = 'lookup',"
+ + " 'merge-engine' = 'partial-update',"
+ + " 'fields.f0.aggregate-function' = 'merge_map',"
+ + " 'fields.f1.sequence-group' = 'f0'"
+ + ")");
+
+ List<Row> input =
+ Arrays.asList(
+ Row.ofKind(RowKind.INSERT, 1, null, 1),
+ Row.ofKind(RowKind.INSERT, 1, Collections.singletonMap(1, "A"), 2),
+ Row.ofKind(RowKind.UPDATE_BEFORE, 1, null, 1),
+ Row.ofKind(RowKind.UPDATE_AFTER, 1, Collections.singletonMap(2, "B"), 3));
+ sEnv.executeSql(
+ String.format(
+ "CREATE TEMPORARY TABLE input ("
+ + " id INT PRIMARY KEY NOT ENFORCED,"
+ + " f0 MAP<INT, STRING>,"
+ + " f1 INT"
+ + ") WITH ("
+ + " 'connector' = 'values',"
+ + " 'data-id' = '%s',"
+ + " 'bounded' = 'true',"
+ + " 'changelog-mode' = 'UB,UA'"
+ + ")",
+ TestValuesTableFactory.registerData(input)))
+ .await();
+ sEnv.executeSql("INSERT INTO test_merge_map1 SELECT * FROM input").await();
+
+ assertThat(sql("SELECT * FROM test_merge_map1"))
+ .containsExactly(Row.of(1, toMap(1, "A", 2, "B"), 3));
+ }
+
+ private Map<Object, Object> toMap(Object... kvs) {
+ Map<Object, Object> result = new HashMap<>();
+ for (int i = 0; i < kvs.length; i += 2) {
+ result.put(kvs[i], kvs[i + 1]);
+ }
+ return result;
+ }
+
+ private void checkOneRecord(Row row, int id, Map<Object, Object> map) {
+ assertThat(row.getField(0)).isEqualTo(id);
+ if (map == null || map.isEmpty()) {
+ assertThat(row.getField(1)).isNull();
+ } else {
+ assertThat((Map<Object, Object>) row.getField(1))
+ .containsExactlyInAnyOrderEntriesOf(map);
+ }
+ }
+}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/MinAggregationITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/MinAggregationITCase.java
new file mode 100644
index 0000000000..009e607f43
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/MinAggregationITCase.java
@@ -0,0 +1,206 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.aggregation;
+
+import org.apache.paimon.flink.CatalogITCaseBase;
+
+import org.apache.flink.types.Row;
+import org.junit.jupiter.api.Test;
+
+import java.math.BigDecimal;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.util.Collections;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** ITCase for min aggregate function. */
+public class MinAggregationITCase extends CatalogITCaseBase {
+ @Override
+ protected List<String> ddl() {
+ return Collections.singletonList(
+ "CREATE TABLE IF NOT EXISTS T3 ("
+ + "j INT, k INT, "
+ + "a INT, "
+ + "b Decimal(4,2), "
+ + "c TINYINT,"
+ + "d SMALLINT,"
+ + "e BIGINT,"
+ + "f FLOAT,"
+ + "h DOUBLE,"
+ + "i DATE,"
+ + "l TIMESTAMP,"
+ + "m CHAR(1),"
+ + "n VARCHAR,"
+ + "PRIMARY KEY (j,k) NOT ENFORCED)"
+ + " WITH ('merge-engine'='aggregation', "
+ + "'fields.a.aggregate-function'='min', "
+ + "'fields.b.aggregate-function'='min', "
+ + "'fields.c.aggregate-function'='min', "
+ + "'fields.d.aggregate-function'='min', "
+ + "'fields.e.aggregate-function'='min', "
+ + "'fields.f.aggregate-function'='min',"
+ + "'fields.h.aggregate-function'='min',"
+ + "'fields.i.aggregate-function'='min',"
+ + "'fields.l.aggregate-function'='min',"
+ + "'fields.m.aggregate-function'='min',"
+ + "'fields.n.aggregate-function'='min'"
+ + ");");
+ }
+
+ @Test
+ public void testMergeInMemory() {
+ batchSql(
+ "INSERT INTO T3 VALUES "
+ + "(1, 2, CAST(NULL AS INT), 1.01, CAST(-1 AS TINYINT), CAST(-1 AS SMALLINT), "
+ + "CAST(1000 AS BIGINT), 1.11, CAST(1.11 AS DOUBLE), CAST('2020-01-01' AS DATE), "
+ + "CAST('2021-01-01 01:01:01' AS TIMESTAMP), 'a', 'aaa'),"
+ + "(1, 2, 2, 1.10, CAST(2 AS TINYINT), CAST(2 AS SMALLINT), "
+ + "CAST(100000 AS BIGINT), -1.11, CAST(1.21 AS DOUBLE), CAST('2020-01-02' AS DATE), "
+ + "CAST('2022-01-01 01:01:01' AS TIMESTAMP), 'b', 'bbb'), "
+ + "(1, 2, 3, 10.00, CAST(1 AS TINYINT), CAST(1 AS SMALLINT), "
+ + "CAST(10000000 AS BIGINT), 0, CAST(-1.11 AS DOUBLE), CAST('2022-01-02' AS DATE), "
+ + "CAST('2022-01-01 02:00:00' AS TIMESTAMP), 'c', 'ccc')");
+ List<Row> result = batchSql("SELECT * FROM T3");
+ assertThat(result)
+ .containsExactlyInAnyOrder(
+ Row.of(
+ 1,
+ 2,
+ 2,
+ new BigDecimal("1.01"),
+ (byte) -1,
+ (short) -1,
+ (long) 1000,
+ (float) -1.11,
+ -1.11,
+ LocalDate.of(2020, 1, 1),
+ LocalDateTime.of(2021, 1, 1, 1, 1, 1),
+ "a",
+ "aaa"));
+ }
+
+ @Test
+ public void testMergeRead() {
+ batchSql(
+ "INSERT INTO T3 VALUES "
+ + "(1, 2, CAST(NULL AS INT), 1.01, CAST(1 AS TINYINT), CAST(-1 AS SMALLINT), CAST(1000 AS BIGINT), "
+ + "1.11, CAST(1.11 AS DOUBLE), CAST('2020-01-01' AS DATE), CAST('2021-01-01 01:01:01' AS TIMESTAMP), 'a', 'aaa')");
+ batchSql(
+ "INSERT INTO T3 VALUES "
+ + "(1, 2, 2, 1.10, CAST(2 AS TINYINT), CAST(2 AS SMALLINT), CAST(100000 AS BIGINT), "
+ + "-1.11, CAST(1.21 AS DOUBLE), CAST('2020-01-02' AS DATE), CAST('2022-01-01 01:01:01' AS TIMESTAMP), 'b', 'bbb')");
+ batchSql(
+ "INSERT INTO T3 VALUES "
+ + "(1, 2, 3, 10.00, CAST(-1 AS TINYINT), CAST(1 AS SMALLINT), CAST(10000000 AS BIGINT), "
+ + "0, CAST(-1.11 AS DOUBLE), CAST('2022-01-02' AS DATE), CAST('2022-01-01 02:00:00' AS TIMESTAMP), 'c', 'ccc')");
+
+ List<Row> result = batchSql("SELECT * FROM T3");
+ assertThat(result)
+ .containsExactlyInAnyOrder(
+ Row.of(
+ 1,
+ 2,
+ 2,
+ new BigDecimal("1.01"),
+ (byte) -1,
+ (short) -1,
+ (long) 1000,
+ (float) -1.11,
+ -1.11,
+ LocalDate.of(2020, 1, 1),
+ LocalDateTime.of(2021, 1, 1, 1, 1, 1),
+ "a",
+ "aaa"));
+ }
+
+ @Test
+ public void testMergeCompaction() {
+ // Wait compaction
+ batchSql("ALTER TABLE T3 SET ('commit.force-compact'='true')");
+
+ // key 1 2
+ batchSql(
+ "INSERT INTO T3 VALUES "
+ + "(1, 2, CAST(NULL AS INT), 1.01, CAST(1 AS TINYINT), CAST(-1 AS SMALLINT), CAST(1000 AS BIGINT), "
+ + "1.11, CAST(1.11 AS DOUBLE), CAST('2020-01-01' AS DATE), CAST('2021-01-01 01:01:01' AS TIMESTAMP), 'a', 'aaa')");
+ batchSql(
+ "INSERT INTO T3 VALUES "
+ + "(1, 2, 2, 1.10, CAST(2 AS TINYINT), CAST(2 AS SMALLINT), CAST(100000 AS BIGINT), "
+ + "-1.11, CAST(1.21 AS DOUBLE), CAST('2020-01-02' AS DATE), CAST('2022-01-01 01:01:01' AS TIMESTAMP), 'b', 'bbb')");
+ batchSql(
+ "INSERT INTO T3 VALUES "
+ + "(1, 2, 3, 10.00, CAST(-1 AS TINYINT), CAST(1 AS SMALLINT), CAST(10000000 AS BIGINT), "
+ + "0, CAST(-1.11 AS DOUBLE), CAST('2022-01-02' AS DATE), CAST('2022-01-01 02:00:00' AS TIMESTAMP), 'c', 'ccc')");
+
+ // key 1 3
+ batchSql(
+ "INSERT INTO T3 VALUES "
+ + "(1, 3, CAST(NULL AS INT), 1.01, CAST(1 AS TINYINT), CAST(-1 AS SMALLINT), CAST(1000 AS BIGINT), "
+ + "1.11, CAST(1.11 AS DOUBLE), CAST('2020-01-01' AS DATE), CAST('2021-01-01 01:01:01' AS TIMESTAMP), 'a', 'aaa')");
+ batchSql(
+ "INSERT INTO T3 VALUES "
+ + "(1, 3, 6, 1.10, CAST(2 AS TINYINT), CAST(2 AS SMALLINT), CAST(100000 AS BIGINT), "
+ + "-1.11, CAST(1.21 AS DOUBLE), CAST('2020-01-02' AS DATE), CAST('2022-01-01 01:01:01' AS TIMESTAMP), 'b', 'bbb')");
+ batchSql(
+ "INSERT INTO T3 VALUES "
+ + "(1, 3, 3, 10.00, CAST(-1 AS TINYINT), CAST(1 AS SMALLINT), CAST(10000000 AS BIGINT), "
+ + "0, CAST(-1.11 AS DOUBLE), CAST('2022-01-02' AS DATE), CAST('2022-01-01 02:00:00' AS TIMESTAMP), 'c', 'ccc')");
+
+ assertThat(batchSql("SELECT * FROM T3"))
+ .containsExactlyInAnyOrder(
+ Row.of(
+ 1,
+ 2,
+ 2,
+ new BigDecimal("1.01"),
+ (byte) -1,
+ (short) -1,
+ (long) 1000,
+ (float) -1.11,
+ -1.11,
+ LocalDate.of(2020, 1, 1),
+ LocalDateTime.of(2021, 1, 1, 1, 1, 1),
+ "a",
+ "aaa"),
+ Row.of(
+ 1,
+ 3,
+ 3,
+ new BigDecimal("1.01"),
+ (byte) -1,
+ (short) -1,
+ (long) 1000,
+ (float) -1.11,
+ -1.11,
+ LocalDate.of(2020, 1, 1),
+ LocalDateTime.of(2021, 1, 1, 1, 1, 1),
+ "a",
+ "aaa"));
+ }
+
+ @Test
+ public void testStreamingRead() {
+ assertThatThrownBy(
+ () -> sEnv.from("T3").execute().print(),
+ "Pre-aggregate continuous reading is not supported");
+ }
+}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/NestedUpdateAggregationITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/NestedUpdateAggregationITCase.java
new file mode 100644
index 0000000000..c25f23864f
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/NestedUpdateAggregationITCase.java
@@ -0,0 +1,332 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.aggregation;
+
+import org.apache.paimon.flink.CatalogITCaseBase;
+import org.apache.paimon.mergetree.compact.aggregate.FieldNestedUpdateAgg;
+
+import org.apache.flink.table.api.config.ExecutionConfigOptions;
+import org.apache.flink.types.Row;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Objects;
+import java.util.stream.Collectors;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** ITCase for {@link FieldNestedUpdateAgg}. */
+public class NestedUpdateAggregationITCase extends CatalogITCaseBase {
+
+ @Override
+ protected List<String> ddl() {
+ String ordersTable =
+ "CREATE TABLE orders (\n"
+ + " order_id INT PRIMARY KEY NOT ENFORCED,\n"
+ + " user_name STRING,\n"
+ + " address STRING\n"
+ + ");";
+
+ String subordersTable =
+ "CREATE TABLE sub_orders (\n"
+ + " order_id INT,\n"
+ + " daily_id INT,\n"
+ + " today STRING,\n"
+ + " product_name STRING,\n"
+ + " price BIGINT,\n"
+ + " PRIMARY KEY (order_id, daily_id, today) NOT ENFORCED\n"
+ + ");";
+
+ String wideTable =
+ "CREATE TABLE order_wide (\n"
+ + " order_id INT PRIMARY KEY NOT ENFORCED,\n"
+ + " user_name STRING,\n"
+ + " address STRING,\n"
+ + " sub_orders ARRAY<ROW<daily_id INT, today STRING, product_name STRING, price BIGINT>>\n"
+ + ") WITH (\n"
+ + " 'merge-engine' = 'aggregation',\n"
+ + " 'fields.sub_orders.aggregate-function' = 'nested_update',\n"
+ + " 'fields.sub_orders.nested-key' = 'daily_id,today',\n"
+ + " 'fields.sub_orders.ignore-retract' = 'true',"
+ + " 'fields.user_name.ignore-retract' = 'true',"
+ + " 'fields.address.ignore-retract' = 'true'"
+ + ")";
+
+ String wideAppendTable =
+ "CREATE TABLE order_append_wide (\n"
+ + " order_id INT PRIMARY KEY NOT ENFORCED,\n"
+ + " user_name STRING,\n"
+ + " address STRING,\n"
+ + " sub_orders ARRAY<ROW<daily_id INT, today STRING, product_name STRING, price BIGINT>>\n"
+ + ") WITH (\n"
+ + " 'merge-engine' = 'aggregation',\n"
+ + " 'fields.sub_orders.aggregate-function' = 'nested_update',\n"
+ + " 'fields.sub_orders.ignore-retract' = 'true',"
+ + " 'fields.user_name.ignore-retract' = 'true',"
+ + " 'fields.address.ignore-retract' = 'true'"
+ + ")";
+
+ return Arrays.asList(ordersTable, subordersTable, wideTable, wideAppendTable);
+ }
+
+ @Test
+ public void testUseCase() {
+ sql(
+ "INSERT INTO orders VALUES "
+ + "(1, 'Wang', 'HangZhou'),"
+ + "(2, 'Zhao', 'ChengDu'),"
+ + "(3, 'Liu', 'NanJing')");
+
+ sql(
+ "INSERT INTO sub_orders VALUES "
+ + "(1, 1, '12-20', 'Apple', 8000),"
+ + "(1, 2, '12-20', 'Tesla', 400000),"
+ + "(1, 1, '12-21', 'Sangsung', 5000),"
+ + "(2, 1, '12-20', 'Tea', 40),"
+ + "(2, 2, '12-20', 'Pot', 60),"
+ + "(3, 1, '12-25', 'Bat', 15),"
+ + "(3, 1, '12-26', 'Cup', 30)");
+
+ sql(widenSql());
+
+ List<Row> result =
+ sql("SELECT * FROM order_wide").stream()
+ .sorted(Comparator.comparingInt(r -> r.getFieldAs(0)))
+ .collect(Collectors.toList());
+
+ assertThat(
+ checkOneRecord(
+ result.get(0),
+ 1,
+ "Wang",
+ "HangZhou",
+ Row.of(1, "12-20", "Apple", 8000L),
+ Row.of(1, "12-21", "Sangsung", 5000L),
+ Row.of(2, "12-20", "Tesla", 400000L)))
+ .isTrue();
+ assertThat(
+ checkOneRecord(
+ result.get(1),
+ 2,
+ "Zhao",
+ "ChengDu",
+ Row.of(1, "12-20", "Tea", 40L),
+ Row.of(2, "12-20", "Pot", 60L)))
+ .isTrue();
+ assertThat(
+ checkOneRecord(
+ result.get(2),
+ 3,
+ "Liu",
+ "NanJing",
+ Row.of(1, "12-25", "Bat", 15L),
+ Row.of(1, "12-26", "Cup", 30L)))
+ .isTrue();
+
+ // query using UNNEST
+ List<Row> unnested =
+ sql(
+ "SELECT order_id, user_name, address, daily_id, today, product_name, price "
+ + "FROM order_wide, UNNEST(sub_orders) AS so(daily_id, today, product_name, price)");
+
+ assertThat(unnested)
+ .containsExactlyInAnyOrder(
+ Row.of(1, "Wang", "HangZhou", 1, "12-20", "Apple", 8000L),
+ Row.of(1, "Wang", "HangZhou", 2, "12-20", "Tesla", 400000L),
+ Row.of(1, "Wang", "HangZhou", 1, "12-21", "Sangsung", 5000L),
+ Row.of(2, "Zhao", "ChengDu", 1, "12-20", "Tea", 40L),
+ Row.of(2, "Zhao", "ChengDu", 2, "12-20", "Pot", 60L),
+ Row.of(3, "Liu", "NanJing", 1, "12-25", "Bat", 15L),
+ Row.of(3, "Liu", "NanJing", 1, "12-26", "Cup", 30L));
+ }
+
+ @Test
+ public void testUseCaseWithNullValue() {
+ sql(
+ "INSERT INTO order_wide\n"
+ + "SELECT 6, CAST (NULL AS STRING), CAST (NULL AS STRING), "
+ + "ARRAY[cast(null as ROW<daily_id INT, today STRING, product_name STRING, price BIGINT>)]");
+
+ List<Row> result =
+ sql("SELECT * FROM order_wide").stream()
+ .sorted(Comparator.comparingInt(r -> r.getFieldAs(0)))
+ .collect(Collectors.toList());
+
+ assertThat(checkOneRecord(result.get(0), 6, null, null, (Row) null)).isTrue();
+
+ sql(
+ "INSERT INTO order_wide\n"
+ + "SELECT 6, 'Sun', CAST (NULL AS STRING), "
+ + "ARRAY[ROW(1, '01-01','Apple', 6999)]");
+
+ result =
+ sql("SELECT * FROM order_wide").stream()
+ .sorted(Comparator.comparingInt(r -> r.getFieldAs(0)))
+ .collect(Collectors.toList());
+ assertThat(
+ checkOneRecord(
+ result.get(0), 6, "Sun", null, Row.of(1, "01-01", "Apple", 6999L)))
+ .isTrue();
+ }
+
+ @Test
+ public void testUseCaseAppend() {
+ sql(
+ "INSERT INTO orders VALUES "
+ + "(1, 'Wang', 'HangZhou'),"
+ + "(2, 'Zhao', 'ChengDu'),"
+ + "(3, 'Liu', 'NanJing')");
+
+ sql(
+ "INSERT INTO sub_orders VALUES "
+ + "(1, 1, '12-20', 'Apple', 8000),"
+ + "(2, 1, '12-20', 'Tesla', 400000),"
+ + "(3, 1, '12-25', 'Bat', 15),"
+ + "(3, 1, '12-26', 'Cup', 30)");
+
+ sql(widenAppendSql());
+
+ // query using UNNEST
+ List<Row> unnested =
+ sql(
+ "SELECT order_id, user_name, address, daily_id, today, product_name, price "
+ + "FROM order_append_wide, UNNEST(sub_orders) AS so(daily_id, today, product_name, price)");
+
+ assertThat(unnested)
+ .containsExactlyInAnyOrder(
+ Row.of(1, "Wang", "HangZhou", 1, "12-20", "Apple", 8000L),
+ Row.of(2, "Zhao", "ChengDu", 1, "12-20", "Tesla", 400000L),
+ Row.of(3, "Liu", "NanJing", 1, "12-25", "Bat", 15L),
+ Row.of(3, "Liu", "NanJing", 1, "12-26", "Cup", 30L));
+ }
+
+ @Test
+ @Timeout(60)
+ public void testUpdateWithIgnoreRetract() throws Exception {
+ sEnv.getConfig()
+ .set(
+ ExecutionConfigOptions.TABLE_EXEC_SINK_UPSERT_MATERIALIZE,
+ ExecutionConfigOptions.UpsertMaterialize.NONE);
+
+ sql("INSERT INTO orders VALUES (1, 'Wang', 'HangZhou')");
+
+ sql(
+ "INSERT INTO sub_orders VALUES "
+ + "(1, 1, '12-20', 'Apple', 8000),"
+ + "(1, 2, '12-20', 'Tesla', 400000),"
+ + "(1, 1, '12-21', 'Sangsung', 5000)");
+
+ sEnv.executeSql(widenSql());
+
+ boolean checkResult;
+ List<Row> result;
+
+ do {
+ Thread.sleep(500);
+ result = sql("SELECT * FROM order_wide");
+ checkResult =
+ !result.isEmpty()
+ && checkOneRecord(
+ result.get(0),
+ 1,
+ "Wang",
+ "HangZhou",
+ Row.of(1, "12-20", "Apple", 8000L),
+ Row.of(1, "12-21", "Sangsung", 5000L),
+ Row.of(2, "12-20", "Tesla", 400000L));
+ } while (!checkResult);
+
+ sql("INSERT INTO sub_orders VALUES (1, 2, '12-20', 'Benz', 380000)");
+
+ do {
+ Thread.sleep(500);
+ result = sql("SELECT * FROM order_wide");
+ checkResult =
+ !result.isEmpty()
+ && checkOneRecord(
+ result.get(0),
+ 1,
+ "Wang",
+ "HangZhou",
+ Row.of(1, "12-20", "Apple", 8000L),
+ Row.of(1, "12-21", "Sangsung", 5000L),
+ Row.of(2, "12-20", "Benz", 380000L));
+ } while (!checkResult);
+ }
+
+ private String widenSql() {
+ return "INSERT INTO order_wide\n"
+ + "SELECT order_id, user_name, address, "
+ + "CAST (NULL AS ARRAY<ROW<daily_id INT, today STRING, product_name STRING, price BIGINT>>) FROM orders\n"
+ + "UNION ALL\n"
+ + "SELECT order_id, CAST (NULL AS STRING), CAST (NULL AS STRING), "
+ + "ARRAY[ROW(daily_id, today, product_name, price)] FROM sub_orders";
+ }
+
+ private String widenAppendSql() {
+ return "INSERT INTO order_append_wide\n"
+ + "SELECT order_id, user_name, address, "
+ + "CAST (NULL AS ARRAY<ROW<daily_id INT, today STRING, product_name STRING, price BIGINT>>) FROM orders\n"
+ + "UNION ALL\n"
+ + "SELECT order_id, CAST (NULL AS STRING), CAST (NULL AS STRING), "
+ + "ARRAY[ROW(daily_id, today, product_name, price)] FROM sub_orders";
+ }
+
+ private boolean checkOneRecord(
+ Row record, int orderId, String userName, String address, Row... subOrders) {
+ if ((int) record.getField(0) != orderId) {
+ return false;
+ }
+ if (!Objects.equals(record.getFieldAs(1), userName)) {
+ return false;
+ }
+ if (!Objects.equals(record.getFieldAs(2), address)) {
+ return false;
+ }
+
+ return checkNestedTable(record.getFieldAs(3), subOrders);
+ }
+
+ private boolean checkNestedTable(Row[] nestedTable, Row... subOrders) {
+ if (nestedTable.length != subOrders.length) {
+ return false;
+ }
+
+ Comparator<Row> comparator =
+ (Comparator)
+ Comparator.comparingInt(r -> ((Row) r).getFieldAs(0))
+ .thenComparing(r -> (String) ((Row) r).getField(1));
+
+ List<Row> sortedActual =
+ Arrays.stream(nestedTable).sorted(comparator).collect(Collectors.toList());
+ List<Row> sortedExpected =
+ Arrays.stream(subOrders).sorted(comparator).collect(Collectors.toList());
+
+ for (int i = 0; i < sortedActual.size(); i++) {
+ if (!Objects.equals(sortedActual.get(i), sortedExpected.get(i))) {
+ return false;
+ }
+ }
+
+ return true;
+ }
+}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/NotExistAggregationFunctionITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/NotExistAggregationFunctionITCase.java
new file mode 100644
index 0000000000..f5a92cf5ee
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/NotExistAggregationFunctionITCase.java
@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.aggregation;
+
+import org.apache.paimon.flink.CatalogITCaseBase;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** ITCase for testing the aggregation merge engine with not exist aggregation function. */
+public class NotExistAggregationFunctionITCase extends CatalogITCaseBase {
+
+ @Test
+ public void testNotExistAggregationFunction() {
+ assertThatThrownBy(
+ () ->
+ sql(
+ "CREATE TABLE test_not_exist("
+ + " id INT PRIMARY KEY NOT ENFORCED,"
+ + " f0 INT"
+ + ") WITH ("
+ + " 'merge-engine' = 'aggregation',"
+ + " 'fields.f0.aggregate-function' = 'not_exist'"
+ + ")"))
+ .rootCause()
+ .hasMessageContaining("Could not find any factory for identifier 'not_exist'");
+ }
+}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/ProductAggregationITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/ProductAggregationITCase.java
new file mode 100644
index 0000000000..d9af0a6165
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/ProductAggregationITCase.java
@@ -0,0 +1,184 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.aggregation;
+
+import org.apache.paimon.flink.CatalogITCaseBase;
+
+import org.apache.flink.types.Row;
+import org.junit.jupiter.api.Test;
+
+import java.math.BigDecimal;
+import java.util.Collections;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** ITCase for product aggregate function. */
+public class ProductAggregationITCase extends CatalogITCaseBase {
+ @Override
+ protected List<String> ddl() {
+ return Collections.singletonList(
+ "CREATE TABLE IF NOT EXISTS T1 ("
+ + "j INT, k INT, "
+ + "a INT, "
+ + "b Decimal(4,2), "
+ + "c TINYINT,"
+ + "d SMALLINT,"
+ + "e BIGINT,"
+ + "f FLOAT,"
+ + "h DOUBLE,"
+ + "PRIMARY KEY (j,k) NOT ENFORCED)"
+ + " WITH ('merge-engine'='aggregation', "
+ + "'fields.a.aggregate-function'='product', "
+ + "'fields.b.aggregate-function'='product', "
+ + "'fields.c.aggregate-function'='product', "
+ + "'fields.d.aggregate-function'='product', "
+ + "'fields.e.aggregate-function'='product', "
+ + "'fields.f.aggregate-function'='product',"
+ + "'fields.h.aggregate-function'='product'"
+ + ");");
+ }
+
+ @Test
+ public void testMergeInMemory() {
+ batchSql("ALTER TABLE T1 MODIFY b DECIMAL(5, 3)");
+
+ batchSql(
+ "INSERT INTO T1 VALUES "
+ + "(1, 2, CAST(NULL AS INT), 1.01, CAST(1 AS TINYINT), CAST(-1 AS SMALLINT), "
+ + "CAST(1000 AS BIGINT), 1.11, CAST(1.11 AS DOUBLE)),"
+ + "(1, 2, 2, 1.10, CAST(2 AS TINYINT), CAST(2 AS SMALLINT), "
+ + "CAST(100000 AS BIGINT), -1.11, CAST(1.11 AS DOUBLE)), "
+ + "(1, 2, 3, 10.00, CAST(1 AS TINYINT), CAST(1 AS SMALLINT), "
+ + "CAST(10000000 AS BIGINT), 0, CAST(-1.11 AS DOUBLE))");
+ List<Row> result = batchSql("SELECT * FROM T1");
+ assertThat(result)
+ .containsExactlyInAnyOrder(
+ Row.of(
+ 1,
+ 2,
+ 6,
+ new BigDecimal("11.110"),
+ (byte) 2,
+ (short) -2,
+ 1000000000000000L,
+ (float) -0.0,
+ -1.3676310000000003));
+
+ // projection
+ assertThat(batchSql("SELECT f,e FROM T1"))
+ .containsExactlyInAnyOrder(Row.of((float) -0.0, 1000000000000000L));
+ }
+
+ @Test
+ public void testMergeRead() {
+ batchSql(
+ "INSERT INTO T1 VALUES "
+ + "(1, 2, 1, 1.01, CAST(1 AS TINYINT), CAST(-1 AS SMALLINT), CAST(1000 AS BIGINT), "
+ + "1.11, CAST(1.11 AS DOUBLE))");
+ batchSql(
+ "INSERT INTO T1 VALUES "
+ + "(1, 2, 2, 1.10, CAST(2 AS TINYINT), CAST(2 AS SMALLINT), CAST(100000 AS BIGINT), "
+ + "CAST(NULL AS FLOAT), CAST(1.11 AS DOUBLE))");
+ batchSql(
+ "INSERT INTO T1 VALUES "
+ + "(1, 2, 3, 10.00, CAST(1 AS TINYINT), CAST(1 AS SMALLINT), CAST(10000000 AS BIGINT), "
+ + "-1.11, CAST(-1.11 AS DOUBLE))");
+
+ List<Row> result = batchSql("SELECT * FROM T1");
+ assertThat(result)
+ .containsExactlyInAnyOrder(
+ Row.of(
+ 1,
+ 2,
+ 6,
+ new BigDecimal("11.10"),
+ (byte) 2,
+ (short) -2,
+ 1000000000000000L,
+ (float) -1.2321,
+ -1.3676310000000003));
+ }
+
+ @Test
+ public void testMergeCompaction() {
+ // Wait compaction
+ batchSql("ALTER TABLE T1 SET ('commit.force-compact'='true')");
+
+ // key 1 2
+ batchSql(
+ "INSERT INTO T1 VALUES "
+ + "(1, 2, 1, 1.01, CAST(1 AS TINYINT), CAST(-1 AS SMALLINT), CAST(1000 AS BIGINT), "
+ + "1.11, CAST(1.11 AS DOUBLE))");
+ batchSql(
+ "INSERT INTO T1 VALUES "
+ + "(1, 2, 2, 1.10, CAST(2 AS TINYINT), CAST(2 AS SMALLINT), CAST(100000 AS BIGINT), "
+ + "CAST(NULL AS FLOAT), CAST(1.11 AS DOUBLE))");
+ batchSql(
+ "INSERT INTO T1 VALUES "
+ + "(1, 2, 3, 10.00, CAST(1 AS TINYINT), CAST(1 AS SMALLINT), CAST(10000000 AS BIGINT), "
+ + "-1.11, CAST(-1.11 AS DOUBLE))");
+
+ // key 1 3
+ batchSql(
+ "INSERT INTO T1 VALUES "
+ + "(1, 3, 2, 1.01, CAST(1 AS TINYINT), CAST(-1 AS SMALLINT), CAST(1000 AS BIGINT), "
+ + "1.11, CAST(1.11 AS DOUBLE))");
+ batchSql(
+ "INSERT INTO T1 VALUES "
+ + "(1, 3, 2, 1.10, CAST(2 AS TINYINT), CAST(2 AS SMALLINT), CAST(100000 AS BIGINT), "
+ + "CAST(NULL AS FLOAT), CAST(1.11 AS DOUBLE))");
+ batchSql(
+ "INSERT INTO T1 VALUES "
+ + "(1, 3, 3, 10.00, CAST(1 AS TINYINT), CAST(1 AS SMALLINT), CAST(10000000 AS BIGINT), "
+ + "-1.11, CAST(-1.11 AS DOUBLE))");
+
+ List<Row> result = batchSql("SELECT * FROM T1");
+ assertThat(result)
+ .containsExactlyInAnyOrder(
+ Row.of(
+ 1,
+ 2,
+ 6,
+ new BigDecimal("11.10"),
+ (byte) 2,
+ (short) -2,
+ 1000000000000000L,
+ (float) -1.2321,
+ -1.3676310000000003),
+ Row.of(
+ 1,
+ 3,
+ 12,
+ new BigDecimal("11.10"),
+ (byte) 2,
+ (short) -2,
+ 1000000000000000L,
+ (float) -1.2321,
+ -1.3676310000000003));
+ }
+
+ @Test
+ public void testStreamingRead() {
+ assertThatThrownBy(
+ () -> sEnv.from("T1").execute().print(),
+ "Pre-aggregate continuous reading is not supported");
+ }
+}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/RoaringBitmapAggAggregationITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/RoaringBitmapAggAggregationITCase.java
new file mode 100644
index 0000000000..a469a5b2d6
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/RoaringBitmapAggAggregationITCase.java
@@ -0,0 +1,156 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.aggregation;
+
+import org.apache.paimon.flink.CatalogITCaseBase;
+import org.apache.paimon.utils.RoaringBitmap32;
+import org.apache.paimon.utils.RoaringBitmap64;
+
+import org.apache.commons.codec.binary.Hex;
+import org.apache.flink.types.Row;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * ITCase for {@link org.apache.paimon.mergetree.compact.aggregate.FieldRoaringBitmap32Agg} & {@link
+ * org.apache.paimon.mergetree.compact.aggregate.FieldRoaringBitmap64Agg}.
+ */
+public class RoaringBitmapAggAggregationITCase extends CatalogITCaseBase {
+
+ @Test
+ public void testRoaring32BitmapAgg() throws IOException {
+ sql(
+ "CREATE TABLE test_rbm64("
+ + " id INT PRIMARY KEY NOT ENFORCED,"
+ + " f0 VARBINARY"
+ + ") WITH ("
+ + " 'merge-engine' = 'aggregation',"
+ + " 'fields.f0.aggregate-function' = 'rbm32'"
+ + ")");
+
+ byte[] v1Bytes = RoaringBitmap32.bitmapOf(1).serialize();
+ byte[] v2Bytes = RoaringBitmap32.bitmapOf(2).serialize();
+ byte[] v3Bytes = RoaringBitmap32.bitmapOf(3).serialize();
+ byte[] v4Bytes = RoaringBitmap32.bitmapOf(1, 2).serialize();
+ byte[] v5Bytes = RoaringBitmap32.bitmapOf(2, 3).serialize();
+ String v1 = Hex.encodeHexString(v1Bytes).toUpperCase();
+ String v2 = Hex.encodeHexString(v2Bytes).toUpperCase();
+ String v3 = Hex.encodeHexString(v3Bytes).toUpperCase();
+
+ sql(
+ "INSERT INTO test_rbm64 VALUES "
+ + "(1, CAST (NULL AS VARBINARY)), "
+ + "(2, CAST (x'"
+ + v1
+ + "' AS VARBINARY)), "
+ + "(3, CAST (x'"
+ + v2
+ + "' AS VARBINARY))");
+
+ List<Row> result = queryAndSort("SELECT * FROM test_rbm64");
+ checkOneRecord(result.get(0), 1, null);
+ checkOneRecord(result.get(1), 2, v1Bytes);
+ checkOneRecord(result.get(2), 3, v2Bytes);
+
+ sql(
+ "INSERT INTO test_rbm64 VALUES "
+ + "(1, CAST (x'"
+ + v1
+ + "' AS VARBINARY)), "
+ + "(2, CAST (x'"
+ + v2
+ + "' AS VARBINARY)), "
+ + "(2, CAST (x'"
+ + v2
+ + "' AS VARBINARY)), "
+ + "(3, CAST (x'"
+ + v3
+ + "' AS VARBINARY))");
+
+ result = queryAndSort("SELECT * FROM test_rbm64");
+ checkOneRecord(result.get(0), 1, v1Bytes);
+ checkOneRecord(result.get(1), 2, v4Bytes);
+ checkOneRecord(result.get(2), 3, v5Bytes);
+ }
+
+ @Test
+ public void testRoaring64BitmapAgg() throws IOException {
+ sql(
+ "CREATE TABLE test_rbm64("
+ + " id INT PRIMARY KEY NOT ENFORCED,"
+ + " f0 VARBINARY"
+ + ") WITH ("
+ + " 'merge-engine' = 'aggregation',"
+ + " 'fields.f0.aggregate-function' = 'rbm64'"
+ + ")");
+
+ byte[] v1Bytes = RoaringBitmap64.bitmapOf(1L).serialize();
+ byte[] v2Bytes = RoaringBitmap64.bitmapOf(2L).serialize();
+ byte[] v3Bytes = RoaringBitmap64.bitmapOf(3L).serialize();
+ byte[] v4Bytes = RoaringBitmap64.bitmapOf(1L, 2L).serialize();
+ byte[] v5Bytes = RoaringBitmap64.bitmapOf(2L, 3L).serialize();
+ String v1 = Hex.encodeHexString(v1Bytes).toUpperCase();
+ String v2 = Hex.encodeHexString(v2Bytes).toUpperCase();
+ String v3 = Hex.encodeHexString(v3Bytes).toUpperCase();
+
+ sql(
+ "INSERT INTO test_rbm64 VALUES "
+ + "(1, CAST (NULL AS VARBINARY)), "
+ + "(2, CAST (x'"
+ + v1
+ + "' AS VARBINARY)), "
+ + "(3, CAST (x'"
+ + v2
+ + "' AS VARBINARY))");
+
+ List<Row> result = queryAndSort("SELECT * FROM test_rbm64");
+ checkOneRecord(result.get(0), 1, null);
+ checkOneRecord(result.get(1), 2, v1Bytes);
+ checkOneRecord(result.get(2), 3, v2Bytes);
+
+ sql(
+ "INSERT INTO test_rbm64 VALUES "
+ + "(1, CAST (x'"
+ + v1
+ + "' AS VARBINARY)), "
+ + "(2, CAST (x'"
+ + v2
+ + "' AS VARBINARY)), "
+ + "(2, CAST (x'"
+ + v2
+ + "' AS VARBINARY)), "
+ + "(3, CAST (x'"
+ + v3
+ + "' AS VARBINARY))");
+
+ result = queryAndSort("SELECT * FROM test_rbm64");
+ checkOneRecord(result.get(0), 1, v1Bytes);
+ checkOneRecord(result.get(1), 2, v4Bytes);
+ checkOneRecord(result.get(2), 3, v5Bytes);
+ }
+
+ private void checkOneRecord(Row row, int id, byte[] expected) {
+ assertThat(row.getField(0)).isEqualTo(id);
+ assertThat(row.getField(1)).isEqualTo(expected);
+ }
+}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/SumAggregationITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/SumAggregationITCase.java
new file mode 100644
index 0000000000..55f4593ed3
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/SumAggregationITCase.java
@@ -0,0 +1,180 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.aggregation;
+
+import org.apache.paimon.flink.CatalogITCaseBase;
+
+import org.apache.flink.types.Row;
+import org.junit.jupiter.api.Test;
+
+import java.math.BigDecimal;
+import java.util.Collections;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** ITCase for sum aggregate function. */
+public class SumAggregationITCase extends CatalogITCaseBase {
+ @Override
+ protected List<String> ddl() {
+ return Collections.singletonList(
+ "CREATE TABLE IF NOT EXISTS T1 ("
+ + "j INT, k INT, "
+ + "a INT, "
+ + "b Decimal(4,2), "
+ + "c TINYINT,"
+ + "d SMALLINT,"
+ + "e BIGINT,"
+ + "f FLOAT,"
+ + "h DOUBLE,"
+ + "PRIMARY KEY (j,k) NOT ENFORCED)"
+ + " WITH ('merge-engine'='aggregation', "
+ + "'fields.a.aggregate-function'='sum', "
+ + "'fields.b.aggregate-function'='sum', "
+ + "'fields.c.aggregate-function'='sum', "
+ + "'fields.d.aggregate-function'='sum', "
+ + "'fields.e.aggregate-function'='sum', "
+ + "'fields.f.aggregate-function'='sum',"
+ + "'fields.h.aggregate-function'='sum'"
+ + ");");
+ }
+
+ @Test
+ public void testMergeInMemory() {
+ batchSql(
+ "INSERT INTO T1 VALUES "
+ + "(1, 2, CAST(NULL AS INT), 1.01, CAST(1 AS TINYINT), CAST(-1 AS SMALLINT), "
+ + "CAST(1000 AS BIGINT), 1.11, CAST(1.11 AS DOUBLE)),"
+ + "(1, 2, 2, 1.10, CAST(2 AS TINYINT), CAST(2 AS SMALLINT), "
+ + "CAST(100000 AS BIGINT), -1.11, CAST(1.11 AS DOUBLE)), "
+ + "(1, 2, 3, 10.00, CAST(1 AS TINYINT), CAST(1 AS SMALLINT), "
+ + "CAST(10000000 AS BIGINT), 0, CAST(-1.11 AS DOUBLE))");
+ assertThat(batchSql("SELECT * FROM T1"))
+ .containsExactlyInAnyOrder(
+ Row.of(
+ 1,
+ 2,
+ 5,
+ new BigDecimal("12.11"),
+ (byte) 4,
+ (short) 2,
+ (long) 10101000,
+ (float) 0,
+ 1.11));
+
+ // projection
+ assertThat(batchSql("SELECT f,e FROM T1"))
+ .containsExactlyInAnyOrder(Row.of((float) 0, (long) 10101000));
+ }
+
+ @Test
+ public void testMergeRead() {
+ batchSql(
+ "INSERT INTO T1 VALUES "
+ + "(1, 2, 1, 1.01, CAST(1 AS TINYINT), CAST(-1 AS SMALLINT), CAST(1000 AS BIGINT), "
+ + "1.11, CAST(1.11 AS DOUBLE))");
+ batchSql(
+ "INSERT INTO T1 VALUES "
+ + "(1, 2, 2, 1.10, CAST(2 AS TINYINT), CAST(2 AS SMALLINT), CAST(100000 AS BIGINT), "
+ + "CAST(NULL AS FLOAT), CAST(1.11 AS DOUBLE))");
+ batchSql(
+ "INSERT INTO T1 VALUES "
+ + "(1, 2, 3, 10.00, CAST(1 AS TINYINT), CAST(1 AS SMALLINT), CAST(10000000 AS BIGINT), "
+ + "-1.11, CAST(-1.11 AS DOUBLE))");
+
+ List<Row> result = batchSql("SELECT * FROM T1");
+ assertThat(result)
+ .containsExactlyInAnyOrder(
+ Row.of(
+ 1,
+ 2,
+ 6,
+ new BigDecimal("12.11"),
+ (byte) 4,
+ (short) 2,
+ (long) 10101000,
+ (float) 0,
+ 1.11));
+ }
+
+ @Test
+ public void testMergeCompaction() {
+ // Wait compaction
+ batchSql("ALTER TABLE T1 SET ('commit.force-compact'='true')");
+
+ // key 1 2
+ batchSql(
+ "INSERT INTO T1 VALUES "
+ + "(1, 2, 1, 1.01, CAST(1 AS TINYINT), CAST(-1 AS SMALLINT), CAST(1000 AS BIGINT), "
+ + "1.11, CAST(1.11 AS DOUBLE))");
+ batchSql(
+ "INSERT INTO T1 VALUES "
+ + "(1, 2, 2, 1.10, CAST(2 AS TINYINT), CAST(2 AS SMALLINT), CAST(100000 AS BIGINT), "
+ + "CAST(NULL AS FLOAT), CAST(1.11 AS DOUBLE))");
+ batchSql(
+ "INSERT INTO T1 VALUES "
+ + "(1, 2, 3, 10.00, CAST(1 AS TINYINT), CAST(1 AS SMALLINT), CAST(10000000 AS BIGINT), "
+ + "-1.11, CAST(-1.11 AS DOUBLE))");
+
+ // key 1 3
+ batchSql(
+ "INSERT INTO T1 VALUES "
+ + "(1, 3, 2, 1.01, CAST(1 AS TINYINT), CAST(-1 AS SMALLINT), CAST(1000 AS BIGINT), "
+ + "1.11, CAST(1.11 AS DOUBLE))");
+ batchSql(
+ "INSERT INTO T1 VALUES "
+ + "(1, 3, 2, 1.10, CAST(2 AS TINYINT), CAST(2 AS SMALLINT), CAST(100000 AS BIGINT), "
+ + "CAST(NULL AS FLOAT), CAST(1.11 AS DOUBLE))");
+ batchSql(
+ "INSERT INTO T1 VALUES "
+ + "(1, 3, 3, 10.00, CAST(1 AS TINYINT), CAST(1 AS SMALLINT), CAST(10000000 AS BIGINT), "
+ + "-1.11, CAST(-1.11 AS DOUBLE))");
+
+ assertThat(batchSql("SELECT * FROM T1"))
+ .containsExactlyInAnyOrder(
+ Row.of(
+ 1,
+ 2,
+ 6,
+ new BigDecimal("12.11"),
+ (byte) 4,
+ (short) 2,
+ (long) 10101000,
+ (float) 0,
+ 1.11),
+ Row.of(
+ 1,
+ 3,
+ 7,
+ new BigDecimal("12.11"),
+ (byte) 4,
+ (short) 2,
+ (long) 10101000,
+ (float) 0,
+ 1.11));
+ }
+
+ @Test
+ public void testStreamingRead() {
+ assertThatThrownBy(
+ () -> sEnv.from("T1").execute().print(),
+ "Pre-aggregate continuous reading is not supported");
+ }
+}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/SumRetractionAggregationITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/SumRetractionAggregationITCase.java
new file mode 100644
index 0000000000..5eb7a6731a
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/SumRetractionAggregationITCase.java
@@ -0,0 +1,74 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.aggregation;
+
+import org.apache.paimon.flink.CatalogITCaseBase;
+import org.apache.paimon.utils.BlockingIterator;
+
+import org.apache.flink.types.Row;
+import org.apache.flink.types.RowKind;
+import org.apache.flink.util.CloseableIterator;
+import org.junit.jupiter.api.Test;
+
+import java.math.BigDecimal;
+import java.util.Collections;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** ITCase for sum aggregate function retraction. */
+public class SumRetractionAggregationITCase extends CatalogITCaseBase {
+ @Override
+ protected List<String> ddl() {
+ return Collections.singletonList(
+ "CREATE TABLE T ("
+ + "k INT,"
+ + "b Decimal(12, 2),"
+ + "PRIMARY KEY (k) NOT ENFORCED)"
+ + " WITH ('merge-engine'='aggregation', "
+ + "'changelog-producer' = 'full-compaction',"
+ + "'fields.b.aggregate-function'='sum'"
+ + ");");
+ }
+
+ @Test
+ public void testRetraction() throws Exception {
+ sql("CREATE TABLE INPUT (" + "k INT," + "b INT," + "PRIMARY KEY (k) NOT ENFORCED);");
+ CloseableIterator<Row> insert =
+ streamSqlIter("INSERT INTO T SELECT k, SUM(b) FROM INPUT GROUP BY k;");
+ BlockingIterator<Row, Row> select = streamSqlBlockIter("SELECT * FROM T");
+
+ sql("INSERT INTO INPUT VALUES (1, 1), (2, 2)");
+ assertThat(select.collect(2))
+ .containsExactlyInAnyOrder(
+ Row.of(1, BigDecimal.valueOf(100, 2)),
+ Row.of(2, BigDecimal.valueOf(200, 2)));
+
+ sql("INSERT INTO INPUT VALUES (1, 3), (2, 4)");
+ assertThat(select.collect(4))
+ .containsExactlyInAnyOrder(
+ Row.ofKind(RowKind.UPDATE_BEFORE, 1, BigDecimal.valueOf(100, 2)),
+ Row.ofKind(RowKind.UPDATE_AFTER, 1, BigDecimal.valueOf(300, 2)),
+ Row.ofKind(RowKind.UPDATE_BEFORE, 2, BigDecimal.valueOf(200, 2)),
+ Row.ofKind(RowKind.UPDATE_AFTER, 2, BigDecimal.valueOf(400, 2)));
+
+ select.close();
+ insert.close();
+ }
+}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/ThetaSketchAggAggregationITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/ThetaSketchAggAggregationITCase.java
new file mode 100644
index 0000000000..763067e6ec
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/ThetaSketchAggAggregationITCase.java
@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.aggregation;
+
+import org.apache.paimon.flink.CatalogITCaseBase;
+
+import org.apache.commons.codec.binary.Hex;
+import org.apache.flink.types.Row;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+
+import static org.apache.paimon.utils.ThetaSketch.sketchOf;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** ITCase for {@link org.apache.paimon.mergetree.compact.aggregate.FieldThetaSketchAgg}. */
+public class ThetaSketchAggAggregationITCase extends CatalogITCaseBase {
+
+ @Test
+ public void testThetaSketchAgg() {
+ sql(
+ "CREATE TABLE test_collect("
+ + " id INT PRIMARY KEY NOT ENFORCED,"
+ + " f0 VARBINARY"
+ + ") WITH ("
+ + " 'merge-engine' = 'aggregation',"
+ + " 'fields.f0.aggregate-function' = 'theta_sketch'"
+ + ")");
+
+ String str1 = Hex.encodeHexString(sketchOf(1)).toUpperCase();
+ String str2 = Hex.encodeHexString(sketchOf(2)).toUpperCase();
+ String str3 = Hex.encodeHexString(sketchOf(3)).toUpperCase();
+
+ sql(
+ String.format(
+ "INSERT INTO test_collect VALUES (1, CAST (NULL AS VARBINARY)),(2, CAST(x'%s' AS VARBINARY)), (3, CAST(x'%s' AS VARBINARY))",
+ str1, str2));
+
+ List<Row> result = queryAndSort("SELECT * FROM test_collect");
+ checkOneRecord(result.get(0), 1, null);
+ checkOneRecord(result.get(1), 2, sketchOf(1));
+ checkOneRecord(result.get(2), 3, sketchOf(2));
+
+ sql(
+ String.format(
+ "INSERT INTO test_collect VALUES (1, CAST (x'%s' AS VARBINARY)),(2, CAST(x'%s' AS VARBINARY)), (2, CAST(x'%s' AS VARBINARY)), (3, CAST(x'%s' AS VARBINARY))",
+ str1, str2, str2, str3));
+
+ result = queryAndSort("SELECT * FROM test_collect");
+ checkOneRecord(result.get(0), 1, sketchOf(1));
+ checkOneRecord(result.get(1), 2, sketchOf(1, 2));
+ checkOneRecord(result.get(2), 3, sketchOf(2, 3));
+ }
+
+ private void checkOneRecord(Row row, int id, byte[] expected) {
+ assertThat(row.getField(0)).isEqualTo(id);
+ assertThat(row.getField(1)).isEqualTo(expected);
+ }
+}