diff --git a/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q14_1.cte.plan.txt b/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q14_1.cte.plan.txt index d3b62b90d..fce6e0f8c 100644 --- a/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q14_1.cte.plan.txt +++ b/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q14_1.cte.plan.txt @@ -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, []) diff --git a/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q14_1.plan.txt b/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q14_1.plan.txt index 93239a289..0392b8d12 100644 --- a/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q14_1.plan.txt +++ b/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q14_1.plan.txt @@ -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 () diff --git a/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q14_1.push.plan.txt b/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q14_1.push.plan.txt index f70d494fd..979eb65f5 100644 --- a/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q14_1.push.plan.txt +++ b/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q14_1.push.plan.txt @@ -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 () diff --git a/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q14_2.cte.plan.txt b/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q14_2.cte.plan.txt index 22e478c62..f17e21789 100644 --- a/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q14_2.cte.plan.txt +++ b/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q14_2.cte.plan.txt @@ -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, []) diff --git a/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q14_2.plan.txt b/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q14_2.plan.txt index a5b9d7eb2..1e9810b59 100644 --- a/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q14_2.plan.txt +++ b/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q14_2.plan.txt @@ -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, []) diff --git a/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q14_2.push.plan.txt b/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q14_2.push.plan.txt index 659608553..d86e82722 100644 --- a/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q14_2.push.plan.txt +++ b/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q14_2.push.plan.txt @@ -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, []) diff --git a/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q34.push.plan.txt b/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q34.push.plan.txt index 56184d7b0..3478b1527 100644 --- a/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q34.push.plan.txt +++ b/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q34.push.plan.txt @@ -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 diff --git a/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q44.plan.txt b/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q44.plan.txt index bb30a05b4..5ede16641 100644 --- a/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q44.plan.txt +++ b/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q44.plan.txt @@ -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 diff --git a/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q44.push.plan.txt b/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q44.push.plan.txt index a9cc9d12e..ae5e780d1 100644 --- a/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q44.push.plan.txt +++ b/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q44.push.plan.txt @@ -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 diff --git a/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q44.reuse.plan.txt b/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q44.reuse.plan.txt index d23273a2a..af0c0babb 100644 --- a/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q44.reuse.plan.txt +++ b/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q44.reuse.plan.txt @@ -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 diff --git a/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q64.cte.plan.txt b/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q64.cte.plan.txt index bd7278138..33aa65701 100644 --- a/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q64.cte.plan.txt +++ b/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q64.cte.plan.txt @@ -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 diff --git a/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q64.plan.txt b/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q64.plan.txt index 4f143a468..fad8cf858 100644 --- a/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q64.plan.txt +++ b/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q64.plan.txt @@ -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 diff --git a/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q64.push.plan.txt b/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q64.push.plan.txt index 3f59e63be..405f22df4 100644 --- a/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q64.push.plan.txt +++ b/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q64.push.plan.txt @@ -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) diff --git a/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q73.push.plan.txt b/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q73.push.plan.txt index b190a13e0..7d81c4113 100644 --- a/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q73.push.plan.txt +++ b/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q73.push.plan.txt @@ -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 diff --git a/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q95.push.plan.txt b/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q95.push.plan.txt index 2525eb52c..896477e74 100644 --- a/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q95.push.plan.txt +++ b/presto-benchto-benchmarks/src/test/resources/sql/presto/tpcds/q95.push.plan.txt @@ -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 diff --git a/presto-benchto-benchmarks/src/test/resources/sql/presto/tpch/q09.plan.txt b/presto-benchto-benchmarks/src/test/resources/sql/presto/tpch/q09.plan.txt index 035457170..602b085d9 100644 --- a/presto-benchto-benchmarks/src/test/resources/sql/presto/tpch/q09.plan.txt +++ b/presto-benchto-benchmarks/src/test/resources/sql/presto/tpch/q09.plan.txt @@ -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 diff --git a/presto-benchto-benchmarks/src/test/resources/sql/presto/tpch/q21.plan.txt b/presto-benchto-benchmarks/src/test/resources/sql/presto/tpch/q21.plan.txt index f5d75a71f..9f04ccd6e 100644 --- a/presto-benchto-benchmarks/src/test/resources/sql/presto/tpch/q21.plan.txt +++ b/presto-benchto-benchmarks/src/test/resources/sql/presto/tpch/q21.plan.txt @@ -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 diff --git a/presto-main/src/main/java/io/prestosql/sql/planner/iterative/rule/DetermineJoinDistributionType.java b/presto-main/src/main/java/io/prestosql/sql/planner/iterative/rule/DetermineJoinDistributionType.java index 140bfb222..c6632eb5e 100644 --- a/presto-main/src/main/java/io/prestosql/sql/planner/iterative/rule/DetermineJoinDistributionType.java +++ b/presto-main/src/main/java/io/prestosql/sql/planner/iterative/rule/DetermineJoinDistributionType.java @@ -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 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 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); diff --git a/presto-main/src/main/java/io/prestosql/sql/planner/iterative/rule/DetermineSemiJoinDistributionType.java b/presto-main/src/main/java/io/prestosql/sql/planner/iterative/rule/DetermineSemiJoinDistributionType.java index f854f1ff5..e95cb4901 100644 --- a/presto-main/src/main/java/io/prestosql/sql/planner/iterative/rule/DetermineSemiJoinDistributionType.java +++ b/presto-main/src/main/java/io/prestosql/sql/planner/iterative/rule/DetermineSemiJoinDistributionType.java @@ -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 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)