jffwang
/
hetu-core
0
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

[96][I4CVQX] join distribution improvement based on size

This commit is contained in:
Nitin Kashyap 2021-10-13 12:03:24 +05:30
parent e16c83d946
commit 4931f1e4e3
No known key found for this signature in database
GPG Key ID: 7C4222EE7DDEFC35
19 changed files with 755 additions and 752 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q14_1.cte.plan.txt
View File

@ -30,12 +30,9 @@ local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["i_item_sk_1"])
cte cross_items
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["i_brand_id_8", "i_category_id_12", "i_class_id_10"])
Filter
scan item
Filter
final hashaggregation over (expr_207, expr_208, expr_209)
local exchange (GATHER, SINGLE, [])
local exchange (REPARTITION, HASH, ["expr_207", "expr_208", "expr_209"])
remote exchange (REPARTITION, HASH, ["i_brand_id_53", "i_category_id_57", "i_class_id_55"])
partial hashaggregation over (i_brand_id_53, i_category_id_57, i_class_id_55)
join (INNER, REPLICATED):
@ -75,6 +72,9 @@ local exchange (GATHER, SINGLE, [])
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan item
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["i_brand_id_8", "i_category_id_12", "i_class_id_10"])
scan item
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
local exchange (GATHER, SINGLE, [])
@ -132,12 +132,9 @@ local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["i_item_sk_518"])
cte cross_items
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["i_brand_id_525", "i_category_id_529", "i_class_id_527"])
Filter
scan item
Filter
final hashaggregation over (expr_793, expr_794, expr_795)
local exchange (GATHER, SINGLE, [])
local exchange (REPARTITION, HASH, ["expr_793", "expr_794", "expr_795"])
remote exchange (REPARTITION, HASH, ["i_brand_id_570", "i_category_id_574", "i_class_id_572"])
partial hashaggregation over (i_brand_id_570, i_category_id_574, i_class_id_572)
join (INNER, REPLICATED):
@ -177,6 +174,9 @@ local exchange (GATHER, SINGLE, [])
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan item
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["i_brand_id_525", "i_category_id_529", "i_class_id_527"])
scan item
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
local exchange (GATHER, SINGLE, [])
@ -234,12 +234,9 @@ local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["i_item_sk_1111"])
cte cross_items
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["i_brand_id_1118", "i_category_id_1122", "i_class_id_1120"])
Filter
scan item
Filter
final hashaggregation over (expr_1386, expr_1387, expr_1388)
local exchange (GATHER, SINGLE, [])
local exchange (REPARTITION, HASH, ["expr_1386", "expr_1387", "expr_1388"])
remote exchange (REPARTITION, HASH, ["i_brand_id_1163", "i_category_id_1167", "i_class_id_1165"])
partial hashaggregation over (i_brand_id_1163, i_category_id_1167, i_class_id_1165)
join (INNER, REPLICATED):
@ -279,6 +276,9 @@ local exchange (GATHER, SINGLE, [])
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan item
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["i_brand_id_1118", "i_category_id_1122", "i_class_id_1120"])
scan item
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
local exchange (GATHER, SINGLE, [])

21
presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q14_1.plan.txt
View File

@ -24,10 +24,8 @@ local exchange (GATHER, SINGLE, [])
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["i_item_sk_1"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["i_brand_id_8", "i_category_id_12", "i_class_id_10"])
scan item
final hashaggregation over (expr_207, expr_208, expr_209)
local exchange (GATHER, SINGLE, [])
local exchange (REPARTITION, HASH, ["expr_207", "expr_208", "expr_209"])
remote exchange (REPARTITION, HASH, ["i_brand_id_53", "i_category_id_57", "i_class_id_55"])
partial hashaggregation over (i_brand_id_53, i_category_id_57, i_class_id_55)
join (INNER, REPLICATED):
@ -61,6 +59,9 @@ local exchange (GATHER, SINGLE, [])
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan item
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["i_brand_id_8", "i_category_id_12", "i_class_id_10"])
scan item
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
final hashaggregation over ()
@ -103,10 +104,8 @@ local exchange (GATHER, SINGLE, [])
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["i_item_sk_518"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["i_brand_id_525", "i_category_id_529", "i_class_id_527"])
scan item
final hashaggregation over (expr_793, expr_794, expr_795)
local exchange (GATHER, SINGLE, [])
local exchange (REPARTITION, HASH, ["expr_793", "expr_794", "expr_795"])
remote exchange (REPARTITION, HASH, ["i_brand_id_570", "i_category_id_574", "i_class_id_572"])
partial hashaggregation over (i_brand_id_570, i_category_id_574, i_class_id_572)
join (INNER, REPLICATED):
@ -140,6 +139,9 @@ local exchange (GATHER, SINGLE, [])
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan item
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["i_brand_id_525", "i_category_id_529", "i_class_id_527"])
scan item
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
final hashaggregation over ()
@ -182,10 +184,8 @@ local exchange (GATHER, SINGLE, [])
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["i_item_sk_1111"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["i_brand_id_1118", "i_category_id_1122", "i_class_id_1120"])
scan item
final hashaggregation over (expr_1386, expr_1387, expr_1388)
local exchange (GATHER, SINGLE, [])
local exchange (REPARTITION, HASH, ["expr_1386", "expr_1387", "expr_1388"])
remote exchange (REPARTITION, HASH, ["i_brand_id_1163", "i_category_id_1167", "i_class_id_1165"])
partial hashaggregation over (i_brand_id_1163, i_category_id_1167, i_class_id_1165)
join (INNER, REPLICATED):
@ -219,6 +219,9 @@ local exchange (GATHER, SINGLE, [])
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan item
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["i_brand_id_1118", "i_category_id_1122", "i_class_id_1120"])
scan item
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
final hashaggregation over ()

21
presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q14_1.push.plan.txt
View File

@ -24,10 +24,8 @@ local exchange (GATHER, SINGLE, [])
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["i_item_sk_3"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["i_brand_id_10", "i_category_id_14", "i_class_id_12"])
scan item
final hashaggregation over (expr_219, expr_220, expr_221)
local exchange (GATHER, SINGLE, [])
local exchange (REPARTITION, HASH, ["expr_219", "expr_220", "expr_221"])
remote exchange (REPARTITION, HASH, ["i_brand_id_57", "i_category_id_61", "i_class_id_59"])
partial hashaggregation over (i_brand_id_57, i_category_id_61, i_class_id_59)
join (INNER, PARTITIONED):
@ -64,6 +62,9 @@ local exchange (GATHER, SINGLE, [])
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["i_item_sk_164"])
scan item
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["i_brand_id_10", "i_category_id_14", "i_class_id_12"])
scan item
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
final hashaggregation over ()
@ -106,10 +107,8 @@ local exchange (GATHER, SINGLE, [])
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["i_item_sk_539"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["i_brand_id_546", "i_category_id_550", "i_class_id_548"])
scan item
final hashaggregation over (expr_824, expr_825, expr_826)
local exchange (GATHER, SINGLE, [])
local exchange (REPARTITION, HASH, ["expr_824", "expr_825", "expr_826"])
remote exchange (REPARTITION, HASH, ["i_brand_id_593", "i_category_id_597", "i_class_id_595"])
partial hashaggregation over (i_brand_id_593, i_category_id_597, i_class_id_595)
join (INNER, PARTITIONED):
@ -146,6 +145,9 @@ local exchange (GATHER, SINGLE, [])
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["i_item_sk_769"])
scan item
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["i_brand_id_546", "i_category_id_550", "i_class_id_548"])
scan item
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
final hashaggregation over ()
@ -188,10 +190,8 @@ local exchange (GATHER, SINGLE, [])
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["i_item_sk_1151"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["i_brand_id_1158", "i_category_id_1162", "i_class_id_1160"])
scan item
final hashaggregation over (expr_1436, expr_1437, expr_1438)
local exchange (GATHER, SINGLE, [])
local exchange (REPARTITION, HASH, ["expr_1436", "expr_1437", "expr_1438"])
remote exchange (REPARTITION, HASH, ["i_brand_id_1205", "i_category_id_1209", "i_class_id_1207"])
partial hashaggregation over (i_brand_id_1205, i_category_id_1209, i_class_id_1207)
join (INNER, PARTITIONED):
@ -228,6 +228,9 @@ local exchange (GATHER, SINGLE, [])
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["i_item_sk_1381"])
scan item
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["i_brand_id_1158", "i_category_id_1162", "i_class_id_1160"])
scan item
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
final hashaggregation over ()

