This is an automated email from the ASF dual-hosted git repository.
dianfu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new b3f60e1eb14 [FLINK-40753][table-planner] Fix decorrelation of
subqueries over VALUES
b3f60e1eb14 is described below
commit b3f60e1eb14ab38ae3fe391a0d6528561cd2a11e
Author: SulphurFH <[email protected]>
AuthorDate: Thu Sep 24 09:55:18 2026 +0800
[FLINK-40753][table-planner] Fix decorrelation of subqueries over VALUES
This closes #29275.
---
.../plan/rules/logical/SubQueryDecorrelator.java | 9 +-
.../plan/rules/logical/SubQueryValuesTest.java | 71 ++++++++
.../plan/rules/logical/SubQueryValuesTest.xml | 189 +++++++++++++++++++++
3 files changed, 266 insertions(+), 3 deletions(-)
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/SubQueryDecorrelator.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/SubQueryDecorrelator.java
index 9ac57e63973..035d76e667f 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/SubQueryDecorrelator.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/SubQueryDecorrelator.java
@@ -485,7 +485,10 @@ public class SubQueryDecorrelator extends RelShuttleImpl {
}
}
RelNode newProject = RelOptUtil.createProject(newInput, projects,
false);
- newProject = ((LogicalProject)
newProject).withHints(rel.getHints());
+ // The projection may be simplified to a Values node.
+ if (newProject instanceof Hintable) {
+ newProject = ((Hintable) newProject).withHints(rel.getHints());
+ }
final RexNode newCorCondition;
if (frame.c != null) {
@@ -897,8 +900,8 @@ public class SubQueryDecorrelator extends RelShuttleImpl {
* @param rel Values to be rewritten
*/
public Frame decorrelateRel(Values rel) {
- // There are no inputs, so rel does not need to be changed.
- return null;
+ // Preserve the Values node while allowing its parents to be
decorrelated.
+ return new Frame(rel, rel, null,
identityMap(rel.getRowType().getFieldCount()));
}
public Frame decorrelateRel(LogicalCorrelate rel) {
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/SubQueryValuesTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/SubQueryValuesTest.java
new file mode 100644
index 00000000000..9733c02e40c
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/SubQueryValuesTest.java
@@ -0,0 +1,71 @@
+/*
+ * 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.rules.logical;
+
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.planner.plan.optimize.program.FlinkBatchProgram;
+import org.apache.flink.table.planner.utils.BatchTableTestUtil;
+import org.apache.flink.table.planner.utils.TableTestBase;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+/** Plan tests for correlated subqueries over SQL VALUES. */
+class SubQueryValuesTest extends TableTestBase {
+
+ private BatchTableTestUtil util;
+
+ @BeforeEach
+ void setup() {
+ util = batchTestUtil(TableConfig.getDefault());
+ util.buildBatchProgram(FlinkBatchProgram.DEFAULT_REWRITE());
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @ValueSource(strings = {"EXISTS", "NOT EXISTS"})
+ void testValues(String predicate) {
+ util.verifyRelPlan(query(predicate));
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @ValueSource(strings = {"EXISTS", "NOT EXISTS"})
+ void testProjectAndFilter(String predicate) {
+ util.verifyRelPlan(queryWithProjectAndFilter(predicate));
+ }
+
+ private static String query(String predicate) {
+ return "SELECT l.id FROM (VALUES (1), (2)) AS l(id) WHERE "
+ + predicate
+ + " (SELECT 1 FROM (VALUES (1), (2)) AS r(right_id) WHERE l.id
= r.right_id)";
+ }
+
+ private static String queryWithProjectAndFilter(String predicate) {
+ return "SELECT l.id, l.amount "
+ + "FROM (VALUES (1, 10), (1, 30), (2, 20), (3, 40), (CAST(NULL
AS INT), 50)) "
+ + "AS l(id, amount) "
+ + "WHERE "
+ + predicate
+ + " (SELECT 1 FROM ("
+ + "SELECT raw_id + 1 AS id, limit_value "
+ + "FROM (VALUES (25, 0), (10, 1), (CAST(NULL AS INT), 2)) AS
v(limit_value, raw_id) "
+ + "WHERE limit_value IS NOT NULL) AS r "
+ + "WHERE l.id = r.id AND l.amount < r.limit_value)";
+ }
+}
diff --git
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/SubQueryValuesTest.xml
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/SubQueryValuesTest.xml
new file mode 100644
index 00000000000..971d553f6b1
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/SubQueryValuesTest.xml
@@ -0,0 +1,189 @@
+<?xml version="1.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.
+-->
+<Root>
+ <TestCase name="testProjectAndFilter[EXISTS]">
+ <Resource name="sql">
+ <![CDATA[SELECT l.id, l.amount FROM (VALUES (1, 10), (1, 30), (2, 20),
(3, 40), (CAST(NULL AS INT), 50)) AS l(id, amount) WHERE EXISTS (SELECT 1 FROM
(SELECT raw_id + 1 AS id, limit_value FROM (VALUES (25, 0), (10, 1), (CAST(NULL
AS INT), 2)) AS v(limit_value, raw_id) WHERE limit_value IS NOT NULL) AS r
WHERE l.id = r.id AND l.amount < r.limit_value)]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(id=[$0], amount=[$1])
++- LogicalFilter(condition=[EXISTS({
+LogicalFilter(condition=[AND(=($cor0.id, $0), <($cor0.amount, $1))])
+ LogicalProject(id=[+($1, 1)], limit_value=[$0])
+ LogicalFilter(condition=[IS NOT NULL($0)])
+ LogicalProject(limit_value=[$0], raw_id=[$1])
+ LogicalUnion(all=[true])
+ LogicalProject(EXPR$0=[25], EXPR$1=[0])
+ LogicalValues(tuples=[[{ 0 }]])
+ LogicalProject(EXPR$0=[10], EXPR$1=[1])
+ LogicalValues(tuples=[[{ 0 }]])
+ LogicalProject(EXPR$0=[null:INTEGER], EXPR$1=[2])
+ LogicalValues(tuples=[[{ 0 }]])
+})], variablesSet=[[$cor0]])
+ +- LogicalProject(id=[$0], amount=[$1])
+ +- LogicalUnion(all=[true])
+ :- LogicalProject(EXPR$0=[1], EXPR$1=[10])
+ : +- LogicalValues(tuples=[[{ 0 }]])
+ :- LogicalProject(EXPR$0=[1], EXPR$1=[30])
+ : +- LogicalValues(tuples=[[{ 0 }]])
+ :- LogicalProject(EXPR$0=[2], EXPR$1=[20])
+ : +- LogicalValues(tuples=[[{ 0 }]])
+ :- LogicalProject(EXPR$0=[3], EXPR$1=[40])
+ : +- LogicalValues(tuples=[[{ 0 }]])
+ +- LogicalProject(EXPR$0=[null:INTEGER], EXPR$1=[50])
+ +- LogicalValues(tuples=[[{ 0 }]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+LogicalProject(id=[$0], amount=[$1])
++- LogicalJoin(condition=[AND(=($0, $2), <($1, $3))], joinType=[semi])
+ :- LogicalProject(id=[$0], amount=[$1])
+ : +- LogicalUnion(all=[true])
+ : :- LogicalProject(EXPR$0=[1], EXPR$1=[10])
+ : : +- LogicalValues(tuples=[[{ 0 }]])
+ : :- LogicalProject(EXPR$0=[1], EXPR$1=[30])
+ : : +- LogicalValues(tuples=[[{ 0 }]])
+ : :- LogicalProject(EXPR$0=[2], EXPR$1=[20])
+ : : +- LogicalValues(tuples=[[{ 0 }]])
+ : :- LogicalProject(EXPR$0=[3], EXPR$1=[40])
+ : : +- LogicalValues(tuples=[[{ 0 }]])
+ : +- LogicalProject(EXPR$0=[null:INTEGER], EXPR$1=[50])
+ : +- LogicalValues(tuples=[[{ 0 }]])
+ +- LogicalFilter(condition=[true])
+ +- LogicalProject(id=[+($1, 1)], limit_value=[$0])
+ +- LogicalFilter(condition=[IS NOT NULL($0)])
+ +- LogicalProject(limit_value=[$0], raw_id=[$1])
+ +- LogicalUnion(all=[true])
+ :- LogicalValues(tuples=[[{ 25, 0 }]])
+ :- LogicalValues(tuples=[[{ 10, 1 }]])
+ +- LogicalValues(tuples=[[{ null, 2 }]])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testProjectAndFilter[NOT EXISTS]">
+ <Resource name="sql">
+ <![CDATA[SELECT l.id, l.amount FROM (VALUES (1, 10), (1, 30), (2, 20),
(3, 40), (CAST(NULL AS INT), 50)) AS l(id, amount) WHERE NOT EXISTS (SELECT 1
FROM (SELECT raw_id + 1 AS id, limit_value FROM (VALUES (25, 0), (10, 1),
(CAST(NULL AS INT), 2)) AS v(limit_value, raw_id) WHERE limit_value IS NOT
NULL) AS r WHERE l.id = r.id AND l.amount < r.limit_value)]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(id=[$0], amount=[$1])
++- LogicalFilter(condition=[NOT(EXISTS({
+LogicalFilter(condition=[AND(=($cor0.id, $0), <($cor0.amount, $1))])
+ LogicalProject(id=[+($1, 1)], limit_value=[$0])
+ LogicalFilter(condition=[IS NOT NULL($0)])
+ LogicalProject(limit_value=[$0], raw_id=[$1])
+ LogicalUnion(all=[true])
+ LogicalProject(EXPR$0=[25], EXPR$1=[0])
+ LogicalValues(tuples=[[{ 0 }]])
+ LogicalProject(EXPR$0=[10], EXPR$1=[1])
+ LogicalValues(tuples=[[{ 0 }]])
+ LogicalProject(EXPR$0=[null:INTEGER], EXPR$1=[2])
+ LogicalValues(tuples=[[{ 0 }]])
+}))], variablesSet=[[$cor0]])
+ +- LogicalProject(id=[$0], amount=[$1])
+ +- LogicalUnion(all=[true])
+ :- LogicalProject(EXPR$0=[1], EXPR$1=[10])
+ : +- LogicalValues(tuples=[[{ 0 }]])
+ :- LogicalProject(EXPR$0=[1], EXPR$1=[30])
+ : +- LogicalValues(tuples=[[{ 0 }]])
+ :- LogicalProject(EXPR$0=[2], EXPR$1=[20])
+ : +- LogicalValues(tuples=[[{ 0 }]])
+ :- LogicalProject(EXPR$0=[3], EXPR$1=[40])
+ : +- LogicalValues(tuples=[[{ 0 }]])
+ +- LogicalProject(EXPR$0=[null:INTEGER], EXPR$1=[50])
+ +- LogicalValues(tuples=[[{ 0 }]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+LogicalProject(id=[$0], amount=[$1])
++- LogicalJoin(condition=[AND(=($0, $2), <($1, $3))], joinType=[anti])
+ :- LogicalProject(id=[$0], amount=[$1])
+ : +- LogicalUnion(all=[true])
+ : :- LogicalProject(EXPR$0=[1], EXPR$1=[10])
+ : : +- LogicalValues(tuples=[[{ 0 }]])
+ : :- LogicalProject(EXPR$0=[1], EXPR$1=[30])
+ : : +- LogicalValues(tuples=[[{ 0 }]])
+ : :- LogicalProject(EXPR$0=[2], EXPR$1=[20])
+ : : +- LogicalValues(tuples=[[{ 0 }]])
+ : :- LogicalProject(EXPR$0=[3], EXPR$1=[40])
+ : : +- LogicalValues(tuples=[[{ 0 }]])
+ : +- LogicalProject(EXPR$0=[null:INTEGER], EXPR$1=[50])
+ : +- LogicalValues(tuples=[[{ 0 }]])
+ +- LogicalFilter(condition=[true])
+ +- LogicalProject(id=[+($1, 1)], limit_value=[$0])
+ +- LogicalFilter(condition=[IS NOT NULL($0)])
+ +- LogicalProject(limit_value=[$0], raw_id=[$1])
+ +- LogicalUnion(all=[true])
+ :- LogicalValues(tuples=[[{ 25, 0 }]])
+ :- LogicalValues(tuples=[[{ 10, 1 }]])
+ +- LogicalValues(tuples=[[{ null, 2 }]])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testValues[EXISTS]">
+ <Resource name="sql">
+ <![CDATA[SELECT l.id FROM (VALUES (1), (2)) AS l(id) WHERE EXISTS
(SELECT 1 FROM (VALUES (1), (2)) AS r(right_id) WHERE l.id = r.right_id)]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(id=[$0])
++- LogicalFilter(condition=[EXISTS({
+LogicalFilter(condition=[=($cor0.id, $0)])
+ LogicalValues(tuples=[[{ 1 }, { 2 }]])
+})], variablesSet=[[$cor0]])
+ +- LogicalValues(tuples=[[{ 1 }, { 2 }]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+LogicalProject(id=[$0])
++- LogicalJoin(condition=[=($0, $1)], joinType=[semi])
+ :- LogicalValues(tuples=[[{ 1 }, { 2 }]])
+ +- LogicalFilter(condition=[true])
+ +- LogicalValues(tuples=[[{ 1 }, { 2 }]])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testValues[NOT EXISTS]">
+ <Resource name="sql">
+ <![CDATA[SELECT l.id FROM (VALUES (1), (2)) AS l(id) WHERE NOT EXISTS
(SELECT 1 FROM (VALUES (1), (2)) AS r(right_id) WHERE l.id = r.right_id)]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(id=[$0])
++- LogicalFilter(condition=[NOT(EXISTS({
+LogicalFilter(condition=[=($cor0.id, $0)])
+ LogicalValues(tuples=[[{ 1 }, { 2 }]])
+}))], variablesSet=[[$cor0]])
+ +- LogicalValues(tuples=[[{ 1 }, { 2 }]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+LogicalProject(id=[$0])
++- LogicalJoin(condition=[=($0, $1)], joinType=[anti])
+ :- LogicalValues(tuples=[[{ 1 }, { 2 }]])
+ +- LogicalFilter(condition=[true])
+ +- LogicalValues(tuples=[[{ 1 }, { 2 }]])
+]]>
+ </Resource>
+ </TestCase>
+</Root>
