This is an automated email from the ASF dual-hosted git repository. snuyanzin pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit 946c2763eb512bbf3e843ac96cf0c10e671d1d2a Author: Sergey Nuyanzin <[email protected]> AuthorDate: Wed Aug 12 13:18:46 2026 +0200 [FLINK-40377][table] `MatchRecognize` fails with Unsupported `RelDataType`: `UNKNOWN` while json plan generation --- .../calcite/sql/type/SqlTypeFactoryImpl.java | 4 +- .../exec/stream/MatchRecognizeJsonPlanTest.java | 79 +++++ .../testMatchRecognize.out | 370 +++++++++++++++++++++ 3 files changed, 452 insertions(+), 1 deletion(-) diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/type/SqlTypeFactoryImpl.java b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/type/SqlTypeFactoryImpl.java index 010b3afc893..3ef88fb0ab9 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/type/SqlTypeFactoryImpl.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/type/SqlTypeFactoryImpl.java @@ -112,7 +112,9 @@ public class SqlTypeFactoryImpl extends RelDataTypeFactoryImpl { @Override public RelDataType createUnknownType() { - return createSqlType(SqlTypeName.UNKNOWN); + // FLINK MODIFICATION BEGIN + return canonize(new UnknownSqlType(this)); + // FLINK MODIFICATION END } @Override diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/MatchRecognizeJsonPlanTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/MatchRecognizeJsonPlanTest.java new file mode 100644 index 00000000000..e8a0dcff1cd --- /dev/null +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/MatchRecognizeJsonPlanTest.java @@ -0,0 +1,79 @@ +/* + * 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 plan generation for {@link StreamExecMatch}. */ +class MatchRecognizeJsonPlanTest extends TableTestBase { + + private StreamTableTestUtil util; + private TableEnvironment tEnv; + + @BeforeEach + void setup() { + util = streamTestUtil(TableConfig.getDefault()); + tEnv = util.getTableEnv(); + + String srcTableDdl = + "CREATE TABLE MyTable (\n" + + " id bigint,\n" + + " name varchar,\n" + + " proctime as PROCTIME()\n" + + ") with (\n" + + " 'connector' = 'values',\n" + + " 'bounded' = 'false')"; + tEnv.executeSql(srcTableDdl); + + String sinkTableDdl = + "CREATE TABLE MySink (\n" + + " a bigint,\n" + + " b bigint,\n" + + " c bigint\n" + + ") with (\n" + + " 'connector' = 'values',\n" + + " 'table-sink-class' = 'DEFAULT')"; + tEnv.executeSql(sinkTableDdl); + } + + @Test + void testMatchRecognize() { + util.verifyJsonPlan( + "insert into MySink" + + " SELECT T.aid, T.bid, T.cid\n" + + " FROM MyTable MATCH_RECOGNIZE (\n" + + " ORDER BY proctime\n" + + " MEASURES\n" + + " A.id AS aid,\n" + + " B.id AS bid,\n" + + " C.id AS cid\n" + + " PATTERN (A B C)\n" + + " DEFINE\n" + + " A AS name = 'a',\n" + + " B AS name = 'b',\n" + + " C AS name = 'c'\n" + + " ) AS T"); + } +} diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/MatchRecognizeJsonPlanTest_jsonplan/testMatchRecognize.out b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/MatchRecognizeJsonPlanTest_jsonplan/testMatchRecognize.out new file mode 100644 index 00000000000..34669f8c501 --- /dev/null +++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/MatchRecognizeJsonPlanTest_jsonplan/testMatchRecognize.out @@ -0,0 +1,370 @@ +{ + "flinkVersion" : "", + "nodes" : [ { + "id" : 1, + "type" : "stream-exec-table-source-scan_2", + "scanTableSource" : { + "table" : { + "identifier" : "`default_catalog`.`default_database`.`MyTable`", + "resolvedTable" : { + "schema" : { + "columns" : [ { + "name" : "id", + "dataType" : "BIGINT" + }, { + "name" : "name", + "dataType" : "VARCHAR(2147483647)" + }, { + "name" : "proctime", + "kind" : "COMPUTED", + "expression" : { + "rexNode" : { + "kind" : "CALL", + "internalName" : "$PROCTIME$1", + "type" : { + "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE", + "nullable" : false, + "precision" : 3, + "kind" : "PROCTIME" + } + }, + "serializableString" : "PROCTIME()" + } + } ] + }, + "options" : { + "bounded" : "false", + "connector" : "values" + } + } + } + }, + "outputType" : "ROW<`id` BIGINT, `name` VARCHAR(2147483647)>", + "description" : "TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[id, name])" + }, { + "id" : 2, + "type" : "stream-exec-calc_1", + "projection" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 0, + "type" : "BIGINT" + }, { + "kind" : "INPUT_REF", + "inputIndex" : 1, + "type" : "VARCHAR(2147483647)" + }, { + "kind" : "CALL", + "internalName" : "$PROCTIME$1", + "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" : "id", + "fieldType" : "BIGINT" + }, { + "name" : "name", + "fieldType" : "VARCHAR(2147483647)" + }, { + "name" : "proctime", + "fieldType" : { + "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE", + "nullable" : false, + "precision" : 3, + "kind" : "PROCTIME" + } + } ] + }, + "description" : "Calc(select=[id, name, PROCTIME() AS proctime])" + }, { + "id" : 3, + "type" : "stream-exec-exchange_1", + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "SINGLETON" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : { + "type" : "ROW", + "fields" : [ { + "name" : "id", + "fieldType" : "BIGINT" + }, { + "name" : "name", + "fieldType" : "VARCHAR(2147483647)" + }, { + "name" : "proctime", + "fieldType" : { + "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE", + "nullable" : false, + "precision" : 3, + "kind" : "PROCTIME" + } + } ] + }, + "description" : "Exchange(distribution=[single])" + }, { + "id" : 4, + "type" : "stream-exec-match_1", + "matchSpec" : { + "pattern" : { + "kind" : "CALL", + "syntax" : "BINARY", + "sqlKind" : "PATTERN_CONCAT", + "operands" : [ { + "kind" : "CALL", + "syntax" : "BINARY", + "sqlKind" : "PATTERN_CONCAT", + "operands" : [ { + "kind" : "LITERAL", + "value" : "A", + "type" : "CHAR(1) NOT NULL" + }, { + "kind" : "LITERAL", + "value" : "B", + "type" : "CHAR(1) NOT NULL" + } ], + "type" : "NULL" + }, { + "kind" : "LITERAL", + "value" : "C", + "type" : "CHAR(1) NOT NULL" + } ], + "type" : "NULL" + }, + "patternDefinitions" : { + "A" : { + "kind" : "CALL", + "syntax" : "BINARY", + "internalName" : "$=$1", + "operands" : [ { + "kind" : "CALL", + "internalName" : "$LAST$1", + "operands" : [ { + "kind" : "PATTERN_INPUT_REF", + "alpha" : "*", + "inputIndex" : 1, + "type" : "VARCHAR(2147483647)" + }, { + "kind" : "LITERAL", + "value" : 0, + "type" : "INT NOT NULL" + } ], + "type" : "VARCHAR(2147483647)" + }, { + "kind" : "LITERAL", + "value" : "a", + "type" : "VARCHAR(2147483647) NOT NULL" + } ], + "type" : "BOOLEAN" + }, + "B" : { + "kind" : "CALL", + "syntax" : "BINARY", + "internalName" : "$=$1", + "operands" : [ { + "kind" : "CALL", + "internalName" : "$LAST$1", + "operands" : [ { + "kind" : "PATTERN_INPUT_REF", + "alpha" : "*", + "inputIndex" : 1, + "type" : "VARCHAR(2147483647)" + }, { + "kind" : "LITERAL", + "value" : 0, + "type" : "INT NOT NULL" + } ], + "type" : "VARCHAR(2147483647)" + }, { + "kind" : "LITERAL", + "value" : "b", + "type" : "VARCHAR(2147483647) NOT NULL" + } ], + "type" : "BOOLEAN" + }, + "C" : { + "kind" : "CALL", + "syntax" : "BINARY", + "internalName" : "$=$1", + "operands" : [ { + "kind" : "CALL", + "internalName" : "$LAST$1", + "operands" : [ { + "kind" : "PATTERN_INPUT_REF", + "alpha" : "*", + "inputIndex" : 1, + "type" : "VARCHAR(2147483647)" + }, { + "kind" : "LITERAL", + "value" : 0, + "type" : "INT NOT NULL" + } ], + "type" : "VARCHAR(2147483647)" + }, { + "kind" : "LITERAL", + "value" : "c", + "type" : "VARCHAR(2147483647) NOT NULL" + } ], + "type" : "BOOLEAN" + } + }, + "measures" : { + "aid" : { + "kind" : "CALL", + "syntax" : "PREFIX", + "internalName" : "$FINAL$1", + "operands" : [ { + "kind" : "PATTERN_INPUT_REF", + "alpha" : "A", + "inputIndex" : 0, + "type" : "BIGINT" + } ], + "type" : "BIGINT" + }, + "bid" : { + "kind" : "CALL", + "syntax" : "PREFIX", + "internalName" : "$FINAL$1", + "operands" : [ { + "kind" : "PATTERN_INPUT_REF", + "alpha" : "B", + "inputIndex" : 0, + "type" : "BIGINT" + } ], + "type" : "BIGINT" + }, + "cid" : { + "kind" : "CALL", + "syntax" : "PREFIX", + "internalName" : "$FINAL$1", + "operands" : [ { + "kind" : "PATTERN_INPUT_REF", + "alpha" : "C", + "inputIndex" : 0, + "type" : "BIGINT" + } ], + "type" : "BIGINT" + } + }, + "after" : { + "kind" : "LITERAL", + "symbol" : "MATCH_RECOGNIZE_AFTER_OPTION", + "value" : "SKIP_TO_NEXT_ROW", + "type" : { + "type" : "SYMBOL", + "nullable" : false + } + }, + "subsets" : { }, + "allRows" : false, + "partition" : { + "fields" : [ ] + }, + "orderBy" : { + "fields" : [ { + "index" : 2, + "isAscending" : true, + "nullIsLast" : false + } ] + }, + "interval" : null + }, + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : "ROW<`aid` BIGINT, `bid` BIGINT, `cid` BIGINT>", + "description" : "Match(orderBy=[proctime ASC], measures=[FINAL(A.id) AS aid, FINAL(B.id) AS bid, FINAL(C.id) AS cid], rowsPerMatch=[ONE ROW PER MATCH], after=[SKIP TO NEXT ROW], pattern=[((_UTF-16LE'A', _UTF-16LE'B'), _UTF-16LE'C')], define=[{A==(LAST(*.$1, 0), _UTF-16LE'a'), B==(LAST(*.$1, 0), _UTF-16LE'b'), C==(LAST(*.$1, 0), _UTF-16LE'c')}])" + }, { + "id" : 5, + "type" : "stream-exec-sink_2", + "configuration" : { + "table.exec.sink.keyed-shuffle" : "AUTO", + "table.exec.sink.not-null-enforcer" : "ERROR", + "table.exec.sink.rowtime-inserter" : "ENABLED", + "table.exec.sink.type-length-enforcer" : "IGNORE", + "table.exec.sink.upsert-materialize" : "AUTO" + }, + "dynamicTableSink" : { + "table" : { + "identifier" : "`default_catalog`.`default_database`.`MySink`", + "resolvedTable" : { + "schema" : { + "columns" : [ { + "name" : "a", + "dataType" : "BIGINT" + }, { + "name" : "b", + "dataType" : "BIGINT" + }, { + "name" : "c", + "dataType" : "BIGINT" + } ] + }, + "options" : { + "connector" : "values", + "table-sink-class" : "DEFAULT" + } + } + } + }, + "inputChangelogMode" : [ "INSERT" ], + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : "ROW<`aid` BIGINT, `bid` BIGINT, `cid` BIGINT>", + "description" : "Sink(table=[default_catalog.default_database.MySink], fields=[aid, bid, cid])" + } ], + "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" + } ] +} \ No newline at end of file