16
presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q14_2.cte.plan.txt
View File

@ -27,12 +27,9 @@ local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["i_item_sk_1"])
cte cross_items
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["i_brand_id_8", "i_category_id_12", "i_class_id_10"])
Filter
scan item
Filter
final hashaggregation over (expr_207, expr_208, expr_209)
local exchange (GATHER, SINGLE, [])
local exchange (REPARTITION, HASH, ["expr_207", "expr_208", "expr_209"])
remote exchange (REPARTITION, HASH, ["i_brand_id_53", "i_category_id_57", "i_class_id_55"])
partial hashaggregation over (i_brand_id_53, i_category_id_57, i_class_id_55)
join (INNER, REPLICATED):
@ -72,6 +69,9 @@ local exchange (GATHER, SINGLE, [])
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan item
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["i_brand_id_8", "i_category_id_12", "i_class_id_10"])
scan item
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
local exchange (GATHER, SINGLE, [])
@ -136,12 +136,9 @@ local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["i_item_sk_536"])
cte cross_items
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["i_brand_id_543", "i_category_id_547", "i_class_id_545"])
Filter
scan item
Filter
final hashaggregation over (expr_811, expr_812, expr_813)
local exchange (GATHER, SINGLE, [])
local exchange (REPARTITION, HASH, ["expr_811", "expr_812", "expr_813"])
remote exchange (REPARTITION, HASH, ["i_brand_id_588", "i_category_id_592", "i_class_id_590"])
partial hashaggregation over (i_brand_id_588, i_category_id_592, i_class_id_590)
join (INNER, REPLICATED):
@ -181,6 +178,9 @@ local exchange (GATHER, SINGLE, [])
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan item
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["i_brand_id_543", "i_category_id_547", "i_class_id_545"])
scan item
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
local exchange (GATHER, SINGLE, [])

14
presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q14_2.plan.txt
View File

@ -21,10 +21,8 @@ local exchange (GATHER, SINGLE, [])
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["i_item_sk_1"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["i_brand_id_8", "i_category_id_12", "i_class_id_10"])
scan item
final hashaggregation over (expr_207, expr_208, expr_209)
local exchange (GATHER, SINGLE, [])
local exchange (REPARTITION, HASH, ["expr_207", "expr_208", "expr_209"])
remote exchange (REPARTITION, HASH, ["i_brand_id_53", "i_category_id_57", "i_class_id_55"])
partial hashaggregation over (i_brand_id_53, i_category_id_57, i_class_id_55)
join (INNER, REPLICATED):
@ -58,6 +56,9 @@ local exchange (GATHER, SINGLE, [])
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan item
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["i_brand_id_8", "i_category_id_12", "i_class_id_10"])
scan item
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
local exchange (GATHER, SINGLE, [])
@ -106,10 +107,8 @@ local exchange (GATHER, SINGLE, [])
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["i_item_sk_536"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["i_brand_id_543", "i_category_id_547", "i_class_id_545"])
scan item
final hashaggregation over (expr_811, expr_812, expr_813)
local exchange (GATHER, SINGLE, [])
local exchange (REPARTITION, HASH, ["expr_811", "expr_812", "expr_813"])
remote exchange (REPARTITION, HASH, ["i_brand_id_588", "i_category_id_592", "i_class_id_590"])
partial hashaggregation over (i_brand_id_588, i_category_id_592, i_class_id_590)
join (INNER, REPLICATED):
@ -143,6 +142,9 @@ local exchange (GATHER, SINGLE, [])
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan item
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["i_brand_id_543", "i_category_id_547", "i_class_id_545"])
scan item
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
local exchange (GATHER, SINGLE, [])

14
presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q14_2.push.plan.txt
View File

@ -22,10 +22,8 @@ local exchange (GATHER, SINGLE, [])
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["i_item_sk_3"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["i_brand_id_10", "i_category_id_14", "i_class_id_12"])
scan item
final hashaggregation over (expr_219, expr_220, expr_221)
local exchange (GATHER, SINGLE, [])
local exchange (REPARTITION, HASH, ["expr_219", "expr_220", "expr_221"])
remote exchange (REPARTITION, HASH, ["i_brand_id_57", "i_category_id_61", "i_class_id_59"])
partial hashaggregation over (i_brand_id_57, i_category_id_61, i_class_id_59)
join (INNER, PARTITIONED):
@ -62,6 +60,9 @@ local exchange (GATHER, SINGLE, [])
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["i_item_sk_164"])
scan item
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["i_brand_id_10", "i_category_id_14", "i_class_id_12"])
scan item
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
local exchange (GATHER, SINGLE, [])
@ -111,10 +112,8 @@ local exchange (GATHER, SINGLE, [])
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["i_item_sk_558"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["i_brand_id_565", "i_category_id_569", "i_class_id_567"])
scan item
final hashaggregation over (expr_843, expr_844, expr_845)
local exchange (GATHER, SINGLE, [])
local exchange (REPARTITION, HASH, ["expr_843", "expr_844", "expr_845"])
remote exchange (REPARTITION, HASH, ["i_brand_id_612", "i_category_id_616", "i_class_id_614"])
partial hashaggregation over (i_brand_id_612, i_category_id_616, i_class_id_614)
join (INNER, PARTITIONED):
@ -151,6 +150,9 @@ local exchange (GATHER, SINGLE, [])
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["i_item_sk_788"])
scan item
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["i_brand_id_565", "i_category_id_569", "i_class_id_567"])
scan item
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
local exchange (GATHER, SINGLE, [])

27
presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q34.push.plan.txt
View File

@ -1,25 +1,24 @@
remote exchange (GATHER, SINGLE, [])
local exchange (GATHER, UNKNOWN, [])
remote exchange (REPARTITION, ROUND_ROBIN, [])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_customer_sk"])
final hashaggregation over (ss_customer_sk, ss_ticket_number)
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ss_customer_sk", "ss_ticket_number"])
partial hashaggregation over (ss_customer_sk, ss_ticket_number)
join (INNER, REPLICATED):
final hashaggregation over (ss_customer_sk, ss_ticket_number)
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ss_customer_sk", "ss_ticket_number"])
partial hashaggregation over (ss_customer_sk, ss_ticket_number)
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, REPLICATED):
scan store_sales
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan date_dim (pushdown = true)
scan store_sales
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan household_demographics (pushdown = true)
scan date_dim (pushdown = true)
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan store (pushdown = true)
scan household_demographics (pushdown = true)
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan store (pushdown = true)
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["c_customer_sk"])
remote exchange (REPLICATE, BROADCAST, [])
scan customer

82
presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q44.plan.txt
View File

