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/11/21 17:28:22 UTC
(flink) branch master updated (278504a2787 -> e0d87b3a9a5)
This is an automated email from the ASF dual-hosted git repository.
dwysakowicz pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
from 278504a2787 [FLINK-27882][tests][table] Migrate flink-scala to JUnit5
new 01864277016 [FLINK-33601] Implement restore tests for Expand node
new e0d87b3a9a5 [FLINK-33601] Deleting ExpandJsonPlanTest.java and ExpandJsonPlanITCase.java
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../plan/nodes/exec/stream/ExpandJsonPlanTest.java | 78 -------------
.../ExpandRestoreTest.java} | 16 +--
.../nodes/exec/testutils/ExpandTestPrograms.java | 76 ++++++++++++
.../plan/nodes/exec/testutils/RestoreTestBase.java | 2 +
.../stream/jsonplan/ExpandJsonPlanITCase.java | 64 -----------
.../stream-exec-expand_1/expand/plan/expand.json} | 127 ++++++++++++++-------
.../expand/savepoint/_metadata | Bin 0 -> 33424 bytes
7 files changed, 170 insertions(+), 193 deletions(-)
delete mode 100644 flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ExpandJsonPlanTest.java
copy flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/{stream/LimitRestoreTest.java => testutils/ExpandRestoreTest.java} (69%)
create mode 100644 flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/testutils/ExpandTestPrograms.java
delete mode 100644 flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/jsonplan/ExpandJsonPlanITCase.java
rename flink-table/flink-table-planner/src/test/resources/{org/apache/flink/table/planner/plan/nodes/exec/stream/ExpandJsonPlanTest_jsonplan/testExpand.out => restore-tests/stream-exec-expand_1/expand/plan/expand.json} (78%)
create mode 100644 flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-expand_1/expand/savepoint/_metadata
(flink) 02/02: [FLINK-33601] Deleting ExpandJsonPlanTest.java and ExpandJsonPlanITCase.java
Posted by dw...@apache.org.
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 e0d87b3a9a587815f1c986a3de5bc60842dc617e
Author: Jim Hughes <jh...@confluent.io>
AuthorDate: Mon Nov 20 17:59:40 2023 -0500
[FLINK-33601] Deleting ExpandJsonPlanTest.java and ExpandJsonPlanITCase.java
---
.../plan/nodes/exec/stream/ExpandJsonPlanTest.java | 78 ----
.../stream/jsonplan/ExpandJsonPlanITCase.java | 64 ---
.../ExpandJsonPlanTest_jsonplan/testExpand.out | 439 ---------------------
3 files changed, 581 deletions(-)
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ExpandJsonPlanTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ExpandJsonPlanTest.java
deleted file mode 100644
index 9d568e6417e..00000000000
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ExpandJsonPlanTest.java
+++ /dev/null
@@ -1,78 +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.flink.table.planner.plan.nodes.exec.stream;
-
-import org.apache.flink.table.api.TableConfig;
-import org.apache.flink.table.api.TableEnvironment;
-import org.apache.flink.table.api.config.OptimizerConfigOptions;
-import org.apache.flink.table.planner.plan.rules.physical.stream.IncrementalAggregateRule;
-import org.apache.flink.table.planner.utils.AggregatePhaseStrategy;
-import org.apache.flink.table.planner.utils.StreamTableTestUtil;
-import org.apache.flink.table.planner.utils.TableTestBase;
-
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-
-/** Test json serialization/deserialization for expand. */
-class ExpandJsonPlanTest extends TableTestBase {
-
- private StreamTableTestUtil util;
- private TableEnvironment tEnv;
-
- @BeforeEach
- void setup() {
- util = streamTestUtil(TableConfig.getDefault());
- tEnv = util.getTableEnv();
- tEnv.getConfig()
- .set(
- OptimizerConfigOptions.TABLE_OPTIMIZER_AGG_PHASE_STRATEGY,
- AggregatePhaseStrategy.ONE_PHASE.name())
- .set(OptimizerConfigOptions.TABLE_OPTIMIZER_DISTINCT_AGG_SPLIT_ENABLED, true)
- .set(IncrementalAggregateRule.TABLE_OPTIMIZER_INCREMENTAL_AGG_ENABLED(), false);
-
- String srcTableDdl =
- "CREATE TABLE MyTable (\n"
- + " a bigint,\n"
- + " b int not null,\n"
- + " c varchar\n"
- + ") with (\n"
- + " 'connector' = 'values',\n"
- + " 'bounded' = 'false')";
- tEnv.executeSql(srcTableDdl);
- }
-
- @Test
- void testExpand() {
- String sinkTableDdl =
- "CREATE TABLE MySink (\n"
- + " a bigint,\n"
- + " b bigint not null,\n"
- + " c varchar\n"
- + ") with (\n"
- + " 'connector' = 'values',\n"
- + " 'sink-insert-only' = 'false',\n"
- + " 'table-sink-class' = 'DEFAULT')";
- tEnv.executeSql(sinkTableDdl);
- util.verifyJsonPlan(
- "insert into MySink select a, "
- + "count(distinct b) as b, "
- + "first_value(c) c "
- + "from MyTable group by a");
- }
-}
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/jsonplan/ExpandJsonPlanITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/jsonplan/ExpandJsonPlanITCase.java
deleted file mode 100644
index 80109626f7b..00000000000
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/jsonplan/ExpandJsonPlanITCase.java
+++ /dev/null
@@ -1,64 +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.flink.table.planner.runtime.stream.jsonplan;
-
-import org.apache.flink.table.api.config.OptimizerConfigOptions;
-import org.apache.flink.table.planner.factories.TestValuesTableFactory;
-import org.apache.flink.table.planner.plan.rules.physical.stream.IncrementalAggregateRule;
-import org.apache.flink.table.planner.runtime.utils.TestData;
-import org.apache.flink.table.planner.utils.AggregatePhaseStrategy;
-import org.apache.flink.table.planner.utils.JavaScalaConversionUtil;
-import org.apache.flink.table.planner.utils.JsonPlanTestBase;
-
-import org.junit.jupiter.api.Test;
-
-import java.util.Arrays;
-import java.util.List;
-
-/** Test for expand json plan. */
-class ExpandJsonPlanITCase extends JsonPlanTestBase {
-
- @Test
- void testExpand() throws Exception {
- tableEnv.getConfig()
- .set(
- OptimizerConfigOptions.TABLE_OPTIMIZER_AGG_PHASE_STRATEGY,
- AggregatePhaseStrategy.ONE_PHASE.name())
- .set(OptimizerConfigOptions.TABLE_OPTIMIZER_DISTINCT_AGG_SPLIT_ENABLED, true)
- .set(IncrementalAggregateRule.TABLE_OPTIMIZER_INCREMENTAL_AGG_ENABLED(), false);
-
- createTestValuesSourceTable(
- "MyTable",
- JavaScalaConversionUtil.toJava(TestData.smallData3()),
- "a int",
- "b bigint",
- "c varchar");
- createTestNonInsertOnlyValuesSinkTable(
- "MySink", "b bigint", "a bigint", "c varchar", "primary key (b) not enforced");
- compileSqlAndExecutePlan(
- "insert into MySink select b, "
- + "count(distinct a) as a, "
- + "max(c) as c "
- + "from MyTable group by b")
- .await();
-
- List<String> result = TestValuesTableFactory.getResultsAsStrings("MySink");
- assertResult(Arrays.asList("+I[1, 1, Hi]", "+I[2, 2, Hello world]"), result);
- }
-}
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/ExpandJsonPlanTest_jsonplan/testExpand.out b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/ExpandJsonPlanTest_jsonplan/testExpand.out
deleted file mode 100644
index feb6c3119e5..00000000000
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/ExpandJsonPlanTest_jsonplan/testExpand.out
+++ /dev/null
@@ -1,439 +0,0 @@
-{
- "flinkVersion" : "",
- "nodes" : [ {
- "id" : 1,
- "type" : "stream-exec-table-source-scan_1",
- "scanTableSource" : {
- "table" : {
- "identifier" : "`default_catalog`.`default_database`.`MyTable`",
- "resolvedTable" : {
- "schema" : {
- "columns" : [ {
- "name" : "a",
- "dataType" : "BIGINT"
- }, {
- "name" : "b",
- "dataType" : "INT NOT NULL"
- }, {
- "name" : "c",
- "dataType" : "VARCHAR(2147483647)"
- } ],
- "watermarkSpecs" : [ ]
- },
- "partitionKeys" : [ ],
- "options" : {
- "bounded" : "false",
- "connector" : "values"
- }
- }
- }
- },
- "outputType" : "ROW<`a` BIGINT, `b` INT NOT NULL, `c` VARCHAR(2147483647)>",
- "description" : "TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c])",
- "inputProperties" : [ ]
- }, {
- "id" : 2,
- "type" : "stream-exec-calc_1",
- "projection" : [ {
- "kind" : "INPUT_REF",
- "inputIndex" : 0,
- "type" : "BIGINT"
- }, {
- "kind" : "INPUT_REF",
- "inputIndex" : 1,
- "type" : "INT NOT NULL"
- }, {
- "kind" : "INPUT_REF",
- "inputIndex" : 2,
- "type" : "VARCHAR(2147483647)"
- }, {
- "kind" : "CALL",
- "internalName" : "$MOD$1",
- "operands" : [ {
- "kind" : "CALL",
- "internalName" : "$HASH_CODE$1",
- "operands" : [ {
- "kind" : "INPUT_REF",
- "inputIndex" : 1,
- "type" : "INT NOT NULL"
- } ],
- "type" : "INT NOT NULL"
- }, {
- "kind" : "LITERAL",
- "value" : 1024,
- "type" : "INT NOT NULL"
- } ],
- "type" : "INT NOT NULL"
- }, {
- "kind" : "CALL",
- "internalName" : "$MOD$1",
- "operands" : [ {
- "kind" : "CALL",
- "internalName" : "$HASH_CODE$1",
- "operands" : [ {
- "kind" : "INPUT_REF",
- "inputIndex" : 2,
- "type" : "VARCHAR(2147483647)"
- } ],
- "type" : "INT"
- }, {
- "kind" : "LITERAL",
- "value" : 1024,
- "type" : "INT NOT NULL"
- } ],
- "type" : "INT"
- } ],
- "condition" : null,
- "inputProperties" : [ {
- "requiredDistribution" : {
- "type" : "UNKNOWN"
- },
- "damBehavior" : "PIPELINED",
- "priority" : 0
- } ],
- "outputType" : "ROW<`a` BIGINT, `b` INT NOT NULL, `c` VARCHAR(2147483647), `$f3` INT NOT NULL, `$f4` INT>",
- "description" : "Calc(select=[a, b, c, MOD(HASH_CODE(b), 1024) AS $f3, MOD(HASH_CODE(c), 1024) AS $f4])"
- }, {
- "id" : 3,
- "type" : "stream-exec-expand_1",
- "projects" : [ [ {
- "kind" : "INPUT_REF",
- "inputIndex" : 0,
- "type" : "BIGINT"
- }, {
- "kind" : "INPUT_REF",
- "inputIndex" : 1,
- "type" : "INT NOT NULL"
- }, {
- "kind" : "INPUT_REF",
- "inputIndex" : 2,
- "type" : "VARCHAR(2147483647)"
- }, {
- "kind" : "INPUT_REF",
- "inputIndex" : 3,
- "type" : "INT NOT NULL"
- }, {
- "kind" : "LITERAL",
- "value" : null,
- "type" : "INT"
- }, {
- "kind" : "LITERAL",
- "value" : 1,
- "type" : "BIGINT NOT NULL"
- } ], [ {
- "kind" : "INPUT_REF",
- "inputIndex" : 0,
- "type" : "BIGINT"
- }, {
- "kind" : "INPUT_REF",
- "inputIndex" : 1,
- "type" : "INT NOT NULL"
- }, {
- "kind" : "INPUT_REF",
- "inputIndex" : 2,
- "type" : "VARCHAR(2147483647)"
- }, {
- "kind" : "LITERAL",
- "value" : null,
- "type" : "INT"
- }, {
- "kind" : "INPUT_REF",
- "inputIndex" : 4,
- "type" : "INT"
- }, {
- "kind" : "LITERAL",
- "value" : 2,
- "type" : "BIGINT NOT NULL"
- } ] ],
- "inputProperties" : [ {
- "requiredDistribution" : {
- "type" : "UNKNOWN"
- },
- "damBehavior" : "PIPELINED",
- "priority" : 0
- } ],
- "outputType" : "ROW<`a` BIGINT, `b` INT NOT NULL, `c` VARCHAR(2147483647), `$f3` INT, `$f4` INT, `$e` BIGINT NOT NULL>",
- "description" : "Expand(projects=[{a, b, c, $f3, null AS $f4, 1 AS $e}, {a, b, c, null AS $f3, $f4, 2 AS $e}])"
- }, {
- "id" : 4,
- "type" : "stream-exec-calc_1",
- "projection" : [ {
- "kind" : "INPUT_REF",
- "inputIndex" : 0,
- "type" : "BIGINT"
- }, {
- "kind" : "INPUT_REF",
- "inputIndex" : 1,
- "type" : "INT NOT NULL"
- }, {
- "kind" : "INPUT_REF",
- "inputIndex" : 2,
- "type" : "VARCHAR(2147483647)"
- }, {
- "kind" : "INPUT_REF",
- "inputIndex" : 3,
- "type" : "INT"
- }, {
- "kind" : "INPUT_REF",
- "inputIndex" : 4,
- "type" : "INT"
- }, {
- "kind" : "CALL",
- "syntax" : "BINARY",
- "internalName" : "$=$1",
- "operands" : [ {
- "kind" : "INPUT_REF",
- "inputIndex" : 5,
- "type" : "BIGINT NOT NULL"
- }, {
- "kind" : "LITERAL",
- "value" : 1,
- "type" : "INT NOT NULL"
- } ],
- "type" : "BOOLEAN NOT NULL"
- }, {
- "kind" : "CALL",
- "syntax" : "BINARY",
- "internalName" : "$=$1",
- "operands" : [ {
- "kind" : "INPUT_REF",
- "inputIndex" : 5,
- "type" : "BIGINT NOT NULL"
- }, {
- "kind" : "LITERAL",
- "value" : 2,
- "type" : "INT NOT NULL"
- } ],
- "type" : "BOOLEAN NOT NULL"
- } ],
- "condition" : null,
- "inputProperties" : [ {
- "requiredDistribution" : {
- "type" : "UNKNOWN"
- },
- "damBehavior" : "PIPELINED",
- "priority" : 0
- } ],
- "outputType" : "ROW<`a` BIGINT, `b` INT NOT NULL, `c` VARCHAR(2147483647), `$f3` INT, `$f4` INT, `$g_1` BOOLEAN NOT NULL, `$g_2` BOOLEAN NOT NULL>",
- "description" : "Calc(select=[a, b, c, $f3, $f4, ($e = 1) AS $g_1, ($e = 2) AS $g_2])"
- }, {
- "id" : 5,
- "type" : "stream-exec-exchange_1",
- "inputProperties" : [ {
- "requiredDistribution" : {
- "type" : "HASH",
- "keys" : [ 0, 3, 4 ]
- },
- "damBehavior" : "PIPELINED",
- "priority" : 0
- } ],
- "outputType" : "ROW<`a` BIGINT, `b` INT NOT NULL, `c` VARCHAR(2147483647), `$f3` INT, `$f4` INT, `$g_1` BOOLEAN NOT NULL, `$g_2` BOOLEAN NOT NULL>",
- "description" : "Exchange(distribution=[hash[a, $f3, $f4]])"
- }, {
- "id" : 6,
- "type" : "stream-exec-group-aggregate_1",
- "configuration" : {
- "table.exec.mini-batch.enabled" : "false",
- "table.exec.mini-batch.size" : "-1"
- },
- "grouping" : [ 0, 3, 4 ],
- "aggCalls" : [ {
- "name" : null,
- "syntax" : "FUNCTION_STAR",
- "internalName" : "$COUNT$1",
- "argList" : [ 1 ],
- "filterArg" : 5,
- "distinct" : true,
- "approximate" : false,
- "ignoreNulls" : false,
- "type" : "BIGINT NOT NULL"
- }, {
- "name" : null,
- "internalName" : "$FIRST_VALUE$1",
- "argList" : [ 2 ],
- "filterArg" : 6,
- "distinct" : false,
- "approximate" : false,
- "ignoreNulls" : false,
- "type" : "VARCHAR(2147483647)"
- } ],
- "aggCallNeedRetractions" : [ false, false ],
- "generateUpdateBefore" : true,
- "needRetraction" : false,
- "state" : [ {
- "index" : 0,
- "ttl" : "0 ms",
- "name" : "groupAggregateState"
- } ],
- "inputProperties" : [ {
- "requiredDistribution" : {
- "type" : "UNKNOWN"
- },
- "damBehavior" : "PIPELINED",
- "priority" : 0
- } ],
- "outputType" : "ROW<`a` BIGINT, `$f3` INT, `$f4` INT, `$f3_0` BIGINT NOT NULL, `$f4_0` VARCHAR(2147483647)>",
- "description" : "GroupAggregate(groupBy=[a, $f3, $f4], partialFinalType=[PARTIAL], select=[a, $f3, $f4, COUNT(DISTINCT b) FILTER $g_1 AS $f3_0, FIRST_VALUE(c) FILTER $g_2 AS $f4_0])"
- }, {
- "id" : 7,
- "type" : "stream-exec-exchange_1",
- "inputProperties" : [ {
- "requiredDistribution" : {
- "type" : "HASH",
- "keys" : [ 0 ]
- },
- "damBehavior" : "PIPELINED",
- "priority" : 0
- } ],
- "outputType" : "ROW<`a` BIGINT, `$f3` INT, `$f4` INT, `$f3_0` BIGINT NOT NULL, `$f4_0` VARCHAR(2147483647)>",
- "description" : "Exchange(distribution=[hash[a]])"
- }, {
- "id" : 8,
- "type" : "stream-exec-group-aggregate_1",
- "configuration" : {
- "table.exec.mini-batch.enabled" : "false",
- "table.exec.mini-batch.size" : "-1"
- },
- "grouping" : [ 0 ],
- "aggCalls" : [ {
- "name" : null,
- "internalName" : "$$SUM0$1",
- "argList" : [ 3 ],
- "filterArg" : -1,
- "distinct" : false,
- "approximate" : false,
- "ignoreNulls" : false,
- "type" : "BIGINT NOT NULL"
- }, {
- "name" : null,
- "internalName" : "$FIRST_VALUE$1",
- "argList" : [ 4 ],
- "filterArg" : -1,
- "distinct" : false,
- "approximate" : false,
- "ignoreNulls" : false,
- "type" : "VARCHAR(2147483647)"
- } ],
- "aggCallNeedRetractions" : [ true, true ],
- "generateUpdateBefore" : true,
- "needRetraction" : true,
- "state" : [ {
- "index" : 0,
- "ttl" : "0 ms",
- "name" : "groupAggregateState"
- } ],
- "inputProperties" : [ {
- "requiredDistribution" : {
- "type" : "UNKNOWN"
- },
- "damBehavior" : "PIPELINED",
- "priority" : 0
- } ],
- "outputType" : "ROW<`a` BIGINT, `$f1` BIGINT NOT NULL, `$f2` VARCHAR(2147483647)>",
- "description" : "GroupAggregate(groupBy=[a], partialFinalType=[FINAL], select=[a, $SUM0_RETRACT($f3_0) AS $f1, FIRST_VALUE_RETRACT($f4_0) AS $f2])"
- }, {
- "id" : 9,
- "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`.`MySink`",
- "resolvedTable" : {
- "schema" : {
- "columns" : [ {
- "name" : "a",
- "dataType" : "BIGINT"
- }, {
- "name" : "b",
- "dataType" : "BIGINT NOT NULL"
- }, {
- "name" : "c",
- "dataType" : "VARCHAR(2147483647)"
- } ],
- "watermarkSpecs" : [ ]
- },
- "partitionKeys" : [ ],
- "options" : {
- "connector" : "values",
- "sink-insert-only" : "false",
- "table-sink-class" : "DEFAULT"
- }
- }
- }
- },
- "inputChangelogMode" : [ "INSERT", "UPDATE_BEFORE", "UPDATE_AFTER", "DELETE" ],
- "inputUpsertKey" : [ 0 ],
- "inputProperties" : [ {
- "requiredDistribution" : {
- "type" : "UNKNOWN"
- },
- "damBehavior" : "PIPELINED",
- "priority" : 0
- } ],
- "outputType" : "ROW<`a` BIGINT, `$f1` BIGINT NOT NULL, `$f2` VARCHAR(2147483647)>",
- "description" : "Sink(table=[default_catalog.default_database.MySink], fields=[a, $f1, $f2])"
- } ],
- "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"
- }, {
- "source" : 6,
- "target" : 7,
- "shuffle" : {
- "type" : "FORWARD"
- },
- "shuffleMode" : "PIPELINED"
- }, {
- "source" : 7,
- "target" : 8,
- "shuffle" : {
- "type" : "FORWARD"
- },
- "shuffleMode" : "PIPELINED"
- }, {
- "source" : 8,
- "target" : 9,
- "shuffle" : {
- "type" : "FORWARD"
- },
- "shuffleMode" : "PIPELINED"
- } ]
-}
(flink) 01/02: [FLINK-33601] Implement restore tests for Expand node
Posted by dw...@apache.org.
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 01864277016544641be81663d6274dd726f25843
Author: Jim Hughes <jh...@confluent.io>
AuthorDate: Mon Nov 20 17:40:32 2023 -0500
[FLINK-33601] Implement restore tests for Expand node
---
.../nodes/exec/testutils/ExpandRestoreTest.java | 38 ++
.../nodes/exec/testutils/ExpandTestPrograms.java | 76 ++++
.../plan/nodes/exec/testutils/RestoreTestBase.java | 2 +
.../stream-exec-expand_1/expand/plan/expand.json | 480 +++++++++++++++++++++
.../expand/savepoint/_metadata | Bin 0 -> 33424 bytes
5 files changed, 596 insertions(+)
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/testutils/ExpandRestoreTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/testutils/ExpandRestoreTest.java
new file mode 100644
index 00000000000..5ffe8a01792
--- /dev/null
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/testutils/ExpandRestoreTest.java
@@ -0,0 +1,38 @@
+/*
+ * 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.testutils;
+
+import org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecExpand;
+import org.apache.flink.table.test.program.TableTestProgram;
+
+import java.util.Collections;
+import java.util.List;
+
+/** Restore tests for {@link StreamExecExpand}. */
+public class ExpandRestoreTest extends RestoreTestBase {
+
+ public ExpandRestoreTest() {
+ super(StreamExecExpand.class);
+ }
+
+ @Override
+ public List<TableTestProgram> programs() {
+ return Collections.singletonList(ExpandTestPrograms.EXPAND);
+ }
+}
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/testutils/ExpandTestPrograms.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/testutils/ExpandTestPrograms.java
new file mode 100644
index 00000000000..216dce2b523
--- /dev/null
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/testutils/ExpandTestPrograms.java
@@ -0,0 +1,76 @@
+/*
+ * 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.testutils;
+
+import org.apache.flink.table.api.config.OptimizerConfigOptions;
+import org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecExpand;
+import org.apache.flink.table.planner.plan.rules.physical.stream.IncrementalAggregateRule;
+import org.apache.flink.table.planner.utils.AggregatePhaseStrategy;
+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 org.apache.flink.types.RowKind;
+
+/** {@link TableTestProgram} definitions for testing {@link StreamExecExpand}. */
+public class ExpandTestPrograms {
+
+ static final TableTestProgram EXPAND =
+ TableTestProgram.of("expand", "validates expand node")
+ .setupConfig(
+ OptimizerConfigOptions.TABLE_OPTIMIZER_AGG_PHASE_STRATEGY,
+ AggregatePhaseStrategy.ONE_PHASE.name())
+ .setupConfig(
+ OptimizerConfigOptions.TABLE_OPTIMIZER_DISTINCT_AGG_SPLIT_ENABLED, true)
+ .setupConfig(
+ IncrementalAggregateRule.TABLE_OPTIMIZER_INCREMENTAL_AGG_ENABLED(),
+ false)
+ .setupTableSource(
+ SourceTestStep.newBuilder("MyTable")
+ .addSchema("a int", "b bigint", "c varchar")
+ .producedBeforeRestore(
+ Row.of(1, 1L, "Hi"),
+ Row.of(2, 2L, "Hello"),
+ Row.of(2, 3L, "Hello world"))
+ .producedAfterRestore(Row.of(5, 6L, "Hello there"))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("MySink")
+ .addSchema(
+ "b bigint",
+ "a bigint",
+ "c varchar",
+ "primary key (b) not enforced")
+ .consumedBeforeRestore(
+ Row.of(1, 1L, null),
+ Row.ofKind(RowKind.UPDATE_AFTER, 1, 1L, "Hi"),
+ Row.of(2, 1L, null),
+ Row.ofKind(RowKind.UPDATE_AFTER, 2, 1L, "Hello"),
+ Row.ofKind(RowKind.UPDATE_AFTER, 2, 2L, "Hello"))
+ .consumedAfterRestore(
+ Row.of(5, 1L, null),
+ Row.ofKind(RowKind.UPDATE_AFTER, 5, 1L, "Hello there"))
+ .build())
+ .runSql(
+ "insert into MySink select a, "
+ + "count(distinct b) as b, "
+ + "first_value(c) c "
+ + "from MyTable group by a")
+ .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 f5c370668e2..59a72d28fff 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
@@ -89,6 +89,7 @@ public abstract class RestoreTestBase implements TableTestProgramRunner {
@Override
public EnumSet<TestKind> supportedSetupSteps() {
return EnumSet.of(
+ TestKind.CONFIG,
TestKind.FUNCTION,
TestKind.SOURCE_WITH_RESTORE_DATA,
TestKind.SINK_WITH_RESTORE_DATA);
@@ -126,6 +127,7 @@ public abstract class RestoreTestBase implements TableTestProgramRunner {
public void generateTestSetupFiles(TableTestProgram program) throws Exception {
final TableEnvironment tEnv =
TableEnvironment.create(EnvironmentSettings.inStreamingMode());
+ program.getSetupConfigOptionTestSteps().forEach(s -> s.apply(tEnv));
tEnv.getConfig()
.set(
TableConfigOptions.PLAN_COMPILE_CATALOG_OBJECTS,
diff --git a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-expand_1/expand/plan/expand.json b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-expand_1/expand/plan/expand.json
new file mode 100644
index 00000000000..d3805482fa3
--- /dev/null
+++ b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-expand_1/expand/plan/expand.json
@@ -0,0 +1,480 @@
+{
+ "flinkVersion" : "1.19",
+ "nodes" : [ {
+ "id" : 1,
+ "type" : "stream-exec-table-source-scan_1",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`MyTable`",
+ "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, MyTable]], fields=[a, b, c])",
+ "inputProperties" : [ ]
+ }, {
+ "id" : 2,
+ "type" : "stream-exec-calc_1",
+ "projection" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "INT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "BIGINT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "VARCHAR(2147483647)"
+ }, {
+ "kind" : "CALL",
+ "internalName" : "$MOD$1",
+ "operands" : [ {
+ "kind" : "CALL",
+ "internalName" : "$HASH_CODE$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "BIGINT"
+ } ],
+ "type" : "INT"
+ }, {
+ "kind" : "LITERAL",
+ "value" : 1024,
+ "type" : "INT NOT NULL"
+ } ],
+ "type" : "INT"
+ }, {
+ "kind" : "CALL",
+ "internalName" : "$MOD$1",
+ "operands" : [ {
+ "kind" : "CALL",
+ "internalName" : "$HASH_CODE$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "VARCHAR(2147483647)"
+ } ],
+ "type" : "INT"
+ }, {
+ "kind" : "LITERAL",
+ "value" : 1024,
+ "type" : "INT NOT NULL"
+ } ],
+ "type" : "INT"
+ } ],
+ "condition" : null,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`a` INT, `b` BIGINT, `c` VARCHAR(2147483647), `$f3` INT, `$f4` INT>",
+ "description" : "Calc(select=[a, b, c, MOD(HASH_CODE(b), 1024) AS $f3, MOD(HASH_CODE(c), 1024) AS $f4])"
+ }, {
+ "id" : 3,
+ "type" : "stream-exec-expand_1",
+ "projects" : [ [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "INT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "BIGINT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "VARCHAR(2147483647)"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 3,
+ "type" : "INT"
+ }, {
+ "kind" : "LITERAL",
+ "value" : null,
+ "type" : "INT"
+ }, {
+ "kind" : "LITERAL",
+ "value" : 1,
+ "type" : "BIGINT NOT NULL"
+ } ], [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "INT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "BIGINT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "VARCHAR(2147483647)"
+ }, {
+ "kind" : "LITERAL",
+ "value" : null,
+ "type" : "INT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 4,
+ "type" : "INT"
+ }, {
+ "kind" : "LITERAL",
+ "value" : 2,
+ "type" : "BIGINT NOT NULL"
+ } ] ],
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`a` INT, `b` BIGINT, `c` VARCHAR(2147483647), `$f3` INT, `$f4` INT, `$e` BIGINT NOT NULL>",
+ "description" : "Expand(projects=[{a, b, c, $f3, null AS $f4, 1 AS $e}, {a, b, c, null AS $f3, $f4, 2 AS $e}])"
+ }, {
+ "id" : 4,
+ "type" : "stream-exec-calc_1",
+ "projection" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "INT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "BIGINT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "VARCHAR(2147483647)"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 3,
+ "type" : "INT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 4,
+ "type" : "INT"
+ }, {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$=$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 5,
+ "type" : "BIGINT NOT NULL"
+ }, {
+ "kind" : "LITERAL",
+ "value" : 1,
+ "type" : "INT NOT NULL"
+ } ],
+ "type" : "BOOLEAN NOT NULL"
+ }, {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$=$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 5,
+ "type" : "BIGINT NOT NULL"
+ }, {
+ "kind" : "LITERAL",
+ "value" : 2,
+ "type" : "INT NOT NULL"
+ } ],
+ "type" : "BOOLEAN NOT NULL"
+ } ],
+ "condition" : null,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`a` INT, `b` BIGINT, `c` VARCHAR(2147483647), `$f3` INT, `$f4` INT, `$g_1` BOOLEAN NOT NULL, `$g_2` BOOLEAN NOT NULL>",
+ "description" : "Calc(select=[a, b, c, $f3, $f4, ($e = 1) AS $g_1, ($e = 2) AS $g_2])"
+ }, {
+ "id" : 5,
+ "type" : "stream-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0, 3, 4 ]
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`a` INT, `b` BIGINT, `c` VARCHAR(2147483647), `$f3` INT, `$f4` INT, `$g_1` BOOLEAN NOT NULL, `$g_2` BOOLEAN NOT NULL>",
+ "description" : "Exchange(distribution=[hash[a, $f3, $f4]])"
+ }, {
+ "id" : 6,
+ "type" : "stream-exec-group-aggregate_1",
+ "configuration" : {
+ "table.exec.mini-batch.enabled" : "false",
+ "table.exec.mini-batch.size" : "-1"
+ },
+ "grouping" : [ 0, 3, 4 ],
+ "aggCalls" : [ {
+ "name" : null,
+ "syntax" : "FUNCTION_STAR",
+ "internalName" : "$COUNT$1",
+ "argList" : [ 1 ],
+ "filterArg" : 5,
+ "distinct" : true,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "BIGINT NOT NULL"
+ }, {
+ "name" : null,
+ "internalName" : "$FIRST_VALUE$1",
+ "argList" : [ 2 ],
+ "filterArg" : 6,
+ "distinct" : false,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "VARCHAR(2147483647)"
+ } ],
+ "aggCallNeedRetractions" : [ false, false ],
+ "generateUpdateBefore" : true,
+ "needRetraction" : false,
+ "state" : [ {
+ "index" : 0,
+ "ttl" : "0 ms",
+ "name" : "groupAggregateState"
+ } ],
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`a` INT, `$f3` INT, `$f4` INT, `$f3_0` BIGINT NOT NULL, `$f4_0` VARCHAR(2147483647)>",
+ "description" : "GroupAggregate(groupBy=[a, $f3, $f4], partialFinalType=[PARTIAL], select=[a, $f3, $f4, COUNT(DISTINCT b) FILTER $g_1 AS $f3_0, FIRST_VALUE(c) FILTER $g_2 AS $f4_0])"
+ }, {
+ "id" : 7,
+ "type" : "stream-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`a` INT, `$f3` INT, `$f4` INT, `$f3_0` BIGINT NOT NULL, `$f4_0` VARCHAR(2147483647)>",
+ "description" : "Exchange(distribution=[hash[a]])"
+ }, {
+ "id" : 8,
+ "type" : "stream-exec-group-aggregate_1",
+ "configuration" : {
+ "table.exec.mini-batch.enabled" : "false",
+ "table.exec.mini-batch.size" : "-1"
+ },
+ "grouping" : [ 0 ],
+ "aggCalls" : [ {
+ "name" : null,
+ "internalName" : "$$SUM0$1",
+ "argList" : [ 3 ],
+ "filterArg" : -1,
+ "distinct" : false,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "BIGINT NOT NULL"
+ }, {
+ "name" : null,
+ "internalName" : "$FIRST_VALUE$1",
+ "argList" : [ 4 ],
+ "filterArg" : -1,
+ "distinct" : false,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "VARCHAR(2147483647)"
+ } ],
+ "aggCallNeedRetractions" : [ true, true ],
+ "generateUpdateBefore" : false,
+ "needRetraction" : true,
+ "state" : [ {
+ "index" : 0,
+ "ttl" : "0 ms",
+ "name" : "groupAggregateState"
+ } ],
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`a` INT, `$f1` BIGINT NOT NULL, `$f2` VARCHAR(2147483647)>",
+ "description" : "GroupAggregate(groupBy=[a], partialFinalType=[FINAL], select=[a, $SUM0_RETRACT($f3_0) AS $f1, FIRST_VALUE_RETRACT($f4_0) AS $f2])"
+ }, {
+ "id" : 9,
+ "type" : "stream-exec-calc_1",
+ "projection" : [ {
+ "kind" : "CALL",
+ "syntax" : "SPECIAL",
+ "internalName" : "$CAST$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "INT"
+ } ],
+ "type" : "BIGINT"
+ }, {
+ "kind" : "CALL",
+ "syntax" : "SPECIAL",
+ "internalName" : "$CAST$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "BIGINT NOT NULL"
+ } ],
+ "type" : "BIGINT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "VARCHAR(2147483647)"
+ } ],
+ "condition" : null,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`b` BIGINT, `a` BIGINT, `c` VARCHAR(2147483647)>",
+ "description" : "Calc(select=[CAST(a AS BIGINT) AS b, CAST($f1 AS BIGINT) AS a, $f2 AS c])"
+ }, {
+ "id" : 10,
+ "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`.`MySink`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "b",
+ "dataType" : "BIGINT NOT NULL"
+ }, {
+ "name" : "a",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "c",
+ "dataType" : "VARCHAR(2147483647)"
+ } ],
+ "watermarkSpecs" : [ ],
+ "primaryKey" : {
+ "name" : "PK_129",
+ "type" : "PRIMARY_KEY",
+ "columns" : [ "b" ]
+ }
+ },
+ "partitionKeys" : [ ]
+ }
+ }
+ },
+ "inputChangelogMode" : [ "INSERT", "UPDATE_AFTER", "DELETE" ],
+ "inputUpsertKey" : [ 0 ],
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`b` BIGINT, `a` BIGINT, `c` VARCHAR(2147483647)>",
+ "description" : "Sink(table=[default_catalog.default_database.MySink], fields=[b, a, 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"
+ }, {
+ "source" : 6,
+ "target" : 7,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "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"
+ } ]
+}
\ No newline at end of file
diff --git a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-expand_1/expand/savepoint/_metadata b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-expand_1/expand/savepoint/_metadata
new file mode 100644
index 00000000000..87e1d794329
Binary files /dev/null and b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-expand_1/expand/savepoint/_metadata differ