You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by dw...@apache.org on 2023/12/07 12:58:59 UTC
(flink) 01/02: [FLINK-33480] Implement restore tests for GroupAggregate node
This is an automated email from the ASF dual-hosted git repository.
dwysakowicz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 193b1c68976cdfbd66147278f23d7d427d9b5562
Author: bvarghese1 <bv...@confluent.io>
AuthorDate: Tue Nov 7 17:03:53 2023 -0800
[FLINK-33480] Implement restore tests for GroupAggregate node
---
.../exec/stream/GroupAggregateRestoreTest.java | 46 ++
.../exec/stream/GroupAggregateTestPrograms.java | 374 +++++++++++++
.../plan/nodes/exec/testutils/RestoreTestBase.java | 1 +
.../plan/group-aggregate-distinct-mini-batch.json | 621 +++++++++++++++++++++
.../savepoint/_metadata | Bin 0 -> 25689 bytes
.../plan/group-aggregate-distinct.json | 359 ++++++++++++
.../group-aggregate-distinct/savepoint/_metadata | Bin 0 -> 27875 bytes
.../plan/group-aggregate-simple-mini-batch.json | 351 ++++++++++++
.../savepoint/_metadata | Bin 0 -> 9479 bytes
.../plan/group-aggregate-simple.json | 272 +++++++++
.../group-aggregate-simple/savepoint/_metadata | Bin 0 -> 10954 bytes
.../group-aggregate-udf-with-merge-mini-batch.json | 400 +++++++++++++
.../savepoint/_metadata | Bin 0 -> 17762 bytes
.../plan/group-aggregate-udf-with-merge.json | 211 +++++++
.../savepoint/_metadata | Bin 0 -> 19881 bytes
...oup-aggregate-udf-without-merge-mini-batch.json | 254 +++++++++
.../savepoint/_metadata | Bin 0 -> 12731 bytes
.../plan/group-aggregate-udf-without-merge.json | 231 ++++++++
.../savepoint/_metadata | Bin 0 -> 14618 bytes
19 files changed, 3120 insertions(+)
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupAggregateRestoreTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupAggregateRestoreTest.java
new file mode 100644
index 00000000000..2c5043d9f4d
--- /dev/null
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupAggregateRestoreTest.java
@@ -0,0 +1,46 @@
+/*
+ * 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.flink.table.planner.plan.nodes.exec.stream;
+
+import org.apache.flink.table.planner.plan.nodes.exec.testutils.RestoreTestBase;
+import org.apache.flink.table.test.program.TableTestProgram;
+
+import java.util.Arrays;
+import java.util.List;
+
+/** Restore tests for {@link StreamExecGroupAggregate}. */
+public class GroupAggregateRestoreTest extends RestoreTestBase {
+
+ public GroupAggregateRestoreTest() {
+ super(StreamExecGroupAggregate.class);
+ }
+
+ @Override
+ public List<TableTestProgram> programs() {
+ return Arrays.asList(
+ GroupAggregateTestPrograms.GROUP_BY_SIMPLE,
+ GroupAggregateTestPrograms.GROUP_BY_SIMPLE_MINI_BATCH,
+ GroupAggregateTestPrograms.GROUP_BY_DISTINCT,
+ GroupAggregateTestPrograms.GROUP_BY_DISTINCT_MINI_BATCH,
+ GroupAggregateTestPrograms.GROUP_BY_UDF_WITH_MERGE,
+ GroupAggregateTestPrograms.GROUP_BY_UDF_WITH_MERGE_MINI_BATCH,
+ GroupAggregateTestPrograms.GROUP_BY_UDF_WITHOUT_MERGE,
+ GroupAggregateTestPrograms.GROUP_BY_UDF_WITHOUT_MERGE_MINI_BATCH);
+ }
+}
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupAggregateTestPrograms.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupAggregateTestPrograms.java
new file mode 100644
index 00000000000..b6a61a72144
--- /dev/null
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupAggregateTestPrograms.java
@@ -0,0 +1,374 @@
+/*
+ * 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.flink.table.planner.plan.nodes.exec.stream;
+
+import org.apache.flink.table.api.config.ExecutionConfigOptions;
+import org.apache.flink.table.planner.plan.utils.JavaUserDefinedAggFunctions.ConcatDistinctAggFunction;
+import org.apache.flink.table.planner.plan.utils.JavaUserDefinedAggFunctions.VarSum1AggFunction;
+import org.apache.flink.table.planner.plan.utils.JavaUserDefinedAggFunctions.VarSum2AggFunction;
+import org.apache.flink.table.planner.plan.utils.JavaUserDefinedAggFunctions.WeightedAvg;
+import org.apache.flink.table.planner.plan.utils.JavaUserDefinedAggFunctions.WeightedAvgWithMerge;
+import org.apache.flink.table.test.program.SinkTestStep;
+import org.apache.flink.table.test.program.SourceTestStep;
+import org.apache.flink.table.test.program.TableTestProgram;
+import org.apache.flink.types.Row;
+
+import java.time.Duration;
+
+/** {@link TableTestProgram} definitions for testing {@link StreamExecGroupAggregate}. */
+public class GroupAggregateTestPrograms {
+
+ static final SourceTestStep SOURCE_ONE =
+ SourceTestStep.newBuilder("source_t")
+ .addSchema("a INT", "b BIGINT", "c VARCHAR")
+ .producedBeforeRestore(
+ Row.of(1, 1L, "Hi"),
+ Row.of(2, 2L, "Hello"),
+ Row.of(2, 2L, "Hello World"))
+ .producedAfterRestore(
+ Row.of(1, 1L, "Hi Again!"),
+ Row.of(2, 2L, "Hello Again!"),
+ Row.of(2, 2L, "Hello World Again!"))
+ .build();
+
+ static final SourceTestStep SOURCE_TWO =
+ SourceTestStep.newBuilder("source_t")
+ .addSchema("a INT", "b BIGINT", "c INT", "d VARCHAR", "e BIGINT")
+ .producedBeforeRestore(
+ Row.of(2, 3L, 2, "Hello World Like", 1L),
+ Row.of(3, 4L, 3, "Hello World Its nice", 2L),
+ Row.of(2, 2L, 1, "Hello World", 2L),
+ Row.of(1, 1L, 0, "Hello", 1L),
+ Row.of(5, 11L, 10, "GHI", 1L),
+ Row.of(3, 5L, 4, "ABC", 2L),
+ Row.of(4, 10L, 9, "FGH", 2L),
+ Row.of(4, 7L, 6, "CDE", 2L),
+ Row.of(5, 14L, 13, "JKL", 2L),
+ Row.of(4, 9L, 8, "EFG", 1L),
+ Row.of(5, 15L, 14, "KLM", 2L),
+ Row.of(5, 12L, 11, "HIJ", 3L),
+ Row.of(4, 8L, 7, "DEF", 1L),
+ Row.of(5, 13L, 12, "IJK", 3L),
+ Row.of(3, 6L, 5, "BCD", 3L))
+ .producedAfterRestore(
+ Row.of(1, 1L, 0, "Hello", 1L),
+ Row.of(3, 5L, 4, "ABC", 2L),
+ Row.of(4, 10L, 9, "FGH", 2L),
+ Row.of(4, 7L, 6, "CDE", 2L),
+ Row.of(7, 7L, 7, "MNO", 7L),
+ Row.of(3, 6L, 5, "BCD", 3L),
+ Row.of(7, 7L, 7, "XYZ", 7L))
+ .build();
+
+ static final TableTestProgram GROUP_BY_SIMPLE =
+ TableTestProgram.of(
+ "group-aggregate-simple", "validates basic aggregation using group by")
+ .setupTableSource(SOURCE_ONE)
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink_t")
+ .addSchema(
+ "b BIGINT",
+ "cnt BIGINT",
+ "avg_a DOUBLE",
+ "min_c VARCHAR",
+ "PRIMARY KEY (b) NOT ENFORCED")
+ .consumedBeforeRestore(
+ "+I[1, 1, null, Hi]",
+ "+I[2, 1, 2.0, Hello]",
+ "+U[2, 2, 2.0, Hello]")
+ .consumedAfterRestore(
+ "+U[1, 2, null, Hi]",
+ "+U[2, 3, 2.0, Hello]",
+ "+U[2, 4, 2.0, Hello]")
+ .build())
+ .runSql(
+ "INSERT INTO sink_t SELECT "
+ + "b, "
+ + "COUNT(*) AS cnt, "
+ + "AVG(a) FILTER (WHERE a > 1) AS avg_a, "
+ + "MIN(c) AS min_c "
+ + "FROM source_t GROUP BY b")
+ .build();
+
+ static final TableTestProgram GROUP_BY_SIMPLE_MINI_BATCH =
+ TableTestProgram.of(
+ "group-aggregate-simple-mini-batch",
+ "validates basic aggregation using group by with mini batch")
+ .setupTableSource(SOURCE_ONE)
+ .setupConfig(ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_ENABLED, true)
+ .setupConfig(
+ ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_ALLOW_LATENCY,
+ Duration.ofSeconds(10))
+ .setupConfig(ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_SIZE, 5L)
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink_t")
+ .addSchema(
+ GROUP_BY_SIMPLE
+ .getSetupSinkTestSteps()
+ .get(0)
+ .schemaComponents)
+ .consumedBeforeRestore(
+ "+I[1, 1, null, Hi]", "+I[2, 2, 2.0, Hello]")
+ .consumedAfterRestore(
+ "+U[1, 2, null, Hi]", "+U[2, 4, 2.0, Hello]")
+ .build())
+ .runSql(GroupAggregateTestPrograms.GROUP_BY_SIMPLE.getRunSqlTestStep().sql)
+ .build();
+
+ static final TableTestProgram GROUP_BY_DISTINCT =
+ TableTestProgram.of("group-aggregate-distinct", "validates group by distinct")
+ .setupTableSource(SOURCE_TWO)
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink_t")
+ .addSchema(
+ "e BIGINT",
+ "cnt_a1 BIGINT",
+ "cnt_a2 BIGINT",
+ "sum_a BIGINT",
+ "sum_b BIGINT",
+ "avg_b DOUBLE",
+ "cnt_d BIGINT",
+ "PRIMARY KEY (e) NOT ENFORCED")
+ .consumedBeforeRestore(
+ "+I[1, 0, 1, 2, 3, 3.0, 1]",
+ "+I[2, 0, 1, 3, 4, 4.0, 1]",
+ "+U[2, 0, 2, 5, 6, 3.0, 2]",
+ "+U[1, 0, 2, 3, 4, 2.0, 2]",
+ "+U[1, 1, 3, 8, 15, 5.0, 3]",
+ "+U[2, 0, 2, 5, 11, 3.0, 3]",
+ "+U[2, 0, 3, 9, 21, 5.0, 4]",
+ "+U[2, 0, 3, 9, 28, 5.0, 5]",
+ "+U[2, 1, 4, 14, 42, 7.0, 6]",
+ "+U[1, 1, 4, 12, 24, 6.0, 4]",
+ "+U[2, 1, 4, 14, 57, 8.0, 7]",
+ "+I[3, 1, 1, 5, 12, 12.0, 1]",
+ "+U[1, 1, 4, 12, 32, 6.0, 5]",
+ "+U[3, 1, 1, 5, 25, 12.0, 2]",
+ "+U[3, 1, 2, 8, 31, 10.0, 3]")
+ .consumedAfterRestore(
+ "+U[1, 1, 4, 12, 32, 5.0, 5]",
+ "+U[2, 1, 4, 14, 57, 7.0, 7]",
+ "+U[2, 1, 4, 14, 57, 8.0, 7]",
+ "+U[2, 1, 4, 14, 57, 7.0, 7]",
+ "+I[7, 0, 1, 7, 7, 7.0, 1]",
+ "+U[3, 1, 2, 8, 31, 9.0, 3]",
+ "+U[7, 0, 1, 7, 7, 7.0, 2]")
+ .build())
+ .runSql(
+ "INSERT INTO sink_t SELECT "
+ + "e, "
+ + "COUNT(DISTINCT a) FILTER (WHERE b > 10) AS cnt_a1, "
+ + "COUNT(DISTINCT a) AS cnt_a2, "
+ + "SUM(DISTINCT a) AS sum_a, "
+ + "SUM(DISTINCT b) AS sum_b, "
+ + "AVG(b) AS avg_b, "
+ + "COUNT(DISTINCT d) AS concat_d "
+ + "FROM source_t GROUP BY e")
+ .build();
+
+ static final TableTestProgram GROUP_BY_DISTINCT_MINI_BATCH =
+ TableTestProgram.of(
+ "group-aggregate-distinct-mini-batch",
+ "validates group by distinct with mini batch")
+ .setupConfig(ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_ENABLED, true)
+ .setupConfig(
+ ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_ALLOW_LATENCY,
+ Duration.ofSeconds(10))
+ .setupConfig(ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_SIZE, 5L)
+ .setupTableSource(SOURCE_TWO)
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink_t")
+ .addSchema(
+ GROUP_BY_DISTINCT
+ .getSetupSinkTestSteps()
+ .get(0)
+ .schemaComponents)
+ .consumedBeforeRestore(
+ "+I[3, 1, 2, 8, 31, 10.0, 3]",
+ "+I[2, 1, 4, 14, 42, 7.0, 6]",
+ "+I[1, 1, 4, 12, 24, 6.0, 4]",
+ "+U[2, 1, 4, 14, 57, 8.0, 7]",
+ "+U[1, 1, 4, 12, 32, 6.0, 5]")
+ .consumedAfterRestore(
+ "+U[3, 1, 2, 8, 31, 9.0, 3]",
+ "+U[2, 1, 4, 14, 57, 7.0, 7]",
+ "+I[7, 0, 1, 7, 7, 7.0, 2]",
+ "+U[1, 1, 4, 12, 32, 5.0, 5]")
+ .build())
+ .runSql(GROUP_BY_DISTINCT.getRunSqlTestStep().sql)
+ .build();
+
+ static final TableTestProgram GROUP_BY_UDF_WITH_MERGE =
+ TableTestProgram.of(
+ "group-aggregate-udf-with-merge",
+ "validates udfs with merging using group by")
+ .setupCatalogFunction("my_avg", WeightedAvgWithMerge.class)
+ .setupTemporarySystemFunction("my_concat", ConcatDistinctAggFunction.class)
+ .setupTableSource(SOURCE_TWO)
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink_t")
+ .addSchema(
+ "d BIGINT",
+ "s1 BIGINT",
+ "c1 VARCHAR",
+ "PRIMARY KEY (d) NOT ENFORCED")
+ .consumedBeforeRestore(
+ "+I[1, 1, Hello World Like]",
+ "+I[2, 2, Hello World Its nice]",
+ "+U[2, 2, Hello World Its nice|Hello World]",
+ "+U[1, 1, Hello World Like|Hello]",
+ "+U[1, 1, Hello World Like|Hello|GHI]",
+ "+U[2, 2, Hello World Its nice|Hello World|ABC]",
+ "+U[2, 2, Hello World Its nice|Hello World|ABC|FGH]",
+ "+U[2, 2, Hello World Its nice|Hello World|ABC|FGH|CDE]",
+ "+U[2, 2, Hello World Its nice|Hello World|ABC|FGH|CDE|JKL]",
+ "+U[1, 1, Hello World Like|Hello|GHI|EFG]",
+ "+U[2, 2, Hello World Its nice|Hello World|ABC|FGH|CDE|JKL|KLM]",
+ "+I[3, 3, HIJ]",
+ "+U[1, 1, Hello World Like|Hello|GHI|EFG|DEF]",
+ "+U[3, 3, HIJ|IJK]",
+ "+U[3, 3, HIJ|IJK|BCD]")
+ .consumedAfterRestore("+I[7, 7, MNO]", "+U[7, 7, MNO|XYZ]")
+ .build())
+ .runSql(
+ "INSERT INTO sink_t SELECT "
+ + "e, "
+ + "my_avg(e, a) as s1, "
+ + "my_concat(d) as c1 "
+ + "FROM source_t GROUP BY e")
+ .build();
+
+ static final TableTestProgram GROUP_BY_UDF_WITH_MERGE_MINI_BATCH =
+ TableTestProgram.of(
+ "group-aggregate-udf-with-merge-mini-batch",
+ "validates udfs with merging using group by with mini batch")
+ .setupConfig(ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_ENABLED, true)
+ .setupConfig(
+ ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_ALLOW_LATENCY,
+ Duration.ofSeconds(10))
+ .setupConfig(ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_SIZE, 5L)
+ .setupCatalogFunction("my_avg", WeightedAvgWithMerge.class)
+ .setupTemporarySystemFunction("my_concat", ConcatDistinctAggFunction.class)
+ .setupTableSource(SOURCE_TWO)
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink_t")
+ .addSchema(
+ GROUP_BY_UDF_WITH_MERGE
+ .getSetupSinkTestSteps()
+ .get(0)
+ .schemaComponents)
+ .consumedBeforeRestore(
+ "+I[2, 2, Hello World Its nice|Hello World|ABC|FGH|CDE|JKL]",
+ "+I[1, 1, Hello World Like|Hello|GHI|EFG]",
+ "+U[2, 2, Hello World Its nice|Hello World|ABC|FGH|CDE|JKL|KLM]",
+ "+U[1, 1, Hello World Like|Hello|GHI|EFG|DEF]",
+ "+I[3, 3, HIJ|IJK|BCD]")
+ .consumedAfterRestore("+I[7, 7, MNO|XYZ]")
+ .build())
+ .runSql(GROUP_BY_UDF_WITH_MERGE.getRunSqlTestStep().sql)
+ .build();
+
+ static final TableTestProgram GROUP_BY_UDF_WITHOUT_MERGE =
+ TableTestProgram.of(
+ "group-aggregate-udf-without-merge",
+ "validates udfs without merging using group by")
+ .setupTemporarySystemFunction("my_sum1", VarSum1AggFunction.class)
+ .setupCatalogFunction("my_avg", WeightedAvg.class)
+ .setupTemporarySystemFunction("my_sum2", VarSum2AggFunction.class)
+ .setupTableSource(SOURCE_TWO)
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink_t")
+ .addSchema(
+ "d BIGINT",
+ "s1 BIGINT",
+ "s2 BIGINT",
+ "s3 BIGINT",
+ "PRIMARY KEY (d) NOT ENFORCED")
+ .consumedBeforeRestore(
+ "+I[1, 12, 0, 1]",
+ "+I[2, 13, 0, 2]",
+ "+U[2, 24, 0, 2]",
+ "+U[1, 22, 0, 1]",
+ "+U[1, 42, 0, 1]",
+ "+U[2, 38, 0, 2]",
+ "+U[2, 57, 0, 2]",
+ "+U[2, 73, 0, 2]",
+ "+U[2, 96, 0, 2]",
+ "+U[1, 60, 0, 1]",
+ "+U[2, 120, 0, 2]",
+ "+I[3, 21, 0, 3]",
+ "+U[1, 77, 0, 1]",
+ "+U[3, 43, 0, 3]",
+ "+U[3, 58, 0, 3]")
+ .consumedAfterRestore(
+ "+U[1, 87, 0, 1]",
+ "+U[2, 134, 0, 2]",
+ "+U[2, 153, 0, 2]",
+ "+U[2, 169, 0, 2]",
+ "+I[7, 17, 0, 7]",
+ "+U[3, 73, 0, 3]",
+ "+U[7, 34, 0, 7]")
+ .build())
+ .runSql(
+ "INSERT INTO sink_t SELECT "
+ + "e, "
+ + "my_sum1(c, 10) as s1, "
+ + "my_sum2(5, c) as s2, "
+ + "my_avg(e, a) as s3 "
+ + "FROM source_t GROUP BY e")
+ .build();
+
+ static final TableTestProgram GROUP_BY_UDF_WITHOUT_MERGE_MINI_BATCH =
+ TableTestProgram.of(
+ "group-aggregate-udf-without-merge-mini-batch",
+ "validates udfs without merging using group by with mini batch")
+ .setupConfig(ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_ENABLED, true)
+ .setupConfig(
+ ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_ALLOW_LATENCY,
+ Duration.ofSeconds(10))
+ .setupConfig(ExecutionConfigOptions.TABLE_EXEC_MINIBATCH_SIZE, 5L)
+ .setupTemporarySystemFunction("my_sum1", VarSum1AggFunction.class)
+ .setupCatalogFunction("my_avg", WeightedAvg.class)
+ .setupTemporarySystemFunction("my_sum2", VarSum2AggFunction.class)
+ .setupTableSource(SOURCE_TWO)
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink_t")
+ .addSchema(
+ GROUP_BY_UDF_WITHOUT_MERGE
+ .getSetupSinkTestSteps()
+ .get(0)
+ .schemaComponents)
+ .consumedBeforeRestore(
+ "+I[2, 24, 0, 2]",
+ "+I[1, 42, 0, 1]",
+ "+U[2, 96, 0, 2]",
+ "+U[1, 60, 0, 1]",
+ "+I[3, 58, 0, 3]",
+ "+U[2, 120, 0, 2]",
+ "+U[1, 77, 0, 1]")
+ .consumedAfterRestore(
+ "+I[7, 17, 0, 7]",
+ "+U[1, 87, 0, 1]",
+ "+U[2, 169, 0, 2]",
+ "+U[3, 73, 0, 3]",
+ "+U[7, 34, 0, 7]")
+ .build())
+ .runSql(GROUP_BY_UDF_WITHOUT_MERGE.getRunSqlTestStep().sql)
+ .build();
+}
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/testutils/RestoreTestBase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/testutils/RestoreTestBase.java
index a10c5b8202f..7c07bf9bf51 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/testutils/RestoreTestBase.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/testutils/RestoreTestBase.java
@@ -186,6 +186,7 @@ public abstract class RestoreTestBase implements TableTestProgramRunner {
.set(
TableConfigOptions.PLAN_COMPILE_CATALOG_OBJECTS,
TableConfigOptions.CatalogPlanCompilation.SCHEMA);
+
for (SourceTestStep sourceTestStep : program.getSetupSourceTestSteps()) {
final String id = TestValuesTableFactory.registerData(sourceTestStep.dataBeforeRestore);
final Map<String, String> options = new HashMap<>();
diff --git a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-group-aggregate_1/group-aggregate-distinct-mini-batch/plan/group-aggregate-distinct-mini-batch.json b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-group-aggregate_1/group-aggregate-distinct-mini-batch/plan/group-aggregate-distinct-mini-batch.json
new file mode 100644
index 00000000000..8dceb51f726
--- /dev/null
+++ b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-group-aggregate_1/group-aggregate-distinct-mini-batch/plan/group-aggregate-distinct-mini-batch.json
@@ -0,0 +1,621 @@
+{
+ "flinkVersion" : "1.19",
+ "nodes" : [ {
+ "id" : 21,
+ "type" : "stream-exec-table-source-scan_1",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`source_t`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "a",
+ "dataType" : "INT"
+ }, {
+ "name" : "b",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "c",
+ "dataType" : "INT"
+ }, {
+ "name" : "d",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "e",
+ "dataType" : "BIGINT"
+ } ],
+ "watermarkSpecs" : [ ]
+ },
+ "partitionKeys" : [ ]
+ }
+ },
+ "abilities" : [ {
+ "type" : "ProjectPushDown",
+ "projectedFields" : [ [ 4 ], [ 0 ], [ 1 ], [ 3 ] ],
+ "producedType" : "ROW<`e` BIGINT, `a` INT, `b` BIGINT, `d` VARCHAR(2147483647)> NOT NULL"
+ }, {
+ "type" : "ReadingMetadata",
+ "metadataKeys" : [ ],
+ "producedType" : "ROW<`e` BIGINT, `a` INT, `b` BIGINT, `d` VARCHAR(2147483647)> NOT NULL"
+ } ]
+ },
+ "outputType" : "ROW<`e` BIGINT, `a` INT, `b` BIGINT, `d` VARCHAR(2147483647)>",
+ "description" : "TableSourceScan(table=[[default_catalog, default_database, source_t, project=[e, a, b, d], metadata=[]]], fields=[e, a, b, d])",
+ "inputProperties" : [ ]
+ }, {
+ "id" : 22,
+ "type" : "stream-exec-mini-batch-assigner_1",
+ "miniBatchInterval" : {
+ "interval" : 10000,
+ "mode" : "ProcTime"
+ },
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`e` BIGINT, `a` INT, `b` BIGINT, `d` VARCHAR(2147483647)>",
+ "description" : "MiniBatchAssigner(interval=[10000ms], mode=[ProcTime])"
+ }, {
+ "id" : 23,
+ "type" : "stream-exec-calc_1",
+ "projection" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "BIGINT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "INT"
+ }, {
+ "kind" : "CALL",
+ "syntax" : "POSTFIX",
+ "internalName" : "$IS TRUE$1",
+ "operands" : [ {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$>$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "BIGINT"
+ }, {
+ "kind" : "LITERAL",
+ "value" : 10,
+ "type" : "INT NOT NULL"
+ } ],
+ "type" : "BOOLEAN"
+ } ],
+ "type" : "BOOLEAN NOT NULL"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "BIGINT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 3,
+ "type" : "VARCHAR(2147483647)"
+ } ],
+ "condition" : null,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`e` BIGINT, `a` INT, `$f2` BOOLEAN NOT NULL, `b` BIGINT, `d` VARCHAR(2147483647)>",
+ "description" : "Calc(select=[e, a, (b > 10) IS TRUE AS $f2, b, d])"
+ }, {
+ "id" : 24,
+ "type" : "stream-exec-local-group-aggregate_1",
+ "configuration" : {
+ "table.exec.mini-batch.enabled" : "true",
+ "table.exec.mini-batch.size" : "5"
+ },
+ "grouping" : [ 0 ],
+ "aggCalls" : [ {
+ "name" : "cnt_a1",
+ "syntax" : "FUNCTION_STAR",
+ "internalName" : "$COUNT$1",
+ "argList" : [ 1 ],
+ "filterArg" : 2,
+ "distinct" : true,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "BIGINT NOT NULL"
+ }, {
+ "name" : "cnt_a2",
+ "syntax" : "FUNCTION_STAR",
+ "internalName" : "$COUNT$1",
+ "argList" : [ 1 ],
+ "filterArg" : -1,
+ "distinct" : true,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "BIGINT NOT NULL"
+ }, {
+ "name" : "sum_a",
+ "internalName" : "$SUM$1",
+ "argList" : [ 1 ],
+ "filterArg" : -1,
+ "distinct" : true,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "INT"
+ }, {
+ "name" : "sum_b",
+ "internalName" : "$SUM$1",
+ "argList" : [ 3 ],
+ "filterArg" : -1,
+ "distinct" : true,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "BIGINT"
+ }, {
+ "name" : "avg_b",
+ "internalName" : "$AVG$1",
+ "argList" : [ 3 ],
+ "filterArg" : -1,
+ "distinct" : false,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "BIGINT"
+ }, {
+ "name" : "concat_d",
+ "syntax" : "FUNCTION_STAR",
+ "internalName" : "$COUNT$1",
+ "argList" : [ 4 ],
+ "filterArg" : -1,
+ "distinct" : true,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "BIGINT NOT NULL"
+ } ],
+ "aggCallNeedRetractions" : [ false, false, false, false, false, false ],
+ "needRetraction" : false,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : {
+ "type" : "ROW",
+ "fields" : [ {
+ "name" : "e",
+ "fieldType" : "BIGINT"
+ }, {
+ "name" : "count$0",
+ "fieldType" : "BIGINT"
+ }, {
+ "name" : "count$1",
+ "fieldType" : "BIGINT"
+ }, {
+ "name" : "sum$2",
+ "fieldType" : "INT"
+ }, {
+ "name" : "sum$3",
+ "fieldType" : "BIGINT"
+ }, {
+ "name" : "sum$4",
+ "fieldType" : "BIGINT"
+ }, {
+ "name" : "count$5",
+ "fieldType" : "BIGINT"
+ }, {
+ "name" : "count$6",
+ "fieldType" : "BIGINT"
+ }, {
+ "name" : "distinct$0",
+ "fieldType" : {
+ "type" : "RAW",
+ "class" : "org.apache.flink.table.api.dataview.MapView",
+ "externalDataType" : {
+ "type" : "STRUCTURED_TYPE",
+ "implementationClass" : "org.apache.flink.table.api.dataview.MapView",
+ "attributes" : [ {
+ "name" : "map",
+ "attributeType" : "MAP<INT, BIGINT NOT NULL>"
+ } ]
+ }
+ }
+ }, {
+ "name" : "distinct$1",
+ "fieldType" : {
+ "type" : "RAW",
+ "class" : "org.apache.flink.table.api.dataview.MapView",
+ "externalDataType" : {
+ "type" : "STRUCTURED_TYPE",
+ "implementationClass" : "org.apache.flink.table.api.dataview.MapView",
+ "attributes" : [ {
+ "name" : "map",
+ "attributeType" : "MAP<BIGINT, BIGINT NOT NULL>"
+ } ]
+ }
+ }
+ }, {
+ "name" : "distinct$2",
+ "fieldType" : {
+ "type" : "RAW",
+ "class" : "org.apache.flink.table.api.dataview.MapView",
+ "externalDataType" : {
+ "logicalType" : {
+ "type" : "STRUCTURED_TYPE",
+ "implementationClass" : "org.apache.flink.table.api.dataview.MapView",
+ "attributes" : [ {
+ "name" : "map",
+ "attributeType" : "MAP<VARCHAR(2147483647), BIGINT NOT NULL>"
+ } ]
+ },
+ "fields" : [ {
+ "name" : "map",
+ "keyClass" : {
+ "conversionClass" : "org.apache.flink.table.data.StringData"
+ }
+ } ]
+ }
+ }
+ } ]
+ },
+ "description" : "LocalGroupAggregate(groupBy=[e], select=[e, COUNT(distinct$0 a) FILTER $f2 AS count$0, COUNT(distinct$0 a) AS count$1, SUM(distinct$0 a) AS sum$2, SUM(distinct$1 b) AS sum$3, AVG(b) AS (sum$4, count$5), COUNT(distinct$2 d) AS count$6, DISTINCT(a) AS distinct$0, DISTINCT(b) AS distinct$1, DISTINCT(d) AS distinct$2])"
+ }, {
+ "id" : 25,
+ "type" : "stream-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : {
+ "type" : "ROW",
+ "fields" : [ {
+ "name" : "e",
+ "fieldType" : "BIGINT"
+ }, {
+ "name" : "count$0",
+ "fieldType" : "BIGINT"
+ }, {
+ "name" : "count$1",
+ "fieldType" : "BIGINT"
+ }, {
+ "name" : "sum$2",
+ "fieldType" : "INT"
+ }, {
+ "name" : "sum$3",
+ "fieldType" : "BIGINT"
+ }, {
+ "name" : "sum$4",
+ "fieldType" : "BIGINT"
+ }, {
+ "name" : "count$5",
+ "fieldType" : "BIGINT"
+ }, {
+ "name" : "count$6",
+ "fieldType" : "BIGINT"
+ }, {
+ "name" : "distinct$0",
+ "fieldType" : {
+ "type" : "RAW",
+ "class" : "org.apache.flink.table.api.dataview.MapView",
+ "externalDataType" : {
+ "type" : "STRUCTURED_TYPE",
+ "implementationClass" : "org.apache.flink.table.api.dataview.MapView",
+ "attributes" : [ {
+ "name" : "map",
+ "attributeType" : "MAP<INT, BIGINT NOT NULL>"
+ } ]
+ }
+ }
+ }, {
+ "name" : "distinct$1",
+ "fieldType" : {
+ "type" : "RAW",
+ "class" : "org.apache.flink.table.api.dataview.MapView",
+ "externalDataType" : {
+ "type" : "STRUCTURED_TYPE",
+ "implementationClass" : "org.apache.flink.table.api.dataview.MapView",
+ "attributes" : [ {
+ "name" : "map",
+ "attributeType" : "MAP<BIGINT, BIGINT NOT NULL>"
+ } ]
+ }
+ }
+ }, {
+ "name" : "distinct$2",
+ "fieldType" : {
+ "type" : "RAW",
+ "class" : "org.apache.flink.table.api.dataview.MapView",
+ "externalDataType" : {
+ "logicalType" : {
+ "type" : "STRUCTURED_TYPE",
+ "implementationClass" : "org.apache.flink.table.api.dataview.MapView",
+ "attributes" : [ {
+ "name" : "map",
+ "attributeType" : "MAP<VARCHAR(2147483647), BIGINT NOT NULL>"
+ } ]
+ },
+ "fields" : [ {
+ "name" : "map",
+ "keyClass" : {
+ "conversionClass" : "org.apache.flink.table.data.StringData"
+ }
+ } ]
+ }
+ }
+ } ]
+ },
+ "description" : "Exchange(distribution=[hash[e]])"
+ }, {
+ "id" : 26,
+ "type" : "stream-exec-global-group-aggregate_1",
+ "configuration" : {
+ "table.exec.mini-batch.enabled" : "true",
+ "table.exec.mini-batch.size" : "5"
+ },
+ "grouping" : [ 0 ],
+ "aggCalls" : [ {
+ "name" : "cnt_a1",
+ "syntax" : "FUNCTION_STAR",
+ "internalName" : "$COUNT$1",
+ "argList" : [ 1 ],
+ "filterArg" : 2,
+ "distinct" : true,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "BIGINT NOT NULL"
+ }, {
+ "name" : "cnt_a2",
+ "syntax" : "FUNCTION_STAR",
+ "internalName" : "$COUNT$1",
+ "argList" : [ 1 ],
+ "filterArg" : -1,
+ "distinct" : true,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "BIGINT NOT NULL"
+ }, {
+ "name" : "sum_a",
+ "internalName" : "$SUM$1",
+ "argList" : [ 1 ],
+ "filterArg" : -1,
+ "distinct" : true,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "INT"
+ }, {
+ "name" : "sum_b",
+ "internalName" : "$SUM$1",
+ "argList" : [ 3 ],
+ "filterArg" : -1,
+ "distinct" : true,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "BIGINT"
+ }, {
+ "name" : "avg_b",
+ "internalName" : "$AVG$1",
+ "argList" : [ 3 ],
+ "filterArg" : -1,
+ "distinct" : false,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "BIGINT"
+ }, {
+ "name" : "concat_d",
+ "syntax" : "FUNCTION_STAR",
+ "internalName" : "$COUNT$1",
+ "argList" : [ 4 ],
+ "filterArg" : -1,
+ "distinct" : true,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "BIGINT NOT NULL"
+ } ],
+ "aggCallNeedRetractions" : [ false, false, false, false, false, false ],
+ "localAggInputRowType" : "ROW<`e` BIGINT, `a` INT, `$f2` BOOLEAN NOT NULL, `b` BIGINT, `d` VARCHAR(2147483647)>",
+ "generateUpdateBefore" : false,
+ "needRetraction" : false,
+ "state" : [ {
+ "index" : 0,
+ "ttl" : "0 ms",
+ "name" : "globalGroupAggregateState"
+ } ],
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`e` BIGINT, `cnt_a1` BIGINT NOT NULL, `cnt_a2` BIGINT NOT NULL, `sum_a` INT, `sum_b` BIGINT, `avg_b` BIGINT, `concat_d` BIGINT NOT NULL>",
+ "description" : "GlobalGroupAggregate(groupBy=[e], select=[e, COUNT(distinct$0 count$0) AS cnt_a1, COUNT(distinct$0 count$1) AS cnt_a2, SUM(distinct$0 sum$2) AS sum_a, SUM(distinct$1 sum$3) AS sum_b, AVG((sum$4, count$5)) AS avg_b, COUNT(distinct$2 count$6) AS concat_d])"
+ }, {
+ "id" : 27,
+ "type" : "stream-exec-calc_1",
+ "projection" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "BIGINT"
+ }, {
+ "kind" : "CALL",
+ "syntax" : "SPECIAL",
+ "internalName" : "$CAST$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "BIGINT NOT NULL"
+ } ],
+ "type" : "BIGINT"
+ }, {
+ "kind" : "CALL",
+ "syntax" : "SPECIAL",
+ "internalName" : "$CAST$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "BIGINT NOT NULL"
+ } ],
+ "type" : "BIGINT"
+ }, {
+ "kind" : "CALL",
+ "syntax" : "SPECIAL",
+ "internalName" : "$CAST$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 3,
+ "type" : "INT"
+ } ],
+ "type" : "BIGINT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 4,
+ "type" : "BIGINT"
+ }, {
+ "kind" : "CALL",
+ "syntax" : "SPECIAL",
+ "internalName" : "$CAST$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 5,
+ "type" : "BIGINT"
+ } ],
+ "type" : "DOUBLE"
+ }, {
+ "kind" : "CALL",
+ "syntax" : "SPECIAL",
+ "internalName" : "$CAST$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 6,
+ "type" : "BIGINT NOT NULL"
+ } ],
+ "type" : "BIGINT"
+ } ],
+ "condition" : null,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`e` BIGINT, `cnt_a1` BIGINT, `cnt_a2` BIGINT, `sum_a` BIGINT, `sum_b` BIGINT, `avg_b` DOUBLE, `cnt_d` BIGINT>",
+ "description" : "Calc(select=[e, CAST(cnt_a1 AS BIGINT) AS cnt_a1, CAST(cnt_a2 AS BIGINT) AS cnt_a2, CAST(sum_a AS BIGINT) AS sum_a, sum_b, CAST(avg_b AS DOUBLE) AS avg_b, CAST(concat_d AS BIGINT) AS cnt_d])"
+ }, {
+ "id" : 28,
+ "type" : "stream-exec-sink_1",
+ "configuration" : {
+ "table.exec.sink.keyed-shuffle" : "AUTO",
+ "table.exec.sink.not-null-enforcer" : "ERROR",
+ "table.exec.sink.rowtime-inserter" : "ENABLED",
+ "table.exec.sink.type-length-enforcer" : "IGNORE",
+ "table.exec.sink.upsert-materialize" : "AUTO"
+ },
+ "dynamicTableSink" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`sink_t`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "e",
+ "dataType" : "BIGINT NOT NULL"
+ }, {
+ "name" : "cnt_a1",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "cnt_a2",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "sum_a",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "sum_b",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "avg_b",
+ "dataType" : "DOUBLE"
+ }, {
+ "name" : "cnt_d",
+ "dataType" : "BIGINT"
+ } ],
+ "watermarkSpecs" : [ ],
+ "primaryKey" : {
+ "name" : "PK_132",
+ "type" : "PRIMARY_KEY",
+ "columns" : [ "e" ]
+ }
+ },
+ "partitionKeys" : [ ]
+ }
+ }
+ },
+ "inputChangelogMode" : [ "INSERT", "UPDATE_AFTER" ],
+ "inputUpsertKey" : [ 0 ],
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`e` BIGINT, `cnt_a1` BIGINT, `cnt_a2` BIGINT, `sum_a` BIGINT, `sum_b` BIGINT, `avg_b` DOUBLE, `cnt_d` BIGINT>",
+ "description" : "Sink(table=[default_catalog.default_database.sink_t], fields=[e, cnt_a1, cnt_a2, sum_a, sum_b, avg_b, cnt_d])"
+ } ],
+ "edges" : [ {
+ "source" : 21,
+ "target" : 22,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 22,
+ "target" : 23,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 23,
+ "target" : 24,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 24,
+ "target" : 25,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 25,
+ "target" : 26,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 26,
+ "target" : 27,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 27,
+ "target" : 28,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ } ]
+}
\ No newline at end of file
diff --git a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-group-aggregate_1/group-aggregate-distinct-mini-batch/savepoint/_metadata b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-group-aggregate_1/group-aggregate-distinct-mini-batch/savepoint/_metadata
new file mode 100644
index 00000000000..83d86c0bb29
Binary files /dev/null and b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-group-aggregate_1/group-aggregate-distinct-mini-batch/savepoint/_metadata differ
diff --git a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-group-aggregate_1/group-aggregate-distinct/plan/group-aggregate-distinct.json b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-group-aggregate_1/group-aggregate-distinct/plan/group-aggregate-distinct.json
new file mode 100644
index 00000000000..b1801290fb3
--- /dev/null
+++ b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-group-aggregate_1/group-aggregate-distinct/plan/group-aggregate-distinct.json
@@ -0,0 +1,359 @@
+{
+ "flinkVersion" : "1.19",
+ "nodes" : [ {
+ "id" : 15,
+ "type" : "stream-exec-table-source-scan_1",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`source_t`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "a",
+ "dataType" : "INT"
+ }, {
+ "name" : "b",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "c",
+ "dataType" : "INT"
+ }, {
+ "name" : "d",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "e",
+ "dataType" : "BIGINT"
+ } ],
+ "watermarkSpecs" : [ ]
+ },
+ "partitionKeys" : [ ]
+ }
+ },
+ "abilities" : [ {
+ "type" : "ProjectPushDown",
+ "projectedFields" : [ [ 4 ], [ 0 ], [ 1 ], [ 3 ] ],
+ "producedType" : "ROW<`e` BIGINT, `a` INT, `b` BIGINT, `d` VARCHAR(2147483647)> NOT NULL"
+ }, {
+ "type" : "ReadingMetadata",
+ "metadataKeys" : [ ],
+ "producedType" : "ROW<`e` BIGINT, `a` INT, `b` BIGINT, `d` VARCHAR(2147483647)> NOT NULL"
+ } ]
+ },
+ "outputType" : "ROW<`e` BIGINT, `a` INT, `b` BIGINT, `d` VARCHAR(2147483647)>",
+ "description" : "TableSourceScan(table=[[default_catalog, default_database, source_t, project=[e, a, b, d], metadata=[]]], fields=[e, a, b, d])",
+ "inputProperties" : [ ]
+ }, {
+ "id" : 16,
+ "type" : "stream-exec-calc_1",
+ "projection" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "BIGINT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "INT"
+ }, {
+ "kind" : "CALL",
+ "syntax" : "POSTFIX",
+ "internalName" : "$IS TRUE$1",
+ "operands" : [ {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$>$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "BIGINT"
+ }, {
+ "kind" : "LITERAL",
+ "value" : 10,
+ "type" : "INT NOT NULL"
+ } ],
+ "type" : "BOOLEAN"
+ } ],
+ "type" : "BOOLEAN NOT NULL"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "BIGINT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 3,
+ "type" : "VARCHAR(2147483647)"
+ } ],
+ "condition" : null,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`e` BIGINT, `a` INT, `$f2` BOOLEAN NOT NULL, `b` BIGINT, `d` VARCHAR(2147483647)>",
+ "description" : "Calc(select=[e, a, (b > 10) IS TRUE AS $f2, b, d])"
+ }, {
+ "id" : 17,
+ "type" : "stream-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`e` BIGINT, `a` INT, `$f2` BOOLEAN NOT NULL, `b` BIGINT, `d` VARCHAR(2147483647)>",
+ "description" : "Exchange(distribution=[hash[e]])"
+ }, {
+ "id" : 18,
+ "type" : "stream-exec-group-aggregate_1",
+ "configuration" : {
+ "table.exec.mini-batch.enabled" : "false",
+ "table.exec.mini-batch.size" : "-1"
+ },
+ "grouping" : [ 0 ],
+ "aggCalls" : [ {
+ "name" : "cnt_a1",
+ "syntax" : "FUNCTION_STAR",
+ "internalName" : "$COUNT$1",
+ "argList" : [ 1 ],
+ "filterArg" : 2,
+ "distinct" : true,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "BIGINT NOT NULL"
+ }, {
+ "name" : "cnt_a2",
+ "syntax" : "FUNCTION_STAR",
+ "internalName" : "$COUNT$1",
+ "argList" : [ 1 ],
+ "filterArg" : -1,
+ "distinct" : true,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "BIGINT NOT NULL"
+ }, {
+ "name" : "sum_a",
+ "internalName" : "$SUM$1",
+ "argList" : [ 1 ],
+ "filterArg" : -1,
+ "distinct" : true,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "INT"
+ }, {
+ "name" : "sum_b",
+ "internalName" : "$SUM$1",
+ "argList" : [ 3 ],
+ "filterArg" : -1,
+ "distinct" : true,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "BIGINT"
+ }, {
+ "name" : "avg_b",
+ "internalName" : "$AVG$1",
+ "argList" : [ 3 ],
+ "filterArg" : -1,
+ "distinct" : false,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "BIGINT"
+ }, {
+ "name" : "concat_d",
+ "syntax" : "FUNCTION_STAR",
+ "internalName" : "$COUNT$1",
+ "argList" : [ 4 ],
+ "filterArg" : -1,
+ "distinct" : true,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "BIGINT NOT NULL"
+ } ],
+ "aggCallNeedRetractions" : [ false, false, false, false, false, false ],
+ "generateUpdateBefore" : false,
+ "needRetraction" : false,
+ "state" : [ {
+ "index" : 0,
+ "ttl" : "0 ms",
+ "name" : "groupAggregateState"
+ } ],
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`e` BIGINT, `cnt_a1` BIGINT NOT NULL, `cnt_a2` BIGINT NOT NULL, `sum_a` INT, `sum_b` BIGINT, `avg_b` BIGINT, `concat_d` BIGINT NOT NULL>",
+ "description" : "GroupAggregate(groupBy=[e], select=[e, COUNT(DISTINCT a) FILTER $f2 AS cnt_a1, COUNT(DISTINCT a) AS cnt_a2, SUM(DISTINCT a) AS sum_a, SUM(DISTINCT b) AS sum_b, AVG(b) AS avg_b, COUNT(DISTINCT d) AS concat_d])"
+ }, {
+ "id" : 19,
+ "type" : "stream-exec-calc_1",
+ "projection" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "BIGINT"
+ }, {
+ "kind" : "CALL",
+ "syntax" : "SPECIAL",
+ "internalName" : "$CAST$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "BIGINT NOT NULL"
+ } ],
+ "type" : "BIGINT"
+ }, {
+ "kind" : "CALL",
+ "syntax" : "SPECIAL",
+ "internalName" : "$CAST$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "BIGINT NOT NULL"
+ } ],
+ "type" : "BIGINT"
+ }, {
+ "kind" : "CALL",
+ "syntax" : "SPECIAL",
+ "internalName" : "$CAST$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 3,
+ "type" : "INT"
+ } ],
+ "type" : "BIGINT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 4,
+ "type" : "BIGINT"
+ }, {
+ "kind" : "CALL",
+ "syntax" : "SPECIAL",
+ "internalName" : "$CAST$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 5,
+ "type" : "BIGINT"
+ } ],
+ "type" : "DOUBLE"
+ }, {
+ "kind" : "CALL",
+ "syntax" : "SPECIAL",
+ "internalName" : "$CAST$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 6,
+ "type" : "BIGINT NOT NULL"
+ } ],
+ "type" : "BIGINT"
+ } ],
+ "condition" : null,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`e` BIGINT, `cnt_a1` BIGINT, `cnt_a2` BIGINT, `sum_a` BIGINT, `sum_b` BIGINT, `avg_b` DOUBLE, `cnt_d` BIGINT>",
+ "description" : "Calc(select=[e, CAST(cnt_a1 AS BIGINT) AS cnt_a1, CAST(cnt_a2 AS BIGINT) AS cnt_a2, CAST(sum_a AS BIGINT) AS sum_a, sum_b, CAST(avg_b AS DOUBLE) AS avg_b, CAST(concat_d AS BIGINT) AS cnt_d])"
+ }, {
+ "id" : 20,
+ "type" : "stream-exec-sink_1",
+ "configuration" : {
+ "table.exec.sink.keyed-shuffle" : "AUTO",
+ "table.exec.sink.not-null-enforcer" : "ERROR",
+ "table.exec.sink.rowtime-inserter" : "ENABLED",
+ "table.exec.sink.type-length-enforcer" : "IGNORE",
+ "table.exec.sink.upsert-materialize" : "AUTO"
+ },
+ "dynamicTableSink" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`sink_t`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "e",
+ "dataType" : "BIGINT NOT NULL"
+ }, {
+ "name" : "cnt_a1",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "cnt_a2",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "sum_a",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "sum_b",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "avg_b",
+ "dataType" : "DOUBLE"
+ }, {
+ "name" : "cnt_d",
+ "dataType" : "BIGINT"
+ } ],
+ "watermarkSpecs" : [ ],
+ "primaryKey" : {
+ "name" : "PK_132",
+ "type" : "PRIMARY_KEY",
+ "columns" : [ "e" ]
+ }
+ },
+ "partitionKeys" : [ ]
+ }
+ }
+ },
+ "inputChangelogMode" : [ "INSERT", "UPDATE_AFTER" ],
+ "inputUpsertKey" : [ 0 ],
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`e` BIGINT, `cnt_a1` BIGINT, `cnt_a2` BIGINT, `sum_a` BIGINT, `sum_b` BIGINT, `avg_b` DOUBLE, `cnt_d` BIGINT>",
+ "description" : "Sink(table=[default_catalog.default_database.sink_t], fields=[e, cnt_a1, cnt_a2, sum_a, sum_b, avg_b, cnt_d])"
+ } ],
+ "edges" : [ {
+ "source" : 15,
+ "target" : 16,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 16,
+ "target" : 17,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 17,
+ "target" : 18,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 18,
+ "target" : 19,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 19,
+ "target" : 20,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ } ]
+}
\ No newline at end of file
diff --git a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-group-aggregate_1/group-aggregate-distinct/savepoint/_metadata b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-group-aggregate_1/group-aggregate-distinct/savepoint/_metadata
new file mode 100644
index 00000000000..0207e623c99
Binary files /dev/null and b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-group-aggregate_1/group-aggregate-distinct/savepoint/_metadata differ
diff --git a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-group-aggregate_1/group-aggregate-simple-mini-batch/plan/group-aggregate-simple-mini-batch.json b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-group-aggregate_1/group-aggregate-simple-mini-batch/plan/group-aggregate-simple-mini-batch.json
new file mode 100644
index 00000000000..6ada7bb67e8
--- /dev/null
+++ b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-group-aggregate_1/group-aggregate-simple-mini-batch/plan/group-aggregate-simple-mini-batch.json
@@ -0,0 +1,351 @@
+{
+ "flinkVersion" : "1.19",
+ "nodes" : [ {
+ "id" : 7,
+ "type" : "stream-exec-table-source-scan_1",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`source_t`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "a",
+ "dataType" : "INT"
+ }, {
+ "name" : "b",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "c",
+ "dataType" : "VARCHAR(2147483647)"
+ } ],
+ "watermarkSpecs" : [ ]
+ },
+ "partitionKeys" : [ ]
+ }
+ }
+ },
+ "outputType" : "ROW<`a` INT, `b` BIGINT, `c` VARCHAR(2147483647)>",
+ "description" : "TableSourceScan(table=[[default_catalog, default_database, source_t]], fields=[a, b, c])",
+ "inputProperties" : [ ]
+ }, {
+ "id" : 8,
+ "type" : "stream-exec-mini-batch-assigner_1",
+ "miniBatchInterval" : {
+ "interval" : 10000,
+ "mode" : "ProcTime"
+ },
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`a` INT, `b` BIGINT, `c` VARCHAR(2147483647)>",
+ "description" : "MiniBatchAssigner(interval=[10000ms], mode=[ProcTime])"
+ }, {
+ "id" : 9,
+ "type" : "stream-exec-calc_1",
+ "projection" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "BIGINT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "INT"
+ }, {
+ "kind" : "CALL",
+ "syntax" : "POSTFIX",
+ "internalName" : "$IS TRUE$1",
+ "operands" : [ {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$>$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "INT"
+ }, {
+ "kind" : "LITERAL",
+ "value" : 1,
+ "type" : "INT NOT NULL"
+ } ],
+ "type" : "BOOLEAN"
+ } ],
+ "type" : "BOOLEAN NOT NULL"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "VARCHAR(2147483647)"
+ } ],
+ "condition" : null,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`b` BIGINT, `a` INT, `$f2` BOOLEAN NOT NULL, `c` VARCHAR(2147483647)>",
+ "description" : "Calc(select=[b, a, (a > 1) IS TRUE AS $f2, c])"
+ }, {
+ "id" : 10,
+ "type" : "stream-exec-local-group-aggregate_1",
+ "configuration" : {
+ "table.exec.mini-batch.enabled" : "true",
+ "table.exec.mini-batch.size" : "5"
+ },
+ "grouping" : [ 0 ],
+ "aggCalls" : [ {
+ "name" : "cnt",
+ "syntax" : "FUNCTION_STAR",
+ "internalName" : "$COUNT$1",
+ "argList" : [ ],
+ "filterArg" : -1,
+ "distinct" : false,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "BIGINT NOT NULL"
+ }, {
+ "name" : "avg_a",
+ "internalName" : "$AVG$1",
+ "argList" : [ 1 ],
+ "filterArg" : 2,
+ "distinct" : false,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "INT"
+ }, {
+ "name" : "min_c",
+ "internalName" : "$MIN$1",
+ "argList" : [ 3 ],
+ "filterArg" : -1,
+ "distinct" : false,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "VARCHAR(2147483647)"
+ } ],
+ "aggCallNeedRetractions" : [ false, false, false ],
+ "needRetraction" : false,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`b` BIGINT, `count1$0` BIGINT, `sum$1` BIGINT, `count$2` BIGINT, `min$3` VARCHAR(2147483647)>",
+ "description" : "LocalGroupAggregate(groupBy=[b], select=[b, COUNT(*) AS count1$0, AVG(a) FILTER $f2 AS (sum$1, count$2), MIN(c) AS min$3])"
+ }, {
+ "id" : 11,
+ "type" : "stream-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`b` BIGINT, `count1$0` BIGINT, `sum$1` BIGINT, `count$2` BIGINT, `min$3` VARCHAR(2147483647)>",
+ "description" : "Exchange(distribution=[hash[b]])"
+ }, {
+ "id" : 12,
+ "type" : "stream-exec-global-group-aggregate_1",
+ "configuration" : {
+ "table.exec.mini-batch.enabled" : "true",
+ "table.exec.mini-batch.size" : "5"
+ },
+ "grouping" : [ 0 ],
+ "aggCalls" : [ {
+ "name" : "cnt",
+ "syntax" : "FUNCTION_STAR",
+ "internalName" : "$COUNT$1",
+ "argList" : [ ],
+ "filterArg" : -1,
+ "distinct" : false,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "BIGINT NOT NULL"
+ }, {
+ "name" : "avg_a",
+ "internalName" : "$AVG$1",
+ "argList" : [ 1 ],
+ "filterArg" : 2,
+ "distinct" : false,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "INT"
+ }, {
+ "name" : "min_c",
+ "internalName" : "$MIN$1",
+ "argList" : [ 3 ],
+ "filterArg" : -1,
+ "distinct" : false,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "VARCHAR(2147483647)"
+ } ],
+ "aggCallNeedRetractions" : [ false, false, false ],
+ "localAggInputRowType" : "ROW<`b` BIGINT, `a` INT, `$f2` BOOLEAN NOT NULL, `c` VARCHAR(2147483647)>",
+ "generateUpdateBefore" : false,
+ "needRetraction" : false,
+ "state" : [ {
+ "index" : 0,
+ "ttl" : "0 ms",
+ "name" : "globalGroupAggregateState"
+ } ],
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`b` BIGINT, `cnt` BIGINT NOT NULL, `avg_a` INT, `min_c` VARCHAR(2147483647)>",
+ "description" : "GlobalGroupAggregate(groupBy=[b], select=[b, COUNT(count1$0) AS cnt, AVG((sum$1, count$2)) AS avg_a, MIN(min$3) AS min_c])"
+ }, {
+ "id" : 13,
+ "type" : "stream-exec-calc_1",
+ "projection" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "BIGINT"
+ }, {
+ "kind" : "CALL",
+ "syntax" : "SPECIAL",
+ "internalName" : "$CAST$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "BIGINT NOT NULL"
+ } ],
+ "type" : "BIGINT"
+ }, {
+ "kind" : "CALL",
+ "syntax" : "SPECIAL",
+ "internalName" : "$CAST$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "INT"
+ } ],
+ "type" : "DOUBLE"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 3,
+ "type" : "VARCHAR(2147483647)"
+ } ],
+ "condition" : null,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`b` BIGINT, `cnt` BIGINT, `avg_a` DOUBLE, `min_c` VARCHAR(2147483647)>",
+ "description" : "Calc(select=[b, CAST(cnt AS BIGINT) AS cnt, CAST(avg_a AS DOUBLE) AS avg_a, min_c])"
+ }, {
+ "id" : 14,
+ "type" : "stream-exec-sink_1",
+ "configuration" : {
+ "table.exec.sink.keyed-shuffle" : "AUTO",
+ "table.exec.sink.not-null-enforcer" : "ERROR",
+ "table.exec.sink.rowtime-inserter" : "ENABLED",
+ "table.exec.sink.type-length-enforcer" : "IGNORE",
+ "table.exec.sink.upsert-materialize" : "AUTO"
+ },
+ "dynamicTableSink" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`sink_t`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "b",
+ "dataType" : "BIGINT NOT NULL"
+ }, {
+ "name" : "cnt",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "avg_a",
+ "dataType" : "DOUBLE"
+ }, {
+ "name" : "min_c",
+ "dataType" : "VARCHAR(2147483647)"
+ } ],
+ "watermarkSpecs" : [ ],
+ "primaryKey" : {
+ "name" : "PK_129",
+ "type" : "PRIMARY_KEY",
+ "columns" : [ "b" ]
+ }
+ },
+ "partitionKeys" : [ ]
+ }
+ }
+ },
+ "inputChangelogMode" : [ "INSERT", "UPDATE_AFTER" ],
+ "inputUpsertKey" : [ 0 ],
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`b` BIGINT, `cnt` BIGINT, `avg_a` DOUBLE, `min_c` VARCHAR(2147483647)>",
+ "description" : "Sink(table=[default_catalog.default_database.sink_t], fields=[b, cnt, avg_a, min_c])"
+ } ],
+ "edges" : [ {
+ "source" : 7,
+ "target" : 8,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 8,
+ "target" : 9,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 9,
+ "target" : 10,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 10,
+ "target" : 11,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 11,
+ "target" : 12,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 12,
+ "target" : 13,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 13,
+ "target" : 14,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ } ]
+}
\ No newline at end of file
diff --git a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-group-aggregate_1/group-aggregate-simple-mini-batch/savepoint/_metadata b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-group-aggregate_1/group-aggregate-simple-mini-batch/savepoint/_metadata
new file mode 100644
index 00000000000..4170ed4293a
Binary files /dev/null and b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-group-aggregate_1/group-aggregate-simple-mini-batch/savepoint/_metadata differ
diff --git a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-group-aggregate_1/group-aggregate-simple/plan/group-aggregate-simple.json b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-group-aggregate_1/group-aggregate-simple/plan/group-aggregate-simple.json
new file mode 100644
index 00000000000..3785d4e87f1
--- /dev/null
+++ b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-group-aggregate_1/group-aggregate-simple/plan/group-aggregate-simple.json
@@ -0,0 +1,272 @@
+{
+ "flinkVersion" : "1.19",
+ "nodes" : [ {
+ "id" : 1,
+ "type" : "stream-exec-table-source-scan_1",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`source_t`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "a",
+ "dataType" : "INT"
+ }, {
+ "name" : "b",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "c",
+ "dataType" : "VARCHAR(2147483647)"
+ } ],
+ "watermarkSpecs" : [ ]
+ },
+ "partitionKeys" : [ ]
+ }
+ }
+ },
+ "outputType" : "ROW<`a` INT, `b` BIGINT, `c` VARCHAR(2147483647)>",
+ "description" : "TableSourceScan(table=[[default_catalog, default_database, source_t]], fields=[a, b, c])",
+ "inputProperties" : [ ]
+ }, {
+ "id" : 2,
+ "type" : "stream-exec-calc_1",
+ "projection" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "BIGINT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "INT"
+ }, {
+ "kind" : "CALL",
+ "syntax" : "POSTFIX",
+ "internalName" : "$IS TRUE$1",
+ "operands" : [ {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$>$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "INT"
+ }, {
+ "kind" : "LITERAL",
+ "value" : 1,
+ "type" : "INT NOT NULL"
+ } ],
+ "type" : "BOOLEAN"
+ } ],
+ "type" : "BOOLEAN NOT NULL"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "VARCHAR(2147483647)"
+ } ],
+ "condition" : null,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`b` BIGINT, `a` INT, `$f2` BOOLEAN NOT NULL, `c` VARCHAR(2147483647)>",
+ "description" : "Calc(select=[b, a, (a > 1) IS TRUE AS $f2, c])"
+ }, {
+ "id" : 3,
+ "type" : "stream-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`b` BIGINT, `a` INT, `$f2` BOOLEAN NOT NULL, `c` VARCHAR(2147483647)>",
+ "description" : "Exchange(distribution=[hash[b]])"
+ }, {
+ "id" : 4,
+ "type" : "stream-exec-group-aggregate_1",
+ "configuration" : {
+ "table.exec.mini-batch.enabled" : "false",
+ "table.exec.mini-batch.size" : "-1"
+ },
+ "grouping" : [ 0 ],
+ "aggCalls" : [ {
+ "name" : "cnt",
+ "syntax" : "FUNCTION_STAR",
+ "internalName" : "$COUNT$1",
+ "argList" : [ ],
+ "filterArg" : -1,
+ "distinct" : false,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "BIGINT NOT NULL"
+ }, {
+ "name" : "avg_a",
+ "internalName" : "$AVG$1",
+ "argList" : [ 1 ],
+ "filterArg" : 2,
+ "distinct" : false,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "INT"
+ }, {
+ "name" : "min_c",
+ "internalName" : "$MIN$1",
+ "argList" : [ 3 ],
+ "filterArg" : -1,
+ "distinct" : false,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "VARCHAR(2147483647)"
+ } ],
+ "aggCallNeedRetractions" : [ false, false, false ],
+ "generateUpdateBefore" : false,
+ "needRetraction" : false,
+ "state" : [ {
+ "index" : 0,
+ "ttl" : "0 ms",
+ "name" : "groupAggregateState"
+ } ],
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`b` BIGINT, `cnt` BIGINT NOT NULL, `avg_a` INT, `min_c` VARCHAR(2147483647)>",
+ "description" : "GroupAggregate(groupBy=[b], select=[b, COUNT(*) AS cnt, AVG(a) FILTER $f2 AS avg_a, MIN(c) AS min_c])"
+ }, {
+ "id" : 5,
+ "type" : "stream-exec-calc_1",
+ "projection" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "BIGINT"
+ }, {
+ "kind" : "CALL",
+ "syntax" : "SPECIAL",
+ "internalName" : "$CAST$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "BIGINT NOT NULL"
+ } ],
+ "type" : "BIGINT"
+ }, {
+ "kind" : "CALL",
+ "syntax" : "SPECIAL",
+ "internalName" : "$CAST$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "INT"
+ } ],
+ "type" : "DOUBLE"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 3,
+ "type" : "VARCHAR(2147483647)"
+ } ],
+ "condition" : null,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`b` BIGINT, `cnt` BIGINT, `avg_a` DOUBLE, `min_c` VARCHAR(2147483647)>",
+ "description" : "Calc(select=[b, CAST(cnt AS BIGINT) AS cnt, CAST(avg_a AS DOUBLE) AS avg_a, min_c])"
+ }, {
+ "id" : 6,
+ "type" : "stream-exec-sink_1",
+ "configuration" : {
+ "table.exec.sink.keyed-shuffle" : "AUTO",
+ "table.exec.sink.not-null-enforcer" : "ERROR",
+ "table.exec.sink.rowtime-inserter" : "ENABLED",
+ "table.exec.sink.type-length-enforcer" : "IGNORE",
+ "table.exec.sink.upsert-materialize" : "AUTO"
+ },
+ "dynamicTableSink" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`sink_t`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "b",
+ "dataType" : "BIGINT NOT NULL"
+ }, {
+ "name" : "cnt",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "avg_a",
+ "dataType" : "DOUBLE"
+ }, {
+ "name" : "min_c",
+ "dataType" : "VARCHAR(2147483647)"
+ } ],
+ "watermarkSpecs" : [ ],
+ "primaryKey" : {
+ "name" : "PK_129",
+ "type" : "PRIMARY_KEY",
+ "columns" : [ "b" ]
+ }
+ },
+ "partitionKeys" : [ ]
+ }
+ }
+ },
+ "inputChangelogMode" : [ "INSERT", "UPDATE_AFTER" ],
+ "inputUpsertKey" : [ 0 ],
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`b` BIGINT, `cnt` BIGINT, `avg_a` DOUBLE, `min_c` VARCHAR(2147483647)>",
+ "description" : "Sink(table=[default_catalog.default_database.sink_t], fields=[b, cnt, avg_a, min_c])"
+ } ],
+ "edges" : [ {
+ "source" : 1,
+ "target" : 2,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 2,
+ "target" : 3,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 3,
+ "target" : 4,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 4,
+ "target" : 5,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 5,
+ "target" : 6,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ } ]
+}
\ No newline at end of file
diff --git a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-group-aggregate_1/group-aggregate-simple/savepoint/_metadata b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-group-aggregate_1/group-aggregate-simple/savepoint/_metadata
new file mode 100644
index 00000000000..d7fbd4184b1
Binary files /dev/null and b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-group-aggregate_1/group-aggregate-simple/savepoint/_metadata differ
diff --git a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-group-aggregate_1/group-aggregate-udf-with-merge-mini-batch/plan/group-aggregate-udf-with-merge-mini-batch.json b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-group-aggregate_1/group-aggregate-udf-with-merge-mini-batch/plan/group-aggregate-udf-with-merge-mini-batch.json
new file mode 100644
index 00000000000..198b0216a19
--- /dev/null
+++ b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-group-aggregate_1/group-aggregate-udf-with-merge-mini-batch/plan/group-aggregate-udf-with-merge-mini-batch.json
@@ -0,0 +1,400 @@
+{
+ "flinkVersion" : "1.19",
+ "nodes" : [ {
+ "id" : 34,
+ "type" : "stream-exec-table-source-scan_1",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`source_t`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "a",
+ "dataType" : "INT"
+ }, {
+ "name" : "b",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "c",
+ "dataType" : "INT"
+ }, {
+ "name" : "d",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "e",
+ "dataType" : "BIGINT"
+ } ],
+ "watermarkSpecs" : [ ]
+ },
+ "partitionKeys" : [ ]
+ }
+ },
+ "abilities" : [ {
+ "type" : "ProjectPushDown",
+ "projectedFields" : [ [ 4 ], [ 0 ], [ 3 ] ],
+ "producedType" : "ROW<`e` BIGINT, `a` INT, `d` VARCHAR(2147483647)> NOT NULL"
+ }, {
+ "type" : "ReadingMetadata",
+ "metadataKeys" : [ ],
+ "producedType" : "ROW<`e` BIGINT, `a` INT, `d` VARCHAR(2147483647)> NOT NULL"
+ } ]
+ },
+ "outputType" : "ROW<`e` BIGINT, `a` INT, `d` VARCHAR(2147483647)>",
+ "description" : "TableSourceScan(table=[[default_catalog, default_database, source_t, project=[e, a, d], metadata=[]]], fields=[e, a, d])",
+ "inputProperties" : [ ]
+ }, {
+ "id" : 35,
+ "type" : "stream-exec-mini-batch-assigner_1",
+ "miniBatchInterval" : {
+ "interval" : 10000,
+ "mode" : "ProcTime"
+ },
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`e` BIGINT, `a` INT, `d` VARCHAR(2147483647)>",
+ "description" : "MiniBatchAssigner(interval=[10000ms], mode=[ProcTime])"
+ }, {
+ "id" : 36,
+ "type" : "stream-exec-calc_1",
+ "projection" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "BIGINT"
+ }, {
+ "kind" : "CALL",
+ "syntax" : "SPECIAL",
+ "internalName" : "$CAST$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "INT"
+ } ],
+ "type" : "BIGINT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "VARCHAR(2147483647)"
+ } ],
+ "condition" : null,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`e` BIGINT, `$f1` BIGINT, `d` VARCHAR(2147483647)>",
+ "description" : "Calc(select=[e, CAST(a AS BIGINT) AS $f1, d])"
+ }, {
+ "id" : 37,
+ "type" : "stream-exec-local-group-aggregate_1",
+ "configuration" : {
+ "table.exec.mini-batch.enabled" : "true",
+ "table.exec.mini-batch.size" : "5"
+ },
+ "grouping" : [ 0 ],
+ "aggCalls" : [ {
+ "name" : "s1",
+ "catalogName" : "`default_catalog`.`default_database`.`my_avg`",
+ "argList" : [ 0, 1 ],
+ "filterArg" : -1,
+ "distinct" : false,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "BIGINT"
+ }, {
+ "name" : "c1",
+ "systemName" : "my_concat",
+ "argList" : [ 2 ],
+ "filterArg" : -1,
+ "distinct" : false,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "VARCHAR(2147483647)"
+ } ],
+ "aggCallNeedRetractions" : [ false, false ],
+ "needRetraction" : false,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : {
+ "type" : "ROW",
+ "fields" : [ {
+ "name" : "e",
+ "fieldType" : "BIGINT"
+ }, {
+ "name" : "my_avg$0",
+ "fieldType" : {
+ "type" : "STRUCTURED_TYPE",
+ "implementationClass" : "org.apache.flink.table.planner.plan.utils.JavaUserDefinedAggFunctions$WeightedAvgAccum",
+ "attributes" : [ {
+ "name" : "count",
+ "attributeType" : "INT NOT NULL"
+ }, {
+ "name" : "sum",
+ "attributeType" : "BIGINT NOT NULL"
+ } ]
+ }
+ }, {
+ "name" : "my_concat$1",
+ "fieldType" : {
+ "type" : "STRUCTURED_TYPE",
+ "implementationClass" : "org.apache.flink.table.planner.plan.utils.JavaUserDefinedAggFunctions$ConcatAcc",
+ "attributes" : [ {
+ "name" : "list",
+ "attributeType" : {
+ "type" : "RAW",
+ "class" : "org.apache.flink.table.api.dataview.ListView",
+ "externalDataType" : {
+ "logicalType" : {
+ "type" : "STRUCTURED_TYPE",
+ "implementationClass" : "org.apache.flink.table.api.dataview.ListView",
+ "attributes" : [ {
+ "name" : "list",
+ "attributeType" : "ARRAY<VARCHAR(2147483647)>"
+ } ]
+ },
+ "fields" : [ {
+ "name" : "list",
+ "conversionClass" : "java.util.List"
+ } ]
+ }
+ }
+ }, {
+ "name" : "map",
+ "attributeType" : {
+ "type" : "RAW",
+ "class" : "org.apache.flink.table.api.dataview.MapView",
+ "externalDataType" : {
+ "type" : "STRUCTURED_TYPE",
+ "implementationClass" : "org.apache.flink.table.api.dataview.MapView",
+ "attributes" : [ {
+ "name" : "map",
+ "attributeType" : "MAP<VARCHAR(2147483647), BOOLEAN>"
+ } ]
+ }
+ }
+ } ]
+ }
+ } ]
+ },
+ "description" : "LocalGroupAggregate(groupBy=[e], select=[e, my_avg(e, $f1) AS my_avg$0, my_concat(d) AS my_concat$1])"
+ }, {
+ "id" : 38,
+ "type" : "stream-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : {
+ "type" : "ROW",
+ "fields" : [ {
+ "name" : "e",
+ "fieldType" : "BIGINT"
+ }, {
+ "name" : "my_avg$0",
+ "fieldType" : {
+ "type" : "STRUCTURED_TYPE",
+ "implementationClass" : "org.apache.flink.table.planner.plan.utils.JavaUserDefinedAggFunctions$WeightedAvgAccum",
+ "attributes" : [ {
+ "name" : "count",
+ "attributeType" : "INT NOT NULL"
+ }, {
+ "name" : "sum",
+ "attributeType" : "BIGINT NOT NULL"
+ } ]
+ }
+ }, {
+ "name" : "my_concat$1",
+ "fieldType" : {
+ "type" : "STRUCTURED_TYPE",
+ "implementationClass" : "org.apache.flink.table.planner.plan.utils.JavaUserDefinedAggFunctions$ConcatAcc",
+ "attributes" : [ {
+ "name" : "list",
+ "attributeType" : {
+ "type" : "RAW",
+ "class" : "org.apache.flink.table.api.dataview.ListView",
+ "externalDataType" : {
+ "logicalType" : {
+ "type" : "STRUCTURED_TYPE",
+ "implementationClass" : "org.apache.flink.table.api.dataview.ListView",
+ "attributes" : [ {
+ "name" : "list",
+ "attributeType" : "ARRAY<VARCHAR(2147483647)>"
+ } ]
+ },
+ "fields" : [ {
+ "name" : "list",
+ "conversionClass" : "java.util.List"
+ } ]
+ }
+ }
+ }, {
+ "name" : "map",
+ "attributeType" : {
+ "type" : "RAW",
+ "class" : "org.apache.flink.table.api.dataview.MapView",
+ "externalDataType" : {
+ "type" : "STRUCTURED_TYPE",
+ "implementationClass" : "org.apache.flink.table.api.dataview.MapView",
+ "attributes" : [ {
+ "name" : "map",
+ "attributeType" : "MAP<VARCHAR(2147483647), BOOLEAN>"
+ } ]
+ }
+ }
+ } ]
+ }
+ } ]
+ },
+ "description" : "Exchange(distribution=[hash[e]])"
+ }, {
+ "id" : 39,
+ "type" : "stream-exec-global-group-aggregate_1",
+ "configuration" : {
+ "table.exec.mini-batch.enabled" : "true",
+ "table.exec.mini-batch.size" : "5"
+ },
+ "grouping" : [ 0 ],
+ "aggCalls" : [ {
+ "name" : "s1",
+ "catalogName" : "`default_catalog`.`default_database`.`my_avg`",
+ "argList" : [ 0, 1 ],
+ "filterArg" : -1,
+ "distinct" : false,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "BIGINT"
+ }, {
+ "name" : "c1",
+ "systemName" : "my_concat",
+ "argList" : [ 2 ],
+ "filterArg" : -1,
+ "distinct" : false,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "VARCHAR(2147483647)"
+ } ],
+ "aggCallNeedRetractions" : [ false, false ],
+ "localAggInputRowType" : "ROW<`e` BIGINT, `$f1` BIGINT, `d` VARCHAR(2147483647)>",
+ "generateUpdateBefore" : false,
+ "needRetraction" : false,
+ "state" : [ {
+ "index" : 0,
+ "ttl" : "0 ms",
+ "name" : "globalGroupAggregateState"
+ } ],
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`e` BIGINT, `s1` BIGINT, `c1` VARCHAR(2147483647)>",
+ "description" : "GlobalGroupAggregate(groupBy=[e], select=[e, my_avg(my_avg$0) AS s1, my_concat(my_concat$1) AS c1])"
+ }, {
+ "id" : 40,
+ "type" : "stream-exec-sink_1",
+ "configuration" : {
+ "table.exec.sink.keyed-shuffle" : "AUTO",
+ "table.exec.sink.not-null-enforcer" : "ERROR",
+ "table.exec.sink.rowtime-inserter" : "ENABLED",
+ "table.exec.sink.type-length-enforcer" : "IGNORE",
+ "table.exec.sink.upsert-materialize" : "AUTO"
+ },
+ "dynamicTableSink" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`sink_t`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "d",
+ "dataType" : "BIGINT NOT NULL"
+ }, {
+ "name" : "s1",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "c1",
+ "dataType" : "VARCHAR(2147483647)"
+ } ],
+ "watermarkSpecs" : [ ],
+ "primaryKey" : {
+ "name" : "PK_131",
+ "type" : "PRIMARY_KEY",
+ "columns" : [ "d" ]
+ }
+ },
+ "partitionKeys" : [ ]
+ }
+ }
+ },
+ "inputChangelogMode" : [ "INSERT", "UPDATE_AFTER" ],
+ "inputUpsertKey" : [ 0 ],
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`e` BIGINT, `s1` BIGINT, `c1` VARCHAR(2147483647)>",
+ "description" : "Sink(table=[default_catalog.default_database.sink_t], fields=[e, s1, c1])"
+ } ],
+ "edges" : [ {
+ "source" : 34,
+ "target" : 35,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 35,
+ "target" : 36,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 36,
+ "target" : 37,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 37,
+ "target" : 38,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 38,
+ "target" : 39,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 39,
+ "target" : 40,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ } ]
+}
\ No newline at end of file
diff --git a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-group-aggregate_1/group-aggregate-udf-with-merge-mini-batch/savepoint/_metadata b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-group-aggregate_1/group-aggregate-udf-with-merge-mini-batch/savepoint/_metadata
new file mode 100644
index 00000000000..e9e16e36ff9
Binary files /dev/null and b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-group-aggregate_1/group-aggregate-udf-with-merge-mini-batch/savepoint/_metadata differ
diff --git a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-group-aggregate_1/group-aggregate-udf-with-merge/plan/group-aggregate-udf-with-merge.json b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-group-aggregate_1/group-aggregate-udf-with-merge/plan/group-aggregate-udf-with-merge.json
new file mode 100644
index 00000000000..e1024872cee
--- /dev/null
+++ b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-group-aggregate_1/group-aggregate-udf-with-merge/plan/group-aggregate-udf-with-merge.json
@@ -0,0 +1,211 @@
+{
+ "flinkVersion" : "1.19",
+ "nodes" : [ {
+ "id" : 29,
+ "type" : "stream-exec-table-source-scan_1",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`source_t`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "a",
+ "dataType" : "INT"
+ }, {
+ "name" : "b",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "c",
+ "dataType" : "INT"
+ }, {
+ "name" : "d",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "e",
+ "dataType" : "BIGINT"
+ } ],
+ "watermarkSpecs" : [ ]
+ },
+ "partitionKeys" : [ ]
+ }
+ },
+ "abilities" : [ {
+ "type" : "ProjectPushDown",
+ "projectedFields" : [ [ 4 ], [ 0 ], [ 3 ] ],
+ "producedType" : "ROW<`e` BIGINT, `a` INT, `d` VARCHAR(2147483647)> NOT NULL"
+ }, {
+ "type" : "ReadingMetadata",
+ "metadataKeys" : [ ],
+ "producedType" : "ROW<`e` BIGINT, `a` INT, `d` VARCHAR(2147483647)> NOT NULL"
+ } ]
+ },
+ "outputType" : "ROW<`e` BIGINT, `a` INT, `d` VARCHAR(2147483647)>",
+ "description" : "TableSourceScan(table=[[default_catalog, default_database, source_t, project=[e, a, d], metadata=[]]], fields=[e, a, d])",
+ "inputProperties" : [ ]
+ }, {
+ "id" : 30,
+ "type" : "stream-exec-calc_1",
+ "projection" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "BIGINT"
+ }, {
+ "kind" : "CALL",
+ "syntax" : "SPECIAL",
+ "internalName" : "$CAST$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "INT"
+ } ],
+ "type" : "BIGINT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "VARCHAR(2147483647)"
+ } ],
+ "condition" : null,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`e` BIGINT, `$f1` BIGINT, `d` VARCHAR(2147483647)>",
+ "description" : "Calc(select=[e, CAST(a AS BIGINT) AS $f1, d])"
+ }, {
+ "id" : 31,
+ "type" : "stream-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`e` BIGINT, `$f1` BIGINT, `d` VARCHAR(2147483647)>",
+ "description" : "Exchange(distribution=[hash[e]])"
+ }, {
+ "id" : 32,
+ "type" : "stream-exec-group-aggregate_1",
+ "configuration" : {
+ "table.exec.mini-batch.enabled" : "false",
+ "table.exec.mini-batch.size" : "-1"
+ },
+ "grouping" : [ 0 ],
+ "aggCalls" : [ {
+ "name" : "s1",
+ "catalogName" : "`default_catalog`.`default_database`.`my_avg`",
+ "argList" : [ 0, 1 ],
+ "filterArg" : -1,
+ "distinct" : false,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "BIGINT"
+ }, {
+ "name" : "c1",
+ "systemName" : "my_concat",
+ "argList" : [ 2 ],
+ "filterArg" : -1,
+ "distinct" : false,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "VARCHAR(2147483647)"
+ } ],
+ "aggCallNeedRetractions" : [ false, false ],
+ "generateUpdateBefore" : false,
+ "needRetraction" : false,
+ "state" : [ {
+ "index" : 0,
+ "ttl" : "0 ms",
+ "name" : "groupAggregateState"
+ } ],
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`e` BIGINT, `s1` BIGINT, `c1` VARCHAR(2147483647)>",
+ "description" : "GroupAggregate(groupBy=[e], select=[e, my_avg(e, $f1) AS s1, my_concat(d) AS c1])"
+ }, {
+ "id" : 33,
+ "type" : "stream-exec-sink_1",
+ "configuration" : {
+ "table.exec.sink.keyed-shuffle" : "AUTO",
+ "table.exec.sink.not-null-enforcer" : "ERROR",
+ "table.exec.sink.rowtime-inserter" : "ENABLED",
+ "table.exec.sink.type-length-enforcer" : "IGNORE",
+ "table.exec.sink.upsert-materialize" : "AUTO"
+ },
+ "dynamicTableSink" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`sink_t`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "d",
+ "dataType" : "BIGINT NOT NULL"
+ }, {
+ "name" : "s1",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "c1",
+ "dataType" : "VARCHAR(2147483647)"
+ } ],
+ "watermarkSpecs" : [ ],
+ "primaryKey" : {
+ "name" : "PK_131",
+ "type" : "PRIMARY_KEY",
+ "columns" : [ "d" ]
+ }
+ },
+ "partitionKeys" : [ ]
+ }
+ }
+ },
+ "inputChangelogMode" : [ "INSERT", "UPDATE_AFTER" ],
+ "inputUpsertKey" : [ 0 ],
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`e` BIGINT, `s1` BIGINT, `c1` VARCHAR(2147483647)>",
+ "description" : "Sink(table=[default_catalog.default_database.sink_t], fields=[e, s1, c1])"
+ } ],
+ "edges" : [ {
+ "source" : 29,
+ "target" : 30,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 30,
+ "target" : 31,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 31,
+ "target" : 32,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 32,
+ "target" : 33,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ } ]
+}
\ No newline at end of file
diff --git a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-group-aggregate_1/group-aggregate-udf-with-merge/savepoint/_metadata b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-group-aggregate_1/group-aggregate-udf-with-merge/savepoint/_metadata
new file mode 100644
index 00000000000..4a255b04b39
Binary files /dev/null and b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-group-aggregate_1/group-aggregate-udf-with-merge/savepoint/_metadata differ
diff --git a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-group-aggregate_1/group-aggregate-udf-without-merge-mini-batch/plan/group-aggregate-udf-without-merge-mini-batch.json b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-group-aggregate_1/group-aggregate-udf-without-merge-mini-batch/plan/group-aggregate-udf-without-merge-mini-batch.json
new file mode 100644
index 00000000000..baa54a95dac
--- /dev/null
+++ b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-group-aggregate_1/group-aggregate-udf-without-merge-mini-batch/plan/group-aggregate-udf-without-merge-mini-batch.json
@@ -0,0 +1,254 @@
+{
+ "flinkVersion" : "1.19",
+ "nodes" : [ {
+ "id" : 46,
+ "type" : "stream-exec-table-source-scan_1",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`source_t`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "a",
+ "dataType" : "INT"
+ }, {
+ "name" : "b",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "c",
+ "dataType" : "INT"
+ }, {
+ "name" : "d",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "e",
+ "dataType" : "BIGINT"
+ } ],
+ "watermarkSpecs" : [ ]
+ },
+ "partitionKeys" : [ ]
+ }
+ },
+ "abilities" : [ {
+ "type" : "ProjectPushDown",
+ "projectedFields" : [ [ 4 ], [ 2 ], [ 0 ] ],
+ "producedType" : "ROW<`e` BIGINT, `c` INT, `a` INT> NOT NULL"
+ }, {
+ "type" : "ReadingMetadata",
+ "metadataKeys" : [ ],
+ "producedType" : "ROW<`e` BIGINT, `c` INT, `a` INT> NOT NULL"
+ } ]
+ },
+ "outputType" : "ROW<`e` BIGINT, `c` INT, `a` INT>",
+ "description" : "TableSourceScan(table=[[default_catalog, default_database, source_t, project=[e, c, a], metadata=[]]], fields=[e, c, a])",
+ "inputProperties" : [ ]
+ }, {
+ "id" : 47,
+ "type" : "stream-exec-mini-batch-assigner_1",
+ "miniBatchInterval" : {
+ "interval" : 10000,
+ "mode" : "ProcTime"
+ },
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`e` BIGINT, `c` INT, `a` INT>",
+ "description" : "MiniBatchAssigner(interval=[10000ms], mode=[ProcTime])"
+ }, {
+ "id" : 48,
+ "type" : "stream-exec-calc_1",
+ "projection" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "BIGINT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "INT"
+ }, {
+ "kind" : "LITERAL",
+ "value" : 10,
+ "type" : "INT NOT NULL"
+ }, {
+ "kind" : "LITERAL",
+ "value" : 5,
+ "type" : "INT NOT NULL"
+ }, {
+ "kind" : "CALL",
+ "syntax" : "SPECIAL",
+ "internalName" : "$CAST$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "INT"
+ } ],
+ "type" : "BIGINT"
+ } ],
+ "condition" : null,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`e` BIGINT, `c` INT, `$f2` INT NOT NULL, `$f3` INT NOT NULL, `$f4` BIGINT>",
+ "description" : "Calc(select=[e, c, 10 AS $f2, 5 AS $f3, CAST(a AS BIGINT) AS $f4])"
+ }, {
+ "id" : 49,
+ "type" : "stream-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`e` BIGINT, `c` INT, `$f2` INT NOT NULL, `$f3` INT NOT NULL, `$f4` BIGINT>",
+ "description" : "Exchange(distribution=[hash[e]])"
+ }, {
+ "id" : 50,
+ "type" : "stream-exec-group-aggregate_1",
+ "configuration" : {
+ "table.exec.mini-batch.enabled" : "true",
+ "table.exec.mini-batch.size" : "5"
+ },
+ "grouping" : [ 0 ],
+ "aggCalls" : [ {
+ "name" : "s1",
+ "systemName" : "my_sum1",
+ "argList" : [ 1, 2 ],
+ "filterArg" : -1,
+ "distinct" : false,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "BIGINT"
+ }, {
+ "name" : "s2",
+ "systemName" : "my_sum2",
+ "argList" : [ 3, 1 ],
+ "filterArg" : -1,
+ "distinct" : false,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "BIGINT"
+ }, {
+ "name" : "s3",
+ "catalogName" : "`default_catalog`.`default_database`.`my_avg`",
+ "argList" : [ 0, 4 ],
+ "filterArg" : -1,
+ "distinct" : false,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "BIGINT"
+ } ],
+ "aggCallNeedRetractions" : [ false, false, false ],
+ "generateUpdateBefore" : false,
+ "needRetraction" : false,
+ "state" : [ {
+ "index" : 0,
+ "ttl" : "0 ms",
+ "name" : "groupAggregateState"
+ } ],
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`e` BIGINT, `s1` BIGINT, `s2` BIGINT, `s3` BIGINT>",
+ "description" : "GroupAggregate(groupBy=[e], select=[e, my_sum1(c, $f2) AS s1, my_sum2($f3, c) AS s2, my_avg(e, $f4) AS s3])"
+ }, {
+ "id" : 51,
+ "type" : "stream-exec-sink_1",
+ "configuration" : {
+ "table.exec.sink.keyed-shuffle" : "AUTO",
+ "table.exec.sink.not-null-enforcer" : "ERROR",
+ "table.exec.sink.rowtime-inserter" : "ENABLED",
+ "table.exec.sink.type-length-enforcer" : "IGNORE",
+ "table.exec.sink.upsert-materialize" : "AUTO"
+ },
+ "dynamicTableSink" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`sink_t`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "d",
+ "dataType" : "BIGINT NOT NULL"
+ }, {
+ "name" : "s1",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "s2",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "s3",
+ "dataType" : "BIGINT"
+ } ],
+ "watermarkSpecs" : [ ],
+ "primaryKey" : {
+ "name" : "PK_131",
+ "type" : "PRIMARY_KEY",
+ "columns" : [ "d" ]
+ }
+ },
+ "partitionKeys" : [ ]
+ }
+ }
+ },
+ "inputChangelogMode" : [ "INSERT", "UPDATE_AFTER" ],
+ "inputUpsertKey" : [ 0 ],
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`e` BIGINT, `s1` BIGINT, `s2` BIGINT, `s3` BIGINT>",
+ "description" : "Sink(table=[default_catalog.default_database.sink_t], fields=[e, s1, s2, s3])"
+ } ],
+ "edges" : [ {
+ "source" : 46,
+ "target" : 47,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 47,
+ "target" : 48,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 48,
+ "target" : 49,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 49,
+ "target" : 50,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 50,
+ "target" : 51,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ } ]
+}
\ No newline at end of file
diff --git a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-group-aggregate_1/group-aggregate-udf-without-merge-mini-batch/savepoint/_metadata b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-group-aggregate_1/group-aggregate-udf-without-merge-mini-batch/savepoint/_metadata
new file mode 100644
index 00000000000..ecae71f5db5
Binary files /dev/null and b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-group-aggregate_1/group-aggregate-udf-without-merge-mini-batch/savepoint/_metadata differ
diff --git a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-group-aggregate_1/group-aggregate-udf-without-merge/plan/group-aggregate-udf-without-merge.json b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-group-aggregate_1/group-aggregate-udf-without-merge/plan/group-aggregate-udf-without-merge.json
new file mode 100644
index 00000000000..53b2e005acb
--- /dev/null
+++ b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-group-aggregate_1/group-aggregate-udf-without-merge/plan/group-aggregate-udf-without-merge.json
@@ -0,0 +1,231 @@
+{
+ "flinkVersion" : "1.19",
+ "nodes" : [ {
+ "id" : 41,
+ "type" : "stream-exec-table-source-scan_1",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`source_t`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "a",
+ "dataType" : "INT"
+ }, {
+ "name" : "b",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "c",
+ "dataType" : "INT"
+ }, {
+ "name" : "d",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "e",
+ "dataType" : "BIGINT"
+ } ],
+ "watermarkSpecs" : [ ]
+ },
+ "partitionKeys" : [ ]
+ }
+ },
+ "abilities" : [ {
+ "type" : "ProjectPushDown",
+ "projectedFields" : [ [ 4 ], [ 2 ], [ 0 ] ],
+ "producedType" : "ROW<`e` BIGINT, `c` INT, `a` INT> NOT NULL"
+ }, {
+ "type" : "ReadingMetadata",
+ "metadataKeys" : [ ],
+ "producedType" : "ROW<`e` BIGINT, `c` INT, `a` INT> NOT NULL"
+ } ]
+ },
+ "outputType" : "ROW<`e` BIGINT, `c` INT, `a` INT>",
+ "description" : "TableSourceScan(table=[[default_catalog, default_database, source_t, project=[e, c, a], metadata=[]]], fields=[e, c, a])",
+ "inputProperties" : [ ]
+ }, {
+ "id" : 42,
+ "type" : "stream-exec-calc_1",
+ "projection" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "BIGINT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "INT"
+ }, {
+ "kind" : "LITERAL",
+ "value" : 10,
+ "type" : "INT NOT NULL"
+ }, {
+ "kind" : "LITERAL",
+ "value" : 5,
+ "type" : "INT NOT NULL"
+ }, {
+ "kind" : "CALL",
+ "syntax" : "SPECIAL",
+ "internalName" : "$CAST$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "INT"
+ } ],
+ "type" : "BIGINT"
+ } ],
+ "condition" : null,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`e` BIGINT, `c` INT, `$f2` INT NOT NULL, `$f3` INT NOT NULL, `$f4` BIGINT>",
+ "description" : "Calc(select=[e, c, 10 AS $f2, 5 AS $f3, CAST(a AS BIGINT) AS $f4])"
+ }, {
+ "id" : 43,
+ "type" : "stream-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`e` BIGINT, `c` INT, `$f2` INT NOT NULL, `$f3` INT NOT NULL, `$f4` BIGINT>",
+ "description" : "Exchange(distribution=[hash[e]])"
+ }, {
+ "id" : 44,
+ "type" : "stream-exec-group-aggregate_1",
+ "configuration" : {
+ "table.exec.mini-batch.enabled" : "false",
+ "table.exec.mini-batch.size" : "-1"
+ },
+ "grouping" : [ 0 ],
+ "aggCalls" : [ {
+ "name" : "s1",
+ "systemName" : "my_sum1",
+ "argList" : [ 1, 2 ],
+ "filterArg" : -1,
+ "distinct" : false,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "BIGINT"
+ }, {
+ "name" : "s2",
+ "systemName" : "my_sum2",
+ "argList" : [ 3, 1 ],
+ "filterArg" : -1,
+ "distinct" : false,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "BIGINT"
+ }, {
+ "name" : "s3",
+ "catalogName" : "`default_catalog`.`default_database`.`my_avg`",
+ "argList" : [ 0, 4 ],
+ "filterArg" : -1,
+ "distinct" : false,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "BIGINT"
+ } ],
+ "aggCallNeedRetractions" : [ false, false, false ],
+ "generateUpdateBefore" : false,
+ "needRetraction" : false,
+ "state" : [ {
+ "index" : 0,
+ "ttl" : "0 ms",
+ "name" : "groupAggregateState"
+ } ],
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`e` BIGINT, `s1` BIGINT, `s2` BIGINT, `s3` BIGINT>",
+ "description" : "GroupAggregate(groupBy=[e], select=[e, my_sum1(c, $f2) AS s1, my_sum2($f3, c) AS s2, my_avg(e, $f4) AS s3])"
+ }, {
+ "id" : 45,
+ "type" : "stream-exec-sink_1",
+ "configuration" : {
+ "table.exec.sink.keyed-shuffle" : "AUTO",
+ "table.exec.sink.not-null-enforcer" : "ERROR",
+ "table.exec.sink.rowtime-inserter" : "ENABLED",
+ "table.exec.sink.type-length-enforcer" : "IGNORE",
+ "table.exec.sink.upsert-materialize" : "AUTO"
+ },
+ "dynamicTableSink" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`sink_t`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "d",
+ "dataType" : "BIGINT NOT NULL"
+ }, {
+ "name" : "s1",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "s2",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "s3",
+ "dataType" : "BIGINT"
+ } ],
+ "watermarkSpecs" : [ ],
+ "primaryKey" : {
+ "name" : "PK_131",
+ "type" : "PRIMARY_KEY",
+ "columns" : [ "d" ]
+ }
+ },
+ "partitionKeys" : [ ]
+ }
+ }
+ },
+ "inputChangelogMode" : [ "INSERT", "UPDATE_AFTER" ],
+ "inputUpsertKey" : [ 0 ],
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`e` BIGINT, `s1` BIGINT, `s2` BIGINT, `s3` BIGINT>",
+ "description" : "Sink(table=[default_catalog.default_database.sink_t], fields=[e, s1, s2, s3])"
+ } ],
+ "edges" : [ {
+ "source" : 41,
+ "target" : 42,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 42,
+ "target" : 43,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 43,
+ "target" : 44,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 44,
+ "target" : 45,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ } ]
+}
\ No newline at end of file
diff --git a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-group-aggregate_1/group-aggregate-udf-without-merge/savepoint/_metadata b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-group-aggregate_1/group-aggregate-udf-without-merge/savepoint/_metadata
new file mode 100644
index 00000000000..c41a3a06ef4
Binary files /dev/null and b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-group-aggregate_1/group-aggregate-udf-without-merge/savepoint/_metadata differ