@ -1,50 +1,48 @@
local exchange (GATHER, SINGLE, [])
remote exchange (GATHER, SINGLE, [])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_item_sk_47"])
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_item_sk"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["rank"])
local exchange (GATHER, SINGLE, [])
remote exchange (GATHER, SINGLE, [])
cross join:
final hashaggregation over (ss_item_sk)
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ss_item_sk"])
partial hashaggregation over (ss_item_sk)
scan store_sales
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
local exchange (GATHER, SINGLE, [])
remote exchange (GATHER, SINGLE, [])
final hashaggregation over (ss_store_sk_13)
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ss_store_sk_13"])
partial hashaggregation over (ss_store_sk_13)
scan store_sales
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["rank_111"])
remote exchange (REPARTITION, HASH, ["rank"])
local exchange (GATHER, SINGLE, [])
remote exchange (GATHER, SINGLE, [])
cross join:
final hashaggregation over (ss_item_sk)
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ss_item_sk"])
partial hashaggregation over (ss_item_sk)
scan store_sales
local exchange (GATHER, SINGLE, [])
remote exchange (GATHER, SINGLE, [])
cross join:
final hashaggregation over (ss_item_sk_47)
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ss_item_sk_47"])
partial hashaggregation over (ss_item_sk_47)
scan store_sales
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
remote exchange (REPLICATE, BROADCAST, [])
local exchange (GATHER, SINGLE, [])
remote exchange (GATHER, SINGLE, [])
final hashaggregation over (ss_store_sk_13)
local exchange (GATHER, SINGLE, [])
remote exchange (GATHER, SINGLE, [])
final hashaggregation over (ss_store_sk_84)
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ss_store_sk_84"])
partial hashaggregation over (ss_store_sk_84)
scan store_sales
remote exchange (REPARTITION, HASH, ["ss_store_sk_13"])
partial hashaggregation over (ss_store_sk_13)
scan store_sales
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["i_item_sk"])
scan item
remote exchange (REPARTITION, HASH, ["rank_111"])
local exchange (GATHER, SINGLE, [])
remote exchange (GATHER, SINGLE, [])
cross join:
final hashaggregation over (ss_item_sk_47)
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ss_item_sk_47"])
partial hashaggregation over (ss_item_sk_47)
scan store_sales
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
local exchange (GATHER, SINGLE, [])
remote exchange (GATHER, SINGLE, [])
final hashaggregation over (ss_store_sk_84)
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ss_store_sk_84"])
partial hashaggregation over (ss_store_sk_84)
scan store_sales
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan item
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["i_item_sk_120"])
remote exchange (REPLICATE, BROADCAST, [])
scan item

72
presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q44.push.plan.txt
View File

@ -1,46 +1,44 @@
local exchange (GATHER, SINGLE, [])
remote exchange (GATHER, SINGLE, [])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_item_sk_48"])
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_item_sk"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["rank"])
cross join:
local exchange (GATHER, SINGLE, [])
remote exchange (GATHER, SINGLE, [])
final hashaggregation over (ss_store_sk_13)
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ss_store_sk_13"])
partial hashaggregation over (ss_store_sk_13)
scan store_sales (pushdown = true)
local exchange (GATHER, SINGLE, [])
remote exchange (GATHER, SINGLE, [])
final hashaggregation over (ss_item_sk)
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ss_item_sk"])
partial hashaggregation over (ss_item_sk)
scan store_sales (pushdown = true)
remote exchange (REPARTITION, HASH, ["rank"])
cross join:
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["rank_114"])
cross join:
remote exchange (GATHER, SINGLE, [])
final hashaggregation over (ss_store_sk_13)
local exchange (GATHER, SINGLE, [])
remote exchange (GATHER, SINGLE, [])
final hashaggregation over (ss_store_sk_86)
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ss_store_sk_86"])
partial hashaggregation over (ss_store_sk_86)
scan store_sales (pushdown = true)
remote exchange (REPARTITION, HASH, ["ss_store_sk_13"])
partial hashaggregation over (ss_store_sk_13)
scan store_sales (pushdown = true)
local exchange (GATHER, SINGLE, [])
remote exchange (GATHER, SINGLE, [])
final hashaggregation over (ss_item_sk)
local exchange (GATHER, SINGLE, [])
remote exchange (GATHER, SINGLE, [])
final hashaggregation over (ss_item_sk_48)
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ss_item_sk_48"])
partial hashaggregation over (ss_item_sk_48)
scan store_sales (pushdown = true)
remote exchange (REPARTITION, HASH, ["ss_item_sk"])
partial hashaggregation over (ss_item_sk)
scan store_sales (pushdown = true)
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["i_item_sk"])
scan item
remote exchange (REPARTITION, HASH, ["rank_114"])
cross join:
local exchange (GATHER, SINGLE, [])
remote exchange (GATHER, SINGLE, [])
final hashaggregation over (ss_store_sk_86)
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ss_store_sk_86"])
partial hashaggregation over (ss_store_sk_86)
scan store_sales (pushdown = true)
local exchange (GATHER, SINGLE, [])
remote exchange (GATHER, SINGLE, [])
final hashaggregation over (ss_item_sk_48)
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ss_item_sk_48"])
partial hashaggregation over (ss_item_sk_48)
scan store_sales (pushdown = true)
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan item
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["i_item_sk_124"])
remote exchange (REPLICATE, BROADCAST, [])
scan item

72
presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q44.reuse.plan.txt
View File

@ -1,46 +1,44 @@
local exchange (GATHER, SINGLE, [])
remote exchange (GATHER, SINGLE, [])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_item_sk_48"])
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_item_sk"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["rank"])
cross join:
local exchange (GATHER, SINGLE, [])
remote exchange (GATHER, SINGLE, [])
final hashaggregation over (ss_store_sk_13)
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ss_store_sk_13"])
partial hashaggregation over (ss_store_sk_13)
scan store_sales
local exchange (GATHER, SINGLE, [])
remote exchange (GATHER, SINGLE, [])
final hashaggregation over (ss_item_sk)
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ss_item_sk"])
partial hashaggregation over (ss_item_sk)
ReuseTableScan store_sales (Consumer)
remote exchange (REPARTITION, HASH, ["rank"])
cross join:
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["rank_114"])
cross join:
remote exchange (GATHER, SINGLE, [])
final hashaggregation over (ss_store_sk_13)
local exchange (GATHER, SINGLE, [])
remote exchange (GATHER, SINGLE, [])
final hashaggregation over (ss_store_sk_86)
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ss_store_sk_86"])
partial hashaggregation over (ss_store_sk_86)
scan store_sales
remote exchange (REPARTITION, HASH, ["ss_store_sk_13"])
partial hashaggregation over (ss_store_sk_13)
scan store_sales
local exchange (GATHER, SINGLE, [])
remote exchange (GATHER, SINGLE, [])
final hashaggregation over (ss_item_sk)
local exchange (GATHER, SINGLE, [])
remote exchange (GATHER, SINGLE, [])
final hashaggregation over (ss_item_sk_48)
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ss_item_sk_48"])
partial hashaggregation over (ss_item_sk_48)
ReuseTableScan store_sales (Producer)
remote exchange (REPARTITION, HASH, ["ss_item_sk"])
partial hashaggregation over (ss_item_sk)
ReuseTableScan store_sales (Consumer)
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["i_item_sk"])
scan item
remote exchange (REPARTITION, HASH, ["rank_114"])
cross join:
local exchange (GATHER, SINGLE, [])
remote exchange (GATHER, SINGLE, [])
final hashaggregation over (ss_store_sk_86)
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ss_store_sk_86"])
partial hashaggregation over (ss_store_sk_86)
scan store_sales
local exchange (GATHER, SINGLE, [])
remote exchange (GATHER, SINGLE, [])
final hashaggregation over (ss_item_sk_48)
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ss_item_sk_48"])
partial hashaggregation over (ss_item_sk_48)
ReuseTableScan store_sales (Producer)
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan item
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["i_item_sk_124"])
remote exchange (REPLICATE, BROADCAST, [])
scan item

