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/12 12:54:22 UTC
(flink) branch master updated (560ada286d3 -> be5cf3c9d67)
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 560ada286d3 [FLINK-33470] Moving (Join|Expand)TestPrograms and RestoreTests to the right package.
new 314b418efea [FLINK-33757] Implement restore tests for Rank node
new be5cf3c9d67 [FLINK-33757] Deleting RankJsonPlanTest.java and RankJsonPlanITCase.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/RankJsonPlanTest.java | 71 -----
.../plan/nodes/exec/stream/RankRestoreTest.java | 42 +++
.../plan/nodes/exec/stream/RankTestPrograms.java | 187 +++++++++++++
.../stream/jsonplan/RankJsonPlanITCase.java | 75 -----
.../rank-n-test/plan/rank-n-test.json | 301 +++++++++++++++++++++
.../rank-n-test/savepoint/_metadata | Bin 0 -> 12161 bytes
.../plan/rank-test-append-fast-strategy.json} | 111 +++-----
.../savepoint/_metadata | Bin 0 -> 13768 bytes
.../plan/rank-test-retract-strategy.json} | 133 ++++-----
.../rank-test-retract-strategy/savepoint/_metadata | Bin 0 -> 18147 bytes
.../plan/rank-test-update-fast-strategy.json} | 186 +++++++------
.../savepoint/_metadata | Bin 0 -> 22387 bytes
12 files changed, 728 insertions(+), 378 deletions(-)
delete mode 100644 flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/RankJsonPlanTest.java
create mode 100644 flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/RankRestoreTest.java
create mode 100644 flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/RankTestPrograms.java
delete mode 100644 flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/jsonplan/RankJsonPlanITCase.java
create mode 100644 flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-rank_1/rank-n-test/plan/rank-n-test.json
create mode 100644 flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-rank_1/rank-n-test/savepoint/_metadata
copy flink-table/flink-table-planner/src/test/resources/{org/apache/flink/table/planner/plan/nodes/exec/stream/RankJsonPlanTest_jsonplan/testRank.out => restore-tests/stream-exec-rank_1/rank-test-append-fast-strategy/plan/rank-test-append-fast-strategy.json} (65%)
create mode 100644 flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-rank_1/rank-test-append-fast-strategy/savepoint/_metadata
copy flink-table/flink-table-planner/src/test/resources/{org/apache/flink/table/planner/plan/nodes/exec/stream/RankJsonPlanTest_jsonplan/testRank.out => restore-tests/stream-exec-rank_1/rank-test-retract-strategy/plan/rank-test-retract-strategy.json} (59%)
create mode 100644 flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-rank_1/rank-test-retract-strategy/savepoint/_metadata
rename flink-table/flink-table-planner/src/test/resources/{org/apache/flink/table/planner/plan/nodes/exec/stream/RankJsonPlanTest_jsonplan/testRank.out => restore-tests/stream-exec-rank_1/rank-test-update-fast-strategy/plan/rank-test-update-fast-strategy.json} (50%)
create mode 100644 flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-rank_1/rank-test-update-fast-strategy/savepoint/_metadata
(flink) 02/02: [FLINK-33757] Deleting RankJsonPlanTest.java and RankJsonPlanITCase.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 be5cf3c9d679ff141a1041774070c66b46b866a7
Author: Jim Hughes <jh...@confluent.io>
AuthorDate: Thu Dec 7 08:12:55 2023 -0500
[FLINK-33757] Deleting RankJsonPlanTest.java and RankJsonPlanITCase.java
---
.../plan/nodes/exec/stream/RankJsonPlanTest.java | 71 --------------------
.../stream/jsonplan/RankJsonPlanITCase.java | 75 ----------------------
2 files changed, 146 deletions(-)
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/RankJsonPlanTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/RankJsonPlanTest.java
deleted file mode 100644
index ba4bdd7c6f1..00000000000
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/RankJsonPlanTest.java
+++ /dev/null
@@ -1,71 +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.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 for rank. */
-class RankJsonPlanTest extends TableTestBase {
-
- private StreamTableTestUtil util;
- private TableEnvironment tEnv;
-
- @BeforeEach
- void setup() {
- util = streamTestUtil(TableConfig.getDefault());
- tEnv = util.getTableEnv();
-
- String srcTableDdl =
- "CREATE TABLE MyTable (\n"
- + " a bigint,\n"
- + " b int not null,\n"
- + " c varchar,\n"
- + " d timestamp(3)\n"
- + ") with (\n"
- + " 'connector' = 'values',\n"
- + " 'bounded' = 'false')";
- tEnv.executeSql(srcTableDdl);
- }
-
- @Test
- void testRank() {
- String sinkTableDdl =
- "CREATE TABLE MySink (\n"
- + " a bigint,\n"
- + " b bigint\n"
- + ") with (\n"
- + " 'connector' = 'values',\n"
- + " 'sink-insert-only' = 'false',\n"
- + " 'table-sink-class' = 'DEFAULT')";
- tEnv.executeSql(sinkTableDdl);
- String sql =
- "insert into MySink SELECT a, row_num\n"
- + "FROM (\n"
- + " SELECT a, ROW_NUMBER() OVER (PARTITION BY b ORDER BY a) as row_num\n"
- + " FROM MyTable)\n"
- + "WHERE row_num <= a";
- util.verifyJsonPlan(sql);
- }
-}
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/jsonplan/RankJsonPlanITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/jsonplan/RankJsonPlanITCase.java
deleted file mode 100644
index b6ce7733ebf..00000000000
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/jsonplan/RankJsonPlanITCase.java
+++ /dev/null
@@ -1,75 +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.planner.factories.TestValuesTableFactory;
-import org.apache.flink.table.planner.runtime.utils.TestData;
-import org.apache.flink.table.planner.utils.JavaScalaConversionUtil;
-import org.apache.flink.table.planner.utils.JsonPlanTestBase;
-
-import org.junit.jupiter.api.Test;
-
-import java.io.IOException;
-import java.util.Arrays;
-import java.util.List;
-import java.util.concurrent.ExecutionException;
-
-/** Test for Rank JsonPlan ser/de. */
-class RankJsonPlanITCase extends JsonPlanTestBase {
- @Test
- void testRank() throws ExecutionException, InterruptedException, IOException {
- createTestValuesSourceTable(
- "MyTable",
- JavaScalaConversionUtil.toJava(TestData.data1()),
- "a int",
- "b varchar",
- "c int");
- createTestNonInsertOnlyValuesSinkTable("`result`", "a int", "b varchar", "c bigint");
- String sql =
- "insert into `result` select * from "
- + "(select a, b, row_number() over(partition by b order by c) as c from MyTable)"
- + " where c = 1";
- compileSqlAndExecutePlan(sql).await();
-
- List<String> expected = Arrays.asList("+I[1, a, 1]", "+I[3, b, 1]", "+I[5, c, 1]");
- assertResult(expected, TestValuesTableFactory.getResultsAsStrings("result"));
- }
-
- @Test
- void testFirstN() throws ExecutionException, InterruptedException, IOException {
- createTestValuesSourceTable(
- "MyTable1",
- JavaScalaConversionUtil.toJava(TestData.data4()),
- "a varchar",
- "b int",
- "c int",
- "t as proctime()");
- createTestNonInsertOnlyValuesSinkTable("`result1`", "a varchar", "b int", "c bigint");
- String sql =
- "insert into `result1` select * from "
- + "(select a, b, row_number() over(partition by a order by t asc) as c from MyTable1)"
- + " where c <= 2";
- compileSqlAndExecutePlan(sql).await();
-
- List<String> expected =
- Arrays.asList(
- "+I[book, 1, 1]", "+I[book, 2, 2]", "+I[fruit, 4, 1]", "+I[fruit, 3, 2]");
- assertResult(expected, TestValuesTableFactory.getResultsAsStrings("result1"));
- }
-}
(flink) 01/02: [FLINK-33757] Implement restore tests for Rank 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 314b418efea8f35d39b05abef5361289b054b6a7
Author: Jim Hughes <jh...@confluent.io>
AuthorDate: Thu Dec 7 08:12:42 2023 -0500
[FLINK-33757] Implement restore tests for Rank node
---
.../plan/nodes/exec/stream/RankRestoreTest.java | 42 +++
.../plan/nodes/exec/stream/RankTestPrograms.java | 187 +++++++++++++
.../rank-n-test/plan/rank-n-test.json | 301 +++++++++++++++++++++
.../rank-n-test/savepoint/_metadata | Bin 0 -> 12161 bytes
.../plan/rank-test-append-fast-strategy.json} | 111 +++-----
.../savepoint/_metadata | Bin 0 -> 13768 bytes
.../plan/rank-test-retract-strategy.json} | 133 ++++-----
.../rank-test-retract-strategy/savepoint/_metadata | Bin 0 -> 18147 bytes
.../plan/rank-test-update-fast-strategy.json} | 186 +++++++------
.../savepoint/_metadata | Bin 0 -> 22387 bytes
10 files changed, 728 insertions(+), 232 deletions(-)
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/RankRestoreTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/RankRestoreTest.java
new file mode 100644
index 00000000000..cb76c1ca723
--- /dev/null
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/RankRestoreTest.java
@@ -0,0 +1,42 @@
+/*
+ * 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 StreamExecRank}. */
+public class RankRestoreTest extends RestoreTestBase {
+
+ public RankRestoreTest() {
+ super(StreamExecRank.class);
+ }
+
+ @Override
+ public List<TableTestProgram> programs() {
+ return Arrays.asList(
+ RankTestPrograms.RANK_TEST_APPEND_FAST_STRATEGY,
+ RankTestPrograms.RANK_TEST_RETRACT_STRATEGY,
+ RankTestPrograms.RANK_TEST_UPDATE_FAST_STRATEGY,
+ RankTestPrograms.RANK_N_TEST);
+ }
+}
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/RankTestPrograms.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/RankTestPrograms.java
new file mode 100644
index 00000000000..979e2adbb52
--- /dev/null
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/RankTestPrograms.java
@@ -0,0 +1,187 @@
+/*
+ * 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.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 static org.apache.flink.table.factories.TestFormatFactory.CHANGELOG_MODE;
+
+/** {@link TableTestProgram} definitions for testing {@link StreamExecRank}. */
+public class RankTestPrograms {
+
+ static final TableTestProgram RANK_TEST_APPEND_FAST_STRATEGY =
+ getTableTestProgram(
+ "rank-test-append-fast-strategy",
+ "I",
+ new String[] {
+ "+I[2, a, 1]",
+ "+I[4, b, 1]",
+ "+I[6, c, 1]",
+ "-U[2, a, 1]",
+ "+U[1, a, 1]",
+ "-U[4, b, 1]",
+ "+U[3, b, 1]",
+ "-U[6, c, 1]",
+ "+U[5, c, 1]"
+ },
+ new String[] {"+I[4, d, 1]", "+I[3, e, 1]"});
+
+ static final TableTestProgram RANK_TEST_RETRACT_STRATEGY =
+ getTableTestProgram(
+ "rank-test-retract-strategy",
+ "I,UA,UB",
+ new String[] {
+ "+I[2, a, 1]",
+ "+I[4, b, 1]",
+ "+I[6, c, 1]",
+ "-D[2, a, 1]",
+ "+I[1, a, 1]",
+ "-D[4, b, 1]",
+ "+I[3, b, 1]",
+ "-D[6, c, 1]",
+ "+I[5, c, 1]"
+ },
+ new String[] {"+I[4, d, 1]", "+I[3, e, 1]"});
+
+ static final TableTestProgram RANK_TEST_UPDATE_FAST_STRATEGY =
+ TableTestProgram.of("rank-test-update-fast-strategy", "validates rank exec node")
+ .setupTableSource(
+ SourceTestStep.newBuilder("MyTable")
+ .addSchema(
+ "a INT primary key not enforced", "b VARCHAR", "c INT")
+ .addOption(CHANGELOG_MODE, "I")
+ .producedBeforeRestore(
+ Row.of(2, "a", 6),
+ Row.of(4, "b", 8),
+ Row.of(6, "c", 10),
+ Row.of(1, "a", 5),
+ Row.of(3, "b", 7),
+ Row.of(5, "c", 9))
+ .producedAfterRestore(Row.of(4, "d", 7), Row.of(0, "a", 8))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink_t")
+ .addSchema(
+ "a INT NOT NULL",
+ "b STRING",
+ "count_c BIGINT NOT NULL",
+ "row_num BIGINT NOT NULL")
+ .consumedBeforeRestore(
+ "+I[2, a, 1, 1]",
+ "+I[4, b, 1, 1]",
+ "+I[6, c, 1, 1]",
+ "-U[2, a, 1, 1]",
+ "+U[1, a, 1, 1]",
+ "+I[2, a, 1, 2]",
+ "-U[4, b, 1, 1]",
+ "+U[3, b, 1, 1]",
+ "+I[4, b, 1, 2]",
+ "-U[6, c, 1, 1]",
+ "+U[5, c, 1, 1]",
+ "+I[6, c, 1, 2]")
+ .consumedAfterRestore(
+ new String[] {
+ "+I[4, d, 1, 1]",
+ "-U[1, a, 1, 1]",
+ "+U[0, a, 1, 1]",
+ "-U[2, a, 1, 2]",
+ "+U[1, a, 1, 2]",
+ "+I[2, a, 1, 3]"
+ })
+ .build())
+ .runSql(
+ "INSERT INTO sink_t SELECT * FROM ("
+ + "SELECT a, b, count_c, ROW_NUMBER() "
+ + " OVER (PARTITION BY b ORDER BY count_c DESC, a ASC) AS row_num"
+ + " FROM ("
+ + " SELECT a, b, COUNT(*) AS count_c"
+ + " FROM MyTable"
+ + " GROUP BY a, b"
+ + " )"
+ + ") WHERE row_num <= 10")
+ .build();
+
+ private static TableTestProgram getTableTestProgram(
+ final String name,
+ final String changelogMode,
+ final String[] resultsBeforeRestore,
+ final String[] resultsAfterRestore) {
+ return TableTestProgram.of(name, "validates rank exec node")
+ .setupTableSource(
+ SourceTestStep.newBuilder("MyTable")
+ .addSchema("a INT", "b VARCHAR", "c INT primary key not enforced")
+ .addOption(CHANGELOG_MODE, changelogMode)
+ .producedBeforeRestore(
+ Row.of(2, "a", 6),
+ Row.of(4, "b", 8),
+ Row.of(6, "c", 10),
+ Row.of(1, "a", 5),
+ Row.of(3, "b", 7),
+ Row.of(5, "c", 9))
+ .producedAfterRestore(Row.of(4, "d", 7), Row.of(3, "e", 8))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink_t")
+ .addSchema("a INT", "b VARCHAR", "c BIGINT")
+ .consumedBeforeRestore(resultsBeforeRestore)
+ .consumedAfterRestore(resultsAfterRestore)
+ .build())
+ .runSql(
+ "insert into `sink_t` select * from "
+ + "(select a, b, row_number() over(partition by b order by c) as c from MyTable)"
+ + " where c = 1")
+ .build();
+ }
+
+ static final TableTestProgram RANK_N_TEST =
+ TableTestProgram.of("rank-n-test", "validates rank node can handle multiple outputs")
+ .setupTableSource(
+ SourceTestStep.newBuilder("MyTable1")
+ .addSchema("a STRING", "b INT", "c INT", "t as proctime()")
+ .producedBeforeRestore(
+ Row.of("book", 1, 12),
+ Row.of("book", 2, 19),
+ Row.of("book", 4, 11),
+ Row.of("fruit", 4, 33))
+ .producedAfterRestore(
+ Row.of("cereal", 6, 21),
+ Row.of("cereal", 7, 23),
+ Row.of("apple", 8, 31),
+ Row.of("fruit", 9, 41))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("result1")
+ .addSchema("a varchar", "b int", "c bigint")
+ .consumedBeforeRestore(
+ "+I[book, 1, 1]", "+I[book, 2, 2]", "+I[fruit, 4, 1]")
+ .consumedAfterRestore(
+ "+I[cereal, 6, 1]",
+ "+I[cereal, 7, 2]",
+ "+I[apple, 8, 1]",
+ "+I[fruit, 9, 2]")
+ .build())
+ .runSql(
+ "insert into `result1` select * from "
+ + "(select a, b, row_number() over(partition by a order by t asc) as c from MyTable1)"
+ + " where c <= 2")
+ .build();
+}
diff --git a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-rank_1/rank-n-test/plan/rank-n-test.json b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-rank_1/rank-n-test/plan/rank-n-test.json
new file mode 100644
index 00000000000..11f4b21e9e3
--- /dev/null
+++ b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-rank_1/rank-n-test/plan/rank-n-test.json
@@ -0,0 +1,301 @@
+{
+ "flinkVersion" : "1.19",
+ "nodes" : [ {
+ "id" : 17,
+ "type" : "stream-exec-table-source-scan_1",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`MyTable1`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "a",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "b",
+ "dataType" : "INT"
+ }, {
+ "name" : "c",
+ "dataType" : "INT"
+ }, {
+ "name" : "t",
+ "kind" : "COMPUTED",
+ "expression" : {
+ "rexNode" : {
+ "kind" : "CALL",
+ "internalName" : "$PROCTIME$1",
+ "operands" : [ ],
+ "type" : {
+ "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
+ "nullable" : false,
+ "precision" : 3,
+ "kind" : "PROCTIME"
+ }
+ },
+ "serializableString" : "PROCTIME()"
+ }
+ } ],
+ "watermarkSpecs" : [ ]
+ },
+ "partitionKeys" : [ ]
+ }
+ },
+ "abilities" : [ {
+ "type" : "ProjectPushDown",
+ "projectedFields" : [ [ 0 ], [ 1 ] ],
+ "producedType" : "ROW<`a` VARCHAR(2147483647), `b` INT> NOT NULL"
+ }, {
+ "type" : "ReadingMetadata",
+ "metadataKeys" : [ ],
+ "producedType" : "ROW<`a` VARCHAR(2147483647), `b` INT> NOT NULL"
+ } ]
+ },
+ "outputType" : "ROW<`a` VARCHAR(2147483647), `b` INT>",
+ "description" : "TableSourceScan(table=[[default_catalog, default_database, MyTable1, project=[a, b], metadata=[]]], fields=[a, b])",
+ "inputProperties" : [ ]
+ }, {
+ "id" : 18,
+ "type" : "stream-exec-calc_1",
+ "projection" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "VARCHAR(2147483647)"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "INT"
+ }, {
+ "kind" : "CALL",
+ "internalName" : "$PROCTIME$1",
+ "operands" : [ ],
+ "type" : {
+ "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
+ "nullable" : false,
+ "precision" : 3,
+ "kind" : "PROCTIME"
+ }
+ } ],
+ "condition" : null,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : {
+ "type" : "ROW",
+ "fields" : [ {
+ "name" : "a",
+ "fieldType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "b",
+ "fieldType" : "INT"
+ }, {
+ "name" : "$2",
+ "fieldType" : {
+ "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
+ "nullable" : false,
+ "precision" : 3,
+ "kind" : "PROCTIME"
+ }
+ } ]
+ },
+ "description" : "Calc(select=[a, b, PROCTIME() AS $2])"
+ }, {
+ "id" : 19,
+ "type" : "stream-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : {
+ "type" : "ROW",
+ "fields" : [ {
+ "name" : "a",
+ "fieldType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "b",
+ "fieldType" : "INT"
+ }, {
+ "name" : "$2",
+ "fieldType" : {
+ "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
+ "nullable" : false,
+ "precision" : 3,
+ "kind" : "PROCTIME"
+ }
+ } ]
+ },
+ "description" : "Exchange(distribution=[hash[a]])"
+ }, {
+ "id" : 20,
+ "type" : "stream-exec-rank_1",
+ "configuration" : {
+ "table.exec.rank.topn-cache-size" : "10000"
+ },
+ "rankType" : "ROW_NUMBER",
+ "partition" : {
+ "fields" : [ 0 ]
+ },
+ "orderBy" : {
+ "fields" : [ {
+ "index" : 2,
+ "isAscending" : true,
+ "nullIsLast" : false
+ } ]
+ },
+ "rankRange" : {
+ "type" : "Constant",
+ "start" : 1,
+ "end" : 2
+ },
+ "rankStrategy" : {
+ "type" : "AppendFast"
+ },
+ "outputRowNumber" : true,
+ "generateUpdateBefore" : true,
+ "state" : [ {
+ "index" : 0,
+ "ttl" : "0 ms",
+ "name" : "rankState"
+ } ],
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : {
+ "type" : "ROW",
+ "fields" : [ {
+ "name" : "a",
+ "fieldType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "b",
+ "fieldType" : "INT"
+ }, {
+ "name" : "$2",
+ "fieldType" : {
+ "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
+ "nullable" : false,
+ "precision" : 3,
+ "kind" : "PROCTIME"
+ }
+ }, {
+ "name" : "w0$o0",
+ "fieldType" : "BIGINT NOT NULL"
+ } ]
+ },
+ "description" : "Rank(strategy=[AppendFastStrategy], rankType=[ROW_NUMBER], rankRange=[rankStart=1, rankEnd=2], partitionBy=[a], orderBy=[$2 ASC], select=[a, b, $2, w0$o0])"
+ }, {
+ "id" : 21,
+ "type" : "stream-exec-calc_1",
+ "projection" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "VARCHAR(2147483647)"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "INT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 3,
+ "type" : "BIGINT NOT NULL"
+ } ],
+ "condition" : null,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`a` VARCHAR(2147483647), `b` INT, `w0$o0` BIGINT NOT NULL>",
+ "description" : "Calc(select=[a, b, w0$o0])"
+ }, {
+ "id" : 22,
+ "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`.`result1`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "a",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "b",
+ "dataType" : "INT"
+ }, {
+ "name" : "c",
+ "dataType" : "BIGINT"
+ } ],
+ "watermarkSpecs" : [ ]
+ },
+ "partitionKeys" : [ ]
+ }
+ }
+ },
+ "inputChangelogMode" : [ "INSERT", "UPDATE_BEFORE", "UPDATE_AFTER", "DELETE" ],
+ "inputUpsertKey" : [ 0, 2 ],
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`a` VARCHAR(2147483647), `b` INT, `w0$o0` BIGINT NOT NULL>",
+ "description" : "Sink(table=[default_catalog.default_database.result1], fields=[a, b, w0$o0])"
+ } ],
+ "edges" : [ {
+ "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"
+ }, {
+ "source" : 20,
+ "target" : 21,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 21,
+ "target" : 22,
+ "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-rank_1/rank-n-test/savepoint/_metadata b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-rank_1/rank-n-test/savepoint/_metadata
new file mode 100644
index 00000000000..8c9ecaab640
Binary files /dev/null and b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-rank_1/rank-n-test/savepoint/_metadata differ
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/RankJsonPlanTest_jsonplan/testRank.out b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-rank_1/rank-test-append-fast-strategy/plan/rank-test-append-fast-strategy.json
similarity index 65%
copy from flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/RankJsonPlanTest_jsonplan/testRank.out
copy to flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-rank_1/rank-test-append-fast-strategy/plan/rank-test-append-fast-strategy.json
index 7209ea0be4a..790da11a5b3 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/RankJsonPlanTest_jsonplan/testRank.out
+++ b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-rank_1/rank-test-append-fast-strategy/plan/rank-test-append-fast-strategy.json
@@ -1,5 +1,5 @@
{
- "flinkVersion" : "",
+ "flinkVersion" : "1.19",
"nodes" : [ {
"id" : 1,
"type" : "stream-exec-table-source-scan_1",
@@ -10,54 +10,30 @@
"schema" : {
"columns" : [ {
"name" : "a",
- "dataType" : "BIGINT"
+ "dataType" : "INT"
}, {
"name" : "b",
- "dataType" : "INT NOT NULL"
- }, {
- "name" : "c",
"dataType" : "VARCHAR(2147483647)"
}, {
- "name" : "d",
- "dataType" : "TIMESTAMP(3)"
+ "name" : "c",
+ "dataType" : "INT NOT NULL"
} ],
- "watermarkSpecs" : [ ]
+ "watermarkSpecs" : [ ],
+ "primaryKey" : {
+ "name" : "PK_c",
+ "type" : "PRIMARY_KEY",
+ "columns" : [ "c" ]
+ }
},
- "partitionKeys" : [ ],
- "options" : {
- "bounded" : "false",
- "connector" : "values"
- }
+ "partitionKeys" : [ ]
}
}
},
- "outputType" : "ROW<`a` BIGINT, `b` INT NOT NULL, `c` VARCHAR(2147483647), `d` TIMESTAMP(3)>",
- "description" : "TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d])",
+ "outputType" : "ROW<`a` INT, `b` VARCHAR(2147483647), `c` INT NOT NULL>",
+ "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"
- } ],
- "condition" : null,
- "inputProperties" : [ {
- "requiredDistribution" : {
- "type" : "UNKNOWN"
- },
- "damBehavior" : "PIPELINED",
- "priority" : 0
- } ],
- "outputType" : "ROW<`a` BIGINT, `b` INT NOT NULL>",
- "description" : "Calc(select=[a, b])"
- }, {
- "id" : 3,
"type" : "stream-exec-exchange_1",
"inputProperties" : [ {
"requiredDistribution" : {
@@ -67,10 +43,10 @@
"damBehavior" : "PIPELINED",
"priority" : 0
} ],
- "outputType" : "ROW<`a` BIGINT, `b` INT NOT NULL>",
+ "outputType" : "ROW<`a` INT, `b` VARCHAR(2147483647), `c` INT NOT NULL>",
"description" : "Exchange(distribution=[hash[b]])"
}, {
- "id" : 4,
+ "id" : 3,
"type" : "stream-exec-rank_1",
"configuration" : {
"table.exec.rank.topn-cache-size" : "10000"
@@ -81,19 +57,20 @@
},
"orderBy" : {
"fields" : [ {
- "index" : 0,
+ "index" : 2,
"isAscending" : true,
"nullIsLast" : false
} ]
},
"rankRange" : {
- "type" : "Variable",
- "endIndex" : 0
+ "type" : "Constant",
+ "start" : 1,
+ "end" : 1
},
"rankStrategy" : {
"type" : "AppendFast"
},
- "outputRowNumber" : true,
+ "outputRowNumber" : false,
"generateUpdateBefore" : true,
"state" : [ {
"index" : 0,
@@ -107,18 +84,22 @@
"damBehavior" : "PIPELINED",
"priority" : 0
} ],
- "outputType" : "ROW<`a` BIGINT, `b` INT NOT NULL, `w0$o0` BIGINT NOT NULL>",
- "description" : "Rank(strategy=[AppendFastStrategy], rankType=[ROW_NUMBER], rankRange=[rankEnd=a], partitionBy=[b], orderBy=[a ASC], select=[a, b, w0$o0])"
+ "outputType" : "ROW<`a` INT, `b` VARCHAR(2147483647), `c` INT NOT NULL>",
+ "description" : "Rank(strategy=[AppendFastStrategy], rankType=[ROW_NUMBER], rankRange=[rankStart=1, rankEnd=1], partitionBy=[b], orderBy=[c ASC], select=[a, b, c])"
}, {
- "id" : 5,
+ "id" : 4,
"type" : "stream-exec-calc_1",
"projection" : [ {
"kind" : "INPUT_REF",
"inputIndex" : 0,
- "type" : "BIGINT"
+ "type" : "INT"
}, {
"kind" : "INPUT_REF",
- "inputIndex" : 2,
+ "inputIndex" : 1,
+ "type" : "VARCHAR(2147483647)"
+ }, {
+ "kind" : "LITERAL",
+ "value" : 1,
"type" : "BIGINT NOT NULL"
} ],
"condition" : null,
@@ -129,10 +110,10 @@
"damBehavior" : "PIPELINED",
"priority" : 0
} ],
- "outputType" : "ROW<`a` BIGINT, `w0$o0` BIGINT NOT NULL>",
- "description" : "Calc(select=[a, w0$o0])"
+ "outputType" : "ROW<`a` INT, `b` VARCHAR(2147483647), `$2` BIGINT NOT NULL>",
+ "description" : "Calc(select=[a, b, 1 AS $2])"
}, {
- "id" : 6,
+ "id" : 5,
"type" : "stream-exec-sink_1",
"configuration" : {
"table.exec.sink.keyed-shuffle" : "AUTO",
@@ -143,28 +124,27 @@
},
"dynamicTableSink" : {
"table" : {
- "identifier" : "`default_catalog`.`default_database`.`MySink`",
+ "identifier" : "`default_catalog`.`default_database`.`sink_t`",
"resolvedTable" : {
"schema" : {
"columns" : [ {
"name" : "a",
- "dataType" : "BIGINT"
+ "dataType" : "INT"
}, {
"name" : "b",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "c",
"dataType" : "BIGINT"
} ],
"watermarkSpecs" : [ ]
},
- "partitionKeys" : [ ],
- "options" : {
- "connector" : "values",
- "sink-insert-only" : "false",
- "table-sink-class" : "DEFAULT"
- }
+ "partitionKeys" : [ ]
}
}
},
"inputChangelogMode" : [ "INSERT", "UPDATE_BEFORE", "UPDATE_AFTER", "DELETE" ],
+ "inputUpsertKey" : [ 1 ],
"inputProperties" : [ {
"requiredDistribution" : {
"type" : "UNKNOWN"
@@ -172,8 +152,8 @@
"damBehavior" : "PIPELINED",
"priority" : 0
} ],
- "outputType" : "ROW<`a` BIGINT, `w0$o0` BIGINT NOT NULL>",
- "description" : "Sink(table=[default_catalog.default_database.MySink], fields=[a, w0$o0])"
+ "outputType" : "ROW<`a` INT, `b` VARCHAR(2147483647), `$2` BIGINT NOT NULL>",
+ "description" : "Sink(table=[default_catalog.default_database.sink_t], fields=[a, b, $2])"
} ],
"edges" : [ {
"source" : 1,
@@ -203,12 +183,5 @@
"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-rank_1/rank-test-append-fast-strategy/savepoint/_metadata b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-rank_1/rank-test-append-fast-strategy/savepoint/_metadata
new file mode 100644
index 00000000000..9d5f2be6c14
Binary files /dev/null and b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-rank_1/rank-test-append-fast-strategy/savepoint/_metadata differ
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/RankJsonPlanTest_jsonplan/testRank.out b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-rank_1/rank-test-retract-strategy/plan/rank-test-retract-strategy.json
similarity index 59%
copy from flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/RankJsonPlanTest_jsonplan/testRank.out
copy to flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-rank_1/rank-test-retract-strategy/plan/rank-test-retract-strategy.json
index 7209ea0be4a..e79e6fc6cbc 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/RankJsonPlanTest_jsonplan/testRank.out
+++ b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-rank_1/rank-test-retract-strategy/plan/rank-test-retract-strategy.json
@@ -1,7 +1,7 @@
{
- "flinkVersion" : "",
+ "flinkVersion" : "1.19",
"nodes" : [ {
- "id" : 1,
+ "id" : 6,
"type" : "stream-exec-table-source-scan_1",
"scanTableSource" : {
"table" : {
@@ -10,54 +10,30 @@
"schema" : {
"columns" : [ {
"name" : "a",
- "dataType" : "BIGINT"
+ "dataType" : "INT"
}, {
"name" : "b",
- "dataType" : "INT NOT NULL"
- }, {
- "name" : "c",
"dataType" : "VARCHAR(2147483647)"
}, {
- "name" : "d",
- "dataType" : "TIMESTAMP(3)"
+ "name" : "c",
+ "dataType" : "INT NOT NULL"
} ],
- "watermarkSpecs" : [ ]
+ "watermarkSpecs" : [ ],
+ "primaryKey" : {
+ "name" : "PK_c",
+ "type" : "PRIMARY_KEY",
+ "columns" : [ "c" ]
+ }
},
- "partitionKeys" : [ ],
- "options" : {
- "bounded" : "false",
- "connector" : "values"
- }
+ "partitionKeys" : [ ]
}
}
},
- "outputType" : "ROW<`a` BIGINT, `b` INT NOT NULL, `c` VARCHAR(2147483647), `d` TIMESTAMP(3)>",
- "description" : "TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d])",
+ "outputType" : "ROW<`a` INT, `b` VARCHAR(2147483647), `c` INT NOT NULL>",
+ "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"
- } ],
- "condition" : null,
- "inputProperties" : [ {
- "requiredDistribution" : {
- "type" : "UNKNOWN"
- },
- "damBehavior" : "PIPELINED",
- "priority" : 0
- } ],
- "outputType" : "ROW<`a` BIGINT, `b` INT NOT NULL>",
- "description" : "Calc(select=[a, b])"
- }, {
- "id" : 3,
+ "id" : 7,
"type" : "stream-exec-exchange_1",
"inputProperties" : [ {
"requiredDistribution" : {
@@ -67,10 +43,10 @@
"damBehavior" : "PIPELINED",
"priority" : 0
} ],
- "outputType" : "ROW<`a` BIGINT, `b` INT NOT NULL>",
+ "outputType" : "ROW<`a` INT, `b` VARCHAR(2147483647), `c` INT NOT NULL>",
"description" : "Exchange(distribution=[hash[b]])"
}, {
- "id" : 4,
+ "id" : 8,
"type" : "stream-exec-rank_1",
"configuration" : {
"table.exec.rank.topn-cache-size" : "10000"
@@ -81,19 +57,20 @@
},
"orderBy" : {
"fields" : [ {
- "index" : 0,
+ "index" : 2,
"isAscending" : true,
"nullIsLast" : false
} ]
},
"rankRange" : {
- "type" : "Variable",
- "endIndex" : 0
+ "type" : "Constant",
+ "start" : 1,
+ "end" : 1
},
"rankStrategy" : {
- "type" : "AppendFast"
+ "type" : "Retract"
},
- "outputRowNumber" : true,
+ "outputRowNumber" : false,
"generateUpdateBefore" : true,
"state" : [ {
"index" : 0,
@@ -107,18 +84,22 @@
"damBehavior" : "PIPELINED",
"priority" : 0
} ],
- "outputType" : "ROW<`a` BIGINT, `b` INT NOT NULL, `w0$o0` BIGINT NOT NULL>",
- "description" : "Rank(strategy=[AppendFastStrategy], rankType=[ROW_NUMBER], rankRange=[rankEnd=a], partitionBy=[b], orderBy=[a ASC], select=[a, b, w0$o0])"
+ "outputType" : "ROW<`a` INT, `b` VARCHAR(2147483647), `c` INT NOT NULL>",
+ "description" : "Rank(strategy=[RetractStrategy], rankType=[ROW_NUMBER], rankRange=[rankStart=1, rankEnd=1], partitionBy=[b], orderBy=[c ASC], select=[a, b, c])"
}, {
- "id" : 5,
+ "id" : 9,
"type" : "stream-exec-calc_1",
"projection" : [ {
"kind" : "INPUT_REF",
"inputIndex" : 0,
- "type" : "BIGINT"
+ "type" : "INT"
}, {
"kind" : "INPUT_REF",
- "inputIndex" : 2,
+ "inputIndex" : 1,
+ "type" : "VARCHAR(2147483647)"
+ }, {
+ "kind" : "LITERAL",
+ "value" : 1,
"type" : "BIGINT NOT NULL"
} ],
"condition" : null,
@@ -129,10 +110,10 @@
"damBehavior" : "PIPELINED",
"priority" : 0
} ],
- "outputType" : "ROW<`a` BIGINT, `w0$o0` BIGINT NOT NULL>",
- "description" : "Calc(select=[a, w0$o0])"
+ "outputType" : "ROW<`a` INT, `b` VARCHAR(2147483647), `$2` BIGINT NOT NULL>",
+ "description" : "Calc(select=[a, b, 1 AS $2])"
}, {
- "id" : 6,
+ "id" : 10,
"type" : "stream-exec-sink_1",
"configuration" : {
"table.exec.sink.keyed-shuffle" : "AUTO",
@@ -143,28 +124,27 @@
},
"dynamicTableSink" : {
"table" : {
- "identifier" : "`default_catalog`.`default_database`.`MySink`",
+ "identifier" : "`default_catalog`.`default_database`.`sink_t`",
"resolvedTable" : {
"schema" : {
"columns" : [ {
"name" : "a",
- "dataType" : "BIGINT"
+ "dataType" : "INT"
}, {
"name" : "b",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "c",
"dataType" : "BIGINT"
} ],
"watermarkSpecs" : [ ]
},
- "partitionKeys" : [ ],
- "options" : {
- "connector" : "values",
- "sink-insert-only" : "false",
- "table-sink-class" : "DEFAULT"
- }
+ "partitionKeys" : [ ]
}
}
},
"inputChangelogMode" : [ "INSERT", "UPDATE_BEFORE", "UPDATE_AFTER", "DELETE" ],
+ "inputUpsertKey" : [ 1 ],
"inputProperties" : [ {
"requiredDistribution" : {
"type" : "UNKNOWN"
@@ -172,43 +152,36 @@
"damBehavior" : "PIPELINED",
"priority" : 0
} ],
- "outputType" : "ROW<`a` BIGINT, `w0$o0` BIGINT NOT NULL>",
- "description" : "Sink(table=[default_catalog.default_database.MySink], fields=[a, w0$o0])"
+ "outputType" : "ROW<`a` INT, `b` VARCHAR(2147483647), `$2` BIGINT NOT NULL>",
+ "description" : "Sink(table=[default_catalog.default_database.sink_t], fields=[a, b, $2])"
} ],
"edges" : [ {
- "source" : 1,
- "target" : 2,
- "shuffle" : {
- "type" : "FORWARD"
- },
- "shuffleMode" : "PIPELINED"
- }, {
- "source" : 2,
- "target" : 3,
+ "source" : 6,
+ "target" : 7,
"shuffle" : {
"type" : "FORWARD"
},
"shuffleMode" : "PIPELINED"
}, {
- "source" : 3,
- "target" : 4,
+ "source" : 7,
+ "target" : 8,
"shuffle" : {
"type" : "FORWARD"
},
"shuffleMode" : "PIPELINED"
}, {
- "source" : 4,
- "target" : 5,
+ "source" : 8,
+ "target" : 9,
"shuffle" : {
"type" : "FORWARD"
},
"shuffleMode" : "PIPELINED"
}, {
- "source" : 5,
- "target" : 6,
+ "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-rank_1/rank-test-retract-strategy/savepoint/_metadata b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-rank_1/rank-test-retract-strategy/savepoint/_metadata
new file mode 100644
index 00000000000..4e542ae3a77
Binary files /dev/null and b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-rank_1/rank-test-retract-strategy/savepoint/_metadata differ
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/RankJsonPlanTest_jsonplan/testRank.out b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-rank_1/rank-test-update-fast-strategy/plan/rank-test-update-fast-strategy.json
similarity index 50%
rename from flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/RankJsonPlanTest_jsonplan/testRank.out
rename to flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-rank_1/rank-test-update-fast-strategy/plan/rank-test-update-fast-strategy.json
index 7209ea0be4a..ce4974b9939 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/RankJsonPlanTest_jsonplan/testRank.out
+++ b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-rank_1/rank-test-update-fast-strategy/plan/rank-test-update-fast-strategy.json
@@ -1,7 +1,7 @@
{
- "flinkVersion" : "",
+ "flinkVersion" : "1.19",
"nodes" : [ {
- "id" : 1,
+ "id" : 11,
"type" : "stream-exec-table-source-scan_1",
"scanTableSource" : {
"table" : {
@@ -10,43 +10,77 @@
"schema" : {
"columns" : [ {
"name" : "a",
- "dataType" : "BIGINT"
- }, {
- "name" : "b",
"dataType" : "INT NOT NULL"
}, {
- "name" : "c",
+ "name" : "b",
"dataType" : "VARCHAR(2147483647)"
}, {
- "name" : "d",
- "dataType" : "TIMESTAMP(3)"
+ "name" : "c",
+ "dataType" : "INT"
} ],
- "watermarkSpecs" : [ ]
+ "watermarkSpecs" : [ ],
+ "primaryKey" : {
+ "name" : "PK_a",
+ "type" : "PRIMARY_KEY",
+ "columns" : [ "a" ]
+ }
},
- "partitionKeys" : [ ],
- "options" : {
- "bounded" : "false",
- "connector" : "values"
- }
+ "partitionKeys" : [ ]
}
- }
+ },
+ "abilities" : [ {
+ "type" : "ProjectPushDown",
+ "projectedFields" : [ [ 0 ], [ 1 ] ],
+ "producedType" : "ROW<`a` INT NOT NULL, `b` VARCHAR(2147483647)> NOT NULL"
+ }, {
+ "type" : "ReadingMetadata",
+ "metadataKeys" : [ ],
+ "producedType" : "ROW<`a` INT NOT NULL, `b` VARCHAR(2147483647)> NOT NULL"
+ } ]
},
- "outputType" : "ROW<`a` BIGINT, `b` INT NOT NULL, `c` VARCHAR(2147483647), `d` TIMESTAMP(3)>",
- "description" : "TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d])",
+ "outputType" : "ROW<`a` INT NOT NULL, `b` VARCHAR(2147483647)>",
+ "description" : "TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, b], metadata=[]]], fields=[a, b])",
"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"
+ "id" : 12,
+ "type" : "stream-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0, 1 ]
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`a` INT NOT NULL, `b` VARCHAR(2147483647)>",
+ "description" : "Exchange(distribution=[hash[a, b]])"
+ }, {
+ "id" : 13,
+ "type" : "stream-exec-group-aggregate_1",
+ "configuration" : {
+ "table.exec.mini-batch.enabled" : "false",
+ "table.exec.mini-batch.size" : "-1"
+ },
+ "grouping" : [ 0, 1 ],
+ "aggCalls" : [ {
+ "name" : "count_c",
+ "syntax" : "FUNCTION_STAR",
+ "internalName" : "$COUNT$1",
+ "argList" : [ ],
+ "filterArg" : -1,
+ "distinct" : false,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "BIGINT NOT NULL"
+ } ],
+ "aggCallNeedRetractions" : [ false ],
+ "generateUpdateBefore" : false,
+ "needRetraction" : false,
+ "state" : [ {
+ "index" : 0,
+ "ttl" : "0 ms",
+ "name" : "groupAggregateState"
} ],
- "condition" : null,
"inputProperties" : [ {
"requiredDistribution" : {
"type" : "UNKNOWN"
@@ -54,10 +88,10 @@
"damBehavior" : "PIPELINED",
"priority" : 0
} ],
- "outputType" : "ROW<`a` BIGINT, `b` INT NOT NULL>",
- "description" : "Calc(select=[a, b])"
+ "outputType" : "ROW<`a` INT NOT NULL, `b` VARCHAR(2147483647), `count_c` BIGINT NOT NULL>",
+ "description" : "GroupAggregate(groupBy=[a, b], select=[a, b, COUNT(*) AS count_c])"
}, {
- "id" : 3,
+ "id" : 14,
"type" : "stream-exec-exchange_1",
"inputProperties" : [ {
"requiredDistribution" : {
@@ -67,10 +101,10 @@
"damBehavior" : "PIPELINED",
"priority" : 0
} ],
- "outputType" : "ROW<`a` BIGINT, `b` INT NOT NULL>",
+ "outputType" : "ROW<`a` INT NOT NULL, `b` VARCHAR(2147483647), `count_c` BIGINT NOT NULL>",
"description" : "Exchange(distribution=[hash[b]])"
}, {
- "id" : 4,
+ "id" : 15,
"type" : "stream-exec-rank_1",
"configuration" : {
"table.exec.rank.topn-cache-size" : "10000"
@@ -81,17 +115,23 @@
},
"orderBy" : {
"fields" : [ {
+ "index" : 2,
+ "isAscending" : false,
+ "nullIsLast" : true
+ }, {
"index" : 0,
"isAscending" : true,
"nullIsLast" : false
} ]
},
"rankRange" : {
- "type" : "Variable",
- "endIndex" : 0
+ "type" : "Constant",
+ "start" : 1,
+ "end" : 10
},
"rankStrategy" : {
- "type" : "AppendFast"
+ "type" : "UpdateFast",
+ "primaryKeys" : [ 0, 1 ]
},
"outputRowNumber" : true,
"generateUpdateBefore" : true,
@@ -107,32 +147,10 @@
"damBehavior" : "PIPELINED",
"priority" : 0
} ],
- "outputType" : "ROW<`a` BIGINT, `b` INT NOT NULL, `w0$o0` BIGINT NOT NULL>",
- "description" : "Rank(strategy=[AppendFastStrategy], rankType=[ROW_NUMBER], rankRange=[rankEnd=a], partitionBy=[b], orderBy=[a ASC], select=[a, b, w0$o0])"
+ "outputType" : "ROW<`a` INT NOT NULL, `b` VARCHAR(2147483647), `count_c` BIGINT NOT NULL, `w0$o0` BIGINT NOT NULL>",
+ "description" : "Rank(strategy=[UpdateFastStrategy[0,1]], rankType=[ROW_NUMBER], rankRange=[rankStart=1, rankEnd=10], partitionBy=[b], orderBy=[count_c DESC, a ASC], select=[a, b, count_c, w0$o0])"
}, {
- "id" : 5,
- "type" : "stream-exec-calc_1",
- "projection" : [ {
- "kind" : "INPUT_REF",
- "inputIndex" : 0,
- "type" : "BIGINT"
- }, {
- "kind" : "INPUT_REF",
- "inputIndex" : 2,
- "type" : "BIGINT NOT NULL"
- } ],
- "condition" : null,
- "inputProperties" : [ {
- "requiredDistribution" : {
- "type" : "UNKNOWN"
- },
- "damBehavior" : "PIPELINED",
- "priority" : 0
- } ],
- "outputType" : "ROW<`a` BIGINT, `w0$o0` BIGINT NOT NULL>",
- "description" : "Calc(select=[a, w0$o0])"
- }, {
- "id" : 6,
+ "id" : 16,
"type" : "stream-exec-sink_1",
"configuration" : {
"table.exec.sink.keyed-shuffle" : "AUTO",
@@ -143,28 +161,30 @@
},
"dynamicTableSink" : {
"table" : {
- "identifier" : "`default_catalog`.`default_database`.`MySink`",
+ "identifier" : "`default_catalog`.`default_database`.`sink_t`",
"resolvedTable" : {
"schema" : {
"columns" : [ {
"name" : "a",
- "dataType" : "BIGINT"
+ "dataType" : "INT NOT NULL"
}, {
"name" : "b",
- "dataType" : "BIGINT"
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "count_c",
+ "dataType" : "BIGINT NOT NULL"
+ }, {
+ "name" : "row_num",
+ "dataType" : "BIGINT NOT NULL"
} ],
"watermarkSpecs" : [ ]
},
- "partitionKeys" : [ ],
- "options" : {
- "connector" : "values",
- "sink-insert-only" : "false",
- "table-sink-class" : "DEFAULT"
- }
+ "partitionKeys" : [ ]
}
}
},
"inputChangelogMode" : [ "INSERT", "UPDATE_BEFORE", "UPDATE_AFTER", "DELETE" ],
+ "inputUpsertKey" : [ 0, 1 ],
"inputProperties" : [ {
"requiredDistribution" : {
"type" : "UNKNOWN"
@@ -172,43 +192,43 @@
"damBehavior" : "PIPELINED",
"priority" : 0
} ],
- "outputType" : "ROW<`a` BIGINT, `w0$o0` BIGINT NOT NULL>",
- "description" : "Sink(table=[default_catalog.default_database.MySink], fields=[a, w0$o0])"
+ "outputType" : "ROW<`a` INT NOT NULL, `b` VARCHAR(2147483647), `count_c` BIGINT NOT NULL, `w0$o0` BIGINT NOT NULL>",
+ "description" : "Sink(table=[default_catalog.default_database.sink_t], fields=[a, b, count_c, w0$o0])"
} ],
"edges" : [ {
- "source" : 1,
- "target" : 2,
+ "source" : 11,
+ "target" : 12,
"shuffle" : {
"type" : "FORWARD"
},
"shuffleMode" : "PIPELINED"
}, {
- "source" : 2,
- "target" : 3,
+ "source" : 12,
+ "target" : 13,
"shuffle" : {
"type" : "FORWARD"
},
"shuffleMode" : "PIPELINED"
}, {
- "source" : 3,
- "target" : 4,
+ "source" : 13,
+ "target" : 14,
"shuffle" : {
"type" : "FORWARD"
},
"shuffleMode" : "PIPELINED"
}, {
- "source" : 4,
- "target" : 5,
+ "source" : 14,
+ "target" : 15,
"shuffle" : {
"type" : "FORWARD"
},
"shuffleMode" : "PIPELINED"
}, {
- "source" : 5,
- "target" : 6,
+ "source" : 15,
+ "target" : 16,
"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-rank_1/rank-test-update-fast-strategy/savepoint/_metadata b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-rank_1/rank-test-update-fast-strategy/savepoint/_metadata
new file mode 100644
index 00000000000..b3d7adb0e5e
Binary files /dev/null and b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-rank_1/rank-test-update-fast-strategy/savepoint/_metadata differ