godfrey he created FLINK-12321:
----------------------------------
Summary: Supports sub-plan reuse
Key: FLINK-12321
URL:
https://issues.apache.org/jira/browse/FLINK-12321 Project: Flink
Issue Type: Bug
Components: Table SQL / Planner
Reporter: godfrey he
Assignee: godfrey he
Many query plans have duplicated sub-plan, and their computing logic is identical. So we could reuse duplicated sub-plans to reduce computing.
Digest of RelNode is an identifier to distinguish RelNode, and the digest of sub-plan contains all inner-nodes' digests. we could use the digest of sub-plan to find duplicated sub-plan.
e.g.
{code:sql}
WITH r AS (SELECT a FROM x LIMIT 10)
SELECT r1.a FROM r r1, r r2 WHERE r1.a = r2.a
the physical plan after sub-plan reuse:
Calc(select=[a])
+- HashJoin(joinType=[InnerJoin], where=[=(a, a0)], select=[a, a0], isBroadcast=[true], build=[right])
:- Exchange(distribution=[hash[a]])
: +- Calc(select=[a], reuse_id=[1])
: +- Limit(offset=[0], fetch=[10], global=[true])
: +- Exchange(distribution=[single])
: +- Limit(offset=[0], fetch=[10], global=[false])
: +- TableSourceScan(table=[[x, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
+- Exchange(distribution=[broadcast])
+- Reused(reference_id=[1])
{code}
sub-plan: Calc-Limit-Exchange-Limit-TableSourceScan is reused.
For batch job, reused node might lead to a deadlock when a HashJoin or NestedLoopJoin have same reused inputs. The probe side of HashJoin could start to read data only after build side has finished. If there is no full dam (DamBehavior#FULL_DAM) operators (e.g. Sort, HashAggregate) in probe side, the data will be blocked by probe side. In this case, we could set Exchange node (if it does not exist, add new one) as BATCH mode to break up the deadlock.(The Exchange node is a full dam operator now)
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
Locked