338
presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q64.cte.plan.txt
View File

@ -7,110 +7,98 @@ remote exchange (GATHER, SINGLE, [])
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ca_city", "ca_city_92", "ca_street_name", "ca_street_name_89", "ca_street_number", "ca_street_number_88", "ca_zip", "ca_zip_95", "d_year", "d_year_22", "d_year_50", "i_product_name", "s_store_name", "s_zip", "ss_item_sk"])
partial hashaggregation over (ca_city, ca_city_92, ca_street_name, ca_street_name_89, ca_street_number, ca_street_number_88, ca_zip, ca_zip_95, d_year, d_year_22, d_year_50, i_product_name, s_store_name, s_zip, ss_item_sk)
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_item_sk"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["hd_income_band_sk_82"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["hd_income_band_sk"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["c_current_addr_sk"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_addr_sk"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["c_current_hdemo_sk"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_hdemo_sk"])
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["c_current_addr_sk"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_addr_sk"])
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_promo_sk"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["c_current_cdemo_sk"])
remote exchange (REPARTITION, HASH, ["ss_customer_sk"])
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_cdemo_sk"])
remote exchange (REPARTITION, HASH, ["ss_item_sk"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["c_first_shipto_date_sk"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["c_first_sales_date_sk"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_customer_sk"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_store_sk"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_sold_date_sk"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_item_sk"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_item_sk", "ss_ticket_number"])
Filter
scan store_sales
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["sr_item_sk", "sr_ticket_number"])
Filter
scan store_returns
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["cs_item_sk"])
remote exchange (REPARTITION, HASH, ["cs_item_sk", "sum_5", "sum_6"])
cte cs_ui
Filter
final hashaggregation over (cs_item_sk)
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["cs_item_sk"])
partial hashaggregation over (cs_item_sk)
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["cs_item_sk", "cs_order_number"])
Filter
scan catalog_sales
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["cr_item_sk", "cr_order_number"])
Filter
scan catalog_returns
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["d_date_sk"])
Filter
scan date_dim
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["s_store_sk"])
Filter
scan store
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["c_customer_sk"])
Filter
scan customer
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["d_date_sk_16"])
scan date_dim
remote exchange (REPARTITION, HASH, ["ss_item_sk", "ss_ticket_number"])
Filter
scan store_sales
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["d_date_sk_44"])
scan date_dim
remote exchange (REPARTITION, HASH, ["sr_item_sk", "sr_ticket_number"])
Filter
scan store_returns
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["cd_demo_sk"])
scan customer_demographics
remote exchange (REPARTITION, HASH, ["cs_item_sk"])
remote exchange (REPARTITION, HASH, ["cs_item_sk", "sum_5", "sum_6"])
cte cs_ui
Filter
final hashaggregation over (cs_item_sk)
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["cs_item_sk"])
partial hashaggregation over (cs_item_sk)
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["cs_item_sk", "cs_order_number"])
Filter
scan catalog_sales
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["cr_item_sk", "cr_order_number"])
Filter
scan catalog_returns
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
Filter
scan date_dim
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["cd_demo_sk_72"])
scan customer_demographics
remote exchange (REPLICATE, BROADCAST, [])
Filter
scan store
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["p_promo_sk"])
scan promotion
remote exchange (REPARTITION, HASH, ["c_customer_sk"])
Filter
scan customer
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan date_dim
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["hd_demo_sk"])
scan household_demographics
remote exchange (REPLICATE, BROADCAST, [])
scan date_dim
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan customer_demographics
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["hd_demo_sk_81"])
scan household_demographics
remote exchange (REPLICATE, BROADCAST, [])
scan customer_demographics
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan promotion
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ca_address_sk"])
scan customer_address
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ca_address_sk_86"])
scan customer_address
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ib_income_band_sk"])
scan income_band
remote exchange (REPLICATE, BROADCAST, [])
scan household_demographics
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan household_demographics
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ca_address_sk"])
scan customer_address
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ca_address_sk_86"])
scan customer_address
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ib_income_band_sk_99"])
remote exchange (REPLICATE, BROADCAST, [])
scan income_band
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan income_band
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["i_item_sk"])
remote exchange (REPLICATE, BROADCAST, [])
Filter
scan item
local exchange (GATHER, SINGLE, [])
@ -119,108 +107,96 @@ remote exchange (GATHER, SINGLE, [])
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ca_city_497", "ca_city_510", "ca_street_name_494", "ca_street_name_507", "ca_street_number_493", "ca_street_number_506", "ca_zip_500", "ca_zip_513", "d_year_319", "d_year_347", "d_year_375", "i_product_name_544", "s_store_name_402", "s_zip_422", "ss_item_sk_192"])
partial hashaggregation over (ca_city_497, ca_city_510, ca_street_name_494, ca_street_name_507, ca_street_number_493, ca_street_number_506, ca_zip_500, ca_zip_513, d_year_319, d_year_347, d_year_375, i_product_name_544, s_store_name_402, s_zip_422, ss_item_sk_192)
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_item_sk_192"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["hd_income_band_sk_487"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["hd_income_band_sk_482"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["c_current_addr_sk_430"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_addr_sk_196"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["c_current_hdemo_sk_429"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_hdemo_sk_195"])
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["c_current_addr_sk_430"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_addr_sk_196"])
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_promo_sk_198"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["c_current_cdemo_sk_428"])
remote exchange (REPARTITION, HASH, ["ss_customer_sk_193"])
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_cdemo_sk_194"])
remote exchange (REPARTITION, HASH, ["ss_item_sk_192"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["c_first_shipto_date_sk_431"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["c_first_sales_date_sk_432"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_customer_sk_193"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_store_sk_197"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_sold_date_sk_190"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_item_sk_192"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_item_sk_192", "ss_ticket_number_199"])
Filter
scan store_sales
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["sr_item_sk_215", "sr_ticket_number_222"])
Filter
scan store_returns
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["cs_item_sk_248"])
remote exchange (REPARTITION, HASH, ["cs_item_sk_248", "sum_302", "sum_303"])
cte cs_ui
Filter
final hashaggregation over (cs_item_sk_248)
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["cs_item_sk_248"])
partial hashaggregation over (cs_item_sk_248)
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["cs_item_sk_248", "cs_order_number_250"])
Filter
scan catalog_sales
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["cr_item_sk_269", "cr_order_number_283"])
Filter
scan catalog_returns
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["d_date_sk_313"])
Filter
scan date_dim
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["s_store_sk_397"])
scan store
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["c_customer_sk_426"])
Filter
scan customer
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["d_date_sk_341"])
scan date_dim
remote exchange (REPARTITION, HASH, ["ss_item_sk_192", "ss_ticket_number_199"])
Filter
scan store_sales
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["d_date_sk_369"])
scan date_dim
remote exchange (REPARTITION, HASH, ["sr_item_sk_215", "sr_ticket_number_222"])
Filter
scan store_returns
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["cd_demo_sk_444"])
scan customer_demographics
remote exchange (REPARTITION, HASH, ["cs_item_sk_248"])
remote exchange (REPARTITION, HASH, ["cs_item_sk_248", "sum_302", "sum_303"])
cte cs_ui
Filter
final hashaggregation over (cs_item_sk_248)
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["cs_item_sk_248"])
partial hashaggregation over (cs_item_sk_248)
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["cs_item_sk_248", "cs_order_number_250"])
Filter
scan catalog_sales
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["cr_item_sk_269", "cr_order_number_283"])
Filter
scan catalog_returns
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
Filter
scan date_dim
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["cd_demo_sk_453"])
scan customer_demographics
remote exchange (REPLICATE, BROADCAST, [])
scan store
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["p_promo_sk_462"])
scan promotion
remote exchange (REPARTITION, HASH, ["c_customer_sk_426"])
Filter
scan customer
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan date_dim
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["hd_demo_sk_481"])
scan household_demographics
remote exchange (REPLICATE, BROADCAST, [])
scan date_dim
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan customer_demographics
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["hd_demo_sk_486"])
scan household_demographics
remote exchange (REPLICATE, BROADCAST, [])
scan customer_demographics
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan promotion
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ca_address_sk_491"])
scan customer_address
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ca_address_sk_504"])
scan customer_address
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ib_income_band_sk_517"])
scan income_band
remote exchange (REPLICATE, BROADCAST, [])
scan household_demographics
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan household_demographics
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ca_address_sk_491"])
scan customer_address
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ca_address_sk_504"])
scan customer_address
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ib_income_band_sk_520"])
remote exchange (REPLICATE, BROADCAST, [])
scan income_band
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan income_band
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["i_item_sk_523"])
remote exchange (REPLICATE, BROADCAST, [])
Filter
scan item

