[
https://issues.apache.org/jira/browse/FLINK-33446?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17802090#comment-17802090
]
Jeyhun Karimov commented on FLINK-33446:
----------------------------------------
Hi [~fsk119] I scanned through the issue a bit.
I can confirm that {{Sql2RelConverter}} generates the correct plan. The issue
IMO is related to the {{RelDecorrelator.decorrelateQuery}}
({{FlinkDecorrelateProgram::optimize}}), specifically this line:
{code:java}
val result = RelDecorrelator.decorrelateQuery(root)
{code}
the input plan ({{root}}) is:
{code:java}
LogicalProject(d2=[$0], d3=[$1])
LogicalProject(d2=[$0], d3=[$1])
LogicalFilter(condition=[IS NULL($2)])
LogicalCorrelate(correlation=[$cor0], joinType=[left],
requiredColumns=[{0, 1}])
LogicalProject(d2=[+(2, $0)], d3=[+(3, $1)])
LogicalTableScan(table=[[default_catalog, default_database, l,
source: [TestTableSource(a, b, c)]]])
LogicalAggregate(group=[{0}])
LogicalProject(i=[true])
LogicalProject(d1=[$0])
LogicalFilter(condition=[AND(=($0, $cor0.d2), IS NULL($1))])
LogicalJoin(condition=[true], joinType=[left],
variablesSet=[[$cor1, $cor0]])
LogicalProject(d1=[+($0, 1)])
LogicalTableScan(table=[[default_catalog, default_database,
r, source: [TestTableSource(d, e, f)]]])
LogicalAggregate(group=[{0}])
LogicalProject(i=[true])
LogicalFilter(condition=[AND(=($0, $cor1.d1), =($1,
$cor1.d1), =(CAST($2):BIGINT, $cor0.d3))])
LogicalProject(d4=[+($0, 4)], d5=[+($0, 5)], d6=[+($0,
6)])
LogicalTableScan(table=[[default_catalog,
default_database, t, source: [TestTableSource(i, j, k)]]])
{code}
and the output of the function call ({{RelDecorrelator.decorrelateQuery}}) is:
{code:java}
LogicalProject(d2=[$0], d3=[$1], d4=[$4])
LogicalFilter(condition=[IS NULL($5)])
LogicalJoin(condition=[AND(=($0, $2), =($1, $3))], joinType=[left])
LogicalProject(d2=[+(2, $0)], d3=[+(3, $1)])
LogicalTableScan(table=[[default_catalog, default_database, l, source:
[TestTableSource(a, b, c)]]])
LogicalProject(d11=[$0], $f3=[$1], d4=[$2], $f4=[true])
LogicalAggregate(group=[{0, 1, 2}])
LogicalProject(d11=[$0], $f3=[$1], d4=[$2])
LogicalFilter(condition=[IS NULL($3)])
LogicalJoin(condition=[true], joinType=[left])
LogicalFilter(condition=[IS NOT NULL($0)])
LogicalProject(d1=[+($0, 1)])
LogicalTableScan(table=[[default_catalog, default_database,
r, source: [TestTableSource(d, e, f)]]])
LogicalProject($f3=[$0], d4=[$1], $f2=[true])
LogicalAggregate(group=[{0, 1}])
LogicalProject($f3=[$3], d4=[$0])
LogicalFilter(condition=[AND(=($1, $0),
=(CAST($2):BIGINT, $3))])
LogicalProject(d4=[+($0, 4)], d5=[+($0, 5)], d6=[+($0,
6)], $f3=[CAST(+($0, 6)):BIGINT])
LogicalTableScan(table=[[default_catalog,
default_database, t, source: [TestTableSource(i, j, k)]]])
{code}
WDYT?
> SubQueryAntiJoinTest#testMultiNotExistsWithCorrelatedOnWhere_NestedCorrelation
> doesn't produce the correct plan
> ---------------------------------------------------------------------------------------------------------------
>
> Key: FLINK-33446
> URL: https://issues.apache.org/jira/browse/FLINK-33446
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
> Affects Versions: 1.17.2, 1.19.0, 1.18.1
> Reporter: Shengkai Fang
> Priority: Major
>
> Although this test doesn't throw an exception, the final plan produces 3
> columns rather than 2 after optimization.
> {code:java}
> LogicalProject(inputs=[0..1], exprs=[[$4]])
> +- LogicalFilter(condition=[IS NULL($5)])
> +- LogicalJoin(condition=[AND(=($0, $2), =($1, $3))], joinType=[left])
> :- LogicalProject(exprs=[[+(2, $0), +(3, $1)]])
> : +- LogicalTableScan(table=[[default_catalog, default_database, l,
> source: [TestTableSource(a, b, c)]]])
> +- LogicalProject(inputs=[0..2], exprs=[[true]])
> +- LogicalAggregate(group=[{0, 1, 2}])
> +- LogicalProject(inputs=[0..2])
> +- LogicalFilter(condition=[IS NULL($3)])
> +- LogicalJoin(condition=[true], joinType=[left])
> :- LogicalFilter(condition=[IS NOT NULL($0)])
> : +- LogicalProject(exprs=[[+($0, 1)]])
> : +- LogicalTableScan(table=[[default_catalog,
> default_database, r, source: [TestTableSource(d, e, f)]]])
> +- LogicalProject(inputs=[0..1], exprs=[[true]])
> +- LogicalAggregate(group=[{0, 1}])
> +- LogicalProject(exprs=[[$3, $0]])
> +- LogicalFilter(condition=[AND(=($1, $0),
> =(CAST($2):BIGINT, $3))])
> +- LogicalProject(exprs=[[+($0, 4), +($0,
> 5), +($0, 6), CAST(+($0, 6)):BIGINT]])
> +-
> LogicalTableScan(table=[[default_catalog, default_database, t, source:
> [TestTableSource(i, j, k)]]])
> {code}
> After digging, I think it's the SubQueryRemoveRule doesn't generate the
> Correlate but generates the Join node, which causes the failure of the
> decorrelation. For a quick fix, I think we should throw an exception to
> notify users it's not a supported feature in the Flink.
> There might exist 2 ways to fix this issue:
> 1. Expand subquery when converting SQL to rel. After experimenting with
> calcite, I found that the Sql2RelConverter generates the correct plan.
> {code:java}
> LogicalProject(inputs=[0..1])
> +- LogicalFilter(condition=[IS NULL($2)])
> +- LogicalCorrelate(correlation=[$cor7], joinType=[left],
> requiredColumns=[{0, 1}])
> :- LogicalProject(exprs=[[+(2, $0), +(3, $1)]])
> : +- LogicalTableScan(table=[[default_catalog, default_database, l,
> source: [TestTableSource(a, b, c)]]])
> +- LogicalAggregate(group=[{}], agg#0=[MIN($0)])
> +- LogicalProject(exprs=[[true]])
> +- LogicalFilter(condition=[AND(=($0, $cor7.d2), IS NULL($1))])
> +- LogicalCorrelate(correlation=[$cor4], joinType=[left],
> requiredColumns=[{0}])
> :- LogicalProject(inputs=[0])
> : +- LogicalTableScan(table=[[default_catalog,
> default_database, r, source: [TestTableSource(d1, e, f)]]])
> +- LogicalAggregate(group=[{}], agg#0=[MIN($0)])
> +- LogicalProject(exprs=[[true]])
> +- LogicalFilter(condition=[AND(=($0, $cor4.d1),
> =($1, $cor4.d1), =(CAST($2):BIGINT, $cor7.d3))])
> +- LogicalProject(exprs=[[+($0, 4), +($0, 5),
> +($0, 6)]])
> +- LogicalTableScan(table=[[default_catalog,
> default_database, t, source: [TestTableSource(i, j, k)]]])
> {code}
> You can find the new plan uses a correlate node rather than a join node.
> 2. CALCITE-5789 has fix this problem by removing the nested correlation node.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
