This is an automated email from the ASF dual-hosted git repository. godfrey 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 bdffb6bd4ae [FLINK-28986][table-planner] UNNEST function with nested filter fails to generate plan bdffb6bd4ae is described below commit bdffb6bd4ae3ea32bdcd6ec6bce6c6e0e8b92a11 Author: Jane Chan <qingyue....@gmail.com> AuthorDate: Tue Aug 16 17:35:54 2022 +0800 [FLINK-28986][table-planner] UNNEST function with nested filter fails to generate plan This closes #20601 --- .../planner/plan/rules/FlinkBatchRuleSets.scala | 2 ++ .../planner/plan/rules/FlinkStreamRuleSets.scala | 2 ++ .../table/planner/plan/batch/sql/UnnestTest.xml | 36 +++++++++++++++++++ .../plan/rules/logical/LogicalUnnestRuleTest.xml | 42 ++++++++++++++++++++++ .../table/planner/plan/stream/sql/UnnestTest.xml | 36 +++++++++++++++++++ .../table/planner/plan/common/UnnestTestBase.scala | 17 +++++++++ .../planner/runtime/stream/sql/UnnestITCase.scala | 31 ++++++++++++++++ 7 files changed, 166 insertions(+) diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkBatchRuleSets.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkBatchRuleSets.scala index a5b1b197dec..691ca43f339 100644 --- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkBatchRuleSets.scala +++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkBatchRuleSets.scala @@ -116,6 +116,8 @@ object FlinkBatchRuleSets { ConvertToNotInOrInRule.INSTANCE, // optimize limit 0 FlinkLimit0RemoveRule.INSTANCE, + // fix: FLINK-28986 nested filter pattern causes unnest rule mismatch + CoreRules.FILTER_MERGE, // unnest rule LogicalUnnestRule.INSTANCE, // Wrap arguments for JSON aggregate functions diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkStreamRuleSets.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkStreamRuleSets.scala index d5462682964..97312367287 100644 --- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkStreamRuleSets.scala +++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkStreamRuleSets.scala @@ -122,6 +122,8 @@ object FlinkStreamRuleSets { ConvertToNotInOrInRule.INSTANCE, // optimize limit 0 FlinkLimit0RemoveRule.INSTANCE, + // fix: FLINK-28986 nested filter pattern causes unnest rule mismatch + CoreRules.FILTER_MERGE, // unnest rule LogicalUnnestRule.INSTANCE, // rewrite constant table function scan to correlate diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/UnnestTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/UnnestTest.xml index 8896c7ee62d..ef3bc2e29d0 100644 --- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/UnnestTest.xml +++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/UnnestTest.xml @@ -291,6 +291,42 @@ LogicalProject(a=[$0], b=[$1], _1=[$2], _2=[$3]) <![CDATA[ Correlate(invocation=[$UNNEST_ROWS$1($cor0.b)], correlate=[table($UNNEST_ROWS$1($cor0.b))], select=[a,b,_1,_2], rowType=[RecordType(INTEGER a, RecordType:peek_no_expand(INTEGER _1, VARCHAR(2147483647) _2) ARRAY b, INTEGER _1, VARCHAR(2147483647) _2)], joinType=[INNER], condition=[>($0, 1)]) +- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b)]]], fields=[a, b]) +]]> + </Resource> + </TestCase> + <TestCase name="testUnnestWithNestedFilter"> + <Resource name="sql"> + <![CDATA[ +SELECT * FROM ( + SELECT a, b1, b2 FROM + (SELECT a, b FROM MyTable) T + CROSS JOIN + UNNEST(T.b) AS S(b1, b2) + WHERE S.b1 >= 12 + ) tmp +WHERE b2 <> 'Hello' + ]]> + </Resource> + <Resource name="ast"> + <![CDATA[ +LogicalProject(a=[$0], b1=[$1], b2=[$2]) ++- LogicalFilter(condition=[<>($2, _UTF-16LE'Hello')]) + +- LogicalProject(a=[$0], b1=[$2], b2=[$3]) + +- LogicalFilter(condition=[>=($2, 12)]) + +- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{1}]) + :- LogicalProject(a=[$0], b=[$1]) + : +- LogicalTableScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b)]]]) + +- LogicalProject(b1=[$0], b2=[$1]) + +- Uncollect + +- LogicalProject(b=[$cor0.b]) + +- LogicalValues(tuples=[[{ 0 }]]) +]]> + </Resource> + <Resource name="optimized exec plan"> + <![CDATA[ +Calc(select=[a, _1 AS b1, _2 AS b2]) ++- Correlate(invocation=[$UNNEST_ROWS$1($cor0.b)], correlate=[table($UNNEST_ROWS$1($cor0.b))], select=[a,b,_1,_2], rowType=[RecordType(INTEGER a, RecordType:peek_no_expand(INTEGER _1, VARCHAR(2147483647) _2) ARRAY b, INTEGER _1, VARCHAR(2147483647) _2)], joinType=[INNER], condition=[AND(>=($0, 12), <>($1, _UTF-16LE'Hello'))]) + +- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b)]]], fields=[a, b]) ]]> </Resource> </TestCase> diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/LogicalUnnestRuleTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/LogicalUnnestRuleTest.xml index ddbdb8d61e3..d2e281ba526 100644 --- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/LogicalUnnestRuleTest.xml +++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/LogicalUnnestRuleTest.xml @@ -308,6 +308,48 @@ LogicalProject(a=[$0], b=[$1], _1=[$2], _2=[$3]) +- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{1}]) :- LogicalTableScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b)]]]) +- LogicalTableFunctionScan(invocation=[$UNNEST_ROWS$1($cor0.b)], rowType=[RecordType:peek_no_expand(INTEGER _1, VARCHAR(2147483647) _2)]) +]]> + </Resource> + </TestCase> + <TestCase name="testUnnestWithNestedFilter"> + <Resource name="sql"> + <![CDATA[ +SELECT * FROM ( + SELECT a, b1, b2 FROM + (SELECT a, b FROM MyTable) T + CROSS JOIN + UNNEST(T.b) AS S(b1, b2) + WHERE S.b1 >= 12 + ) tmp +WHERE b2 <> 'Hello' + ]]> + </Resource> + <Resource name="ast"> + <![CDATA[ +LogicalProject(a=[$0], b1=[$1], b2=[$2]) ++- LogicalFilter(condition=[<>($2, _UTF-16LE'Hello')]) + +- LogicalProject(a=[$0], b1=[$2], b2=[$3]) + +- LogicalFilter(condition=[>=($2, 12)]) + +- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{1}]) + :- LogicalProject(a=[$0], b=[$1]) + : +- LogicalTableScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b)]]]) + +- LogicalProject(b1=[$0], b2=[$1]) + +- Uncollect + +- LogicalProject(b=[$cor0.b]) + +- LogicalValues(tuples=[[{ 0 }]]) +]]> + </Resource> + <Resource name="optimized rel plan"> + <![CDATA[ +LogicalProject(a=[$0], b1=[$1], b2=[$2]) ++- LogicalFilter(condition=[<>($2, _UTF-16LE'Hello')]) + +- LogicalProject(a=[$0], b1=[$2], b2=[$3]) + +- LogicalFilter(condition=[>=($2, 12)]) + +- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{1}]) + :- LogicalProject(a=[$0], b=[$1]) + : +- LogicalTableScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b)]]]) + +- LogicalProject(b1=[$0], b2=[$1]) + +- LogicalTableFunctionScan(invocation=[$UNNEST_ROWS$1($cor0.b)], rowType=[RecordType:peek_no_expand(INTEGER _1, VARCHAR(2147483647) _2)]) ]]> </Resource> </TestCase> diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/UnnestTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/UnnestTest.xml index 1dbe3458b14..f266ab21419 100644 --- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/UnnestTest.xml +++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/UnnestTest.xml @@ -282,6 +282,42 @@ LogicalProject(a=[$0], b=[$1], _1=[$2], _2=[$3]) <![CDATA[ Correlate(invocation=[$UNNEST_ROWS$1($cor0.b)], correlate=[table($UNNEST_ROWS$1($cor0.b))], select=[a,b,_1,_2], rowType=[RecordType(INTEGER a, RecordType:peek_no_expand(INTEGER _1, VARCHAR(2147483647) _2) ARRAY b, INTEGER _1, VARCHAR(2147483647) _2)], joinType=[INNER], condition=[>($0, 1)]) +- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b)]]], fields=[a, b]) +]]> + </Resource> + </TestCase> + <TestCase name="testUnnestWithNestedFilter"> + <Resource name="sql"> + <![CDATA[ +SELECT * FROM ( + SELECT a, b1, b2 FROM + (SELECT a, b FROM MyTable) T + CROSS JOIN + UNNEST(T.b) AS S(b1, b2) + WHERE S.b1 >= 12 + ) tmp +WHERE b2 <> 'Hello' + ]]> + </Resource> + <Resource name="ast"> + <![CDATA[ +LogicalProject(a=[$0], b1=[$1], b2=[$2]) ++- LogicalFilter(condition=[<>($2, _UTF-16LE'Hello')]) + +- LogicalProject(a=[$0], b1=[$2], b2=[$3]) + +- LogicalFilter(condition=[>=($2, 12)]) + +- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{1}]) + :- LogicalProject(a=[$0], b=[$1]) + : +- LogicalTableScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b)]]]) + +- LogicalProject(b1=[$0], b2=[$1]) + +- Uncollect + +- LogicalProject(b=[$cor0.b]) + +- LogicalValues(tuples=[[{ 0 }]]) +]]> + </Resource> + <Resource name="optimized exec plan"> + <![CDATA[ +Calc(select=[a, _1 AS b1, _2 AS b2]) ++- Correlate(invocation=[$UNNEST_ROWS$1($cor0.b)], correlate=[table($UNNEST_ROWS$1($cor0.b))], select=[a,b,_1,_2], rowType=[RecordType(INTEGER a, RecordType:peek_no_expand(INTEGER _1, VARCHAR(2147483647) _2) ARRAY b, INTEGER _1, VARCHAR(2147483647) _2)], joinType=[INNER], condition=[AND(>=($0, 12), <>($1, _UTF-16LE'Hello'))]) + +- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b)]]], fields=[a, b]) ]]> </Resource> </TestCase> diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/common/UnnestTestBase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/common/UnnestTestBase.scala index 04403b4e96c..c7d76a4c8da 100644 --- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/common/UnnestTestBase.scala +++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/common/UnnestTestBase.scala @@ -121,6 +121,23 @@ abstract class UnnestTestBase(withExecPlan: Boolean) extends TableTestBase { verifyPlan("SELECT a, b, A._1, A._2 FROM MyTable, UNNEST(MyTable.b) AS A where A._1 > 1") } + @Test + def testUnnestWithNestedFilter(): Unit = { + util.addTableSource[(Int, Array[(Int, String)])]("MyTable", 'a, 'b) + val sqlQuery = + """ + |SELECT * FROM ( + | SELECT a, b1, b2 FROM + | (SELECT a, b FROM MyTable) T + | CROSS JOIN + | UNNEST(T.b) AS S(b1, b2) + | WHERE S.b1 >= 12 + | ) tmp + |WHERE b2 <> 'Hello' + """.stripMargin + verifyPlan(sqlQuery) + } + private def verifyPlan(sql: String): Unit = { if (withExecPlan) { util.verifyExecPlan(sql) diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/UnnestITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/UnnestITCase.scala index 863bf3984f8..81214a4cf2e 100644 --- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/UnnestITCase.scala +++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/UnnestITCase.scala @@ -313,4 +313,35 @@ class UnnestITCase(mode: StateBackendMode) extends StreamingWithStateTestBase(mo assertEquals(expected.sorted, sink.getAppendResults.sorted) } + @Test + def testUnnestWithNestedFilter(): Unit = { + val data = List( + (1, Array((12, "45.6"), (12, "45.612"))), + (2, Array((13, "41.6"), (14, "45.2136"))), + (3, Array((18, "42.6"))) + ) + val t = env.fromCollection(data).toTable(tEnv, 'a, 'b) + tEnv.registerTable("MyTable", t) + + val sqlQuery = + """ + |SELECT * FROM ( + | SELECT a, b1, b2 FROM + | (SELECT a, b FROM MyTable) T + | CROSS JOIN + | UNNEST(T.b) as S(b1, b2) + | WHERE S.b1 >= 12 + | ) tmp + |WHERE b2 <> '42.6' + """.stripMargin + + val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row] + val sink = new TestingAppendSink + result.addSink(sink) + env.execute() + + val expected = List("1,12,45.612", "1,12,45.6", "2,13,41.6", "2,14,45.2136") + assertEquals(expected.sorted, sink.getAppendResults.sorted) + } + }