You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "lincoln lee (Jira)" <ji...@apache.org> on 2022/08/26 09:51:00 UTC
[jira] [Updated] (FLINK-29120) Unexpected join hint propagation into view
[ https://issues.apache.org/jira/browse/FLINK-29120?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
lincoln lee updated FLINK-29120:
--------------------------------
Description:
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}
was:
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}
> 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
> Priority: Blocker
>
> 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)