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/20 15:46:25 UTC
(flink) 02/02: [FLINK-33860] Remove WindowTableFunction JsonPlan & JsonIT tests
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 cabb28d25c4c58af3ee23fc4a63f9564aefd6146
Author: bvarghese1 <bv...@confluent.io>
AuthorDate: Fri Dec 15 13:15:33 2023 -0800
[FLINK-33860] Remove WindowTableFunction JsonPlan & JsonIT tests
- These are covered by the new restore tests
---
.../stream/WindowTableFunctionJsonPlanTest.java | 210 -----
.../jsonplan/WindowTableFunctionJsonITCase.java | 109 ---
.../testFollowedByWindowDeduplicate.out | 584 --------------
.../testFollowedByWindowJoin.out | 842 ---------------------
.../testFollowedByWindowRank.out | 515 -------------
.../testIndividualWindowTVF.out | 351 ---------
.../testIndividualWindowTVFProcessingTime.out | 456 -----------
7 files changed, 3067 deletions(-)
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowTableFunctionJsonPlanTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowTableFunctionJsonPlanTest.java
deleted file mode 100644
index 2b9ec7ba1e4..00000000000
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowTableFunctionJsonPlanTest.java
+++ /dev/null
@@ -1,210 +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/deserialization for window table function. */
-class WindowTableFunctionJsonPlanTest extends TableTestBase {
-
- private StreamTableTestUtil util;
- private TableEnvironment tEnv;
-
- @BeforeEach
- void setup() {
- util = streamTestUtil(TableConfig.getDefault());
- tEnv = util.getTableEnv();
-
- String srcTable1Ddl =
- "CREATE TABLE MyTable (\n"
- + " a INT,\n"
- + " b BIGINT,\n"
- + " c VARCHAR,\n"
- + " `rowtime` AS TO_TIMESTAMP(c),\n"
- + " proctime as PROCTIME(),\n"
- + " WATERMARK for `rowtime` AS `rowtime` - INTERVAL '1' SECOND\n"
- + ") WITH (\n"
- + " 'connector' = 'values')\n";
- tEnv.executeSql(srcTable1Ddl);
-
- String srcTable2Ddl =
- "CREATE TABLE MyTable2 (\n"
- + " a INT,\n"
- + " b BIGINT,\n"
- + " c VARCHAR,\n"
- + " `rowtime` AS TO_TIMESTAMP(c),\n"
- + " proctime as PROCTIME(),\n"
- + " WATERMARK for `rowtime` AS `rowtime` - INTERVAL '1' SECOND\n"
- + ") WITH (\n"
- + " 'connector' = 'values')\n";
- tEnv.executeSql(srcTable2Ddl);
- }
-
- @Test
- void testIndividualWindowTVF() {
- String sinkTableDdl =
- "CREATE TABLE MySink (\n"
- + " window_start TIMESTAMP(3),\n"
- + " window_end TIMESTAMP(3),\n"
- + " a INT,\n"
- + " b BIGINT,\n"
- + " c VARCHAR\n"
- + ") WITH (\n"
- + " 'connector' = 'values')\n";
- tEnv.executeSql(sinkTableDdl);
- util.verifyJsonPlan(
- "insert into MySink select\n"
- + " window_start,\n"
- + " window_end,\n"
- + " a,\n"
- + " b,\n"
- + " c\n"
- + "FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(rowtime), INTERVAL '15' MINUTE))");
- }
-
- @Test
- void testIndividualWindowTVFProcessingTime() {
- String sinkTableDdl =
- "CREATE TABLE MySink (\n"
- + " window_start TIMESTAMP(3),\n"
- + " window_end TIMESTAMP(3),\n"
- + " a INT,\n"
- + " b BIGINT,\n"
- + " c VARCHAR\n"
- + ") WITH (\n"
- + " 'connector' = 'values')\n";
- tEnv.executeSql(sinkTableDdl);
- util.verifyJsonPlan(
- "insert into MySink select\n"
- + " window_start,\n"
- + " window_end,\n"
- + " a,\n"
- + " b,\n"
- + " c\n"
- + "FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(proctime), INTERVAL '15' MINUTE))");
- }
-
- @Test
- void testFollowedByWindowJoin() {
- String sinkTableDdl =
- "CREATE TABLE MySink (\n"
- + " window_start TIMESTAMP(3) NOT NULL,\n"
- + " window_end TIMESTAMP(3) NOT NULL,\n"
- + " l_a INT,\n"
- + " l_b BIGINT,\n"
- + " l_c VARCHAR,\n"
- + " r_a INT,\n"
- + " r_b BIGINT,\n"
- + " r_c VARCHAR\n"
- + ") WITH (\n"
- + " 'connector' = 'values')\n";
- tEnv.executeSql(sinkTableDdl);
- util.verifyJsonPlan(
- "insert into MySink select\n"
- + " L.window_start,\n"
- + " L.window_end,\n"
- + " L.a,\n"
- + " L.b,\n"
- + " L.c,\n"
- + " R.a,\n"
- + " R.b,\n"
- + " R.c\n"
- + "FROM (\n"
- + " SELECT\n"
- + " window_start,\n"
- + " window_end,\n"
- + " a,\n"
- + " b,\n"
- + " c\n"
- + " FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(rowtime), INTERVAL '15' MINUTE))\n"
- + " WHERE b > 10\n"
- + ") L\n"
- + "JOIN (\n"
- + " SELECT\n"
- + " window_start,\n"
- + " window_end,\n"
- + " a,\n"
- + " b,\n"
- + " c\n"
- + " FROM TABLE(TUMBLE(TABLE MyTable2, DESCRIPTOR(rowtime), INTERVAL '15' MINUTE))\n"
- + " WHERE b > 10\n"
- + ") R\n"
- + "ON L.window_start = R.window_start AND L.window_end = R.window_end AND L.a = R.a");
- }
-
- @Test
- void testFollowedByWindowRank() {
- String sinkTableDdl =
- "CREATE TABLE MySink (\n"
- + " window_start TIMESTAMP(3),\n"
- + " window_end TIMESTAMP(3),\n"
- + " a INT,\n"
- + " b BIGINT,\n"
- + " c VARCHAR\n"
- + ") WITH (\n"
- + " 'connector' = 'values')\n";
- tEnv.executeSql(sinkTableDdl);
- util.verifyJsonPlan(
- "insert into MySink select\n"
- + " window_start,\n"
- + " window_end,\n"
- + " a,\n"
- + " b,\n"
- + " c\n"
- + "FROM (\n"
- + " SELECT\n"
- + " *,\n"
- + " ROW_NUMBER() OVER(PARTITION BY a, window_start, window_end ORDER BY b DESC) as rownum\n"
- + " FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(rowtime), INTERVAL '15' MINUTE)))\n"
- + "WHERE rownum <= 3");
- }
-
- @Test
- void testFollowedByWindowDeduplicate() {
- String sinkTableDdl =
- "CREATE TABLE MySink (\n"
- + " window_start TIMESTAMP(3),\n"
- + " window_end TIMESTAMP(3),\n"
- + " a INT,\n"
- + " b BIGINT,\n"
- + " c VARCHAR\n"
- + ") WITH (\n"
- + " 'connector' = 'values')\n";
- tEnv.executeSql(sinkTableDdl);
- util.verifyJsonPlan(
- "insert into MySink select\n"
- + " window_start,\n"
- + " window_end,\n"
- + " a,\n"
- + " b,\n"
- + " c\n"
- + "FROM (\n"
- + " SELECT\n"
- + " *,\n"
- + " ROW_NUMBER() OVER(PARTITION BY a, window_start, window_end ORDER BY rowtime DESC) as rownum\n"
- + " FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(rowtime), INTERVAL '15' MINUTE)))\n"
- + "WHERE rownum <= 1");
- }
-}
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/jsonplan/WindowTableFunctionJsonITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/jsonplan/WindowTableFunctionJsonITCase.java
deleted file mode 100644
index b816fdc189a..00000000000
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/jsonplan/WindowTableFunctionJsonITCase.java
+++ /dev/null
@@ -1,109 +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.BeforeEach;
-import org.junit.jupiter.api.Test;
-
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.List;
-
-/** Test for window deduplicate json plan. */
-class WindowTableFunctionJsonITCase extends JsonPlanTestBase {
-
- @BeforeEach
- @Override
- protected void setup() throws Exception {
- super.setup();
- createTestValuesSourceTable(
- "MyTable",
- JavaScalaConversionUtil.toJava(TestData.windowDataWithTimestamp()),
- new String[] {
- "ts STRING",
- "`int` INT",
- "`double` DOUBLE",
- "`float` FLOAT",
- "`bigdec` DECIMAL(10, 2)",
- "`string` STRING",
- "`name` STRING",
- "`rowtime` AS TO_TIMESTAMP(`ts`)",
- "WATERMARK for `rowtime` AS `rowtime` - INTERVAL '1' SECOND",
- },
- new HashMap<String, String>() {
- {
- put("enable-watermark-push-down", "true");
- put("failing-source", "true");
- }
- });
- }
-
- @Test
- void testEventTimeTumbleWindow() throws Exception {
- createTestValuesSinkTable(
- "MySink",
- "ts STRING",
- "`int` INT",
- "`double` DOUBLE",
- "`float` FLOAT",
- "`bigdec` DECIMAL(10, 2)",
- "`string` STRING",
- "`name` STRING",
- "`rowtime` STRING",
- "window_start TIMESTAMP(3)",
- "window_end TIMESTAMP(3)",
- "window_time TIMESTAMP(3)");
- compileSqlAndExecutePlan(
- "insert into MySink select\n"
- + " `ts`,\n"
- + " `int`,\n"
- + " `double`,\n"
- + " `float`, \n"
- + " `bigdec`, \n"
- + " `string`, \n"
- + " `name`, \n"
- + " CAST(`rowtime` AS STRING), \n"
- + " window_start, \n"
- + " window_end, \n"
- + " window_time \n"
- + "FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(rowtime), INTERVAL '5' SECOND))")
- .await();
-
- List<String> result = TestValuesTableFactory.getResultsAsStrings("MySink");
- assertResult(
- Arrays.asList(
- "+I[2020-10-10 00:00:01, 1, 1.0, 1.0, 1.11, Hi, a, 2020-10-10 00:00:01.000, 2020-10-10T00:00, 2020-10-10T00:00:05, 2020-10-10T00:00:04.999]",
- "+I[2020-10-10 00:00:02, 2, 2.0, 2.0, 2.22, Comment#1, a, 2020-10-10 00:00:02.000, 2020-10-10T00:00, 2020-10-10T00:00:05, 2020-10-10T00:00:04.999]",
- "+I[2020-10-10 00:00:03, 2, 2.0, 2.0, 2.22, Comment#1, a, 2020-10-10 00:00:03.000, 2020-10-10T00:00, 2020-10-10T00:00:05, 2020-10-10T00:00:04.999]",
- "+I[2020-10-10 00:00:04, 5, 5.0, 5.0, 5.55, null, a, 2020-10-10 00:00:04.000, 2020-10-10T00:00, 2020-10-10T00:00:05, 2020-10-10T00:00:04.999]",
- "+I[2020-10-10 00:00:04, 5, 5.0, null, 5.55, Hi, a, 2020-10-10 00:00:04.000, 2020-10-10T00:00, 2020-10-10T00:00:05, 2020-10-10T00:00:04.999]",
- "+I[2020-10-10 00:00:06, 6, 6.0, 6.0, 6.66, Hi, b, 2020-10-10 00:00:06.000, 2020-10-10T00:00:05, 2020-10-10T00:00:10, 2020-10-10T00:00:09.999]",
- "+I[2020-10-10 00:00:07, 3, 3.0, 3.0, null, Hello, b, 2020-10-10 00:00:07.000, 2020-10-10T00:00:05, 2020-10-10T00:00:10, 2020-10-10T00:00:09.999]",
- "+I[2020-10-10 00:00:08, 3, null, 3.0, 3.33, Comment#2, a, 2020-10-10 00:00:08.000, 2020-10-10T00:00:05, 2020-10-10T00:00:10, 2020-10-10T00:00:09.999]",
- "+I[2020-10-10 00:00:16, 4, 4.0, 4.0, 4.44, Hi, b, 2020-10-10 00:00:16.000, 2020-10-10T00:00:15, 2020-10-10T00:00:20, 2020-10-10T00:00:19.999]",
- "+I[2020-10-10 00:00:32, 7, 7.0, 7.0, 7.77, null, null, 2020-10-10 00:00:32.000, 2020-10-10T00:00:30, 2020-10-10T00:00:35, 2020-10-10T00:00:34.999]",
- "+I[2020-10-10 00:00:34, 1, 3.0, 3.0, 3.33, Comment#3, b, 2020-10-10 00:00:34.000, 2020-10-10T00:00:30, 2020-10-10T00:00:35, 2020-10-10T00:00:34.999]"),
- result);
- }
-}
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowTableFunctionJsonPlanTest_jsonplan/testFollowedByWindowDeduplicate.out b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowTableFunctionJsonPlanTest_jsonplan/testFollowedByWindowDeduplicate.out
deleted file mode 100644
index 32e969333a8..00000000000
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowTableFunctionJsonPlanTest_jsonplan/testFollowedByWindowDeduplicate.out
+++ /dev/null
@@ -1,584 +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" : "INT"
- }, {
- "name" : "b",
- "dataType" : "BIGINT"
- }, {
- "name" : "c",
- "dataType" : "VARCHAR(2147483647)"
- }, {
- "name" : "rowtime",
- "kind" : "COMPUTED",
- "expression" : {
- "rexNode" : {
- "kind" : "CALL",
- "internalName" : "$TO_TIMESTAMP$1",
- "operands" : [ {
- "kind" : "INPUT_REF",
- "inputIndex" : 2,
- "type" : "VARCHAR(2147483647)"
- } ],
- "type" : "TIMESTAMP(3)"
- },
- "serializableString" : "TO_TIMESTAMP(`c`)"
- }
- }, {
- "name" : "proctime",
- "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" : [ {
- "rowtimeAttribute" : "rowtime",
- "expression" : {
- "rexNode" : {
- "kind" : "CALL",
- "syntax" : "SPECIAL",
- "internalName" : "$-$1",
- "operands" : [ {
- "kind" : "INPUT_REF",
- "inputIndex" : 3,
- "type" : "TIMESTAMP(3)"
- }, {
- "kind" : "LITERAL",
- "value" : "1000",
- "type" : "INTERVAL SECOND(6) NOT NULL"
- } ],
- "type" : "TIMESTAMP(3)"
- },
- "serializableString" : "`rowtime` - INTERVAL '1' SECOND"
- }
- } ]
- },
- "partitionKeys" : [ ],
- "options" : {
- "connector" : "values"
- }
- }
- }
- },
- "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" : "$TO_TIMESTAMP$1",
- "operands" : [ {
- "kind" : "INPUT_REF",
- "inputIndex" : 2,
- "type" : "VARCHAR(2147483647)"
- } ],
- "type" : "TIMESTAMP(3)"
- }, {
- "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" : "INT"
- }, {
- "name" : "b",
- "fieldType" : "BIGINT"
- }, {
- "name" : "c",
- "fieldType" : "VARCHAR(2147483647)"
- }, {
- "name" : "rowtime",
- "fieldType" : "TIMESTAMP(3)"
- }, {
- "name" : "proctime",
- "fieldType" : {
- "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
- "nullable" : false,
- "precision" : 3,
- "kind" : "PROCTIME"
- }
- } ]
- },
- "description" : "Calc(select=[a, b, c, TO_TIMESTAMP(c) AS rowtime, PROCTIME() AS proctime])"
- }, {
- "id" : 3,
- "type" : "stream-exec-watermark-assigner_1",
- "watermarkExpr" : {
- "kind" : "CALL",
- "syntax" : "SPECIAL",
- "internalName" : "$-$1",
- "operands" : [ {
- "kind" : "INPUT_REF",
- "inputIndex" : 3,
- "type" : "TIMESTAMP(3)"
- }, {
- "kind" : "LITERAL",
- "value" : "1000",
- "type" : "INTERVAL SECOND(6) NOT NULL"
- } ],
- "type" : "TIMESTAMP(3)"
- },
- "rowtimeFieldIndex" : 3,
- "inputProperties" : [ {
- "requiredDistribution" : {
- "type" : "UNKNOWN"
- },
- "damBehavior" : "PIPELINED",
- "priority" : 0
- } ],
- "outputType" : {
- "type" : "ROW",
- "fields" : [ {
- "name" : "a",
- "fieldType" : "INT"
- }, {
- "name" : "b",
- "fieldType" : "BIGINT"
- }, {
- "name" : "c",
- "fieldType" : "VARCHAR(2147483647)"
- }, {
- "name" : "rowtime",
- "fieldType" : {
- "type" : "TIMESTAMP_WITHOUT_TIME_ZONE",
- "precision" : 3,
- "kind" : "ROWTIME"
- }
- }, {
- "name" : "proctime",
- "fieldType" : {
- "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
- "nullable" : false,
- "precision" : 3,
- "kind" : "PROCTIME"
- }
- } ]
- },
- "description" : "WatermarkAssigner(rowtime=[rowtime], watermark=[(rowtime - 1000:INTERVAL SECOND)])"
- }, {
- "id" : 4,
- "type" : "stream-exec-window-table-function_1",
- "configuration" : {
- "table.local-time-zone" : "default"
- },
- "windowing" : {
- "strategy" : "TimeAttribute",
- "window" : {
- "type" : "TumblingWindow",
- "size" : "PT15M"
- },
- "timeAttributeType" : {
- "type" : "TIMESTAMP_WITHOUT_TIME_ZONE",
- "precision" : 3,
- "kind" : "ROWTIME"
- },
- "timeAttributeIndex" : 3,
- "isRowtime" : true
- },
- "inputProperties" : [ {
- "requiredDistribution" : {
- "type" : "UNKNOWN"
- },
- "damBehavior" : "PIPELINED",
- "priority" : 0
- } ],
- "outputType" : {
- "type" : "ROW",
- "fields" : [ {
- "name" : "a",
- "fieldType" : "INT"
- }, {
- "name" : "b",
- "fieldType" : "BIGINT"
- }, {
- "name" : "c",
- "fieldType" : "VARCHAR(2147483647)"
- }, {
- "name" : "rowtime",
- "fieldType" : {
- "type" : "TIMESTAMP_WITHOUT_TIME_ZONE",
- "precision" : 3,
- "kind" : "ROWTIME"
- }
- }, {
- "name" : "proctime",
- "fieldType" : {
- "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
- "nullable" : false,
- "precision" : 3,
- "kind" : "PROCTIME"
- }
- }, {
- "name" : "window_start",
- "fieldType" : "TIMESTAMP(3) NOT NULL"
- }, {
- "name" : "window_end",
- "fieldType" : "TIMESTAMP(3) NOT NULL"
- }, {
- "name" : "window_time",
- "fieldType" : {
- "type" : "TIMESTAMP_WITHOUT_TIME_ZONE",
- "nullable" : false,
- "precision" : 3,
- "kind" : "ROWTIME"
- }
- } ]
- },
- "description" : "WindowTableFunction(window=[TUMBLE(time_col=[rowtime], size=[15 min])])"
- }, {
- "id" : 5,
- "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" : {
- "type" : "TIMESTAMP_WITHOUT_TIME_ZONE",
- "precision" : 3,
- "kind" : "ROWTIME"
- }
- }, {
- "kind" : "INPUT_REF",
- "inputIndex" : 5,
- "type" : "TIMESTAMP(3) NOT NULL"
- }, {
- "kind" : "INPUT_REF",
- "inputIndex" : 6,
- "type" : "TIMESTAMP(3) NOT NULL"
- } ],
- "condition" : null,
- "inputProperties" : [ {
- "requiredDistribution" : {
- "type" : "UNKNOWN"
- },
- "damBehavior" : "PIPELINED",
- "priority" : 0
- } ],
- "outputType" : {
- "type" : "ROW",
- "fields" : [ {
- "name" : "a",
- "fieldType" : "INT"
- }, {
- "name" : "b",
- "fieldType" : "BIGINT"
- }, {
- "name" : "c",
- "fieldType" : "VARCHAR(2147483647)"
- }, {
- "name" : "rowtime",
- "fieldType" : {
- "type" : "TIMESTAMP_WITHOUT_TIME_ZONE",
- "precision" : 3,
- "kind" : "ROWTIME"
- }
- }, {
- "name" : "window_start",
- "fieldType" : "TIMESTAMP(3) NOT NULL"
- }, {
- "name" : "window_end",
- "fieldType" : "TIMESTAMP(3) NOT NULL"
- } ]
- },
- "description" : "Calc(select=[a, b, c, rowtime, window_start, window_end])"
- }, {
- "id" : 6,
- "type" : "stream-exec-exchange_1",
- "inputProperties" : [ {
- "requiredDistribution" : {
- "type" : "HASH",
- "keys" : [ 0 ]
- },
- "damBehavior" : "PIPELINED",
- "priority" : 0
- } ],
- "outputType" : {
- "type" : "ROW",
- "fields" : [ {
- "name" : "a",
- "fieldType" : "INT"
- }, {
- "name" : "b",
- "fieldType" : "BIGINT"
- }, {
- "name" : "c",
- "fieldType" : "VARCHAR(2147483647)"
- }, {
- "name" : "rowtime",
- "fieldType" : {
- "type" : "TIMESTAMP_WITHOUT_TIME_ZONE",
- "precision" : 3,
- "kind" : "ROWTIME"
- }
- }, {
- "name" : "window_start",
- "fieldType" : "TIMESTAMP(3) NOT NULL"
- }, {
- "name" : "window_end",
- "fieldType" : "TIMESTAMP(3) NOT NULL"
- } ]
- },
- "description" : "Exchange(distribution=[hash[a]])"
- }, {
- "id" : 7,
- "type" : "stream-exec-window-deduplicate_1",
- "configuration" : {
- "table.local-time-zone" : "default"
- },
- "partitionKeys" : [ 0 ],
- "orderKey" : 3,
- "keepLastRow" : true,
- "windowing" : {
- "strategy" : "WindowAttached",
- "window" : {
- "type" : "TumblingWindow",
- "size" : "PT15M"
- },
- "timeAttributeType" : {
- "type" : "TIMESTAMP_WITHOUT_TIME_ZONE",
- "precision" : 3,
- "kind" : "ROWTIME"
- },
- "windowStart" : 4,
- "windowEnd" : 5,
- "isRowtime" : true
- },
- "inputProperties" : [ {
- "requiredDistribution" : {
- "type" : "UNKNOWN"
- },
- "damBehavior" : "PIPELINED",
- "priority" : 0
- } ],
- "outputType" : {
- "type" : "ROW",
- "fields" : [ {
- "name" : "a",
- "fieldType" : "INT"
- }, {
- "name" : "b",
- "fieldType" : "BIGINT"
- }, {
- "name" : "c",
- "fieldType" : "VARCHAR(2147483647)"
- }, {
- "name" : "rowtime",
- "fieldType" : {
- "type" : "TIMESTAMP_WITHOUT_TIME_ZONE",
- "precision" : 3,
- "kind" : "ROWTIME"
- }
- }, {
- "name" : "window_start",
- "fieldType" : "TIMESTAMP(3) NOT NULL"
- }, {
- "name" : "window_end",
- "fieldType" : "TIMESTAMP(3) NOT NULL"
- } ]
- },
- "description" : "WindowDeduplicate(window=[TUMBLE(win_start=[window_start], win_end=[window_end], size=[15 min])], keep=[LastRow], partitionKeys=[a], orderKey=[rowtime], order=[ROWTIME])"
- }, {
- "id" : 8,
- "type" : "stream-exec-calc_1",
- "projection" : [ {
- "kind" : "INPUT_REF",
- "inputIndex" : 4,
- "type" : "TIMESTAMP(3) NOT NULL"
- }, {
- "kind" : "INPUT_REF",
- "inputIndex" : 5,
- "type" : "TIMESTAMP(3) NOT NULL"
- }, {
- "kind" : "INPUT_REF",
- "inputIndex" : 0,
- "type" : "INT"
- }, {
- "kind" : "INPUT_REF",
- "inputIndex" : 1,
- "type" : "BIGINT"
- }, {
- "kind" : "INPUT_REF",
- "inputIndex" : 2,
- "type" : "VARCHAR(2147483647)"
- } ],
- "condition" : null,
- "inputProperties" : [ {
- "requiredDistribution" : {
- "type" : "UNKNOWN"
- },
- "damBehavior" : "PIPELINED",
- "priority" : 0
- } ],
- "outputType" : "ROW<`window_start` TIMESTAMP(3) NOT NULL, `window_end` TIMESTAMP(3) NOT NULL, `a` INT, `b` BIGINT, `c` VARCHAR(2147483647)>",
- "description" : "Calc(select=[window_start, window_end, a, b, c])"
- }, {
- "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" : "window_start",
- "dataType" : "TIMESTAMP(3)"
- }, {
- "name" : "window_end",
- "dataType" : "TIMESTAMP(3)"
- }, {
- "name" : "a",
- "dataType" : "INT"
- }, {
- "name" : "b",
- "dataType" : "BIGINT"
- }, {
- "name" : "c",
- "dataType" : "VARCHAR(2147483647)"
- } ],
- "watermarkSpecs" : [ ]
- },
- "partitionKeys" : [ ],
- "options" : {
- "connector" : "values"
- }
- }
- }
- },
- "inputChangelogMode" : [ "INSERT" ],
- "inputProperties" : [ {
- "requiredDistribution" : {
- "type" : "UNKNOWN"
- },
- "damBehavior" : "PIPELINED",
- "priority" : 0
- } ],
- "outputType" : "ROW<`window_start` TIMESTAMP(3) NOT NULL, `window_end` TIMESTAMP(3) NOT NULL, `a` INT, `b` BIGINT, `c` VARCHAR(2147483647)>",
- "description" : "Sink(table=[default_catalog.default_database.MySink], fields=[window_start, window_end, a, b, 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"
- } ]
-}
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowTableFunctionJsonPlanTest_jsonplan/testFollowedByWindowJoin.out b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowTableFunctionJsonPlanTest_jsonplan/testFollowedByWindowJoin.out
deleted file mode 100644
index 95e775ff06c..00000000000
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowTableFunctionJsonPlanTest_jsonplan/testFollowedByWindowJoin.out
+++ /dev/null
@@ -1,842 +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" : "INT"
- }, {
- "name" : "b",
- "dataType" : "BIGINT"
- }, {
- "name" : "c",
- "dataType" : "VARCHAR(2147483647)"
- }, {
- "name" : "rowtime",
- "kind" : "COMPUTED",
- "expression" : {
- "rexNode" : {
- "kind" : "CALL",
- "internalName" : "$TO_TIMESTAMP$1",
- "operands" : [ {
- "kind" : "INPUT_REF",
- "inputIndex" : 2,
- "type" : "VARCHAR(2147483647)"
- } ],
- "type" : "TIMESTAMP(3)"
- },
- "serializableString" : "TO_TIMESTAMP(`c`)"
- }
- }, {
- "name" : "proctime",
- "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" : [ {
- "rowtimeAttribute" : "rowtime",
- "expression" : {
- "rexNode" : {
- "kind" : "CALL",
- "syntax" : "SPECIAL",
- "internalName" : "$-$1",
- "operands" : [ {
- "kind" : "INPUT_REF",
- "inputIndex" : 3,
- "type" : "TIMESTAMP(3)"
- }, {
- "kind" : "LITERAL",
- "value" : "1000",
- "type" : "INTERVAL SECOND(6) NOT NULL"
- } ],
- "type" : "TIMESTAMP(3)"
- },
- "serializableString" : "`rowtime` - INTERVAL '1' SECOND"
- }
- } ]
- },
- "partitionKeys" : [ ],
- "options" : {
- "connector" : "values"
- }
- }
- }
- },
- "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" : "$TO_TIMESTAMP$1",
- "operands" : [ {
- "kind" : "INPUT_REF",
- "inputIndex" : 2,
- "type" : "VARCHAR(2147483647)"
- } ],
- "type" : "TIMESTAMP(3)"
- } ],
- "condition" : null,
- "inputProperties" : [ {
- "requiredDistribution" : {
- "type" : "UNKNOWN"
- },
- "damBehavior" : "PIPELINED",
- "priority" : 0
- } ],
- "outputType" : "ROW<`a` INT, `b` BIGINT, `c` VARCHAR(2147483647), `rowtime` TIMESTAMP(3)>",
- "description" : "Calc(select=[a, b, c, TO_TIMESTAMP(c) AS rowtime])"
- }, {
- "id" : 3,
- "type" : "stream-exec-watermark-assigner_1",
- "watermarkExpr" : {
- "kind" : "CALL",
- "syntax" : "SPECIAL",
- "internalName" : "$-$1",
- "operands" : [ {
- "kind" : "INPUT_REF",
- "inputIndex" : 3,
- "type" : "TIMESTAMP(3)"
- }, {
- "kind" : "LITERAL",
- "value" : "1000",
- "type" : "INTERVAL SECOND(6) NOT NULL"
- } ],
- "type" : "TIMESTAMP(3)"
- },
- "rowtimeFieldIndex" : 3,
- "inputProperties" : [ {
- "requiredDistribution" : {
- "type" : "UNKNOWN"
- },
- "damBehavior" : "PIPELINED",
- "priority" : 0
- } ],
- "outputType" : {
- "type" : "ROW",
- "fields" : [ {
- "name" : "a",
- "fieldType" : "INT"
- }, {
- "name" : "b",
- "fieldType" : "BIGINT"
- }, {
- "name" : "c",
- "fieldType" : "VARCHAR(2147483647)"
- }, {
- "name" : "rowtime",
- "fieldType" : {
- "type" : "TIMESTAMP_WITHOUT_TIME_ZONE",
- "precision" : 3,
- "kind" : "ROWTIME"
- }
- } ]
- },
- "description" : "WatermarkAssigner(rowtime=[rowtime], watermark=[(rowtime - 1000:INTERVAL SECOND)])"
- }, {
- "id" : 4,
- "type" : "stream-exec-window-table-function_1",
- "configuration" : {
- "table.local-time-zone" : "default"
- },
- "windowing" : {
- "strategy" : "TimeAttribute",
- "window" : {
- "type" : "TumblingWindow",
- "size" : "PT15M"
- },
- "timeAttributeType" : {
- "type" : "TIMESTAMP_WITHOUT_TIME_ZONE",
- "precision" : 3,
- "kind" : "ROWTIME"
- },
- "timeAttributeIndex" : 3,
- "isRowtime" : true
- },
- "inputProperties" : [ {
- "requiredDistribution" : {
- "type" : "UNKNOWN"
- },
- "damBehavior" : "PIPELINED",
- "priority" : 0
- } ],
- "outputType" : {
- "type" : "ROW",
- "fields" : [ {
- "name" : "a",
- "fieldType" : "INT"
- }, {
- "name" : "b",
- "fieldType" : "BIGINT"
- }, {
- "name" : "c",
- "fieldType" : "VARCHAR(2147483647)"
- }, {
- "name" : "rowtime",
- "fieldType" : {
- "type" : "TIMESTAMP_WITHOUT_TIME_ZONE",
- "precision" : 3,
- "kind" : "ROWTIME"
- }
- }, {
- "name" : "window_start",
- "fieldType" : "TIMESTAMP(3) NOT NULL"
- }, {
- "name" : "window_end",
- "fieldType" : "TIMESTAMP(3) NOT NULL"
- }, {
- "name" : "window_time",
- "fieldType" : {
- "type" : "TIMESTAMP_WITHOUT_TIME_ZONE",
- "nullable" : false,
- "precision" : 3,
- "kind" : "ROWTIME"
- }
- } ]
- },
- "description" : "WindowTableFunction(window=[TUMBLE(time_col=[rowtime], size=[15 min])])"
- }, {
- "id" : 5,
- "type" : "stream-exec-calc_1",
- "projection" : [ {
- "kind" : "INPUT_REF",
- "inputIndex" : 4,
- "type" : "TIMESTAMP(3) NOT NULL"
- }, {
- "kind" : "INPUT_REF",
- "inputIndex" : 5,
- "type" : "TIMESTAMP(3) NOT NULL"
- }, {
- "kind" : "INPUT_REF",
- "inputIndex" : 0,
- "type" : "INT"
- }, {
- "kind" : "INPUT_REF",
- "inputIndex" : 1,
- "type" : "BIGINT"
- }, {
- "kind" : "INPUT_REF",
- "inputIndex" : 2,
- "type" : "VARCHAR(2147483647)"
- } ],
- "condition" : {
- "kind" : "CALL",
- "syntax" : "BINARY",
- "internalName" : "$>$1",
- "operands" : [ {
- "kind" : "INPUT_REF",
- "inputIndex" : 1,
- "type" : "BIGINT"
- }, {
- "kind" : "LITERAL",
- "value" : 10,
- "type" : "INT NOT NULL"
- } ],
- "type" : "BOOLEAN"
- },
- "inputProperties" : [ {
- "requiredDistribution" : {
- "type" : "UNKNOWN"
- },
- "damBehavior" : "PIPELINED",
- "priority" : 0
- } ],
- "outputType" : "ROW<`window_start` TIMESTAMP(3) NOT NULL, `window_end` TIMESTAMP(3) NOT NULL, `a` INT, `b` BIGINT, `c` VARCHAR(2147483647)>",
- "description" : "Calc(select=[window_start, window_end, a, b, c], where=[(b > 10)])"
- }, {
- "id" : 6,
- "type" : "stream-exec-exchange_1",
- "inputProperties" : [ {
- "requiredDistribution" : {
- "type" : "HASH",
- "keys" : [ 2 ]
- },
- "damBehavior" : "PIPELINED",
- "priority" : 0
- } ],
- "outputType" : "ROW<`window_start` TIMESTAMP(3) NOT NULL, `window_end` TIMESTAMP(3) NOT NULL, `a` INT, `b` BIGINT, `c` VARCHAR(2147483647)>",
- "description" : "Exchange(distribution=[hash[a]])"
- }, {
- "id" : 7,
- "type" : "stream-exec-table-source-scan_1",
- "scanTableSource" : {
- "table" : {
- "identifier" : "`default_catalog`.`default_database`.`MyTable2`",
- "resolvedTable" : {
- "schema" : {
- "columns" : [ {
- "name" : "a",
- "dataType" : "INT"
- }, {
- "name" : "b",
- "dataType" : "BIGINT"
- }, {
- "name" : "c",
- "dataType" : "VARCHAR(2147483647)"
- }, {
- "name" : "rowtime",
- "kind" : "COMPUTED",
- "expression" : {
- "rexNode" : {
- "kind" : "CALL",
- "internalName" : "$TO_TIMESTAMP$1",
- "operands" : [ {
- "kind" : "INPUT_REF",
- "inputIndex" : 2,
- "type" : "VARCHAR(2147483647)"
- } ],
- "type" : "TIMESTAMP(3)"
- },
- "serializableString" : "TO_TIMESTAMP(`c`)"
- }
- }, {
- "name" : "proctime",
- "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" : [ {
- "rowtimeAttribute" : "rowtime",
- "expression" : {
- "rexNode" : {
- "kind" : "CALL",
- "syntax" : "SPECIAL",
- "internalName" : "$-$1",
- "operands" : [ {
- "kind" : "INPUT_REF",
- "inputIndex" : 3,
- "type" : "TIMESTAMP(3)"
- }, {
- "kind" : "LITERAL",
- "value" : "1000",
- "type" : "INTERVAL SECOND(6) NOT NULL"
- } ],
- "type" : "TIMESTAMP(3)"
- },
- "serializableString" : "`rowtime` - INTERVAL '1' SECOND"
- }
- } ]
- },
- "partitionKeys" : [ ],
- "options" : {
- "connector" : "values"
- }
- }
- }
- },
- "outputType" : "ROW<`a` INT, `b` BIGINT, `c` VARCHAR(2147483647)>",
- "description" : "TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[a, b, c])",
- "inputProperties" : [ ]
- }, {
- "id" : 8,
- "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" : "$TO_TIMESTAMP$1",
- "operands" : [ {
- "kind" : "INPUT_REF",
- "inputIndex" : 2,
- "type" : "VARCHAR(2147483647)"
- } ],
- "type" : "TIMESTAMP(3)"
- } ],
- "condition" : null,
- "inputProperties" : [ {
- "requiredDistribution" : {
- "type" : "UNKNOWN"
- },
- "damBehavior" : "PIPELINED",
- "priority" : 0
- } ],
- "outputType" : "ROW<`a` INT, `b` BIGINT, `c` VARCHAR(2147483647), `rowtime` TIMESTAMP(3)>",
- "description" : "Calc(select=[a, b, c, TO_TIMESTAMP(c) AS rowtime])"
- }, {
- "id" : 9,
- "type" : "stream-exec-watermark-assigner_1",
- "watermarkExpr" : {
- "kind" : "CALL",
- "syntax" : "SPECIAL",
- "internalName" : "$-$1",
- "operands" : [ {
- "kind" : "INPUT_REF",
- "inputIndex" : 3,
- "type" : "TIMESTAMP(3)"
- }, {
- "kind" : "LITERAL",
- "value" : "1000",
- "type" : "INTERVAL SECOND(6) NOT NULL"
- } ],
- "type" : "TIMESTAMP(3)"
- },
- "rowtimeFieldIndex" : 3,
- "inputProperties" : [ {
- "requiredDistribution" : {
- "type" : "UNKNOWN"
- },
- "damBehavior" : "PIPELINED",
- "priority" : 0
- } ],
- "outputType" : {
- "type" : "ROW",
- "fields" : [ {
- "name" : "a",
- "fieldType" : "INT"
- }, {
- "name" : "b",
- "fieldType" : "BIGINT"
- }, {
- "name" : "c",
- "fieldType" : "VARCHAR(2147483647)"
- }, {
- "name" : "rowtime",
- "fieldType" : {
- "type" : "TIMESTAMP_WITHOUT_TIME_ZONE",
- "precision" : 3,
- "kind" : "ROWTIME"
- }
- } ]
- },
- "description" : "WatermarkAssigner(rowtime=[rowtime], watermark=[(rowtime - 1000:INTERVAL SECOND)])"
- }, {
- "id" : 10,
- "type" : "stream-exec-window-table-function_1",
- "configuration" : {
- "table.local-time-zone" : "default"
- },
- "windowing" : {
- "strategy" : "TimeAttribute",
- "window" : {
- "type" : "TumblingWindow",
- "size" : "PT15M"
- },
- "timeAttributeType" : {
- "type" : "TIMESTAMP_WITHOUT_TIME_ZONE",
- "precision" : 3,
- "kind" : "ROWTIME"
- },
- "timeAttributeIndex" : 3,
- "isRowtime" : true
- },
- "inputProperties" : [ {
- "requiredDistribution" : {
- "type" : "UNKNOWN"
- },
- "damBehavior" : "PIPELINED",
- "priority" : 0
- } ],
- "outputType" : {
- "type" : "ROW",
- "fields" : [ {
- "name" : "a",
- "fieldType" : "INT"
- }, {
- "name" : "b",
- "fieldType" : "BIGINT"
- }, {
- "name" : "c",
- "fieldType" : "VARCHAR(2147483647)"
- }, {
- "name" : "rowtime",
- "fieldType" : {
- "type" : "TIMESTAMP_WITHOUT_TIME_ZONE",
- "precision" : 3,
- "kind" : "ROWTIME"
- }
- }, {
- "name" : "window_start",
- "fieldType" : "TIMESTAMP(3) NOT NULL"
- }, {
- "name" : "window_end",
- "fieldType" : "TIMESTAMP(3) NOT NULL"
- }, {
- "name" : "window_time",
- "fieldType" : {
- "type" : "TIMESTAMP_WITHOUT_TIME_ZONE",
- "nullable" : false,
- "precision" : 3,
- "kind" : "ROWTIME"
- }
- } ]
- },
- "description" : "WindowTableFunction(window=[TUMBLE(time_col=[rowtime], size=[15 min])])"
- }, {
- "id" : 11,
- "type" : "stream-exec-calc_1",
- "projection" : [ {
- "kind" : "INPUT_REF",
- "inputIndex" : 4,
- "type" : "TIMESTAMP(3) NOT NULL"
- }, {
- "kind" : "INPUT_REF",
- "inputIndex" : 5,
- "type" : "TIMESTAMP(3) NOT NULL"
- }, {
- "kind" : "INPUT_REF",
- "inputIndex" : 0,
- "type" : "INT"
- }, {
- "kind" : "INPUT_REF",
- "inputIndex" : 1,
- "type" : "BIGINT"
- }, {
- "kind" : "INPUT_REF",
- "inputIndex" : 2,
- "type" : "VARCHAR(2147483647)"
- } ],
- "condition" : {
- "kind" : "CALL",
- "syntax" : "BINARY",
- "internalName" : "$>$1",
- "operands" : [ {
- "kind" : "INPUT_REF",
- "inputIndex" : 1,
- "type" : "BIGINT"
- }, {
- "kind" : "LITERAL",
- "value" : 10,
- "type" : "INT NOT NULL"
- } ],
- "type" : "BOOLEAN"
- },
- "inputProperties" : [ {
- "requiredDistribution" : {
- "type" : "UNKNOWN"
- },
- "damBehavior" : "PIPELINED",
- "priority" : 0
- } ],
- "outputType" : "ROW<`window_start` TIMESTAMP(3) NOT NULL, `window_end` TIMESTAMP(3) NOT NULL, `a` INT, `b` BIGINT, `c` VARCHAR(2147483647)>",
- "description" : "Calc(select=[window_start, window_end, a, b, c], where=[(b > 10)])"
- }, {
- "id" : 12,
- "type" : "stream-exec-exchange_1",
- "inputProperties" : [ {
- "requiredDistribution" : {
- "type" : "HASH",
- "keys" : [ 2 ]
- },
- "damBehavior" : "PIPELINED",
- "priority" : 0
- } ],
- "outputType" : "ROW<`window_start` TIMESTAMP(3) NOT NULL, `window_end` TIMESTAMP(3) NOT NULL, `a` INT, `b` BIGINT, `c` VARCHAR(2147483647)>",
- "description" : "Exchange(distribution=[hash[a]])"
- }, {
- "id" : 13,
- "type" : "stream-exec-window-join_1",
- "configuration" : {
- "table.local-time-zone" : "default"
- },
- "joinSpec" : {
- "joinType" : "INNER",
- "leftKeys" : [ 2 ],
- "rightKeys" : [ 2 ],
- "filterNulls" : [ true ],
- "nonEquiCondition" : null
- },
- "leftWindowing" : {
- "strategy" : "WindowAttached",
- "window" : {
- "type" : "TumblingWindow",
- "size" : "PT15M"
- },
- "timeAttributeType" : {
- "type" : "TIMESTAMP_WITHOUT_TIME_ZONE",
- "precision" : 3,
- "kind" : "ROWTIME"
- },
- "windowStart" : 0,
- "windowEnd" : 1,
- "isRowtime" : true
- },
- "rightWindowing" : {
- "strategy" : "WindowAttached",
- "window" : {
- "type" : "TumblingWindow",
- "size" : "PT15M"
- },
- "timeAttributeType" : {
- "type" : "TIMESTAMP_WITHOUT_TIME_ZONE",
- "precision" : 3,
- "kind" : "ROWTIME"
- },
- "windowStart" : 0,
- "windowEnd" : 1,
- "isRowtime" : true
- },
- "inputProperties" : [ {
- "requiredDistribution" : {
- "type" : "UNKNOWN"
- },
- "damBehavior" : "PIPELINED",
- "priority" : 0
- }, {
- "requiredDistribution" : {
- "type" : "UNKNOWN"
- },
- "damBehavior" : "PIPELINED",
- "priority" : 0
- } ],
- "outputType" : "ROW<`window_start` TIMESTAMP(3) NOT NULL, `window_end` TIMESTAMP(3) NOT NULL, `a` INT, `b` BIGINT, `c` VARCHAR(2147483647), `window_start0` TIMESTAMP(3) NOT NULL, `window_end0` TIMESTAMP(3) NOT NULL, `a0` INT, `b0` BIGINT, `c0` VARCHAR(2147483647)>",
- "description" : "WindowJoin(leftWindow=[TUMBLE(win_start=[window_start], win_end=[window_end], size=[15 min])], rightWindow=[TUMBLE(win_start=[window_start], win_end=[window_end], size=[15 min])], joinType=[InnerJoin], where=[(a = a0)], select=[window_start, window_end, a, b, c, window_start0, window_end0, a0, b0, c0])"
- }, {
- "id" : 14,
- "type" : "stream-exec-calc_1",
- "projection" : [ {
- "kind" : "INPUT_REF",
- "inputIndex" : 0,
- "type" : "TIMESTAMP(3) NOT NULL"
- }, {
- "kind" : "INPUT_REF",
- "inputIndex" : 1,
- "type" : "TIMESTAMP(3) NOT NULL"
- }, {
- "kind" : "INPUT_REF",
- "inputIndex" : 2,
- "type" : "INT"
- }, {
- "kind" : "INPUT_REF",
- "inputIndex" : 3,
- "type" : "BIGINT"
- }, {
- "kind" : "INPUT_REF",
- "inputIndex" : 4,
- "type" : "VARCHAR(2147483647)"
- }, {
- "kind" : "INPUT_REF",
- "inputIndex" : 7,
- "type" : "INT"
- }, {
- "kind" : "INPUT_REF",
- "inputIndex" : 8,
- "type" : "BIGINT"
- }, {
- "kind" : "INPUT_REF",
- "inputIndex" : 9,
- "type" : "VARCHAR(2147483647)"
- } ],
- "condition" : null,
- "inputProperties" : [ {
- "requiredDistribution" : {
- "type" : "UNKNOWN"
- },
- "damBehavior" : "PIPELINED",
- "priority" : 0
- } ],
- "outputType" : "ROW<`window_start` TIMESTAMP(3) NOT NULL, `window_end` TIMESTAMP(3) NOT NULL, `a` INT, `b` BIGINT, `c` VARCHAR(2147483647), `a0` INT, `b0` BIGINT, `c0` VARCHAR(2147483647)>",
- "description" : "Calc(select=[window_start, window_end, a, b, c, a0, b0, c0])"
- }, {
- "id" : 15,
- "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" : "window_start",
- "dataType" : "TIMESTAMP(3) NOT NULL"
- }, {
- "name" : "window_end",
- "dataType" : "TIMESTAMP(3) NOT NULL"
- }, {
- "name" : "l_a",
- "dataType" : "INT"
- }, {
- "name" : "l_b",
- "dataType" : "BIGINT"
- }, {
- "name" : "l_c",
- "dataType" : "VARCHAR(2147483647)"
- }, {
- "name" : "r_a",
- "dataType" : "INT"
- }, {
- "name" : "r_b",
- "dataType" : "BIGINT"
- }, {
- "name" : "r_c",
- "dataType" : "VARCHAR(2147483647)"
- } ],
- "watermarkSpecs" : [ ]
- },
- "partitionKeys" : [ ],
- "options" : {
- "connector" : "values"
- }
- }
- }
- },
- "inputChangelogMode" : [ "INSERT" ],
- "inputProperties" : [ {
- "requiredDistribution" : {
- "type" : "UNKNOWN"
- },
- "damBehavior" : "PIPELINED",
- "priority" : 0
- } ],
- "outputType" : "ROW<`window_start` TIMESTAMP(3) NOT NULL, `window_end` TIMESTAMP(3) NOT NULL, `a` INT, `b` BIGINT, `c` VARCHAR(2147483647), `a0` INT, `b0` BIGINT, `c0` VARCHAR(2147483647)>",
- "description" : "Sink(table=[default_catalog.default_database.MySink], fields=[window_start, window_end, a, b, c, a0, b0, c0])"
- } ],
- "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" : 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" : 6,
- "target" : 13,
- "shuffle" : {
- "type" : "FORWARD"
- },
- "shuffleMode" : "PIPELINED"
- }, {
- "source" : 12,
- "target" : 13,
- "shuffle" : {
- "type" : "FORWARD"
- },
- "shuffleMode" : "PIPELINED"
- }, {
- "source" : 13,
- "target" : 14,
- "shuffle" : {
- "type" : "FORWARD"
- },
- "shuffleMode" : "PIPELINED"
- }, {
- "source" : 14,
- "target" : 15,
- "shuffle" : {
- "type" : "FORWARD"
- },
- "shuffleMode" : "PIPELINED"
- } ]
-}
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowTableFunctionJsonPlanTest_jsonplan/testFollowedByWindowRank.out b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowTableFunctionJsonPlanTest_jsonplan/testFollowedByWindowRank.out
deleted file mode 100644
index d710c772534..00000000000
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowTableFunctionJsonPlanTest_jsonplan/testFollowedByWindowRank.out
+++ /dev/null
@@ -1,515 +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" : "INT"
- }, {
- "name" : "b",
- "dataType" : "BIGINT"
- }, {
- "name" : "c",
- "dataType" : "VARCHAR(2147483647)"
- }, {
- "name" : "rowtime",
- "kind" : "COMPUTED",
- "expression" : {
- "rexNode" : {
- "kind" : "CALL",
- "internalName" : "$TO_TIMESTAMP$1",
- "operands" : [ {
- "kind" : "INPUT_REF",
- "inputIndex" : 2,
- "type" : "VARCHAR(2147483647)"
- } ],
- "type" : "TIMESTAMP(3)"
- },
- "serializableString" : "TO_TIMESTAMP(`c`)"
- }
- }, {
- "name" : "proctime",
- "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" : [ {
- "rowtimeAttribute" : "rowtime",
- "expression" : {
- "rexNode" : {
- "kind" : "CALL",
- "syntax" : "SPECIAL",
- "internalName" : "$-$1",
- "operands" : [ {
- "kind" : "INPUT_REF",
- "inputIndex" : 3,
- "type" : "TIMESTAMP(3)"
- }, {
- "kind" : "LITERAL",
- "value" : "1000",
- "type" : "INTERVAL SECOND(6) NOT NULL"
- } ],
- "type" : "TIMESTAMP(3)"
- },
- "serializableString" : "`rowtime` - INTERVAL '1' SECOND"
- }
- } ]
- },
- "partitionKeys" : [ ],
- "options" : {
- "connector" : "values"
- }
- }
- }
- },
- "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" : "$TO_TIMESTAMP$1",
- "operands" : [ {
- "kind" : "INPUT_REF",
- "inputIndex" : 2,
- "type" : "VARCHAR(2147483647)"
- } ],
- "type" : "TIMESTAMP(3)"
- }, {
- "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" : "INT"
- }, {
- "name" : "b",
- "fieldType" : "BIGINT"
- }, {
- "name" : "c",
- "fieldType" : "VARCHAR(2147483647)"
- }, {
- "name" : "rowtime",
- "fieldType" : "TIMESTAMP(3)"
- }, {
- "name" : "proctime",
- "fieldType" : {
- "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
- "nullable" : false,
- "precision" : 3,
- "kind" : "PROCTIME"
- }
- } ]
- },
- "description" : "Calc(select=[a, b, c, TO_TIMESTAMP(c) AS rowtime, PROCTIME() AS proctime])"
- }, {
- "id" : 3,
- "type" : "stream-exec-watermark-assigner_1",
- "watermarkExpr" : {
- "kind" : "CALL",
- "syntax" : "SPECIAL",
- "internalName" : "$-$1",
- "operands" : [ {
- "kind" : "INPUT_REF",
- "inputIndex" : 3,
- "type" : "TIMESTAMP(3)"
- }, {
- "kind" : "LITERAL",
- "value" : "1000",
- "type" : "INTERVAL SECOND(6) NOT NULL"
- } ],
- "type" : "TIMESTAMP(3)"
- },
- "rowtimeFieldIndex" : 3,
- "inputProperties" : [ {
- "requiredDistribution" : {
- "type" : "UNKNOWN"
- },
- "damBehavior" : "PIPELINED",
- "priority" : 0
- } ],
- "outputType" : {
- "type" : "ROW",
- "fields" : [ {
- "name" : "a",
- "fieldType" : "INT"
- }, {
- "name" : "b",
- "fieldType" : "BIGINT"
- }, {
- "name" : "c",
- "fieldType" : "VARCHAR(2147483647)"
- }, {
- "name" : "rowtime",
- "fieldType" : {
- "type" : "TIMESTAMP_WITHOUT_TIME_ZONE",
- "precision" : 3,
- "kind" : "ROWTIME"
- }
- }, {
- "name" : "proctime",
- "fieldType" : {
- "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
- "nullable" : false,
- "precision" : 3,
- "kind" : "PROCTIME"
- }
- } ]
- },
- "description" : "WatermarkAssigner(rowtime=[rowtime], watermark=[(rowtime - 1000:INTERVAL SECOND)])"
- }, {
- "id" : 4,
- "type" : "stream-exec-window-table-function_1",
- "configuration" : {
- "table.local-time-zone" : "default"
- },
- "windowing" : {
- "strategy" : "TimeAttribute",
- "window" : {
- "type" : "TumblingWindow",
- "size" : "PT15M"
- },
- "timeAttributeType" : {
- "type" : "TIMESTAMP_WITHOUT_TIME_ZONE",
- "precision" : 3,
- "kind" : "ROWTIME"
- },
- "timeAttributeIndex" : 3,
- "isRowtime" : true
- },
- "inputProperties" : [ {
- "requiredDistribution" : {
- "type" : "UNKNOWN"
- },
- "damBehavior" : "PIPELINED",
- "priority" : 0
- } ],
- "outputType" : {
- "type" : "ROW",
- "fields" : [ {
- "name" : "a",
- "fieldType" : "INT"
- }, {
- "name" : "b",
- "fieldType" : "BIGINT"
- }, {
- "name" : "c",
- "fieldType" : "VARCHAR(2147483647)"
- }, {
- "name" : "rowtime",
- "fieldType" : {
- "type" : "TIMESTAMP_WITHOUT_TIME_ZONE",
- "precision" : 3,
- "kind" : "ROWTIME"
- }
- }, {
- "name" : "proctime",
- "fieldType" : {
- "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
- "nullable" : false,
- "precision" : 3,
- "kind" : "PROCTIME"
- }
- }, {
- "name" : "window_start",
- "fieldType" : "TIMESTAMP(3) NOT NULL"
- }, {
- "name" : "window_end",
- "fieldType" : "TIMESTAMP(3) NOT NULL"
- }, {
- "name" : "window_time",
- "fieldType" : {
- "type" : "TIMESTAMP_WITHOUT_TIME_ZONE",
- "nullable" : false,
- "precision" : 3,
- "kind" : "ROWTIME"
- }
- } ]
- },
- "description" : "WindowTableFunction(window=[TUMBLE(time_col=[rowtime], size=[15 min])])"
- }, {
- "id" : 5,
- "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" : 5,
- "type" : "TIMESTAMP(3) NOT NULL"
- }, {
- "kind" : "INPUT_REF",
- "inputIndex" : 6,
- "type" : "TIMESTAMP(3) NOT NULL"
- } ],
- "condition" : null,
- "inputProperties" : [ {
- "requiredDistribution" : {
- "type" : "UNKNOWN"
- },
- "damBehavior" : "PIPELINED",
- "priority" : 0
- } ],
- "outputType" : "ROW<`a` INT, `b` BIGINT, `c` VARCHAR(2147483647), `window_start` TIMESTAMP(3) NOT NULL, `window_end` TIMESTAMP(3) NOT NULL>",
- "description" : "Calc(select=[a, b, c, window_start, window_end])"
- }, {
- "id" : 6,
- "type" : "stream-exec-exchange_1",
- "inputProperties" : [ {
- "requiredDistribution" : {
- "type" : "HASH",
- "keys" : [ 0 ]
- },
- "damBehavior" : "PIPELINED",
- "priority" : 0
- } ],
- "outputType" : "ROW<`a` INT, `b` BIGINT, `c` VARCHAR(2147483647), `window_start` TIMESTAMP(3) NOT NULL, `window_end` TIMESTAMP(3) NOT NULL>",
- "description" : "Exchange(distribution=[hash[a]])"
- }, {
- "id" : 7,
- "type" : "stream-exec-window-rank_1",
- "configuration" : {
- "table.local-time-zone" : "default"
- },
- "rankType" : "ROW_NUMBER",
- "partitionSpec" : {
- "fields" : [ 0 ]
- },
- "sortSpec" : {
- "fields" : [ {
- "index" : 1,
- "isAscending" : false,
- "nullIsLast" : true
- } ]
- },
- "rankRange" : {
- "type" : "Constant",
- "start" : 1,
- "end" : 3
- },
- "outputRowNumber" : false,
- "windowing" : {
- "strategy" : "WindowAttached",
- "window" : {
- "type" : "TumblingWindow",
- "size" : "PT15M"
- },
- "timeAttributeType" : {
- "type" : "TIMESTAMP_WITHOUT_TIME_ZONE",
- "precision" : 3,
- "kind" : "ROWTIME"
- },
- "windowStart" : 3,
- "windowEnd" : 4,
- "isRowtime" : true
- },
- "inputProperties" : [ {
- "requiredDistribution" : {
- "type" : "UNKNOWN"
- },
- "damBehavior" : "PIPELINED",
- "priority" : 0
- } ],
- "outputType" : "ROW<`a` INT, `b` BIGINT, `c` VARCHAR(2147483647), `window_start` TIMESTAMP(3) NOT NULL, `window_end` TIMESTAMP(3) NOT NULL>",
- "description" : "WindowRank(window=[TUMBLE(win_start=[window_start], win_end=[window_end], size=[15 min])], rankType=[ROW_NUMBER], rankRange=[rankStart=1, rankEnd=3], partitionBy=[a], orderBy=[b DESC], select=[a, b, c, window_start, window_end])"
- }, {
- "id" : 8,
- "type" : "stream-exec-calc_1",
- "projection" : [ {
- "kind" : "INPUT_REF",
- "inputIndex" : 3,
- "type" : "TIMESTAMP(3) NOT NULL"
- }, {
- "kind" : "INPUT_REF",
- "inputIndex" : 4,
- "type" : "TIMESTAMP(3) NOT NULL"
- }, {
- "kind" : "INPUT_REF",
- "inputIndex" : 0,
- "type" : "INT"
- }, {
- "kind" : "INPUT_REF",
- "inputIndex" : 1,
- "type" : "BIGINT"
- }, {
- "kind" : "INPUT_REF",
- "inputIndex" : 2,
- "type" : "VARCHAR(2147483647)"
- } ],
- "condition" : null,
- "inputProperties" : [ {
- "requiredDistribution" : {
- "type" : "UNKNOWN"
- },
- "damBehavior" : "PIPELINED",
- "priority" : 0
- } ],
- "outputType" : "ROW<`window_start` TIMESTAMP(3) NOT NULL, `window_end` TIMESTAMP(3) NOT NULL, `a` INT, `b` BIGINT, `c` VARCHAR(2147483647)>",
- "description" : "Calc(select=[window_start, window_end, a, b, c])"
- }, {
- "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" : "window_start",
- "dataType" : "TIMESTAMP(3)"
- }, {
- "name" : "window_end",
- "dataType" : "TIMESTAMP(3)"
- }, {
- "name" : "a",
- "dataType" : "INT"
- }, {
- "name" : "b",
- "dataType" : "BIGINT"
- }, {
- "name" : "c",
- "dataType" : "VARCHAR(2147483647)"
- } ],
- "watermarkSpecs" : [ ]
- },
- "partitionKeys" : [ ],
- "options" : {
- "connector" : "values"
- }
- }
- }
- },
- "inputChangelogMode" : [ "INSERT" ],
- "inputProperties" : [ {
- "requiredDistribution" : {
- "type" : "UNKNOWN"
- },
- "damBehavior" : "PIPELINED",
- "priority" : 0
- } ],
- "outputType" : "ROW<`window_start` TIMESTAMP(3) NOT NULL, `window_end` TIMESTAMP(3) NOT NULL, `a` INT, `b` BIGINT, `c` VARCHAR(2147483647)>",
- "description" : "Sink(table=[default_catalog.default_database.MySink], fields=[window_start, window_end, a, b, 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"
- } ]
-}
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowTableFunctionJsonPlanTest_jsonplan/testIndividualWindowTVF.out b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowTableFunctionJsonPlanTest_jsonplan/testIndividualWindowTVF.out
deleted file mode 100644
index b60dc577b95..00000000000
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowTableFunctionJsonPlanTest_jsonplan/testIndividualWindowTVF.out
+++ /dev/null
@@ -1,351 +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" : "INT"
- }, {
- "name" : "b",
- "dataType" : "BIGINT"
- }, {
- "name" : "c",
- "dataType" : "VARCHAR(2147483647)"
- }, {
- "name" : "rowtime",
- "kind" : "COMPUTED",
- "expression" : {
- "rexNode" : {
- "kind" : "CALL",
- "internalName" : "$TO_TIMESTAMP$1",
- "operands" : [ {
- "kind" : "INPUT_REF",
- "inputIndex" : 2,
- "type" : "VARCHAR(2147483647)"
- } ],
- "type" : "TIMESTAMP(3)"
- },
- "serializableString" : "TO_TIMESTAMP(`c`)"
- }
- }, {
- "name" : "proctime",
- "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" : [ {
- "rowtimeAttribute" : "rowtime",
- "expression" : {
- "rexNode" : {
- "kind" : "CALL",
- "syntax" : "SPECIAL",
- "internalName" : "$-$1",
- "operands" : [ {
- "kind" : "INPUT_REF",
- "inputIndex" : 3,
- "type" : "TIMESTAMP(3)"
- }, {
- "kind" : "LITERAL",
- "value" : "1000",
- "type" : "INTERVAL SECOND(6) NOT NULL"
- } ],
- "type" : "TIMESTAMP(3)"
- },
- "serializableString" : "`rowtime` - INTERVAL '1' SECOND"
- }
- } ]
- },
- "partitionKeys" : [ ],
- "options" : {
- "connector" : "values"
- }
- }
- }
- },
- "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" : "$TO_TIMESTAMP$1",
- "operands" : [ {
- "kind" : "INPUT_REF",
- "inputIndex" : 2,
- "type" : "VARCHAR(2147483647)"
- } ],
- "type" : "TIMESTAMP(3)"
- } ],
- "condition" : null,
- "inputProperties" : [ {
- "requiredDistribution" : {
- "type" : "UNKNOWN"
- },
- "damBehavior" : "PIPELINED",
- "priority" : 0
- } ],
- "outputType" : "ROW<`a` INT, `b` BIGINT, `c` VARCHAR(2147483647), `rowtime` TIMESTAMP(3)>",
- "description" : "Calc(select=[a, b, c, TO_TIMESTAMP(c) AS rowtime])"
- }, {
- "id" : 3,
- "type" : "stream-exec-watermark-assigner_1",
- "watermarkExpr" : {
- "kind" : "CALL",
- "syntax" : "SPECIAL",
- "internalName" : "$-$1",
- "operands" : [ {
- "kind" : "INPUT_REF",
- "inputIndex" : 3,
- "type" : "TIMESTAMP(3)"
- }, {
- "kind" : "LITERAL",
- "value" : "1000",
- "type" : "INTERVAL SECOND(6) NOT NULL"
- } ],
- "type" : "TIMESTAMP(3)"
- },
- "rowtimeFieldIndex" : 3,
- "inputProperties" : [ {
- "requiredDistribution" : {
- "type" : "UNKNOWN"
- },
- "damBehavior" : "PIPELINED",
- "priority" : 0
- } ],
- "outputType" : {
- "type" : "ROW",
- "fields" : [ {
- "name" : "a",
- "fieldType" : "INT"
- }, {
- "name" : "b",
- "fieldType" : "BIGINT"
- }, {
- "name" : "c",
- "fieldType" : "VARCHAR(2147483647)"
- }, {
- "name" : "rowtime",
- "fieldType" : {
- "type" : "TIMESTAMP_WITHOUT_TIME_ZONE",
- "precision" : 3,
- "kind" : "ROWTIME"
- }
- } ]
- },
- "description" : "WatermarkAssigner(rowtime=[rowtime], watermark=[(rowtime - 1000:INTERVAL SECOND)])"
- }, {
- "id" : 4,
- "type" : "stream-exec-window-table-function_1",
- "configuration" : {
- "table.local-time-zone" : "default"
- },
- "windowing" : {
- "strategy" : "TimeAttribute",
- "window" : {
- "type" : "TumblingWindow",
- "size" : "PT15M"
- },
- "timeAttributeType" : {
- "type" : "TIMESTAMP_WITHOUT_TIME_ZONE",
- "precision" : 3,
- "kind" : "ROWTIME"
- },
- "timeAttributeIndex" : 3,
- "isRowtime" : true
- },
- "inputProperties" : [ {
- "requiredDistribution" : {
- "type" : "UNKNOWN"
- },
- "damBehavior" : "PIPELINED",
- "priority" : 0
- } ],
- "outputType" : {
- "type" : "ROW",
- "fields" : [ {
- "name" : "a",
- "fieldType" : "INT"
- }, {
- "name" : "b",
- "fieldType" : "BIGINT"
- }, {
- "name" : "c",
- "fieldType" : "VARCHAR(2147483647)"
- }, {
- "name" : "rowtime",
- "fieldType" : {
- "type" : "TIMESTAMP_WITHOUT_TIME_ZONE",
- "precision" : 3,
- "kind" : "ROWTIME"
- }
- }, {
- "name" : "window_start",
- "fieldType" : "TIMESTAMP(3) NOT NULL"
- }, {
- "name" : "window_end",
- "fieldType" : "TIMESTAMP(3) NOT NULL"
- }, {
- "name" : "window_time",
- "fieldType" : {
- "type" : "TIMESTAMP_WITHOUT_TIME_ZONE",
- "nullable" : false,
- "precision" : 3,
- "kind" : "ROWTIME"
- }
- } ]
- },
- "description" : "WindowTableFunction(window=[TUMBLE(time_col=[rowtime], size=[15 min])])"
- }, {
- "id" : 5,
- "type" : "stream-exec-calc_1",
- "projection" : [ {
- "kind" : "INPUT_REF",
- "inputIndex" : 4,
- "type" : "TIMESTAMP(3) NOT NULL"
- }, {
- "kind" : "INPUT_REF",
- "inputIndex" : 5,
- "type" : "TIMESTAMP(3) NOT NULL"
- }, {
- "kind" : "INPUT_REF",
- "inputIndex" : 0,
- "type" : "INT"
- }, {
- "kind" : "INPUT_REF",
- "inputIndex" : 1,
- "type" : "BIGINT"
- }, {
- "kind" : "INPUT_REF",
- "inputIndex" : 2,
- "type" : "VARCHAR(2147483647)"
- } ],
- "condition" : null,
- "inputProperties" : [ {
- "requiredDistribution" : {
- "type" : "UNKNOWN"
- },
- "damBehavior" : "PIPELINED",
- "priority" : 0
- } ],
- "outputType" : "ROW<`window_start` TIMESTAMP(3) NOT NULL, `window_end` TIMESTAMP(3) NOT NULL, `a` INT, `b` BIGINT, `c` VARCHAR(2147483647)>",
- "description" : "Calc(select=[window_start, window_end, a, b, 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`.`MySink`",
- "resolvedTable" : {
- "schema" : {
- "columns" : [ {
- "name" : "window_start",
- "dataType" : "TIMESTAMP(3)"
- }, {
- "name" : "window_end",
- "dataType" : "TIMESTAMP(3)"
- }, {
- "name" : "a",
- "dataType" : "INT"
- }, {
- "name" : "b",
- "dataType" : "BIGINT"
- }, {
- "name" : "c",
- "dataType" : "VARCHAR(2147483647)"
- } ],
- "watermarkSpecs" : [ ]
- },
- "partitionKeys" : [ ],
- "options" : {
- "connector" : "values"
- }
- }
- }
- },
- "inputChangelogMode" : [ "INSERT" ],
- "inputProperties" : [ {
- "requiredDistribution" : {
- "type" : "UNKNOWN"
- },
- "damBehavior" : "PIPELINED",
- "priority" : 0
- } ],
- "outputType" : "ROW<`window_start` TIMESTAMP(3) NOT NULL, `window_end` TIMESTAMP(3) NOT NULL, `a` INT, `b` BIGINT, `c` VARCHAR(2147483647)>",
- "description" : "Sink(table=[default_catalog.default_database.MySink], fields=[window_start, window_end, a, b, 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"
- } ]
-}
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowTableFunctionJsonPlanTest_jsonplan/testIndividualWindowTVFProcessingTime.out b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowTableFunctionJsonPlanTest_jsonplan/testIndividualWindowTVFProcessingTime.out
deleted file mode 100644
index 2cb86ead482..00000000000
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowTableFunctionJsonPlanTest_jsonplan/testIndividualWindowTVFProcessingTime.out
+++ /dev/null
@@ -1,456 +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" : "INT"
- }, {
- "name" : "b",
- "dataType" : "BIGINT"
- }, {
- "name" : "c",
- "dataType" : "VARCHAR(2147483647)"
- }, {
- "name" : "rowtime",
- "kind" : "COMPUTED",
- "expression" : {
- "rexNode" : {
- "kind" : "CALL",
- "internalName" : "$TO_TIMESTAMP$1",
- "operands" : [ {
- "kind" : "INPUT_REF",
- "inputIndex" : 2,
- "type" : "VARCHAR(2147483647)"
- } ],
- "type" : "TIMESTAMP(3)"
- },
- "serializableString" : "TO_TIMESTAMP(`c`)"
- }
- }, {
- "name" : "proctime",
- "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" : [ {
- "rowtimeAttribute" : "rowtime",
- "expression" : {
- "rexNode" : {
- "kind" : "CALL",
- "syntax" : "SPECIAL",
- "internalName" : "$-$1",
- "operands" : [ {
- "kind" : "INPUT_REF",
- "inputIndex" : 3,
- "type" : "TIMESTAMP(3)"
- }, {
- "kind" : "LITERAL",
- "value" : "1000",
- "type" : "INTERVAL SECOND(6) NOT NULL"
- } ],
- "type" : "TIMESTAMP(3)"
- },
- "serializableString" : "`rowtime` - INTERVAL '1' SECOND"
- }
- } ]
- },
- "partitionKeys" : [ ],
- "options" : {
- "connector" : "values"
- }
- }
- }
- },
- "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" : "$TO_TIMESTAMP$1",
- "operands" : [ {
- "kind" : "INPUT_REF",
- "inputIndex" : 2,
- "type" : "VARCHAR(2147483647)"
- } ],
- "type" : "TIMESTAMP(3)"
- }, {
- "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" : "INT"
- }, {
- "name" : "b",
- "fieldType" : "BIGINT"
- }, {
- "name" : "c",
- "fieldType" : "VARCHAR(2147483647)"
- }, {
- "name" : "rowtime",
- "fieldType" : "TIMESTAMP(3)"
- }, {
- "name" : "proctime",
- "fieldType" : {
- "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
- "nullable" : false,
- "precision" : 3,
- "kind" : "PROCTIME"
- }
- } ]
- },
- "description" : "Calc(select=[a, b, c, TO_TIMESTAMP(c) AS rowtime, PROCTIME() AS proctime])"
- }, {
- "id" : 3,
- "type" : "stream-exec-watermark-assigner_1",
- "watermarkExpr" : {
- "kind" : "CALL",
- "syntax" : "SPECIAL",
- "internalName" : "$-$1",
- "operands" : [ {
- "kind" : "INPUT_REF",
- "inputIndex" : 3,
- "type" : "TIMESTAMP(3)"
- }, {
- "kind" : "LITERAL",
- "value" : "1000",
- "type" : "INTERVAL SECOND(6) NOT NULL"
- } ],
- "type" : "TIMESTAMP(3)"
- },
- "rowtimeFieldIndex" : 3,
- "inputProperties" : [ {
- "requiredDistribution" : {
- "type" : "UNKNOWN"
- },
- "damBehavior" : "PIPELINED",
- "priority" : 0
- } ],
- "outputType" : {
- "type" : "ROW",
- "fields" : [ {
- "name" : "a",
- "fieldType" : "INT"
- }, {
- "name" : "b",
- "fieldType" : "BIGINT"
- }, {
- "name" : "c",
- "fieldType" : "VARCHAR(2147483647)"
- }, {
- "name" : "rowtime",
- "fieldType" : {
- "type" : "TIMESTAMP_WITHOUT_TIME_ZONE",
- "precision" : 3,
- "kind" : "ROWTIME"
- }
- }, {
- "name" : "proctime",
- "fieldType" : {
- "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
- "nullable" : false,
- "precision" : 3,
- "kind" : "PROCTIME"
- }
- } ]
- },
- "description" : "WatermarkAssigner(rowtime=[rowtime], watermark=[(rowtime - 1000:INTERVAL SECOND)])"
- }, {
- "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" : 4,
- "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" : "INT"
- }, {
- "name" : "b",
- "fieldType" : "BIGINT"
- }, {
- "name" : "c",
- "fieldType" : "VARCHAR(2147483647)"
- }, {
- "name" : "proctime",
- "fieldType" : {
- "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
- "nullable" : false,
- "precision" : 3,
- "kind" : "PROCTIME"
- }
- } ]
- },
- "description" : "Calc(select=[a, b, c, proctime])"
- }, {
- "id" : 5,
- "type" : "stream-exec-window-table-function_1",
- "configuration" : {
- "table.local-time-zone" : "default"
- },
- "windowing" : {
- "strategy" : "TimeAttribute",
- "window" : {
- "type" : "TumblingWindow",
- "size" : "PT15M"
- },
- "timeAttributeType" : {
- "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
- "nullable" : false,
- "precision" : 3,
- "kind" : "PROCTIME"
- },
- "timeAttributeIndex" : 3,
- "isRowtime" : false
- },
- "inputProperties" : [ {
- "requiredDistribution" : {
- "type" : "UNKNOWN"
- },
- "damBehavior" : "PIPELINED",
- "priority" : 0
- } ],
- "outputType" : {
- "type" : "ROW",
- "fields" : [ {
- "name" : "a",
- "fieldType" : "INT"
- }, {
- "name" : "b",
- "fieldType" : "BIGINT"
- }, {
- "name" : "c",
- "fieldType" : "VARCHAR(2147483647)"
- }, {
- "name" : "proctime",
- "fieldType" : {
- "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
- "nullable" : false,
- "precision" : 3,
- "kind" : "PROCTIME"
- }
- }, {
- "name" : "window_start",
- "fieldType" : "TIMESTAMP(3) NOT NULL"
- }, {
- "name" : "window_end",
- "fieldType" : "TIMESTAMP(3) NOT NULL"
- }, {
- "name" : "window_time",
- "fieldType" : {
- "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
- "nullable" : false,
- "precision" : 3,
- "kind" : "PROCTIME"
- }
- } ]
- },
- "description" : "WindowTableFunction(window=[TUMBLE(time_col=[proctime], size=[15 min])])"
- }, {
- "id" : 6,
- "type" : "stream-exec-calc_1",
- "projection" : [ {
- "kind" : "INPUT_REF",
- "inputIndex" : 4,
- "type" : "TIMESTAMP(3) NOT NULL"
- }, {
- "kind" : "INPUT_REF",
- "inputIndex" : 5,
- "type" : "TIMESTAMP(3) NOT NULL"
- }, {
- "kind" : "INPUT_REF",
- "inputIndex" : 0,
- "type" : "INT"
- }, {
- "kind" : "INPUT_REF",
- "inputIndex" : 1,
- "type" : "BIGINT"
- }, {
- "kind" : "INPUT_REF",
- "inputIndex" : 2,
- "type" : "VARCHAR(2147483647)"
- } ],
- "condition" : null,
- "inputProperties" : [ {
- "requiredDistribution" : {
- "type" : "UNKNOWN"
- },
- "damBehavior" : "PIPELINED",
- "priority" : 0
- } ],
- "outputType" : "ROW<`window_start` TIMESTAMP(3) NOT NULL, `window_end` TIMESTAMP(3) NOT NULL, `a` INT, `b` BIGINT, `c` VARCHAR(2147483647)>",
- "description" : "Calc(select=[window_start, window_end, a, b, c])"
- }, {
- "id" : 7,
- "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" : "window_start",
- "dataType" : "TIMESTAMP(3)"
- }, {
- "name" : "window_end",
- "dataType" : "TIMESTAMP(3)"
- }, {
- "name" : "a",
- "dataType" : "INT"
- }, {
- "name" : "b",
- "dataType" : "BIGINT"
- }, {
- "name" : "c",
- "dataType" : "VARCHAR(2147483647)"
- } ],
- "watermarkSpecs" : [ ]
- },
- "partitionKeys" : [ ],
- "options" : {
- "connector" : "values"
- }
- }
- }
- },
- "inputChangelogMode" : [ "INSERT" ],
- "inputProperties" : [ {
- "requiredDistribution" : {
- "type" : "UNKNOWN"
- },
- "damBehavior" : "PIPELINED",
- "priority" : 0
- } ],
- "outputType" : "ROW<`window_start` TIMESTAMP(3) NOT NULL, `window_end` TIMESTAMP(3) NOT NULL, `a` INT, `b` BIGINT, `c` VARCHAR(2147483647)>",
- "description" : "Sink(table=[default_catalog.default_database.MySink], fields=[window_start, window_end, a, b, 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"
- } ]
-}