Caizhi Weng created FLINK-19959:
-----------------------------------
Summary: Multiple input creation algorithm will deduce an
incorrect input order if the inputs are related under PIPELINED shuffle mode
Key: FLINK-19959
URL: https://issues.apache.org/jira/browse/FLINK-19959
Project: Flink
Issue Type: Bug
Components: Table SQL / Planner
Reporter: Caizhi Weng
Fix For: 1.12.0
Consider the following SQL
{code:sql}
WITH
T1 AS (SELECT x.a AS a, y.d AS b FROM y LEFT JOIN x ON y.d = x.a),
T2 AS (SELECT a, b FROM (SELECT a, b FROM T1) UNION ALL (SELECT x.a AS a, x.b
AS b FROM x))
SELECT * FROM T2 LEFT JOIN t ON T2.a = t.a
{code}
The multiple input creation algorithm will currently deduce the following plan
under the PIPELINED shuffle mode:
{code}
MultipleInputNode(readOrder=[0,1,1,0],
members=[\nNestedLoopJoin(joinType=[LeftOuterJoin], where=[=(a, a0)],
select=[a, b, a0, b0, c], build=[right])\n:- Union(all=[true], union=[a, b])\n:
:- Calc(select=[a, CAST(d) AS b])\n: : +-
NestedLoopJoin(joinType=[LeftOuterJoin], where=[=(d, a)], select=[d, a],
build=[right])\n: : :- [#3] Calc(select=[d])\n: : +- [#4]
Exchange(distribution=[broadcast])\n: +- [#2] Calc(select=[a, b])\n+- [#1]
Exchange(distribution=[broadcast])\n])
:- Exchange(distribution=[broadcast])
: +- BoundedStreamScan(table=[[default_catalog, default_database, t]],
fields=[a, b, c])
:- Calc(select=[a, b])
: +- LegacyTableSourceScan(table=[[default_catalog, default_database, x,
source: [TestTableSource(a, b, c, nx)]]], fields=[a, b, c, nx], reuse_id=[1])
:- Calc(select=[d])
: +- LegacyTableSourceScan(table=[[default_catalog, default_database, y,
source: [TestTableSource(d, e, f, ny)]]], fields=[d, e, f, ny])
+- Exchange(distribution=[broadcast])
+- Calc(select=[a])
+- Reused(reference_id=[1])
{code}
It's obvious that the 2nd and the 4th input for the multiple input node should
have the same input priority, otherwise a deadlock will occur.
This is because the current algorithm fails to consider the case when the
inputs are related out of the multiple input node.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
