[
https://issues.apache.org/jira/browse/FLINK-29120?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
godfrey he closed FLINK-29120.
------------------------------
Resolution: Fixed
Fixed in master: 6722c89d0df35643dde38c1b8f096aa785579884
in 1.16: 357221b9b3543447b2b439a413639c0ed201ab35
> Unexpected join hint propagation into view
> ------------------------------------------
>
> Key: FLINK-29120
> URL: https://issues.apache.org/jira/browse/FLINK-29120
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
> Affects Versions: 1.16.0
> Reporter: lincoln lee
> Assignee: xuyang
> Priority: Blocker
> Labels: pull-request-available
> Fix For: 1.16.0
>
>
> 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)