299
presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q64.plan.txt
View File

@ -2,182 +2,163 @@ remote exchange (GATHER, SINGLE, [])
local exchange (GATHER, UNKNOWN, [])
remote exchange (REPARTITION, ROUND_ROBIN, [])
join (INNER, PARTITIONED):
final hashaggregation over (ca_city, ca_city_92, ca_street_name, ca_street_name_89, ca_street_number, ca_street_number_88, ca_zip, ca_zip_95, cs_item_sk, d_year, d_year_22, d_year_50, i_product_name, s_store_name, s_zip)
local exchange (GATHER, SINGLE, [])
partial hashaggregation over (ca_city, ca_city_92, ca_street_name, ca_street_name_89, ca_street_number, ca_street_number_88, ca_zip, ca_zip_95, cs_item_sk, d_year, d_year_22, d_year_50, i_product_name, s_store_name, s_zip)
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["cs_item_sk"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["hd_income_band_sk_82"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["hd_income_band_sk"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["c_current_addr_sk"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_addr_sk"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["c_current_hdemo_sk"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_hdemo_sk"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_promo_sk"])
remote exchange (REPARTITION, HASH, ["cs_item_sk", "s_store_name", "s_zip"])
final hashaggregation over (ca_city, ca_city_92, ca_street_name, ca_street_name_89, ca_street_number, ca_street_number_88, ca_zip, ca_zip_95, cs_item_sk, d_year, d_year_22, d_year_50, i_product_name, s_store_name, s_zip)
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ca_city", "ca_city_92", "ca_street_name", "ca_street_name_89", "ca_street_number", "ca_street_number_88", "ca_zip", "ca_zip_95", "cs_item_sk", "d_year", "d_year_22", "d_year_50", "i_product_name", "s_store_name", "s_zip"])
partial hashaggregation over (ca_city, ca_city_92, ca_street_name, ca_street_name_89, ca_street_number, ca_street_number_88, ca_zip, ca_zip_95, cs_item_sk, d_year, d_year_22, d_year_50, i_product_name, s_store_name, s_zip)
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["c_current_addr_sk"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_addr_sk"])
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["c_current_cdemo_sk"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_cdemo_sk"])
remote exchange (REPARTITION, HASH, ["ss_customer_sk"])
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["c_first_shipto_date_sk"])
remote exchange (REPARTITION, HASH, ["ss_item_sk"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["c_first_sales_date_sk"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_customer_sk"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_store_sk"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_sold_date_sk"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_item_sk"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_item_sk", "ss_ticket_number"])
scan store_sales
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["sr_item_sk", "sr_ticket_number"])
scan store_returns
final hashaggregation over (cs_item_sk)
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["cs_item_sk"])
partial hashaggregation over (cs_item_sk)
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["cs_item_sk", "cs_order_number"])
scan catalog_sales
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["cr_item_sk", "cr_order_number"])
scan catalog_returns
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["d_date_sk"])
scan date_dim
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["s_store_sk"])
scan store
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["c_customer_sk"])
scan customer
remote exchange (REPARTITION, HASH, ["ss_item_sk", "ss_ticket_number"])
scan store_sales
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["d_date_sk_16"])
scan date_dim
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["d_date_sk_44"])
scan date_dim
remote exchange (REPARTITION, HASH, ["sr_item_sk", "sr_ticket_number"])
scan store_returns
final hashaggregation over (cs_item_sk)
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["cs_item_sk"])
partial hashaggregation over (cs_item_sk)
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["cs_item_sk", "cs_order_number"])
scan catalog_sales
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["cr_item_sk", "cr_order_number"])
scan catalog_returns
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan date_dim
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["cd_demo_sk"])
scan customer_demographics
remote exchange (REPLICATE, BROADCAST, [])
scan store
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["cd_demo_sk_72"])
scan customer_demographics
remote exchange (REPARTITION, HASH, ["c_customer_sk"])
scan customer
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan date_dim
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["p_promo_sk"])
scan promotion
remote exchange (REPLICATE, BROADCAST, [])
scan date_dim
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan customer_demographics
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["hd_demo_sk"])
scan household_demographics
remote exchange (REPLICATE, BROADCAST, [])
scan customer_demographics
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan promotion
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["hd_demo_sk_81"])
remote exchange (REPLICATE, BROADCAST, [])
scan household_demographics
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ca_address_sk"])
scan customer_address
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ca_address_sk_86"])
scan customer_address
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan household_demographics
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ca_address_sk"])
scan customer_address
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ca_address_sk_86"])
scan customer_address
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ib_income_band_sk"])
remote exchange (REPLICATE, BROADCAST, [])
scan income_band
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan income_band
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ib_income_band_sk_99"])
scan income_band
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["i_item_sk"])
scan item
final hashaggregation over (ca_city_497, ca_city_510, ca_street_name_494, ca_street_name_507, ca_street_number_493, ca_street_number_506, ca_zip_500, ca_zip_513, cs_item_sk_248, d_year_319, d_year_347, d_year_375, i_product_name_544, s_store_name_402, s_zip_422)
local exchange (GATHER, SINGLE, [])
partial hashaggregation over (ca_city_497, ca_city_510, ca_street_name_494, ca_street_name_507, ca_street_number_493, ca_street_number_506, ca_zip_500, ca_zip_513, cs_item_sk_248, d_year_319, d_year_347, d_year_375, i_product_name_544, s_store_name_402, s_zip_422)
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["cs_item_sk_248"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["hd_income_band_sk_487"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["hd_income_band_sk_482"])
remote exchange (REPLICATE, BROADCAST, [])
scan item
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["cs_item_sk_248", "s_store_name_402", "s_zip_422"])
final hashaggregation over (ca_city_497, ca_city_510, ca_street_name_494, ca_street_name_507, ca_street_number_493, ca_street_number_506, ca_zip_500, ca_zip_513, cs_item_sk_248, d_year_319, d_year_347, d_year_375, i_product_name_544, s_store_name_402, s_zip_422)
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ca_city_497", "ca_city_510", "ca_street_name_494", "ca_street_name_507", "ca_street_number_493", "ca_street_number_506", "ca_zip_500", "ca_zip_513", "cs_item_sk_248", "d_year_319", "d_year_347", "d_year_375", "i_product_name_544", "s_store_name_402", "s_zip_422"])
partial hashaggregation over (ca_city_497, ca_city_510, ca_street_name_494, ca_street_name_507, ca_street_number_493, ca_street_number_506, ca_zip_500, ca_zip_513, cs_item_sk_248, d_year_319, d_year_347, d_year_375, i_product_name_544, s_store_name_402, s_zip_422)
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["c_current_addr_sk_430"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_addr_sk_196"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["c_current_hdemo_sk_429"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_hdemo_sk_195"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_promo_sk_198"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["c_current_cdemo_sk_428"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_cdemo_sk_194"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["c_first_shipto_date_sk_431"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["c_first_sales_date_sk_432"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_customer_sk_193"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_store_sk_197"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_sold_date_sk_190"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_item_sk_192"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_item_sk_192", "ss_ticket_number_199"])
scan store_sales
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["sr_item_sk_215", "sr_ticket_number_222"])
scan store_returns
final hashaggregation over (cs_item_sk_248)
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["cs_item_sk_248"])
partial hashaggregation over (cs_item_sk_248)
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["cs_item_sk_248", "cs_order_number_250"])
scan catalog_sales
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["cr_item_sk_269", "cr_order_number_283"])
scan catalog_returns
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["d_date_sk_313"])
scan date_dim
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["s_store_sk_397"])
scan store
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["c_customer_sk_426"])
scan customer
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_customer_sk_193"])
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_item_sk_192"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_item_sk_192", "ss_ticket_number_199"])
scan store_sales
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["sr_item_sk_215", "sr_ticket_number_222"])
scan store_returns
final hashaggregation over (cs_item_sk_248)
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["d_date_sk_341"])
scan date_dim
remote exchange (REPARTITION, HASH, ["cs_item_sk_248"])
partial hashaggregation over (cs_item_sk_248)
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["cs_item_sk_248", "cs_order_number_250"])
scan catalog_sales
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["cr_item_sk_269", "cr_order_number_283"])
scan catalog_returns
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["d_date_sk_369"])
remote exchange (REPLICATE, BROADCAST, [])
scan date_dim
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["cd_demo_sk_444"])
scan customer_demographics
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan store
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["c_customer_sk_426"])
scan customer
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["cd_demo_sk_453"])
scan customer_demographics
remote exchange (REPLICATE, BROADCAST, [])
scan date_dim
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan date_dim
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["p_promo_sk_462"])
scan promotion
remote exchange (REPLICATE, BROADCAST, [])
scan customer_demographics
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan customer_demographics
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["hd_demo_sk_481"])
scan household_demographics
remote exchange (REPLICATE, BROADCAST, [])
scan promotion
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan household_demographics
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["hd_demo_sk_486"])
remote exchange (REPLICATE, BROADCAST, [])
scan household_demographics
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ca_address_sk_491"])
@ -185,12 +166,12 @@ remote exchange (GATHER, SINGLE, [])
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ca_address_sk_504"])
scan customer_address
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan income_band
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ib_income_band_sk_517"])
remote exchange (REPLICATE, BROADCAST, [])
scan income_band
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ib_income_band_sk_520"])
scan income_band
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["i_item_sk_523"])
scan item
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan item

