You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by sn...@apache.org on 2026/08/12 13:37:53 UTC
(flink) branch master updated (de793af8458 -> 1aaf0f26486)
This is an automated email from the ASF dual-hosted git repository.
snuyanzin pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
from de793af8458 [FLINK-40373][table] `JSON_VALUE` might fail on code gen for null input
new 946c2763eb5 [FLINK-40377][table] `MatchRecognize` fails with Unsupported `RelDataType`: `UNKNOWN` while json plan generation
new 1aaf0f26486 [hotfix][ci] Add parallel execution to Maven compile options
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:
.../calcite/sql/type/SqlTypeFactoryImpl.java | 4 +-
...anTest.java => MatchRecognizeJsonPlanTest.java} | 41 +++++++++++-------
.../testMatchRecognize.out} | 49 +++++++++++-----------
tools/ci/test_controller.sh | 2 +-
4 files changed, 54 insertions(+), 42 deletions(-)
copy flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/{PythonGroupAggregateJsonPlanTest.java => MatchRecognizeJsonPlanTest.java} (63%)
copy flink-table/flink-table-planner/src/test/resources/{restore-tests/stream-exec-match_1/match-simple/plan/match-simple.json => org/apache/flink/table/planner/plan/nodes/exec/stream/MatchRecognizeJsonPlanTest_jsonplan/testMatchRecognize.out} (91%)
(flink) 02/02: [hotfix][ci] Add parallel execution to Maven compile options
Posted by sn...@apache.org.
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 1aaf0f26486f77f0cd10e1b8e8c4d76a9ae5a113
Author: Sergey Nuyanzin <sn...@gmail.com>
AuthorDate: Tue Aug 11 09:19:33 2026 +0200
[hotfix][ci] Add parallel execution to Maven compile options
---
tools/ci/test_controller.sh | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/tools/ci/test_controller.sh b/tools/ci/test_controller.sh
index 0a620fb6f5c..5517187f342 100755
--- a/tools/ci/test_controller.sh
+++ b/tools/ci/test_controller.sh
@@ -81,7 +81,7 @@ export LOG4J_PROPERTIES=${HERE}/log4j.properties
MVN_LOGGING_OPTIONS="-Dlog.dir=${DEBUG_FILES_OUTPUT_DIR} -Dlog4j.configurationFile=file://$LOG4J_PROPERTIES"
MVN_COMMON_OPTIONS="-Dfast -Pskip-webui-build $MVN_LOGGING_OPTIONS"
-MVN_COMPILE_OPTIONS="-DskipTests"
+MVN_COMPILE_OPTIONS="-DskipTests -T1C"
MVN_COMPILE_MODULES=$(get_compile_modules_for_stage ${STAGE})
CALLBACK_ON_TIMEOUT="print_stacktraces | tee ${DEBUG_FILES_OUTPUT_DIR}/jps-traces.out"
(flink) 01/02: [FLINK-40377][table] `MatchRecognize` fails with Unsupported `RelDataType`: `UNKNOWN` while json plan generation
Posted by sn...@apache.org.
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 <sn...@gmail.com>
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