lincoln lee created FLINK-29120:
-----------------------------------
Summary: 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
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}
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}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