286
presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q64.push.plan.txt
View File

@ -6,94 +6,79 @@ remote exchange (GATHER, SINGLE, [])
local exchange (GATHER, SINGLE, [])
partial hashaggregation over (ca_city, ca_city_106, ca_street_name, ca_street_name_103, ca_street_number, ca_street_number_102, ca_zip, ca_zip_109, d_year, d_year_26, d_year_55, i_item_sk, i_product_name, s_store_name, s_zip)
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["cs_item_sk"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["hd_income_band_sk_94"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["hd_income_band_sk"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["c_current_addr_sk"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_addr_sk"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["c_current_hdemo_sk"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_hdemo_sk"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_promo_sk"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["c_current_cdemo_sk"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_cdemo_sk"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["c_first_shipto_date_sk"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["c_first_sales_date_sk"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_customer_sk"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_store_sk"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_sold_date_sk"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["sr_item_sk"])
join (INNER, REPLICATED):
scan store_sales
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan store_returns
final hashaggregation over (cs_item_sk)
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["cs_item_sk"])
partial hashaggregation over (cs_item_sk)
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["cs_item_sk", "cs_order_number"])
scan catalog_sales
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["cr_item_sk", "cr_order_number"])
scan catalog_returns
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["d_date_sk"])
scan date_dim (pushdown = true)
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["s_store_sk"])
scan store
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["c_customer_sk"])
scan customer
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["d_date_sk_20"])
scan date_dim
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["d_date_sk_49"])
scan date_dim
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["cd_demo_sk"])
scan customer_demographics
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["sr_item_sk"])
join (INNER, REPLICATED):
scan store_sales
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan store_returns
final hashaggregation over (cs_item_sk)
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["cd_demo_sk_81"])
scan customer_demographics
remote exchange (REPARTITION, HASH, ["cs_item_sk"])
partial hashaggregation over (cs_item_sk)
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["cs_item_sk", "cs_order_number"])
scan catalog_sales
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["cr_item_sk", "cr_order_number"])
scan catalog_returns
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["p_promo_sk"])
scan promotion
remote exchange (REPLICATE, BROADCAST, [])
scan date_dim (pushdown = true)
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan store
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["hd_demo_sk"])
scan household_demographics
remote exchange (REPLICATE, BROADCAST, [])
scan customer
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan date_dim
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["hd_demo_sk_93"])
scan household_demographics
remote exchange (REPLICATE, BROADCAST, [])
scan date_dim
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan customer_demographics
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ca_address_sk"])
scan customer_address
remote exchange (REPLICATE, BROADCAST, [])
scan customer_demographics
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan promotion
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ca_address_sk_100"])
scan customer_address
remote exchange (REPLICATE, BROADCAST, [])
scan household_demographics
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan household_demographics
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ib_income_band_sk"])
scan income_band
remote exchange (REPLICATE, BROADCAST, [])
scan customer_address
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan customer_address
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ib_income_band_sk_115"])
remote exchange (REPLICATE, BROADCAST, [])
scan income_band
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan income_band
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["i_item_sk"])
scan item (pushdown = true)
@ -101,94 +86,79 @@ remote exchange (GATHER, SINGLE, [])
local exchange (GATHER, SINGLE, [])
partial hashaggregation over (ca_city_529, ca_city_543, ca_street_name_526, ca_street_name_540, ca_street_number_525, ca_street_number_539, ca_zip_532, ca_zip_546, d_year_341, d_year_370, d_year_399, i_item_sk_559, i_product_name_580, s_store_name_427, s_zip_447)
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["cs_item_sk_268"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["hd_income_band_sk_518"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["hd_income_band_sk_512"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["c_current_addr_sk_456"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_addr_sk_214"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["c_current_hdemo_sk_455"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_hdemo_sk_213"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_promo_sk_216"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["c_current_cdemo_sk_454"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_cdemo_sk_212"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["c_first_shipto_date_sk_457"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["c_first_sales_date_sk_458"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_customer_sk_211"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_store_sk_215"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_sold_date_sk_208"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["sr_item_sk_234"])
join (INNER, REPLICATED):
scan store_sales
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan store_returns
final hashaggregation over (cs_item_sk_268)
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["cs_item_sk_268"])
partial hashaggregation over (cs_item_sk_268)
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["cs_item_sk_268", "cs_order_number_270"])
scan catalog_sales
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["cr_item_sk_290", "cr_order_number_304"])
scan catalog_returns
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["d_date_sk_335"])
scan date_dim (pushdown = true)
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["s_store_sk_422"])
scan store
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["c_customer_sk_452"])
scan customer
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["d_date_sk_364"])
scan date_dim
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["d_date_sk_393"])
scan date_dim
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["cd_demo_sk_471"])
scan customer_demographics
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["sr_item_sk_234"])
join (INNER, REPLICATED):
scan store_sales
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan store_returns
final hashaggregation over (cs_item_sk_268)
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["cd_demo_sk_481"])
scan customer_demographics
remote exchange (REPARTITION, HASH, ["cs_item_sk_268"])
partial hashaggregation over (cs_item_sk_268)
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["cs_item_sk_268", "cs_order_number_270"])
scan catalog_sales
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["cr_item_sk_290", "cr_order_number_304"])
scan catalog_returns
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["p_promo_sk_491"])
scan promotion
remote exchange (REPLICATE, BROADCAST, [])
scan date_dim (pushdown = true)
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan store
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["hd_demo_sk_511"])
scan household_demographics
remote exchange (REPLICATE, BROADCAST, [])
scan customer
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan date_dim
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["hd_demo_sk_517"])
scan household_demographics
remote exchange (REPLICATE, BROADCAST, [])
scan date_dim
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan customer_demographics
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ca_address_sk_523"])
scan customer_address
remote exchange (REPLICATE, BROADCAST, [])
scan customer_demographics
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan promotion
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ca_address_sk_537"])
scan customer_address
remote exchange (REPLICATE, BROADCAST, [])
scan household_demographics
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan household_demographics
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ib_income_band_sk_551"])
scan income_band
remote exchange (REPLICATE, BROADCAST, [])
scan customer_address
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan customer_address
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ib_income_band_sk_555"])
remote exchange (REPLICATE, BROADCAST, [])
scan income_band
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan income_band
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["i_item_sk_559"])
scan item (pushdown = true)

