[
https://issues.apache.org/jira/browse/FLINK-29120?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Martijn Visser reassigned FLINK-29120:
--------------------------------------
Assignee: xuyang
> Unexpected join hint propagation into view
> ------------------------------------------
>
> Key: FLINK-29120
> URL: https://issues.apache.org/jira/browse/FLINK-29120
> Project: Flink
> Issue Type: Bug
> Reporter: lincoln lee
> Assignee: xuyang
> Priority: Blocker
> Labels: pull-request-available
>
> As expected, Join Hint should only affects the current query block, and does
> not affect the Join strategy in subquery and view.
> But current implementation behaviors inconsistently:
> use source tables of flink-tpch-test, the following join hint takes effect
> unexpectedly
> {code:java}
> Flink SQL> create temporary view v1 as SELECT
> > p_name,
> > p_mfgr,
> > p_brand,
> > p_type,
> > s_name,
> > s_address
> > FROM
> > part,
> > supplier
> > WHERE p_partkey = s_suppkey;
> [INFO] Execute statement succeed.
>
> Flink SQL> explain SELECT /*+ SHUFFLE_MERGE(part) */ * from v1;
> == Abstract Syntax Tree ==
> LogicalProject(p_name=[$1], p_mfgr=[$2], p_brand=[$3], p_type=[$4],
> s_name=[$10], s_address=[$11])
> +- LogicalFilter(condition=[=($0, $9)])
> +- LogicalJoin(condition=[true], joinType=[inner],
> joinHints=[[[SHUFFLE_MERGE inheritPath:[0, 0] options:[part]]]])
> :- LogicalTableScan(table=[[default_catalog, default_database, part]])
> +- LogicalTableScan(table=[[default_catalog, default_database,
> supplier]])
> == Optimized Physical Plan ==
> Calc(select=[p_name, p_mfgr, p_brand, p_type, s_name, s_address])
> +- SortMergeJoin(joinType=[InnerJoin], where=[=(p_partkey, s_suppkey)],
> select=[p_partkey, p_name, p_mfgr, p_brand, p_type, s_suppkey, s_name,
> s_address])
> :- Exchange(distribution=[hash[p_partkey]])
> : +- TableSourceScan(table=[[default_catalog, default_database, part,
> project=[p_partkey, p_name, p_mfgr, p_brand, p_type], metadata=[]]],
> fields=[p_partkey, p_name, p_mfgr, p_brand, p_type])
> +- Exchange(distribution=[hash[s_suppkey]])
> +- TableSourceScan(table=[[default_catalog, default_database, supplier,
> project=[s_suppkey, s_name, s_address], metadata=[]]], fields=[s_suppkey,
> s_name, s_address])
> == Optimized Execution Plan ==
> Calc(select=[p_name, p_mfgr, p_brand, p_type, s_name, s_address])
> +- SortMergeJoin(joinType=[InnerJoin], where=[(p_partkey = s_suppkey)],
> select=[p_partkey, p_name, p_mfgr, p_brand, p_type, s_suppkey, s_name,
> s_address])
> :- Exchange(distribution=[hash[p_partkey]])
> : +- TableSourceScan(table=[[default_catalog, default_database, part,
> project=[p_partkey, p_name, p_mfgr, p_brand, p_type], metadata=[]]],
> fields=[p_partkey, p_name, p_mfgr, p_brand, p_type])
> +- Exchange(distribution=[hash[s_suppkey]])
> +- TableSourceScan(table=[[default_catalog, default_database, supplier,
> project=[s_suppkey, s_name, s_address], metadata=[]]], fields=[s_suppkey,
> s_name, s_address])
> {code}
>
> without hint
> {code}
> Flink SQL> explain SELECT * from v1;
> == Abstract Syntax Tree ==
> LogicalProject(p_name=[$1], p_mfgr=[$2], p_brand=[$3], p_type=[$4],
> s_name=[$10], s_address=[$11])
> +- LogicalFilter(condition=[=($0, $9)])
> +- LogicalJoin(condition=[true], joinType=[inner])
> :- LogicalTableScan(table=[[default_catalog, default_database, part]])
> +- LogicalTableScan(table=[[default_catalog, default_database,
> supplier]])
> == Optimized Physical Plan ==
> Calc(select=[p_name, p_mfgr, p_brand, p_type, s_name, s_address])
> +- HashJoin(joinType=[InnerJoin], where=[=(p_partkey, s_suppkey)],
> select=[p_partkey, p_name, p_mfgr, p_brand, p_type, s_suppkey, s_name,
> s_address], isBroadcast=[true], build=[right])
> :- TableSourceScan(table=[[default_catalog, default_database, part,
> project=[p_partkey, p_name, p_mfgr, p_brand, p_type], metadata=[]]],
> fields=[p_partkey, p_name, p_mfgr, p_brand, p_type])
> +- Exchange(distribution=[broadcast])
> +- TableSourceScan(table=[[default_catalog, default_database, supplier,
> project=[s_suppkey, s_name, s_address], metadata=[]]], fields=[s_suppkey,
> s_name, s_address])
> == Optimized Execution Plan ==
> Calc(select=[p_name, p_mfgr, p_brand, p_type, s_name, s_address])
> +- MultipleInput(readOrder=[1,0], members=[\nHashJoin(joinType=[InnerJoin],
> where=[(p_partkey = s_suppkey)], select=[p_partkey, p_name, p_mfgr, p_brand,
> p_type, s_suppkey, s_name, s_address], isBroadcast=[true], build=[right])\n:-
> [#1] TableSourceScan(table=[[default_catalog, default_database, part,
> project=[p_partkey, p_name, p_mfgr, p_brand, p_type], metadata=[]]],
> fields=[p_partkey, p_name, p_mfgr, p_brand, p_type])\n+- [#2]
> Exchange(distribution=[broadcast])\n])
> :- TableSourceScan(table=[[default_catalog, default_database, part,
> project=[p_partkey, p_name, p_mfgr, p_brand, p_type], metadata=[]]],
> fields=[p_partkey, p_name, p_mfgr, p_brand, p_type])
> +- Exchange(distribution=[broadcast])
> +- TableSourceScan(table=[[default_catalog, default_database, supplier,
> project=[s_suppkey, s_name, s_address], metadata=[]]], fields=[s_suppkey,
> s_name, s_address])
> {code}
>
>
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