39
presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q73.push.plan.txt
View File

@ -1,26 +1,25 @@
remote exchange (GATHER, SINGLE, [])
local exchange (GATHER, UNKNOWN, [])
remote exchange (REPARTITION, ROUND_ROBIN, [])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_customer_sk"])
final hashaggregation over (ss_customer_sk, ss_ticket_number)
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ss_customer_sk", "ss_ticket_number"])
partial hashaggregation over (ss_customer_sk, ss_ticket_number)
join (INNER, REPLICATED):
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_hdemo_sk"])
join (INNER, REPLICATED):
scan store_sales
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan date_dim (pushdown = true)
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["hd_demo_sk"])
scan household_demographics (pushdown = true)
join (INNER, REPLICATED):
final hashaggregation over (ss_customer_sk, ss_ticket_number)
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ss_customer_sk", "ss_ticket_number"])
partial hashaggregation over (ss_customer_sk, ss_ticket_number)
join (INNER, REPLICATED):
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["ss_hdemo_sk"])
join (INNER, REPLICATED):
scan store_sales
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan date_dim (pushdown = true)
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan store (pushdown = true)
remote exchange (REPARTITION, HASH, ["hd_demo_sk"])
scan household_demographics (pushdown = true)
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan store (pushdown = true)
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["c_customer_sk"])
remote exchange (REPLICATE, BROADCAST, [])
scan customer

31
presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q95.push.plan.txt
View File

@ -4,8 +4,8 @@ final hashaggregation over ()
partial hashaggregation over ()
local exchange (GATHER, SINGLE, [])
semijoin (PARTITIONED):
semijoin (PARTITIONED):
remote exchange (REPARTITION, HASH, ["ws_order_number"])
remote exchange (REPARTITION, HASH, ["ws_order_number"])
semijoin (REPLICATED):
join (INNER, REPLICATED):
join (INNER, REPLICATED):
join (INNER, REPLICATED):
@ -19,18 +19,19 @@ final hashaggregation over ()
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan web_site (pushdown = true)
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ws_order_number_20"])
single hashaggregation over (ws_order_number_20)
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ws_order_number_20"])
scan web_sales
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
single hashaggregation over (ws_order_number_20)
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ws_order_number_20"])
scan web_sales
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["wr_order_number"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["wr_order_number"])
scan web_returns
single hashaggregation over (ws_order_number_105)
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ws_order_number_105"])
scan web_sales
join (INNER, REPLICATED):
scan web_returns
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
single hashaggregation over (ws_order_number_105)
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["ws_order_number_105"])
scan web_sales

41
presto-benchto-benchmarks/src/test/resources/sql/presto/tpch/q09.plan.txt
View File

@ -5,28 +5,27 @@ remote exchange (GATHER, SINGLE, [])
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["expr_18", "name_15"])
partial hashaggregation over (expr_18, name_15)
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["nationkey"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["orderkey"])
join (INNER, REPLICATED):
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["orderkey"])
join (INNER, PARTITIONED):
join (INNER, PARTITIONED):
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["suppkey_4"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["partkey"])
scan part
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["partkey_3"])
scan lineitem
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["suppkey"])
scan supplier
remote exchange (REPARTITION, HASH, ["suppkey_4"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["partkey"])
scan part
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["partkey_3"])
scan lineitem
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["suppkey_8"])
scan partsupp
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["orderkey_11"])
scan orders
remote exchange (REPARTITION, HASH, ["suppkey"])
scan supplier
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["suppkey_8"])
scan partsupp
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["orderkey_11"])
scan orders
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["nationkey_14"])
remote exchange (REPLICATE, BROADCAST, [])
scan nation

24
presto-benchto-benchmarks/src/test/resources/sql/presto/tpch/q21.plan.txt
View File

@ -10,23 +10,21 @@ local exchange (GATHER, SINGLE, [])
local exchange (GATHER, SINGLE, [])
partial hashaggregation over (commitdate, name, name_7, nationkey, orderkey, orderstatus, receiptdate, suppkey, unique_123)
join (LEFT, PARTITIONED):
remote exchange (REPARTITION, HASH, ["orderkey"])
join (INNER, REPLICATED):
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["nationkey"])
remote exchange (REPARTITION, HASH, ["orderkey"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["orderkey"])
join (INNER, PARTITIONED):
remote exchange (REPARTITION, HASH, ["suppkey"])
scan supplier
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["suppkey_0"])
scan lineitem
remote exchange (REPARTITION, HASH, ["suppkey"])
scan supplier
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["orderkey_3"])
scan orders
remote exchange (REPARTITION, HASH, ["suppkey_0"])
scan lineitem
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["nationkey_6"])
scan nation
remote exchange (REPARTITION, HASH, ["orderkey_3"])
scan orders
local exchange (GATHER, SINGLE, [])
remote exchange (REPLICATE, BROADCAST, [])
scan nation
local exchange (GATHER, SINGLE, [])
remote exchange (REPARTITION, HASH, ["orderkey_10"])
scan lineitem

68
presto-main/src/main/java/io/prestosql/sql/planner/iterative/rule/DetermineJoinDistributionType.java
View File

@ -14,6 +14,7 @@
package io.prestosql.sql.planner.iterative.rule;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Ordering;
import io.airlift.units.DataSize;
import io.prestosql.cost.CostComparator;
@ -25,9 +26,14 @@ import io.prestosql.matching.Captures;
import io.prestosql.matching.Pattern;
import io.prestosql.spi.plan.JoinNode;
import io.prestosql.spi.plan.PlanNode;
import io.prestosql.spi.plan.TableScanNode;
import io.prestosql.spi.plan.ValuesNode;
import io.prestosql.sql.analyzer.FeaturesConfig.JoinDistributionType;
import io.prestosql.sql.planner.TypeProvider;
import io.prestosql.sql.planner.iterative.Lookup;
import io.prestosql.sql.planner.iterative.Rule;
import io.prestosql.sql.planner.optimizations.PlanNodeSearcher;
import io.prestosql.sql.tree.Unnest;
import java.util.ArrayList;
import java.util.List;
@ -45,6 +51,7 @@ import static io.prestosql.spi.plan.JoinNode.Type.RIGHT;
import static io.prestosql.sql.analyzer.FeaturesConfig.JoinDistributionType.AUTOMATIC;
import static io.prestosql.sql.planner.optimizations.QueryCardinalityUtil.isAtMostScalar;
import static io.prestosql.sql.planner.plan.Patterns.join;
import static io.prestosql.util.MorePredicates.isInstanceOfAny;
import static java.util.Objects.requireNonNull;
public class DetermineJoinDistributionType
@ -92,7 +99,32 @@ public class DetermineJoinDistributionType
PlanNode buildSide = joinNode.getRight();
PlanNodeStatsEstimate buildSideStatsEstimate = context.getStatsProvider().getStats(buildSide);
double buildSideSizeInBytes = buildSideStatsEstimate.getOutputSizeInBytes(buildSide.getOutputSymbols(), context.getSymbolAllocator().getTypes());
return buildSideSizeInBytes <= joinMaxBroadcastTableSize.get().toBytes();
return buildSideSizeInBytes <= joinMaxBroadcastTableSize.get().toBytes()
|| getSourceTablesSizeInBytes(buildSide, context) <= joinMaxBroadcastTableSize.get().toBytes();
}
public static double getSourceTablesSizeInBytes(PlanNode node, Context context)
{
return getSourceTablesSizeInBytes(node, context.getLookup(), context.getStatsProvider(), context.getSymbolAllocator().getTypes());
}
@VisibleForTesting
static double getSourceTablesSizeInBytes(PlanNode node, Lookup lookup, StatsProvider statsProvider, TypeProvider typeProvider)
{
boolean hasExpandingNodes = PlanNodeSearcher.searchFrom(node, lookup)
.where(isInstanceOfAny(JoinNode.class, Unnest.class))
.matches();
if (hasExpandingNodes) {
return Double.NaN;
}
List<PlanNode> sourceNodes = PlanNodeSearcher.searchFrom(node, lookup)
.where(isInstanceOfAny(TableScanNode.class, ValuesNode.class))
.findAll();
return sourceNodes.stream()
.mapToDouble(sourceNode -> statsProvider.getStats(sourceNode).getOutputSizeInBytes(sourceNode.getOutputSymbols(), typeProvider))
.sum();
}
private PlanNode getCostBasedJoin(JoinNode joinNode, Context context)
@ -103,7 +135,7 @@ public class DetermineJoinDistributionType
addJoinsWithDifferentDistributions(joinNode.flipChildren(), possibleJoinNodes, context);
if (possibleJoinNodes.stream().anyMatch(result -> result.getCost().hasUnknownComponents()) || possibleJoinNodes.isEmpty()) {
return getSyntacticOrderJoin(joinNode, context, AUTOMATIC);
return getSizeBasedJoin(joinNode, context);
}
// Using Ordering to facilitate rule determinism
@ -111,6 +143,36 @@ public class DetermineJoinDistributionType
return planNodeOrderings.min(possibleJoinNodes).getPlanNode();
}
private JoinNode getSizeBasedJoin(JoinNode joinNode, Context context)
{
DataSize joinMaxBroadcastTableSize = getJoinMaxBroadcastTableSize(context.getSession()).get();
boolean isRightSideSmall = getSourceTablesSizeInBytes(joinNode.getRight(), context) <= joinMaxBroadcastTableSize.toBytes();
if (isRightSideSmall && !mustPartition(joinNode)) {
// choose right join side with small source tables as replicated build side
return joinNode.withDistributionType(REPLICATED);
}
boolean isLeftSideSmall = getSourceTablesSizeInBytes(joinNode.getLeft(), context) <= joinMaxBroadcastTableSize.toBytes();
if (isLeftSideSmall && !mustPartition(joinNode.flipChildren())) {
// choose join left side with small source tables as replicated build side
return joinNode.flipChildren().withDistributionType(REPLICATED);
}
if (isRightSideSmall) {
// right side is small enough, but must be partitioned
return joinNode.withDistributionType(PARTITIONED);
}
if (isLeftSideSmall) {
// left side is small enough, but must be partitioned
return joinNode.flipChildren().withDistributionType(PARTITIONED);
}
// neither side is small enough, choose syntactic join order
return getSyntacticOrderJoin(joinNode, context, AUTOMATIC);
}
private void addJoinsWithDifferentDistributions(JoinNode joinNode, List<PlanNodeWithCost> possibleJoinNodes, Context context)
{
if (!mustPartition(joinNode) && canReplicate(joinNode, context)) {
@ -121,7 +183,7 @@ public class DetermineJoinDistributionType
}
}
private PlanNode getSyntacticOrderJoin(JoinNode joinNode, Context context, JoinDistributionType joinDistributionType)
private JoinNode getSyntacticOrderJoin(JoinNode joinNode, Context context, JoinDistributionType joinDistributionType)
{
if (mustPartition(joinNode)) {
return joinNode.withDistributionType(PARTITIONED);

18
presto-main/src/main/java/io/prestosql/sql/planner/iterative/rule/DetermineSemiJoinDistributionType.java
View File

@ -49,6 +49,7 @@ import java.util.Optional;
import static io.prestosql.SystemSessionProperties.getJoinDistributionType;
import static io.prestosql.SystemSessionProperties.getJoinMaxBroadcastTableSize;
import static io.prestosql.cost.CostCalculatorWithEstimatedExchanges.calculateJoinCostWithoutOutput;
import static io.prestosql.sql.planner.iterative.rule.DetermineJoinDistributionType.getSourceTablesSizeInBytes;
import static io.prestosql.sql.planner.plan.Patterns.semiJoin;
import static io.prestosql.sql.planner.plan.SemiJoinNode.DistributionType.PARTITIONED;
import static io.prestosql.sql.planner.plan.SemiJoinNode.DistributionType.REPLICATED;
@ -106,7 +107,7 @@ public class DetermineSemiJoinDistributionType
possibleJoinNodes.add(getSemiJoinNodeWithCost(node.withDistributionType(PARTITIONED), context));
if (possibleJoinNodes.stream().anyMatch(result -> result.getCost().hasUnknownComponents())) {
return node.withDistributionType(PARTITIONED);
return getSizeBaseDistributionType(node, context);
}
// Using Ordering to facilitate rule determinism
@ -114,6 +115,18 @@ public class DetermineSemiJoinDistributionType
return planNodeOrderings.min(possibleJoinNodes).getPlanNode();
}
private PlanNode getSizeBaseDistributionType(SemiJoinNode node, Context context)
{
DataSize joinMaxBroadcastTableSize = getJoinMaxBroadcastTableSize(context.getSession()).get();
if (getSourceTablesSizeInBytes(node.getFilteringSource(), context) <= joinMaxBroadcastTableSize.toBytes()) {
// choose replicated distribution type as filtering source contains small source tables only
return node.withDistributionType(REPLICATED);
}
return node.withDistributionType(PARTITIONED);
}
private boolean canReplicate(SemiJoinNode node, Context context)
{
Optional<DataSize> joinMaxBroadcastTableSize = getJoinMaxBroadcastTableSize(context.getSession());
@ -124,7 +137,8 @@ public class DetermineSemiJoinDistributionType
PlanNode buildSide = node.getFilteringSource();
PlanNodeStatsEstimate buildSideStatsEstimate = context.getStatsProvider().getStats(buildSide);
double buildSideSizeInBytes = buildSideStatsEstimate.getOutputSizeInBytes(buildSide.getOutputSymbols(), context.getSymbolAllocator().getTypes());
return buildSideSizeInBytes <= joinMaxBroadcastTableSize.get().toBytes();
return buildSideSizeInBytes <= joinMaxBroadcastTableSize.get().toBytes()
|| getSourceTablesSizeInBytes(buildSide, context) <= joinMaxBroadcastTableSize.get().toBytes();
}
private PlanNodeWithCost getSemiJoinNodeWithCost(SemiJoinNode possibleJoinNode, Context context)