This is an automated email from the ASF dual-hosted git repository.
andygrove pushed a commit to branch branch-54
in repository https://gitbox.apache.org/repos/asf/datafusion-ballista.git
The following commit(s) were added to refs/heads/branch-54 by this push:
new 0a4ed4e07 fix: disable the SortMergeJoinExec broadcast conversion by
default (#2130)
0a4ed4e07 is described below
commit 0a4ed4e07bc8339f296314a68a3abf5e987ff640
Author: Andy Grove <[email protected]>
AuthorDate: Wed Jul 22 06:40:00 2026 -0600
fix: disable the SortMergeJoinExec broadcast conversion by default (#2130)
---
ballista/core/src/config.rs | 17 ++-
.../tests/tpch_plan_stability/approved/q10.txt | 47 ++++---
.../tests/tpch_plan_stability/approved/q11.txt | 74 +++++------
.../tests/tpch_plan_stability/approved/q15.txt | 64 +++++-----
.../tests/tpch_plan_stability/approved/q16.txt | 42 +++----
.../tests/tpch_plan_stability/approved/q2.txt | 140 ++++++++++-----------
.../tests/tpch_plan_stability/approved/q20.txt | 49 ++++----
.../tests/tpch_plan_stability/approved/q21.txt | 61 +++++----
.../tests/tpch_plan_stability/approved/q5.txt | 80 ++++++------
.../tests/tpch_plan_stability/approved/q7.txt | 73 +++++------
.../tests/tpch_plan_stability/approved/q8.txt | 104 +++++++--------
.../tests/tpch_plan_stability/approved/q9.txt | 63 +++++-----
docs/source/upgrading/54.0.0.md | 9 +-
13 files changed, 389 insertions(+), 434 deletions(-)
diff --git a/ballista/core/src/config.rs b/ballista/core/src/config.rs
index 087a646b8..4ea638a29 100644
--- a/ballista/core/src/config.rs
+++ b/ballista/core/src/config.rs
@@ -103,7 +103,8 @@ pub const BALLISTA_BROADCAST_JOIN_THRESHOLD_BYTES: &str =
/// Configuration key to enable broadcasting a small build side of a
/// `SortMergeJoinExec` by converting it to a `CollectLeft` hash join in the
-/// static distributed planner. Enabled by default.
+/// static distributed planner. Disabled by default: the conversion can
+/// produce incorrect results, so it is opt-in only.
pub const BALLISTA_BROADCAST_SORT_MERGE_JOIN_ENABLED: &str =
"ballista.optimizer.broadcast_sort_merge_join_enabled";
@@ -243,9 +244,10 @@ static CONFIG_ENTRIES: LazyLock<HashMap<String,
ConfigEntry>> = LazyLock::new(||
ConfigEntry::new(BALLISTA_BROADCAST_SORT_MERGE_JOIN_ENABLED.to_string(),
"Broadcast a small build side of a SortMergeJoinExec
by converting it \
to a CollectLeft hash join in the static distributed
planner. \
- The build side must also fit under
broadcast_join_threshold_bytes.".to_string(),
+ The build side must also fit under
broadcast_join_threshold_bytes. \
+ Disabled by default: the conversion can produce
incorrect results.".to_string(),
DataType::Boolean,
- Some(true.to_string())),
+ Some(false.to_string())),
ConfigEntry::new(BALLISTA_CLIENT_PULL.to_string(),
"Should client employ pull or push job tracking. In
pull mode client will make a request to server in the loop, until job finishes.
Pull mode is kept for legacy clients.".to_string(),
DataType::Boolean,
@@ -570,7 +572,8 @@ impl BallistaConfig {
/// Returns whether broadcasting a small build side of a
`SortMergeJoinExec`
/// (by converting it to a `CollectLeft` hash join) is enabled in the
static
- /// distributed planner.
+ /// distributed planner. Off unless explicitly opted into, because the
+ /// conversion can produce incorrect results.
pub fn broadcast_sort_merge_join_enabled(&self) -> bool {
self.get_bool_setting(BALLISTA_BROADCAST_SORT_MERGE_JOIN_ENABLED)
}
@@ -821,9 +824,11 @@ mod tests {
Ok(())
}
+ /// The conversion can produce incorrect results, so it must stay off
+ /// unless a user explicitly opts in.
#[test]
- fn broadcast_sort_merge_join_enabled_by_default() {
+ fn broadcast_sort_merge_join_disabled_by_default() {
let config = BallistaConfig::default();
- assert!(config.broadcast_sort_merge_join_enabled());
+ assert!(!config.broadcast_sort_merge_join_enabled());
}
}
diff --git a/ballista/scheduler/tests/tpch_plan_stability/approved/q10.txt
b/ballista/scheduler/tests/tpch_plan_stability/approved/q10.txt
index cf63a2940..7b12a86d2 100644
--- a/ballista/scheduler/tests/tpch_plan_stability/approved/q10.txt
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q10.txt
@@ -1,60 +1,57 @@
=== Stage 1 ===
-SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
- StatsExec: rows=25
-
-=== Stage 2 ===
-ShuffleWriterExec: partitioning: None
- UnresolvedShuffleExec: stage=1, partitioning: Hash([n_nationkey@0], 16)
-
-=== Stage 3 ===
SortShuffleWriterExec: partitioning=Hash([c_custkey@0], 16)
StatsExec: rows=15000000
-=== Stage 4 ===
+=== Stage 2 ===
SortShuffleWriterExec: partitioning=Hash([o_custkey@1], 16)
FilterExec: o_orderdate@2 >= 1993-10-01 AND o_orderdate@2 < 1994-01-01,
projection=[o_orderkey@0, o_custkey@1]
StatsExec: rows=150000000
-=== Stage 5 ===
+=== Stage 3 ===
SortShuffleWriterExec: partitioning=Hash([o_orderkey@7], 16)
ProjectionExec: expr=[c_custkey@0 as c_custkey, c_name@1 as c_name,
c_address@2 as c_address, c_nationkey@3 as c_nationkey, c_phone@4 as c_phone,
c_acctbal@5 as c_acctbal, c_comment@6 as c_comment, o_orderkey@7 as o_orderkey]
SortMergeJoinExec: join_type=Inner, on=[(c_custkey@0, o_custkey@1)]
SortExec: expr=[c_custkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=3, partitioning: Hash([c_custkey@0], 16)
+ UnresolvedShuffleExec: stage=1, partitioning: Hash([c_custkey@0], 16)
SortExec: expr=[o_custkey@1 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=4, partitioning: Hash([o_custkey@1], 16)
+ UnresolvedShuffleExec: stage=2, partitioning: Hash([o_custkey@1], 16)
-=== Stage 6 ===
+=== Stage 4 ===
SortShuffleWriterExec: partitioning=Hash([l_orderkey@0], 16)
FilterExec: l_returnflag@3 = R, projection=[l_orderkey@0, l_extendedprice@1,
l_discount@2]
StatsExec: rows=600037902
-=== Stage 7 ===
+=== Stage 5 ===
SortShuffleWriterExec: partitioning=Hash([c_nationkey@3], 16)
ProjectionExec: expr=[c_custkey@0 as c_custkey, c_name@1 as c_name,
c_address@2 as c_address, c_nationkey@3 as c_nationkey, c_phone@4 as c_phone,
c_acctbal@5 as c_acctbal, c_comment@6 as c_comment, l_extendedprice@9 as
l_extendedprice, l_discount@10 as l_discount]
SortMergeJoinExec: join_type=Inner, on=[(o_orderkey@7, l_orderkey@0)]
SortExec: expr=[o_orderkey@7 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=5, partitioning: Hash([o_orderkey@7], 16)
+ UnresolvedShuffleExec: stage=3, partitioning: Hash([o_orderkey@7], 16)
SortExec: expr=[l_orderkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=6, partitioning: Hash([l_orderkey@0], 16)
+ UnresolvedShuffleExec: stage=4, partitioning: Hash([l_orderkey@0], 16)
-=== Stage 8 ===
+=== Stage 6 ===
+SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
+ StatsExec: rows=25
+
+=== Stage 7 ===
SortShuffleWriterExec: partitioning=Hash([c_custkey@0, c_name@1, c_acctbal@2,
c_phone@3, n_name@4, c_address@5, c_comment@6], 16)
AggregateExec: mode=Partial, gby=[c_custkey@0 as c_custkey, c_name@1 as
c_name, c_acctbal@4 as c_acctbal, c_phone@3 as c_phone, n_name@8 as n_name,
c_address@2 as c_address, c_comment@5 as c_comment],
aggr=[sum(lineitem.l_extendedprice * Some(1),20,0 - lineitem.l_discount) as
sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)]
ProjectionExec: expr=[c_custkey@0 as c_custkey, c_name@1 as c_name,
c_address@2 as c_address, c_phone@4 as c_phone, c_acctbal@5 as c_acctbal,
c_comment@6 as c_comment, l_extendedprice@7 as l_extendedprice, l_discount@8 as
l_discount, n_name@10 as n_name]
- ProjectionExec: expr=[c_custkey@2 as c_custkey, c_name@3 as c_name,
c_address@4 as c_address, c_nationkey@5 as c_nationkey, c_phone@6 as c_phone,
c_acctbal@7 as c_acctbal, c_comment@8 as c_comment, l_extendedprice@9 as
l_extendedprice, l_discount@10 as l_discount, n_nationkey@0 as n_nationkey,
n_name@1 as n_name]
- HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(n_nationkey@0,
c_nationkey@3)]
- UnresolvedShuffleExec: stage=2, broadcast=true, upstream_partitions:
16
- UnresolvedShuffleExec: stage=7, partitioning: Hash([c_nationkey@3],
16)
+ SortMergeJoinExec: join_type=Inner, on=[(c_nationkey@3, n_nationkey@0)]
+ SortExec: expr=[c_nationkey@3 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: stage=5, partitioning: Hash([c_nationkey@3],
16)
+ SortExec: expr=[n_nationkey@0 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: stage=6, partitioning: Hash([n_nationkey@0],
16)
-=== Stage 9 ===
+=== Stage 8 ===
ShuffleWriterExec: partitioning: None
SortExec: TopK(fetch=20), expr=[revenue@2 DESC], preserve_partitioning=[true]
ProjectionExec: expr=[c_custkey@0 as c_custkey, c_name@1 as c_name,
sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)@7 as revenue,
c_acctbal@2 as c_acctbal, n_name@4 as n_name, c_address@5 as c_address,
c_phone@3 as c_phone, c_comment@6 as c_comment]
AggregateExec: mode=FinalPartitioned, gby=[c_custkey@0 as c_custkey,
c_name@1 as c_name, c_acctbal@2 as c_acctbal, c_phone@3 as c_phone, n_name@4 as
n_name, c_address@5 as c_address, c_comment@6 as c_comment],
aggr=[sum(lineitem.l_extendedprice * Some(1),20,0 - lineitem.l_discount) as
sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)]
- UnresolvedShuffleExec: stage=8, partitioning: Hash([c_custkey@0,
c_name@1, c_acctbal@2, c_phone@3, n_name@4, c_address@5, c_comment@6], 16)
+ UnresolvedShuffleExec: stage=7, partitioning: Hash([c_custkey@0,
c_name@1, c_acctbal@2, c_phone@3, n_name@4, c_address@5, c_comment@6], 16)
-=== Stage 10 ===
+=== Stage 9 ===
ShuffleWriterExec: partitioning: None
SortPreservingMergeExec: [revenue@2 DESC], fetch=20
- UnresolvedShuffleExec: stage=9, partitioning: Hash([c_custkey@0, c_name@1,
c_acctbal@3, c_phone@6, n_name@4, c_address@5, c_comment@7], 16)
+ UnresolvedShuffleExec: stage=8, partitioning: Hash([c_custkey@0, c_name@1,
c_acctbal@3, c_phone@6, n_name@4, c_address@5, c_comment@7], 16)
diff --git a/ballista/scheduler/tests/tpch_plan_stability/approved/q11.txt
b/ballista/scheduler/tests/tpch_plan_stability/approved/q11.txt
index 9a68df95d..736c88e26 100644
--- a/ballista/scheduler/tests/tpch_plan_stability/approved/q11.txt
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q11.txt
@@ -1,74 +1,68 @@
=== Stage 1 ===
-SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
- FilterExec: n_name@1 = GERMANY, projection=[n_nationkey@0]
- StatsExec: rows=25
-
-=== Stage 2 ===
-ShuffleWriterExec: partitioning: None
- UnresolvedShuffleExec: stage=1, partitioning: Hash([n_nationkey@0], 16)
-
-=== Stage 3 ===
SortShuffleWriterExec: partitioning=Hash([ps_suppkey@0], 16)
StatsExec: rows=80000000
-=== Stage 4 ===
+=== Stage 2 ===
SortShuffleWriterExec: partitioning=Hash([s_suppkey@0], 16)
StatsExec: rows=1000000
-=== Stage 5 ===
+=== Stage 3 ===
SortShuffleWriterExec: partitioning=Hash([s_nationkey@2], 16)
ProjectionExec: expr=[ps_availqty@1 as ps_availqty, ps_supplycost@2 as
ps_supplycost, s_nationkey@4 as s_nationkey]
SortMergeJoinExec: join_type=Inner, on=[(ps_suppkey@0, s_suppkey@0)]
SortExec: expr=[ps_suppkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=3, partitioning: Hash([ps_suppkey@0], 16)
+ UnresolvedShuffleExec: stage=1, partitioning: Hash([ps_suppkey@0], 16)
SortExec: expr=[s_suppkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=4, partitioning: Hash([s_suppkey@0], 16)
+ UnresolvedShuffleExec: stage=2, partitioning: Hash([s_suppkey@0], 16)
-=== Stage 6 ===
-ShuffleWriterExec: partitioning: None
- AggregateExec: mode=Partial, gby=[], aggr=[sum(partsupp.ps_supplycost *
partsupp.ps_availqty)]
- ProjectionExec: expr=[ps_availqty@0 as ps_availqty, ps_supplycost@1 as
ps_supplycost]
- ProjectionExec: expr=[ps_availqty@1 as ps_availqty, ps_supplycost@2 as
ps_supplycost, s_nationkey@3 as s_nationkey, n_nationkey@0 as n_nationkey]
- HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(n_nationkey@0,
s_nationkey@2)]
- UnresolvedShuffleExec: stage=2, broadcast=true, upstream_partitions:
16
- UnresolvedShuffleExec: stage=5, partitioning: Hash([s_nationkey@2],
16)
-
-=== Stage 7 ===
+=== Stage 4 ===
SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
FilterExec: n_name@1 = GERMANY, projection=[n_nationkey@0]
StatsExec: rows=25
-=== Stage 8 ===
+=== Stage 5 ===
ShuffleWriterExec: partitioning: None
- UnresolvedShuffleExec: stage=7, partitioning: Hash([n_nationkey@0], 16)
+ AggregateExec: mode=Partial, gby=[], aggr=[sum(partsupp.ps_supplycost *
partsupp.ps_availqty)]
+ ProjectionExec: expr=[ps_availqty@0 as ps_availqty, ps_supplycost@1 as
ps_supplycost]
+ SortMergeJoinExec: join_type=Inner, on=[(s_nationkey@2, n_nationkey@0)]
+ SortExec: expr=[s_nationkey@2 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: stage=3, partitioning: Hash([s_nationkey@2],
16)
+ SortExec: expr=[n_nationkey@0 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: stage=4, partitioning: Hash([n_nationkey@0],
16)
-=== Stage 9 ===
+=== Stage 6 ===
SortShuffleWriterExec: partitioning=Hash([ps_suppkey@1], 16)
StatsExec: rows=80000000
-=== Stage 10 ===
+=== Stage 7 ===
SortShuffleWriterExec: partitioning=Hash([s_suppkey@0], 16)
StatsExec: rows=1000000
-=== Stage 11 ===
+=== Stage 8 ===
SortShuffleWriterExec: partitioning=Hash([s_nationkey@3], 16)
ProjectionExec: expr=[ps_partkey@0 as ps_partkey, ps_availqty@2 as
ps_availqty, ps_supplycost@3 as ps_supplycost, s_nationkey@5 as s_nationkey]
SortMergeJoinExec: join_type=Inner, on=[(ps_suppkey@1, s_suppkey@0)]
SortExec: expr=[ps_suppkey@1 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=9, partitioning: Hash([ps_suppkey@1], 16)
+ UnresolvedShuffleExec: stage=6, partitioning: Hash([ps_suppkey@1], 16)
SortExec: expr=[s_suppkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=10, partitioning: Hash([s_suppkey@0], 16)
+ UnresolvedShuffleExec: stage=7, partitioning: Hash([s_suppkey@0], 16)
-=== Stage 12 ===
+=== Stage 9 ===
+SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
+ FilterExec: n_name@1 = GERMANY, projection=[n_nationkey@0]
+ StatsExec: rows=25
+
+=== Stage 10 ===
SortShuffleWriterExec: partitioning=Hash([ps_partkey@0], 16)
AggregateExec: mode=Partial, gby=[ps_partkey@0 as ps_partkey],
aggr=[sum(partsupp.ps_supplycost * partsupp.ps_availqty)]
ProjectionExec: expr=[ps_partkey@0 as ps_partkey, ps_availqty@1 as
ps_availqty, ps_supplycost@2 as ps_supplycost]
- ProjectionExec: expr=[ps_partkey@1 as ps_partkey, ps_availqty@2 as
ps_availqty, ps_supplycost@3 as ps_supplycost, s_nationkey@4 as s_nationkey,
n_nationkey@0 as n_nationkey]
- HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(n_nationkey@0,
s_nationkey@3)]
- UnresolvedShuffleExec: stage=8, broadcast=true, upstream_partitions:
16
- UnresolvedShuffleExec: stage=11, partitioning: Hash([s_nationkey@3],
16)
+ SortMergeJoinExec: join_type=Inner, on=[(s_nationkey@3, n_nationkey@0)]
+ SortExec: expr=[s_nationkey@3 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: stage=8, partitioning: Hash([s_nationkey@3],
16)
+ SortExec: expr=[n_nationkey@0 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: stage=9, partitioning: Hash([n_nationkey@0],
16)
-=== Stage 13 ===
+=== Stage 11 ===
ShuffleWriterExec: partitioning: None
SortExec: expr=[value@1 DESC], preserve_partitioning=[true]
ProjectionExec: expr=[ps_partkey@1 as ps_partkey,
sum(partsupp.ps_supplycost * partsupp.ps_availqty)@2 as value]
@@ -76,12 +70,12 @@ ShuffleWriterExec: partitioning: None
ProjectionExec: expr=[CAST(CAST(sum(partsupp.ps_supplycost *
partsupp.ps_availqty)@0 AS Float64) * 0.0001 AS Decimal128(38, 15)) as
sum(partsupp.ps_supplycost * partsupp.ps_availqty) * Float64(0.0001)]
AggregateExec: mode=Final, gby=[], aggr=[sum(partsupp.ps_supplycost
* partsupp.ps_availqty)]
CoalescePartitionsExec
- UnresolvedShuffleExec: stage=6, partitioning:
Hash([s_nationkey@2], 16)
+ UnresolvedShuffleExec: stage=5, partitioning:
Hash([n_nationkey@3], 16)
ProjectionExec: expr=[ps_partkey@0 as ps_partkey,
sum(partsupp.ps_supplycost * partsupp.ps_availqty)@1 as
sum(partsupp.ps_supplycost * partsupp.ps_availqty),
CAST(sum(partsupp.ps_supplycost * partsupp.ps_availqty)@1 AS Decimal128(38,
15)) as join_proj_push_down_1]
AggregateExec: mode=FinalPartitioned, gby=[ps_partkey@0 as
ps_partkey], aggr=[sum(partsupp.ps_supplycost * partsupp.ps_availqty)]
- UnresolvedShuffleExec: stage=12, partitioning:
Hash([ps_partkey@0], 16)
+ UnresolvedShuffleExec: stage=10, partitioning:
Hash([ps_partkey@0], 16)
-=== Stage 14 ===
+=== Stage 12 ===
ShuffleWriterExec: partitioning: None
SortPreservingMergeExec: [value@1 DESC]
- UnresolvedShuffleExec: stage=13, partitioning: Hash([ps_partkey@0], 16)
+ UnresolvedShuffleExec: stage=11, partitioning: Hash([ps_partkey@0], 16)
diff --git a/ballista/scheduler/tests/tpch_plan_stability/approved/q15.txt
b/ballista/scheduler/tests/tpch_plan_stability/approved/q15.txt
index 3863e7bc8..caf8d0d9d 100644
--- a/ballista/scheduler/tests/tpch_plan_stability/approved/q15.txt
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q15.txt
@@ -1,57 +1,53 @@
=== Stage 1 ===
-SortShuffleWriterExec: partitioning=Hash([l_suppkey@0], 16)
- AggregateExec: mode=Partial, gby=[l_suppkey@0 as l_suppkey],
aggr=[sum(lineitem.l_extendedprice * Some(1),20,0 - lineitem.l_discount) as
sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)]
- FilterExec: l_shipdate@3 >= 1996-01-01 AND l_shipdate@3 < 1996-04-01,
projection=[l_suppkey@0, l_extendedprice@1, l_discount@2]
- StatsExec: rows=600037902
-
-=== Stage 2 ===
-ShuffleWriterExec: partitioning: None
- AggregateExec: mode=Partial, gby=[], aggr=[max(revenue0.total_revenue)]
- ProjectionExec: expr=[sum(lineitem.l_extendedprice * Int64(1) -
lineitem.l_discount)@1 as total_revenue]
- AggregateExec: mode=FinalPartitioned, gby=[l_suppkey@0 as l_suppkey],
aggr=[sum(lineitem.l_extendedprice * Some(1),20,0 - lineitem.l_discount) as
sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)]
- UnresolvedShuffleExec: stage=1, partitioning: Hash([l_suppkey@0], 16)
-
-=== Stage 3 ===
-SortShuffleWriterExec: partitioning=Hash([max(revenue0.total_revenue)@0], 16)
- AggregateExec: mode=Final, gby=[], aggr=[max(revenue0.total_revenue)]
- CoalescePartitionsExec
- UnresolvedShuffleExec: stage=2, partitioning: Hash([l_suppkey@0], 16)
-
-=== Stage 4 ===
-ShuffleWriterExec: partitioning: None
- UnresolvedShuffleExec: stage=3, partitioning:
Hash([max(revenue0.total_revenue)@0], 16)
-
-=== Stage 5 ===
SortShuffleWriterExec: partitioning=Hash([s_suppkey@0], 16)
StatsExec: rows=1000000
-=== Stage 6 ===
+=== Stage 2 ===
SortShuffleWriterExec: partitioning=Hash([l_suppkey@0], 16)
AggregateExec: mode=Partial, gby=[l_suppkey@0 as l_suppkey],
aggr=[sum(lineitem.l_extendedprice * Some(1),20,0 - lineitem.l_discount) as
sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)]
FilterExec: l_shipdate@3 >= 1996-01-01 AND l_shipdate@3 < 1996-04-01,
projection=[l_suppkey@0, l_extendedprice@1, l_discount@2]
StatsExec: rows=600037902
-=== Stage 7 ===
+=== Stage 3 ===
SortShuffleWriterExec: partitioning=Hash([total_revenue@4], 16)
ProjectionExec: expr=[s_suppkey@0 as s_suppkey, s_name@1 as s_name,
s_address@2 as s_address, s_phone@3 as s_phone, total_revenue@5 as
total_revenue]
SortMergeJoinExec: join_type=Inner, on=[(s_suppkey@0, supplier_no@0)]
SortExec: expr=[s_suppkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=5, partitioning: Hash([s_suppkey@0], 16)
+ UnresolvedShuffleExec: stage=1, partitioning: Hash([s_suppkey@0], 16)
SortExec: expr=[supplier_no@0 ASC], preserve_partitioning=[true]
ProjectionExec: expr=[l_suppkey@0 as supplier_no,
sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)@1 as
total_revenue]
AggregateExec: mode=FinalPartitioned, gby=[l_suppkey@0 as
l_suppkey], aggr=[sum(lineitem.l_extendedprice * Some(1),20,0 -
lineitem.l_discount) as sum(lineitem.l_extendedprice * Int64(1) -
lineitem.l_discount)]
- UnresolvedShuffleExec: stage=6, partitioning: Hash([l_suppkey@0],
16)
+ UnresolvedShuffleExec: stage=2, partitioning: Hash([l_suppkey@0],
16)
-=== Stage 8 ===
+=== Stage 4 ===
+SortShuffleWriterExec: partitioning=Hash([l_suppkey@0], 16)
+ AggregateExec: mode=Partial, gby=[l_suppkey@0 as l_suppkey],
aggr=[sum(lineitem.l_extendedprice * Some(1),20,0 - lineitem.l_discount) as
sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)]
+ FilterExec: l_shipdate@3 >= 1996-01-01 AND l_shipdate@3 < 1996-04-01,
projection=[l_suppkey@0, l_extendedprice@1, l_discount@2]
+ StatsExec: rows=600037902
+
+=== Stage 5 ===
+ShuffleWriterExec: partitioning: None
+ AggregateExec: mode=Partial, gby=[], aggr=[max(revenue0.total_revenue)]
+ ProjectionExec: expr=[sum(lineitem.l_extendedprice * Int64(1) -
lineitem.l_discount)@1 as total_revenue]
+ AggregateExec: mode=FinalPartitioned, gby=[l_suppkey@0 as l_suppkey],
aggr=[sum(lineitem.l_extendedprice * Some(1),20,0 - lineitem.l_discount) as
sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)]
+ UnresolvedShuffleExec: stage=4, partitioning: Hash([l_suppkey@0], 16)
+
+=== Stage 6 ===
+SortShuffleWriterExec: partitioning=Hash([max(revenue0.total_revenue)@0], 16)
+ AggregateExec: mode=Final, gby=[], aggr=[max(revenue0.total_revenue)]
+ CoalescePartitionsExec
+ UnresolvedShuffleExec: stage=5, partitioning: Hash([l_suppkey@0], 16)
+
+=== Stage 7 ===
ShuffleWriterExec: partitioning: None
SortExec: expr=[s_suppkey@0 ASC NULLS LAST], preserve_partitioning=[true]
ProjectionExec: expr=[s_suppkey@0 as s_suppkey, s_name@1 as s_name,
s_address@2 as s_address, s_phone@3 as s_phone, total_revenue@4 as
total_revenue]
- ProjectionExec: expr=[s_suppkey@1 as s_suppkey, s_name@2 as s_name,
s_address@3 as s_address, s_phone@4 as s_phone, total_revenue@5 as
total_revenue, max(revenue0.total_revenue)@0 as max(revenue0.total_revenue)]
- HashJoinExec: mode=CollectLeft, join_type=Inner,
on=[(max(revenue0.total_revenue)@0, total_revenue@4)]
- UnresolvedShuffleExec: stage=4, broadcast=true, upstream_partitions:
16
- UnresolvedShuffleExec: stage=7, partitioning:
Hash([total_revenue@4], 16)
+ SortMergeJoinExec: join_type=Inner, on=[(total_revenue@4,
max(revenue0.total_revenue)@0)]
+ SortExec: expr=[total_revenue@4 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: stage=3, partitioning:
Hash([total_revenue@4], 16)
+ UnresolvedShuffleExec: stage=6, partitioning:
Hash([max(revenue0.total_revenue)@0], 16)
-=== Stage 9 ===
+=== Stage 8 ===
ShuffleWriterExec: partitioning: None
SortPreservingMergeExec: [s_suppkey@0 ASC NULLS LAST]
- UnresolvedShuffleExec: stage=8, partitioning: Hash([total_revenue@4], 16)
+ UnresolvedShuffleExec: stage=7, partitioning: Hash([total_revenue@4], 16)
diff --git a/ballista/scheduler/tests/tpch_plan_stability/approved/q16.txt
b/ballista/scheduler/tests/tpch_plan_stability/approved/q16.txt
index 2947632db..2a6265621 100644
--- a/ballista/scheduler/tests/tpch_plan_stability/approved/q16.txt
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q16.txt
@@ -1,46 +1,44 @@
=== Stage 1 ===
-SortShuffleWriterExec: partitioning=Hash([s_suppkey@0], 16)
- FilterExec: s_comment@1 LIKE %Customer%Complaints%, projection=[s_suppkey@0]
- StatsExec: rows=1000000
-
-=== Stage 2 ===
-ShuffleWriterExec: partitioning: None
- UnresolvedShuffleExec: stage=1, partitioning: Hash([s_suppkey@0], 16)
-
-=== Stage 3 ===
SortShuffleWriterExec: partitioning=Hash([ps_partkey@0], 16)
StatsExec: rows=80000000
-=== Stage 4 ===
+=== Stage 2 ===
SortShuffleWriterExec: partitioning=Hash([p_partkey@0], 16)
FilterExec: p_brand@1 != Brand#45 AND p_type@2 NOT LIKE MEDIUM POLISHED% AND
p_size@3 IN (SET) ([49, 14, 23, 45, 19, 3, 36, 9])
StatsExec: rows=20000000
-=== Stage 5 ===
+=== Stage 3 ===
SortShuffleWriterExec: partitioning=Hash([ps_suppkey@0], 16)
ProjectionExec: expr=[ps_suppkey@1 as ps_suppkey, p_brand@3 as p_brand,
p_type@4 as p_type, p_size@5 as p_size]
SortMergeJoinExec: join_type=Inner, on=[(ps_partkey@0, p_partkey@0)]
SortExec: expr=[ps_partkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=3, partitioning: Hash([ps_partkey@0], 16)
+ UnresolvedShuffleExec: stage=1, partitioning: Hash([ps_partkey@0], 16)
SortExec: expr=[p_partkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=4, partitioning: Hash([p_partkey@0], 16)
+ UnresolvedShuffleExec: stage=2, partitioning: Hash([p_partkey@0], 16)
-=== Stage 6 ===
+=== Stage 4 ===
+SortShuffleWriterExec: partitioning=Hash([s_suppkey@0], 16)
+ FilterExec: s_comment@1 LIKE %Customer%Complaints%, projection=[s_suppkey@0]
+ StatsExec: rows=1000000
+
+=== Stage 5 ===
SortShuffleWriterExec: partitioning=Hash([p_brand@0, p_type@1, p_size@2], 16)
AggregateExec: mode=Partial, gby=[p_brand@0 as p_brand, p_type@1 as p_type,
p_size@2 as p_size], aggr=[count(alias1)]
- AggregateExec: mode=SinglePartitioned, gby=[p_brand@1 as p_brand, p_type@2
as p_type, p_size@3 as p_size, ps_suppkey@0 as alias1], aggr=[]
- HashJoinExec: mode=CollectLeft, join_type=RightAnti, on=[(s_suppkey@0,
ps_suppkey@0)]
- UnresolvedShuffleExec: stage=2, broadcast=true, upstream_partitions: 16
- UnresolvedShuffleExec: stage=5, partitioning: Hash([ps_suppkey@0], 16)
+ AggregateExec: mode=SinglePartitioned, gby=[p_brand@1 as p_brand, p_type@2
as p_type, p_size@3 as p_size, ps_suppkey@0 as alias1], aggr=[],
ordering_mode=PartiallySorted([3])
+ SortMergeJoinExec: join_type=LeftAnti, on=[(ps_suppkey@0, s_suppkey@0)]
+ SortExec: expr=[ps_suppkey@0 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: stage=3, partitioning: Hash([ps_suppkey@0],
16)
+ SortExec: expr=[s_suppkey@0 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: stage=4, partitioning: Hash([s_suppkey@0], 16)
-=== Stage 7 ===
+=== Stage 6 ===
ShuffleWriterExec: partitioning: None
SortExec: expr=[supplier_cnt@3 DESC, p_brand@0 ASC NULLS LAST, p_type@1 ASC
NULLS LAST, p_size@2 ASC NULLS LAST], preserve_partitioning=[true]
ProjectionExec: expr=[p_brand@0 as p_brand, p_type@1 as p_type, p_size@2
as p_size, count(alias1)@3 as supplier_cnt]
AggregateExec: mode=FinalPartitioned, gby=[p_brand@0 as p_brand,
p_type@1 as p_type, p_size@2 as p_size], aggr=[count(alias1)]
- UnresolvedShuffleExec: stage=6, partitioning: Hash([p_brand@0,
p_type@1, p_size@2], 16)
+ UnresolvedShuffleExec: stage=5, partitioning: Hash([p_brand@0,
p_type@1, p_size@2], 16)
-=== Stage 8 ===
+=== Stage 7 ===
ShuffleWriterExec: partitioning: None
SortPreservingMergeExec: [supplier_cnt@3 DESC, p_brand@0 ASC NULLS LAST,
p_type@1 ASC NULLS LAST, p_size@2 ASC NULLS LAST]
- UnresolvedShuffleExec: stage=7, partitioning: Hash([p_brand@0, p_type@1,
p_size@2], 16)
+ UnresolvedShuffleExec: stage=6, partitioning: Hash([p_brand@0, p_type@1,
p_size@2], 16)
diff --git a/ballista/scheduler/tests/tpch_plan_stability/approved/q2.txt
b/ballista/scheduler/tests/tpch_plan_stability/approved/q2.txt
index a3448fbdf..ee1015189 100644
--- a/ballista/scheduler/tests/tpch_plan_stability/approved/q2.txt
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q2.txt
@@ -1,137 +1,123 @@
=== Stage 1 ===
-SortShuffleWriterExec: partitioning=Hash([r_regionkey@0], 16)
- FilterExec: r_name@1 = EUROPE, projection=[r_regionkey@0]
- StatsExec: rows=5
-
-=== Stage 2 ===
-ShuffleWriterExec: partitioning: None
- UnresolvedShuffleExec: stage=1, partitioning: Hash([r_regionkey@0], 16)
-
-=== Stage 3 ===
-SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
- StatsExec: rows=25
-
-=== Stage 4 ===
-ShuffleWriterExec: partitioning: None
- UnresolvedShuffleExec: stage=3, partitioning: Hash([n_nationkey@0], 16)
-
-=== Stage 5 ===
SortShuffleWriterExec: partitioning=Hash([p_partkey@0], 16)
FilterExec: p_size@3 = 15 AND p_type@2 LIKE %BRASS, projection=[p_partkey@0,
p_mfgr@1]
StatsExec: rows=20000000
-=== Stage 6 ===
+=== Stage 2 ===
SortShuffleWriterExec: partitioning=Hash([ps_partkey@0], 16)
StatsExec: rows=80000000
-=== Stage 7 ===
+=== Stage 3 ===
SortShuffleWriterExec: partitioning=Hash([ps_suppkey@2], 16)
ProjectionExec: expr=[p_partkey@0 as p_partkey, p_mfgr@1 as p_mfgr,
ps_suppkey@3 as ps_suppkey, ps_supplycost@4 as ps_supplycost]
SortMergeJoinExec: join_type=Inner, on=[(p_partkey@0, ps_partkey@0)]
SortExec: expr=[p_partkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=5, partitioning: Hash([p_partkey@0], 16)
+ UnresolvedShuffleExec: stage=1, partitioning: Hash([p_partkey@0], 16)
SortExec: expr=[ps_partkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=6, partitioning: Hash([ps_partkey@0], 16)
+ UnresolvedShuffleExec: stage=2, partitioning: Hash([ps_partkey@0], 16)
-=== Stage 8 ===
+=== Stage 4 ===
SortShuffleWriterExec: partitioning=Hash([s_suppkey@0], 16)
StatsExec: rows=1000000
-=== Stage 9 ===
+=== Stage 5 ===
SortShuffleWriterExec: partitioning=Hash([s_nationkey@4], 16)
ProjectionExec: expr=[p_partkey@0 as p_partkey, p_mfgr@1 as p_mfgr, s_name@5
as s_name, s_address@6 as s_address, s_nationkey@7 as s_nationkey, s_phone@8 as
s_phone, s_acctbal@9 as s_acctbal, s_comment@10 as s_comment, ps_supplycost@3
as ps_supplycost]
SortMergeJoinExec: join_type=Inner, on=[(ps_suppkey@2, s_suppkey@0)]
SortExec: expr=[ps_suppkey@2 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=7, partitioning: Hash([ps_suppkey@2], 16)
+ UnresolvedShuffleExec: stage=3, partitioning: Hash([ps_suppkey@2], 16)
SortExec: expr=[s_suppkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=8, partitioning: Hash([s_suppkey@0], 16)
+ UnresolvedShuffleExec: stage=4, partitioning: Hash([s_suppkey@0], 16)
-=== Stage 10 ===
+=== Stage 6 ===
+SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
+ StatsExec: rows=25
+
+=== Stage 7 ===
SortShuffleWriterExec: partitioning=Hash([n_regionkey@9], 16)
ProjectionExec: expr=[p_partkey@0 as p_partkey, p_mfgr@1 as p_mfgr, s_name@2
as s_name, s_address@3 as s_address, s_phone@5 as s_phone, s_acctbal@6 as
s_acctbal, s_comment@7 as s_comment, ps_supplycost@8 as ps_supplycost,
n_name@10 as n_name, n_regionkey@11 as n_regionkey]
- ProjectionExec: expr=[p_partkey@3 as p_partkey, p_mfgr@4 as p_mfgr,
s_name@5 as s_name, s_address@6 as s_address, s_nationkey@7 as s_nationkey,
s_phone@8 as s_phone, s_acctbal@9 as s_acctbal, s_comment@10 as s_comment,
ps_supplycost@11 as ps_supplycost, n_nationkey@0 as n_nationkey, n_name@1 as
n_name, n_regionkey@2 as n_regionkey]
- HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(n_nationkey@0,
s_nationkey@4)]
- UnresolvedShuffleExec: stage=4, broadcast=true, upstream_partitions: 16
- UnresolvedShuffleExec: stage=9, partitioning: Hash([s_nationkey@4], 16)
-
-=== Stage 11 ===
-SortShuffleWriterExec: partitioning=Hash([p_partkey@0, ps_supplycost@7], 16)
- ProjectionExec: expr=[p_partkey@0 as p_partkey, p_mfgr@1 as p_mfgr, s_name@2
as s_name, s_address@3 as s_address, s_phone@4 as s_phone, s_acctbal@5 as
s_acctbal, s_comment@6 as s_comment, ps_supplycost@7 as ps_supplycost, n_name@8
as n_name]
- ProjectionExec: expr=[p_partkey@1 as p_partkey, p_mfgr@2 as p_mfgr,
s_name@3 as s_name, s_address@4 as s_address, s_phone@5 as s_phone, s_acctbal@6
as s_acctbal, s_comment@7 as s_comment, ps_supplycost@8 as ps_supplycost,
n_name@9 as n_name, n_regionkey@10 as n_regionkey, r_regionkey@0 as r_regionkey]
- HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(r_regionkey@0,
n_regionkey@9)]
- UnresolvedShuffleExec: stage=2, broadcast=true, upstream_partitions: 16
- UnresolvedShuffleExec: stage=10, partitioning: Hash([n_regionkey@9],
16)
-
-=== Stage 12 ===
-ShuffleWriterExec: partitioning: None
- UnresolvedShuffleExec: stage=11, partitioning: Hash([p_partkey@0,
ps_supplycost@7], 16)
+ SortMergeJoinExec: join_type=Inner, on=[(s_nationkey@4, n_nationkey@0)]
+ SortExec: expr=[s_nationkey@4 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: stage=5, partitioning: Hash([s_nationkey@4], 16)
+ SortExec: expr=[n_nationkey@0 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: stage=6, partitioning: Hash([n_nationkey@0], 16)
-=== Stage 13 ===
+=== Stage 8 ===
SortShuffleWriterExec: partitioning=Hash([r_regionkey@0], 16)
FilterExec: r_name@1 = EUROPE, projection=[r_regionkey@0]
StatsExec: rows=5
-=== Stage 14 ===
-ShuffleWriterExec: partitioning: None
- UnresolvedShuffleExec: stage=13, partitioning: Hash([r_regionkey@0], 16)
-
-=== Stage 15 ===
-SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
- StatsExec: rows=25
-
-=== Stage 16 ===
-ShuffleWriterExec: partitioning: None
- UnresolvedShuffleExec: stage=15, partitioning: Hash([n_nationkey@0], 16)
+=== Stage 9 ===
+SortShuffleWriterExec: partitioning=Hash([p_partkey@0, ps_supplycost@7], 16)
+ ProjectionExec: expr=[p_partkey@0 as p_partkey, p_mfgr@1 as p_mfgr, s_name@2
as s_name, s_address@3 as s_address, s_phone@4 as s_phone, s_acctbal@5 as
s_acctbal, s_comment@6 as s_comment, ps_supplycost@7 as ps_supplycost, n_name@8
as n_name]
+ SortMergeJoinExec: join_type=Inner, on=[(n_regionkey@9, r_regionkey@0)]
+ SortExec: expr=[n_regionkey@9 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: stage=7, partitioning: Hash([n_regionkey@9], 16)
+ SortExec: expr=[r_regionkey@0 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: stage=8, partitioning: Hash([r_regionkey@0], 16)
-=== Stage 17 ===
+=== Stage 10 ===
SortShuffleWriterExec: partitioning=Hash([ps_suppkey@1], 16)
StatsExec: rows=80000000
-=== Stage 18 ===
+=== Stage 11 ===
SortShuffleWriterExec: partitioning=Hash([s_suppkey@0], 16)
StatsExec: rows=1000000
-=== Stage 19 ===
+=== Stage 12 ===
SortShuffleWriterExec: partitioning=Hash([s_nationkey@2], 16)
ProjectionExec: expr=[ps_partkey@0 as ps_partkey, ps_supplycost@2 as
ps_supplycost, s_nationkey@4 as s_nationkey]
SortMergeJoinExec: join_type=Inner, on=[(ps_suppkey@1, s_suppkey@0)]
SortExec: expr=[ps_suppkey@1 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=17, partitioning: Hash([ps_suppkey@1], 16)
+ UnresolvedShuffleExec: stage=10, partitioning: Hash([ps_suppkey@1], 16)
SortExec: expr=[s_suppkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=18, partitioning: Hash([s_suppkey@0], 16)
+ UnresolvedShuffleExec: stage=11, partitioning: Hash([s_suppkey@0], 16)
-=== Stage 20 ===
+=== Stage 13 ===
+SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
+ StatsExec: rows=25
+
+=== Stage 14 ===
SortShuffleWriterExec: partitioning=Hash([n_regionkey@2], 16)
ProjectionExec: expr=[ps_partkey@0 as ps_partkey, ps_supplycost@1 as
ps_supplycost, n_regionkey@4 as n_regionkey]
- ProjectionExec: expr=[ps_partkey@2 as ps_partkey, ps_supplycost@3 as
ps_supplycost, s_nationkey@4 as s_nationkey, n_nationkey@0 as n_nationkey,
n_regionkey@1 as n_regionkey]
- HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(n_nationkey@0,
s_nationkey@2)]
- UnresolvedShuffleExec: stage=16, broadcast=true, upstream_partitions:
16
- UnresolvedShuffleExec: stage=19, partitioning: Hash([s_nationkey@2],
16)
+ SortMergeJoinExec: join_type=Inner, on=[(s_nationkey@2, n_nationkey@0)]
+ SortExec: expr=[s_nationkey@2 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: stage=12, partitioning: Hash([s_nationkey@2],
16)
+ SortExec: expr=[n_nationkey@0 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: stage=13, partitioning: Hash([n_nationkey@0],
16)
-=== Stage 21 ===
+=== Stage 15 ===
+SortShuffleWriterExec: partitioning=Hash([r_regionkey@0], 16)
+ FilterExec: r_name@1 = EUROPE, projection=[r_regionkey@0]
+ StatsExec: rows=5
+
+=== Stage 16 ===
SortShuffleWriterExec: partitioning=Hash([ps_partkey@0], 16)
AggregateExec: mode=Partial, gby=[ps_partkey@0 as ps_partkey],
aggr=[min(partsupp.ps_supplycost)]
ProjectionExec: expr=[ps_partkey@0 as ps_partkey, ps_supplycost@1 as
ps_supplycost]
- ProjectionExec: expr=[ps_partkey@1 as ps_partkey, ps_supplycost@2 as
ps_supplycost, n_regionkey@3 as n_regionkey, r_regionkey@0 as r_regionkey]
- HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(r_regionkey@0,
n_regionkey@2)]
- UnresolvedShuffleExec: stage=14, broadcast=true,
upstream_partitions: 16
- UnresolvedShuffleExec: stage=20, partitioning: Hash([n_regionkey@2],
16)
+ SortMergeJoinExec: join_type=Inner, on=[(n_regionkey@2, r_regionkey@0)]
+ SortExec: expr=[n_regionkey@2 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: stage=14, partitioning: Hash([n_regionkey@2],
16)
+ SortExec: expr=[r_regionkey@0 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: stage=15, partitioning: Hash([r_regionkey@0],
16)
-=== Stage 22 ===
+=== Stage 17 ===
SortShuffleWriterExec: partitioning=Hash([ps_partkey@1,
min(partsupp.ps_supplycost)@0], 16)
ProjectionExec: expr=[min(partsupp.ps_supplycost)@1 as
min(partsupp.ps_supplycost), ps_partkey@0 as ps_partkey]
AggregateExec: mode=FinalPartitioned, gby=[ps_partkey@0 as ps_partkey],
aggr=[min(partsupp.ps_supplycost)]
- UnresolvedShuffleExec: stage=21, partitioning: Hash([ps_partkey@0], 16)
+ UnresolvedShuffleExec: stage=16, partitioning: Hash([ps_partkey@0], 16)
-=== Stage 23 ===
+=== Stage 18 ===
ShuffleWriterExec: partitioning: None
SortExec: TopK(fetch=100), expr=[s_acctbal@0 DESC, n_name@2 ASC NULLS LAST,
s_name@1 ASC NULLS LAST, p_partkey@3 ASC NULLS LAST],
preserve_partitioning=[true]
ProjectionExec: expr=[s_acctbal@5 as s_acctbal, s_name@2 as s_name,
n_name@8 as n_name, p_partkey@0 as p_partkey, p_mfgr@1 as p_mfgr, s_address@3
as s_address, s_phone@4 as s_phone, s_comment@6 as s_comment]
- HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(p_partkey@0,
ps_partkey@1), (ps_supplycost@7, min(partsupp.ps_supplycost)@0)]
- UnresolvedShuffleExec: stage=12, broadcast=true, upstream_partitions:
16
- UnresolvedShuffleExec: stage=22, partitioning: Hash([ps_partkey@1,
min(partsupp.ps_supplycost)@0], 16)
+ SortMergeJoinExec: join_type=Inner, on=[(p_partkey@0, ps_partkey@1),
(ps_supplycost@7, min(partsupp.ps_supplycost)@0)]
+ SortExec: expr=[p_partkey@0 ASC, ps_supplycost@7 ASC],
preserve_partitioning=[true]
+ UnresolvedShuffleExec: stage=9, partitioning: Hash([p_partkey@0,
ps_supplycost@7], 16)
+ SortExec: expr=[ps_partkey@1 ASC, min(partsupp.ps_supplycost)@0 ASC],
preserve_partitioning=[true]
+ UnresolvedShuffleExec: stage=17, partitioning: Hash([ps_partkey@1,
min(partsupp.ps_supplycost)@0], 16)
-=== Stage 24 ===
+=== Stage 19 ===
ShuffleWriterExec: partitioning: None
SortPreservingMergeExec: [s_acctbal@0 DESC, n_name@2 ASC NULLS LAST,
s_name@1 ASC NULLS LAST, p_partkey@3 ASC NULLS LAST], fetch=100
- UnresolvedShuffleExec: stage=23, partitioning: Hash([p_partkey@3,
min(partsupp.ps_supplycost)@9], 16)
+ UnresolvedShuffleExec: stage=18, partitioning: Hash([p_partkey@3,
min(partsupp.ps_supplycost)@9], 16)
diff --git a/ballista/scheduler/tests/tpch_plan_stability/approved/q20.txt
b/ballista/scheduler/tests/tpch_plan_stability/approved/q20.txt
index 0066e4843..de78d20fb 100644
--- a/ballista/scheduler/tests/tpch_plan_stability/approved/q20.txt
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q20.txt
@@ -1,69 +1,66 @@
=== Stage 1 ===
+SortShuffleWriterExec: partitioning=Hash([s_nationkey@3], 16)
+ StatsExec: rows=1000000
+
+=== Stage 2 ===
SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
FilterExec: n_name@1 = CANADA, projection=[n_nationkey@0]
StatsExec: rows=25
-=== Stage 2 ===
-ShuffleWriterExec: partitioning: None
- UnresolvedShuffleExec: stage=1, partitioning: Hash([n_nationkey@0], 16)
-
=== Stage 3 ===
-SortShuffleWriterExec: partitioning=Hash([s_nationkey@3], 16)
- StatsExec: rows=1000000
-
-=== Stage 4 ===
SortShuffleWriterExec: partitioning=Hash([s_suppkey@0], 16)
ProjectionExec: expr=[s_suppkey@0 as s_suppkey, s_name@1 as s_name,
s_address@2 as s_address]
- ProjectionExec: expr=[s_suppkey@1 as s_suppkey, s_name@2 as s_name,
s_address@3 as s_address, s_nationkey@4 as s_nationkey, n_nationkey@0 as
n_nationkey]
- HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(n_nationkey@0,
s_nationkey@3)]
- UnresolvedShuffleExec: stage=2, broadcast=true, upstream_partitions: 16
- UnresolvedShuffleExec: stage=3, partitioning: Hash([s_nationkey@3], 16)
+ SortMergeJoinExec: join_type=Inner, on=[(s_nationkey@3, n_nationkey@0)]
+ SortExec: expr=[s_nationkey@3 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: stage=1, partitioning: Hash([s_nationkey@3], 16)
+ SortExec: expr=[n_nationkey@0 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: stage=2, partitioning: Hash([n_nationkey@0], 16)
-=== Stage 5 ===
+=== Stage 4 ===
SortShuffleWriterExec: partitioning=Hash([ps_partkey@0], 16)
StatsExec: rows=80000000
-=== Stage 6 ===
+=== Stage 5 ===
SortShuffleWriterExec: partitioning=Hash([p_partkey@0], 16)
FilterExec: p_name@1 LIKE forest%, projection=[p_partkey@0]
StatsExec: rows=20000000
-=== Stage 7 ===
+=== Stage 6 ===
SortShuffleWriterExec: partitioning=Hash([ps_partkey@0, ps_suppkey@1], 16)
SortMergeJoinExec: join_type=LeftSemi, on=[(ps_partkey@0, p_partkey@0)]
SortExec: expr=[ps_partkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=5, partitioning: Hash([ps_partkey@0], 16)
+ UnresolvedShuffleExec: stage=4, partitioning: Hash([ps_partkey@0], 16)
SortExec: expr=[p_partkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=6, partitioning: Hash([p_partkey@0], 16)
+ UnresolvedShuffleExec: stage=5, partitioning: Hash([p_partkey@0], 16)
-=== Stage 8 ===
+=== Stage 7 ===
SortShuffleWriterExec: partitioning=Hash([l_partkey@0, l_suppkey@1], 16)
AggregateExec: mode=Partial, gby=[l_partkey@0 as l_partkey, l_suppkey@1 as
l_suppkey], aggr=[sum(lineitem.l_quantity)]
FilterExec: l_shipdate@3 >= 1994-01-01 AND l_shipdate@3 < 1995-01-01,
projection=[l_partkey@0, l_suppkey@1, l_quantity@2]
StatsExec: rows=600037902
-=== Stage 9 ===
+=== Stage 8 ===
SortShuffleWriterExec: partitioning=Hash([ps_suppkey@0], 16)
ProjectionExec: expr=[ps_suppkey@1 as ps_suppkey]
SortMergeJoinExec: join_type=Inner, on=[(ps_partkey@0, l_partkey@1),
(ps_suppkey@1, l_suppkey@2)], filter=CAST(ps_availqty@0 AS Float64) >
Float64(0.5) * sum(lineitem.l_quantity)@1
SortExec: expr=[ps_partkey@0 ASC, ps_suppkey@1 ASC],
preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=7, partitioning: Hash([ps_partkey@0,
ps_suppkey@1], 16)
+ UnresolvedShuffleExec: stage=6, partitioning: Hash([ps_partkey@0,
ps_suppkey@1], 16)
SortExec: expr=[l_partkey@1 ASC, l_suppkey@2 ASC],
preserve_partitioning=[true]
ProjectionExec: expr=[0.5 * CAST(sum(lineitem.l_quantity)@2 AS
Float64) as Float64(0.5) * sum(lineitem.l_quantity), l_partkey@0 as l_partkey,
l_suppkey@1 as l_suppkey]
AggregateExec: mode=FinalPartitioned, gby=[l_partkey@0 as l_partkey,
l_suppkey@1 as l_suppkey], aggr=[sum(lineitem.l_quantity)]
- UnresolvedShuffleExec: stage=8, partitioning: Hash([l_partkey@0,
l_suppkey@1], 16)
+ UnresolvedShuffleExec: stage=7, partitioning: Hash([l_partkey@0,
l_suppkey@1], 16)
-=== Stage 10 ===
+=== Stage 9 ===
ShuffleWriterExec: partitioning: None
SortExec: expr=[s_name@0 ASC NULLS LAST], preserve_partitioning=[true]
ProjectionExec: expr=[s_name@1 as s_name, s_address@2 as s_address]
SortMergeJoinExec: join_type=LeftSemi, on=[(s_suppkey@0, ps_suppkey@0)]
SortExec: expr=[s_suppkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=4, partitioning: Hash([s_suppkey@0], 16)
+ UnresolvedShuffleExec: stage=3, partitioning: Hash([s_suppkey@0], 16)
SortExec: expr=[ps_suppkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=9, partitioning: Hash([ps_suppkey@0],
16)
+ UnresolvedShuffleExec: stage=8, partitioning: Hash([ps_suppkey@0],
16)
-=== Stage 11 ===
+=== Stage 10 ===
ShuffleWriterExec: partitioning: None
SortPreservingMergeExec: [s_name@0 ASC NULLS LAST]
- UnresolvedShuffleExec: stage=10, partitioning: Hash([s_suppkey@0], 16)
+ UnresolvedShuffleExec: stage=9, partitioning: Hash([s_suppkey@0], 16)
diff --git a/ballista/scheduler/tests/tpch_plan_stability/approved/q21.txt
b/ballista/scheduler/tests/tpch_plan_stability/approved/q21.txt
index bd49954af..003c45c93 100644
--- a/ballista/scheduler/tests/tpch_plan_stability/approved/q21.txt
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q21.txt
@@ -1,82 +1,79 @@
=== Stage 1 ===
-SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
- FilterExec: n_name@1 = SAUDI ARABIA, projection=[n_nationkey@0]
- StatsExec: rows=25
-
-=== Stage 2 ===
-ShuffleWriterExec: partitioning: None
- UnresolvedShuffleExec: stage=1, partitioning: Hash([n_nationkey@0], 16)
-
-=== Stage 3 ===
SortShuffleWriterExec: partitioning=Hash([s_suppkey@0], 16)
StatsExec: rows=1000000
-=== Stage 4 ===
+=== Stage 2 ===
SortShuffleWriterExec: partitioning=Hash([l_suppkey@1], 16)
FilterExec: l_receiptdate@3 > l_commitdate@2, projection=[l_orderkey@0,
l_suppkey@1]
StatsExec: rows=600037902
-=== Stage 5 ===
+=== Stage 3 ===
SortShuffleWriterExec: partitioning=Hash([l_orderkey@2], 16)
ProjectionExec: expr=[s_name@1 as s_name, s_nationkey@2 as s_nationkey,
l_orderkey@3 as l_orderkey, l_suppkey@4 as l_suppkey]
SortMergeJoinExec: join_type=Inner, on=[(s_suppkey@0, l_suppkey@1)]
SortExec: expr=[s_suppkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=3, partitioning: Hash([s_suppkey@0], 16)
+ UnresolvedShuffleExec: stage=1, partitioning: Hash([s_suppkey@0], 16)
SortExec: expr=[l_suppkey@1 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=4, partitioning: Hash([l_suppkey@1], 16)
+ UnresolvedShuffleExec: stage=2, partitioning: Hash([l_suppkey@1], 16)
-=== Stage 6 ===
+=== Stage 4 ===
SortShuffleWriterExec: partitioning=Hash([o_orderkey@0], 16)
FilterExec: o_orderstatus@1 = F, projection=[o_orderkey@0]
StatsExec: rows=150000000
-=== Stage 7 ===
+=== Stage 5 ===
SortShuffleWriterExec: partitioning=Hash([s_nationkey@1], 16)
ProjectionExec: expr=[s_name@0 as s_name, s_nationkey@1 as s_nationkey,
l_orderkey@2 as l_orderkey, l_suppkey@3 as l_suppkey]
SortMergeJoinExec: join_type=Inner, on=[(l_orderkey@2, o_orderkey@0)]
SortExec: expr=[l_orderkey@2 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=5, partitioning: Hash([l_orderkey@2], 16)
+ UnresolvedShuffleExec: stage=3, partitioning: Hash([l_orderkey@2], 16)
SortExec: expr=[o_orderkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=6, partitioning: Hash([o_orderkey@0], 16)
+ UnresolvedShuffleExec: stage=4, partitioning: Hash([o_orderkey@0], 16)
-=== Stage 8 ===
+=== Stage 6 ===
+SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
+ FilterExec: n_name@1 = SAUDI ARABIA, projection=[n_nationkey@0]
+ StatsExec: rows=25
+
+=== Stage 7 ===
SortShuffleWriterExec: partitioning=Hash([l_orderkey@1], 16)
ProjectionExec: expr=[s_name@0 as s_name, l_orderkey@2 as l_orderkey,
l_suppkey@3 as l_suppkey]
- ProjectionExec: expr=[s_name@1 as s_name, s_nationkey@2 as s_nationkey,
l_orderkey@3 as l_orderkey, l_suppkey@4 as l_suppkey, n_nationkey@0 as
n_nationkey]
- HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(n_nationkey@0,
s_nationkey@1)]
- UnresolvedShuffleExec: stage=2, broadcast=true, upstream_partitions: 16
- UnresolvedShuffleExec: stage=7, partitioning: Hash([s_nationkey@1], 16)
+ SortMergeJoinExec: join_type=Inner, on=[(s_nationkey@1, n_nationkey@0)]
+ SortExec: expr=[s_nationkey@1 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: stage=5, partitioning: Hash([s_nationkey@1], 16)
+ SortExec: expr=[n_nationkey@0 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: stage=6, partitioning: Hash([n_nationkey@0], 16)
-=== Stage 9 ===
+=== Stage 8 ===
SortShuffleWriterExec: partitioning=Hash([l_orderkey@0], 16)
StatsExec: rows=600037902
-=== Stage 10 ===
+=== Stage 9 ===
SortShuffleWriterExec: partitioning=Hash([l_orderkey@0], 16)
FilterExec: l_receiptdate@3 > l_commitdate@2, projection=[l_orderkey@0,
l_suppkey@1]
StatsExec: rows=600037902
-=== Stage 11 ===
+=== Stage 10 ===
SortShuffleWriterExec: partitioning=Hash([s_name@0], 16)
AggregateExec: mode=Partial, gby=[s_name@0 as s_name], aggr=[count(Int64(1))]
ProjectionExec: expr=[s_name@0 as s_name]
SortMergeJoinExec: join_type=LeftAnti, on=[(l_orderkey@1,
l_orderkey@0)], filter=l_suppkey@1 != l_suppkey@0
SortMergeJoinExec: join_type=LeftSemi, on=[(l_orderkey@1,
l_orderkey@0)], filter=l_suppkey@1 != l_suppkey@0
SortExec: expr=[l_orderkey@1 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=8, partitioning: Hash([l_orderkey@1],
16)
+ UnresolvedShuffleExec: stage=7, partitioning: Hash([l_orderkey@1],
16)
SortExec: expr=[l_orderkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=9, partitioning: Hash([l_orderkey@0],
16)
+ UnresolvedShuffleExec: stage=8, partitioning: Hash([l_orderkey@0],
16)
SortExec: expr=[l_orderkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=10, partitioning: Hash([l_orderkey@0],
16)
+ UnresolvedShuffleExec: stage=9, partitioning: Hash([l_orderkey@0],
16)
-=== Stage 12 ===
+=== Stage 11 ===
ShuffleWriterExec: partitioning: None
SortExec: TopK(fetch=100), expr=[numwait@1 DESC, s_name@0 ASC NULLS LAST],
preserve_partitioning=[true]
ProjectionExec: expr=[s_name@0 as s_name, count(Int64(1))@1 as numwait]
AggregateExec: mode=FinalPartitioned, gby=[s_name@0 as s_name],
aggr=[count(Int64(1))]
- UnresolvedShuffleExec: stage=11, partitioning: Hash([s_name@0], 16)
+ UnresolvedShuffleExec: stage=10, partitioning: Hash([s_name@0], 16)
-=== Stage 13 ===
+=== Stage 12 ===
ShuffleWriterExec: partitioning: None
SortPreservingMergeExec: [numwait@1 DESC, s_name@0 ASC NULLS LAST], fetch=100
- UnresolvedShuffleExec: stage=12, partitioning: Hash([s_name@0], 16)
+ UnresolvedShuffleExec: stage=11, partitioning: Hash([s_name@0], 16)
diff --git a/ballista/scheduler/tests/tpch_plan_stability/approved/q5.txt
b/ballista/scheduler/tests/tpch_plan_stability/approved/q5.txt
index 78c140e22..8849e6c0b 100644
--- a/ballista/scheduler/tests/tpch_plan_stability/approved/q5.txt
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q5.txt
@@ -1,89 +1,83 @@
=== Stage 1 ===
-SortShuffleWriterExec: partitioning=Hash([r_regionkey@0], 16)
- FilterExec: r_name@1 = ASIA, projection=[r_regionkey@0]
- StatsExec: rows=5
-
-=== Stage 2 ===
-ShuffleWriterExec: partitioning: None
- UnresolvedShuffleExec: stage=1, partitioning: Hash([r_regionkey@0], 16)
-
-=== Stage 3 ===
-SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
- StatsExec: rows=25
-
-=== Stage 4 ===
-ShuffleWriterExec: partitioning: None
- UnresolvedShuffleExec: stage=3, partitioning: Hash([n_nationkey@0], 16)
-
-=== Stage 5 ===
SortShuffleWriterExec: partitioning=Hash([c_custkey@0], 16)
StatsExec: rows=15000000
-=== Stage 6 ===
+=== Stage 2 ===
SortShuffleWriterExec: partitioning=Hash([o_custkey@1], 16)
FilterExec: o_orderdate@2 >= 1994-01-01 AND o_orderdate@2 < 1995-01-01,
projection=[o_orderkey@0, o_custkey@1]
StatsExec: rows=150000000
-=== Stage 7 ===
+=== Stage 3 ===
SortShuffleWriterExec: partitioning=Hash([o_orderkey@1], 16)
ProjectionExec: expr=[c_nationkey@1 as c_nationkey, o_orderkey@2 as
o_orderkey]
SortMergeJoinExec: join_type=Inner, on=[(c_custkey@0, o_custkey@1)]
SortExec: expr=[c_custkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=5, partitioning: Hash([c_custkey@0], 16)
+ UnresolvedShuffleExec: stage=1, partitioning: Hash([c_custkey@0], 16)
SortExec: expr=[o_custkey@1 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=6, partitioning: Hash([o_custkey@1], 16)
+ UnresolvedShuffleExec: stage=2, partitioning: Hash([o_custkey@1], 16)
-=== Stage 8 ===
+=== Stage 4 ===
SortShuffleWriterExec: partitioning=Hash([l_orderkey@0], 16)
StatsExec: rows=600037902
-=== Stage 9 ===
+=== Stage 5 ===
SortShuffleWriterExec: partitioning=Hash([l_suppkey@1, c_nationkey@0], 16)
ProjectionExec: expr=[c_nationkey@0 as c_nationkey, l_suppkey@3 as
l_suppkey, l_extendedprice@4 as l_extendedprice, l_discount@5 as l_discount]
SortMergeJoinExec: join_type=Inner, on=[(o_orderkey@1, l_orderkey@0)]
SortExec: expr=[o_orderkey@1 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=7, partitioning: Hash([o_orderkey@1], 16)
+ UnresolvedShuffleExec: stage=3, partitioning: Hash([o_orderkey@1], 16)
SortExec: expr=[l_orderkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=8, partitioning: Hash([l_orderkey@0], 16)
+ UnresolvedShuffleExec: stage=4, partitioning: Hash([l_orderkey@0], 16)
-=== Stage 10 ===
+=== Stage 6 ===
SortShuffleWriterExec: partitioning=Hash([s_suppkey@0, s_nationkey@1], 16)
StatsExec: rows=1000000
-=== Stage 11 ===
+=== Stage 7 ===
SortShuffleWriterExec: partitioning=Hash([s_nationkey@2], 16)
ProjectionExec: expr=[l_extendedprice@2 as l_extendedprice, l_discount@3 as
l_discount, s_nationkey@5 as s_nationkey]
SortMergeJoinExec: join_type=Inner, on=[(l_suppkey@1, s_suppkey@0),
(c_nationkey@0, s_nationkey@1)]
SortExec: expr=[l_suppkey@1 ASC, c_nationkey@0 ASC],
preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=9, partitioning: Hash([l_suppkey@1,
c_nationkey@0], 16)
+ UnresolvedShuffleExec: stage=5, partitioning: Hash([l_suppkey@1,
c_nationkey@0], 16)
SortExec: expr=[s_suppkey@0 ASC, s_nationkey@1 ASC],
preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=10, partitioning: Hash([s_suppkey@0,
s_nationkey@1], 16)
+ UnresolvedShuffleExec: stage=6, partitioning: Hash([s_suppkey@0,
s_nationkey@1], 16)
-=== Stage 12 ===
+=== Stage 8 ===
+SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
+ StatsExec: rows=25
+
+=== Stage 9 ===
SortShuffleWriterExec: partitioning=Hash([n_regionkey@3], 16)
ProjectionExec: expr=[l_extendedprice@0 as l_extendedprice, l_discount@1 as
l_discount, n_name@4 as n_name, n_regionkey@5 as n_regionkey]
- ProjectionExec: expr=[l_extendedprice@3 as l_extendedprice, l_discount@4
as l_discount, s_nationkey@5 as s_nationkey, n_nationkey@0 as n_nationkey,
n_name@1 as n_name, n_regionkey@2 as n_regionkey]
- HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(n_nationkey@0,
s_nationkey@2)]
- UnresolvedShuffleExec: stage=4, broadcast=true, upstream_partitions: 16
- UnresolvedShuffleExec: stage=11, partitioning: Hash([s_nationkey@2],
16)
+ SortMergeJoinExec: join_type=Inner, on=[(s_nationkey@2, n_nationkey@0)]
+ SortExec: expr=[s_nationkey@2 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: stage=7, partitioning: Hash([s_nationkey@2], 16)
+ SortExec: expr=[n_nationkey@0 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: stage=8, partitioning: Hash([n_nationkey@0], 16)
-=== Stage 13 ===
+=== Stage 10 ===
+SortShuffleWriterExec: partitioning=Hash([r_regionkey@0], 16)
+ FilterExec: r_name@1 = ASIA, projection=[r_regionkey@0]
+ StatsExec: rows=5
+
+=== Stage 11 ===
SortShuffleWriterExec: partitioning=Hash([n_name@0], 16)
AggregateExec: mode=Partial, gby=[n_name@2 as n_name],
aggr=[sum(lineitem.l_extendedprice * Some(1),20,0 - lineitem.l_discount) as
sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)]
ProjectionExec: expr=[l_extendedprice@0 as l_extendedprice, l_discount@1
as l_discount, n_name@2 as n_name]
- ProjectionExec: expr=[l_extendedprice@1 as l_extendedprice, l_discount@2
as l_discount, n_name@3 as n_name, n_regionkey@4 as n_regionkey, r_regionkey@0
as r_regionkey]
- HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(r_regionkey@0,
n_regionkey@3)]
- UnresolvedShuffleExec: stage=2, broadcast=true, upstream_partitions:
16
- UnresolvedShuffleExec: stage=12, partitioning: Hash([n_regionkey@3],
16)
+ SortMergeJoinExec: join_type=Inner, on=[(n_regionkey@3, r_regionkey@0)]
+ SortExec: expr=[n_regionkey@3 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: stage=9, partitioning: Hash([n_regionkey@3],
16)
+ SortExec: expr=[r_regionkey@0 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: stage=10, partitioning: Hash([r_regionkey@0],
16)
-=== Stage 14 ===
+=== Stage 12 ===
ShuffleWriterExec: partitioning: None
SortExec: expr=[revenue@1 DESC], preserve_partitioning=[true]
ProjectionExec: expr=[n_name@0 as n_name, sum(lineitem.l_extendedprice *
Int64(1) - lineitem.l_discount)@1 as revenue]
AggregateExec: mode=FinalPartitioned, gby=[n_name@0 as n_name],
aggr=[sum(lineitem.l_extendedprice * Some(1),20,0 - lineitem.l_discount) as
sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)]
- UnresolvedShuffleExec: stage=13, partitioning: Hash([n_name@0], 16)
+ UnresolvedShuffleExec: stage=11, partitioning: Hash([n_name@0], 16)
-=== Stage 15 ===
+=== Stage 13 ===
ShuffleWriterExec: partitioning: None
SortPreservingMergeExec: [revenue@1 DESC]
- UnresolvedShuffleExec: stage=14, partitioning: Hash([n_name@0], 16)
+ UnresolvedShuffleExec: stage=12, partitioning: Hash([n_name@0], 16)
diff --git a/ballista/scheduler/tests/tpch_plan_stability/approved/q7.txt
b/ballista/scheduler/tests/tpch_plan_stability/approved/q7.txt
index abfcad9fb..e2fbadd62 100644
--- a/ballista/scheduler/tests/tpch_plan_stability/approved/q7.txt
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q7.txt
@@ -1,89 +1,84 @@
=== Stage 1 ===
-SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
- FilterExec: n_name@1 = FRANCE OR n_name@1 = GERMANY
- StatsExec: rows=25
-
-=== Stage 2 ===
-ShuffleWriterExec: partitioning: None
- UnresolvedShuffleExec: stage=1, partitioning: Hash([n_nationkey@0], 16)
-
-=== Stage 3 ===
SortShuffleWriterExec: partitioning=Hash([s_suppkey@0], 16)
StatsExec: rows=1000000
-=== Stage 4 ===
+=== Stage 2 ===
SortShuffleWriterExec: partitioning=Hash([l_suppkey@1], 16)
FilterExec: l_shipdate@4 >= 1995-01-01 AND l_shipdate@4 <= 1996-12-31
StatsExec: rows=600037902
-=== Stage 5 ===
+=== Stage 3 ===
SortShuffleWriterExec: partitioning=Hash([l_orderkey@1], 16)
ProjectionExec: expr=[s_nationkey@1 as s_nationkey, l_orderkey@2 as
l_orderkey, l_extendedprice@4 as l_extendedprice, l_discount@5 as l_discount,
l_shipdate@6 as l_shipdate]
SortMergeJoinExec: join_type=Inner, on=[(s_suppkey@0, l_suppkey@1)]
SortExec: expr=[s_suppkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=3, partitioning: Hash([s_suppkey@0], 16)
+ UnresolvedShuffleExec: stage=1, partitioning: Hash([s_suppkey@0], 16)
SortExec: expr=[l_suppkey@1 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=4, partitioning: Hash([l_suppkey@1], 16)
+ UnresolvedShuffleExec: stage=2, partitioning: Hash([l_suppkey@1], 16)
-=== Stage 6 ===
+=== Stage 4 ===
SortShuffleWriterExec: partitioning=Hash([o_orderkey@0], 16)
StatsExec: rows=150000000
-=== Stage 7 ===
+=== Stage 5 ===
SortShuffleWriterExec: partitioning=Hash([o_custkey@4], 16)
ProjectionExec: expr=[s_nationkey@0 as s_nationkey, l_extendedprice@2 as
l_extendedprice, l_discount@3 as l_discount, l_shipdate@4 as l_shipdate,
o_custkey@6 as o_custkey]
SortMergeJoinExec: join_type=Inner, on=[(l_orderkey@1, o_orderkey@0)]
SortExec: expr=[l_orderkey@1 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=5, partitioning: Hash([l_orderkey@1], 16)
+ UnresolvedShuffleExec: stage=3, partitioning: Hash([l_orderkey@1], 16)
SortExec: expr=[o_orderkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=6, partitioning: Hash([o_orderkey@0], 16)
+ UnresolvedShuffleExec: stage=4, partitioning: Hash([o_orderkey@0], 16)
-=== Stage 8 ===
+=== Stage 6 ===
SortShuffleWriterExec: partitioning=Hash([c_custkey@0], 16)
StatsExec: rows=15000000
-=== Stage 9 ===
+=== Stage 7 ===
SortShuffleWriterExec: partitioning=Hash([s_nationkey@0], 16)
ProjectionExec: expr=[s_nationkey@0 as s_nationkey, l_extendedprice@1 as
l_extendedprice, l_discount@2 as l_discount, l_shipdate@3 as l_shipdate,
c_nationkey@6 as c_nationkey]
SortMergeJoinExec: join_type=Inner, on=[(o_custkey@4, c_custkey@0)]
SortExec: expr=[o_custkey@4 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=7, partitioning: Hash([o_custkey@4], 16)
+ UnresolvedShuffleExec: stage=5, partitioning: Hash([o_custkey@4], 16)
SortExec: expr=[c_custkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=8, partitioning: Hash([c_custkey@0], 16)
+ UnresolvedShuffleExec: stage=6, partitioning: Hash([c_custkey@0], 16)
-=== Stage 10 ===
+=== Stage 8 ===
+SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
+ FilterExec: n_name@1 = FRANCE OR n_name@1 = GERMANY
+ StatsExec: rows=25
+
+=== Stage 9 ===
SortShuffleWriterExec: partitioning=Hash([c_nationkey@3], 16)
ProjectionExec: expr=[l_extendedprice@1 as l_extendedprice, l_discount@2 as
l_discount, l_shipdate@3 as l_shipdate, c_nationkey@4 as c_nationkey, n_name@6
as n_name]
- ProjectionExec: expr=[s_nationkey@2 as s_nationkey, l_extendedprice@3 as
l_extendedprice, l_discount@4 as l_discount, l_shipdate@5 as l_shipdate,
c_nationkey@6 as c_nationkey, n_nationkey@0 as n_nationkey, n_name@1 as n_name]
- HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(n_nationkey@0,
s_nationkey@0)]
- UnresolvedShuffleExec: stage=2, broadcast=true, upstream_partitions: 16
- UnresolvedShuffleExec: stage=9, partitioning: Hash([s_nationkey@0], 16)
+ SortMergeJoinExec: join_type=Inner, on=[(s_nationkey@0, n_nationkey@0)]
+ SortExec: expr=[s_nationkey@0 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: stage=7, partitioning: Hash([s_nationkey@0], 16)
+ SortExec: expr=[n_nationkey@0 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: stage=8, partitioning: Hash([n_nationkey@0], 16)
-=== Stage 11 ===
-ShuffleWriterExec: partitioning: None
- UnresolvedShuffleExec: stage=10, partitioning: Hash([c_nationkey@3], 16)
-
-=== Stage 12 ===
+=== Stage 10 ===
SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
FilterExec: n_name@1 = GERMANY OR n_name@1 = FRANCE
StatsExec: rows=25
-=== Stage 13 ===
+=== Stage 11 ===
SortShuffleWriterExec: partitioning=Hash([supp_nation@0, cust_nation@1,
l_year@2], 16)
AggregateExec: mode=Partial, gby=[supp_nation@0 as supp_nation,
cust_nation@1 as cust_nation, l_year@2 as l_year], aggr=[sum(shipping.volume)]
ProjectionExec: expr=[n_name@4 as supp_nation, n_name@6 as cust_nation,
date_part(YEAR, l_shipdate@2) as l_year, l_extendedprice@0 * (Some(1),20,0 -
l_discount@1) as volume]
- HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(c_nationkey@3,
n_nationkey@0)], filter=n_name@0 = FRANCE AND n_name@1 = GERMANY OR n_name@0 =
GERMANY AND n_name@1 = FRANCE
- UnresolvedShuffleExec: stage=11, broadcast=true, upstream_partitions:
16
- UnresolvedShuffleExec: stage=12, partitioning: Hash([n_nationkey@0],
16)
+ SortMergeJoinExec: join_type=Inner, on=[(c_nationkey@3, n_nationkey@0)],
filter=n_name@0 = FRANCE AND n_name@1 = GERMANY OR n_name@0 = GERMANY AND
n_name@1 = FRANCE
+ SortExec: expr=[c_nationkey@3 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: stage=9, partitioning: Hash([c_nationkey@3],
16)
+ SortExec: expr=[n_nationkey@0 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: stage=10, partitioning: Hash([n_nationkey@0],
16)
-=== Stage 14 ===
+=== Stage 12 ===
ShuffleWriterExec: partitioning: None
SortExec: expr=[supp_nation@0 ASC NULLS LAST, cust_nation@1 ASC NULLS LAST,
l_year@2 ASC NULLS LAST], preserve_partitioning=[true]
ProjectionExec: expr=[supp_nation@0 as supp_nation, cust_nation@1 as
cust_nation, l_year@2 as l_year, sum(shipping.volume)@3 as revenue]
AggregateExec: mode=FinalPartitioned, gby=[supp_nation@0 as supp_nation,
cust_nation@1 as cust_nation, l_year@2 as l_year], aggr=[sum(shipping.volume)]
- UnresolvedShuffleExec: stage=13, partitioning: Hash([supp_nation@0,
cust_nation@1, l_year@2], 16)
+ UnresolvedShuffleExec: stage=11, partitioning: Hash([supp_nation@0,
cust_nation@1, l_year@2], 16)
-=== Stage 15 ===
+=== Stage 13 ===
ShuffleWriterExec: partitioning: None
SortPreservingMergeExec: [supp_nation@0 ASC NULLS LAST, cust_nation@1 ASC
NULLS LAST, l_year@2 ASC NULLS LAST]
- UnresolvedShuffleExec: stage=14, partitioning: Hash([supp_nation@0,
cust_nation@1, l_year@2], 16)
+ UnresolvedShuffleExec: stage=12, partitioning: Hash([supp_nation@0,
cust_nation@1, l_year@2], 16)
diff --git a/ballista/scheduler/tests/tpch_plan_stability/approved/q8.txt
b/ballista/scheduler/tests/tpch_plan_stability/approved/q8.txt
index 912a82560..77ca93cd2 100644
--- a/ballista/scheduler/tests/tpch_plan_stability/approved/q8.txt
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q8.txt
@@ -1,118 +1,110 @@
=== Stage 1 ===
-SortShuffleWriterExec: partitioning=Hash([r_regionkey@0], 16)
- FilterExec: r_name@1 = AMERICA, projection=[r_regionkey@0]
- StatsExec: rows=5
-
-=== Stage 2 ===
-ShuffleWriterExec: partitioning: None
- UnresolvedShuffleExec: stage=1, partitioning: Hash([r_regionkey@0], 16)
-
-=== Stage 3 ===
-SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
- StatsExec: rows=25
-
-=== Stage 4 ===
-ShuffleWriterExec: partitioning: None
- UnresolvedShuffleExec: stage=3, partitioning: Hash([n_nationkey@0], 16)
-
-=== Stage 5 ===
SortShuffleWriterExec: partitioning=Hash([p_partkey@0], 16)
FilterExec: p_type@1 = ECONOMY ANODIZED STEEL, projection=[p_partkey@0]
StatsExec: rows=20000000
-=== Stage 6 ===
+=== Stage 2 ===
SortShuffleWriterExec: partitioning=Hash([l_partkey@1], 16)
StatsExec: rows=600037902
-=== Stage 7 ===
+=== Stage 3 ===
SortShuffleWriterExec: partitioning=Hash([l_suppkey@1], 16)
ProjectionExec: expr=[l_orderkey@1 as l_orderkey, l_suppkey@3 as l_suppkey,
l_extendedprice@4 as l_extendedprice, l_discount@5 as l_discount]
SortMergeJoinExec: join_type=Inner, on=[(p_partkey@0, l_partkey@1)]
SortExec: expr=[p_partkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=5, partitioning: Hash([p_partkey@0], 16)
+ UnresolvedShuffleExec: stage=1, partitioning: Hash([p_partkey@0], 16)
SortExec: expr=[l_partkey@1 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=6, partitioning: Hash([l_partkey@1], 16)
+ UnresolvedShuffleExec: stage=2, partitioning: Hash([l_partkey@1], 16)
-=== Stage 8 ===
+=== Stage 4 ===
SortShuffleWriterExec: partitioning=Hash([s_suppkey@0], 16)
StatsExec: rows=1000000
-=== Stage 9 ===
+=== Stage 5 ===
SortShuffleWriterExec: partitioning=Hash([l_orderkey@0], 16)
ProjectionExec: expr=[l_orderkey@0 as l_orderkey, l_extendedprice@2 as
l_extendedprice, l_discount@3 as l_discount, s_nationkey@5 as s_nationkey]
SortMergeJoinExec: join_type=Inner, on=[(l_suppkey@1, s_suppkey@0)]
SortExec: expr=[l_suppkey@1 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=7, partitioning: Hash([l_suppkey@1], 16)
+ UnresolvedShuffleExec: stage=3, partitioning: Hash([l_suppkey@1], 16)
SortExec: expr=[s_suppkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=8, partitioning: Hash([s_suppkey@0], 16)
+ UnresolvedShuffleExec: stage=4, partitioning: Hash([s_suppkey@0], 16)
-=== Stage 10 ===
+=== Stage 6 ===
SortShuffleWriterExec: partitioning=Hash([o_orderkey@0], 16)
FilterExec: o_orderdate@2 >= 1995-01-01 AND o_orderdate@2 <= 1996-12-31
StatsExec: rows=150000000
-=== Stage 11 ===
+=== Stage 7 ===
SortShuffleWriterExec: partitioning=Hash([o_custkey@3], 16)
ProjectionExec: expr=[l_extendedprice@1 as l_extendedprice, l_discount@2 as
l_discount, s_nationkey@3 as s_nationkey, o_custkey@5 as o_custkey,
o_orderdate@6 as o_orderdate]
SortMergeJoinExec: join_type=Inner, on=[(l_orderkey@0, o_orderkey@0)]
SortExec: expr=[l_orderkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=9, partitioning: Hash([l_orderkey@0], 16)
+ UnresolvedShuffleExec: stage=5, partitioning: Hash([l_orderkey@0], 16)
SortExec: expr=[o_orderkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=10, partitioning: Hash([o_orderkey@0], 16)
+ UnresolvedShuffleExec: stage=6, partitioning: Hash([o_orderkey@0], 16)
-=== Stage 12 ===
+=== Stage 8 ===
SortShuffleWriterExec: partitioning=Hash([c_custkey@0], 16)
StatsExec: rows=15000000
-=== Stage 13 ===
+=== Stage 9 ===
SortShuffleWriterExec: partitioning=Hash([c_nationkey@4], 16)
ProjectionExec: expr=[l_extendedprice@0 as l_extendedprice, l_discount@1 as
l_discount, s_nationkey@2 as s_nationkey, o_orderdate@4 as o_orderdate,
c_nationkey@6 as c_nationkey]
SortMergeJoinExec: join_type=Inner, on=[(o_custkey@3, c_custkey@0)]
SortExec: expr=[o_custkey@3 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=11, partitioning: Hash([o_custkey@3], 16)
+ UnresolvedShuffleExec: stage=7, partitioning: Hash([o_custkey@3], 16)
SortExec: expr=[c_custkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=12, partitioning: Hash([c_custkey@0], 16)
+ UnresolvedShuffleExec: stage=8, partitioning: Hash([c_custkey@0], 16)
-=== Stage 14 ===
+=== Stage 10 ===
+SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
+ StatsExec: rows=25
+
+=== Stage 11 ===
SortShuffleWriterExec: partitioning=Hash([s_nationkey@2], 16)
ProjectionExec: expr=[l_extendedprice@0 as l_extendedprice, l_discount@1 as
l_discount, s_nationkey@2 as s_nationkey, o_orderdate@3 as o_orderdate,
n_regionkey@6 as n_regionkey]
- ProjectionExec: expr=[l_extendedprice@2 as l_extendedprice, l_discount@3
as l_discount, s_nationkey@4 as s_nationkey, o_orderdate@5 as o_orderdate,
c_nationkey@6 as c_nationkey, n_nationkey@0 as n_nationkey, n_regionkey@1 as
n_regionkey]
- HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(n_nationkey@0,
c_nationkey@4)]
- UnresolvedShuffleExec: stage=4, broadcast=true, upstream_partitions: 16
- UnresolvedShuffleExec: stage=13, partitioning: Hash([c_nationkey@4],
16)
-
-=== Stage 15 ===
-ShuffleWriterExec: partitioning: None
- UnresolvedShuffleExec: stage=14, partitioning: Hash([s_nationkey@2], 16)
+ SortMergeJoinExec: join_type=Inner, on=[(c_nationkey@4, n_nationkey@0)]
+ SortExec: expr=[c_nationkey@4 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: stage=9, partitioning: Hash([c_nationkey@4], 16)
+ SortExec: expr=[n_nationkey@0 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: stage=10, partitioning: Hash([n_nationkey@0],
16)
-=== Stage 16 ===
+=== Stage 12 ===
SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
StatsExec: rows=25
-=== Stage 17 ===
+=== Stage 13 ===
SortShuffleWriterExec: partitioning=Hash([n_regionkey@3], 16)
ProjectionExec: expr=[l_extendedprice@0 as l_extendedprice, l_discount@1 as
l_discount, o_orderdate@3 as o_orderdate, n_regionkey@4 as n_regionkey,
n_name@6 as n_name]
- HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(s_nationkey@2,
n_nationkey@0)]
- UnresolvedShuffleExec: stage=15, broadcast=true, upstream_partitions: 16
- UnresolvedShuffleExec: stage=16, partitioning: Hash([n_nationkey@0], 16)
+ SortMergeJoinExec: join_type=Inner, on=[(s_nationkey@2, n_nationkey@0)]
+ SortExec: expr=[s_nationkey@2 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: stage=11, partitioning: Hash([s_nationkey@2],
16)
+ SortExec: expr=[n_nationkey@0 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: stage=12, partitioning: Hash([n_nationkey@0],
16)
-=== Stage 18 ===
+=== Stage 14 ===
+SortShuffleWriterExec: partitioning=Hash([r_regionkey@0], 16)
+ FilterExec: r_name@1 = AMERICA, projection=[r_regionkey@0]
+ StatsExec: rows=5
+
+=== Stage 15 ===
SortShuffleWriterExec: partitioning=Hash([o_year@0], 16)
AggregateExec: mode=Partial, gby=[o_year@0 as o_year], aggr=[sum(CASE WHEN
all_nations.nation = BRAZIL THEN all_nations.volume ELSE Some(0),38,4 END) as
sum(CASE WHEN all_nations.nation = Utf8("BRAZIL") THEN all_nations.volume ELSE
Int64(0) END), sum(all_nations.volume)]
ProjectionExec: expr=[date_part(YEAR, o_orderdate@2) as o_year,
l_extendedprice@0 * (Some(1),20,0 - l_discount@1) as volume, n_name@4 as nation]
- ProjectionExec: expr=[l_extendedprice@1 as l_extendedprice, l_discount@2
as l_discount, o_orderdate@3 as o_orderdate, n_regionkey@4 as n_regionkey,
n_name@5 as n_name, r_regionkey@0 as r_regionkey]
- HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(r_regionkey@0,
n_regionkey@3)]
- UnresolvedShuffleExec: stage=2, broadcast=true, upstream_partitions:
16
- UnresolvedShuffleExec: stage=17, partitioning: Hash([n_regionkey@3],
16)
+ SortMergeJoinExec: join_type=Inner, on=[(n_regionkey@3, r_regionkey@0)]
+ SortExec: expr=[n_regionkey@3 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: stage=13, partitioning: Hash([n_regionkey@3],
16)
+ SortExec: expr=[r_regionkey@0 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: stage=14, partitioning: Hash([r_regionkey@0],
16)
-=== Stage 19 ===
+=== Stage 16 ===
ShuffleWriterExec: partitioning: None
SortExec: expr=[o_year@0 ASC NULLS LAST], preserve_partitioning=[true]
ProjectionExec: expr=[o_year@0 as o_year, sum(CASE WHEN all_nations.nation
= Utf8("BRAZIL") THEN all_nations.volume ELSE Int64(0) END)@1 /
sum(all_nations.volume)@2 as mkt_share]
AggregateExec: mode=FinalPartitioned, gby=[o_year@0 as o_year],
aggr=[sum(CASE WHEN all_nations.nation = BRAZIL THEN all_nations.volume ELSE
Some(0),38,4 END) as sum(CASE WHEN all_nations.nation = Utf8("BRAZIL") THEN
all_nations.volume ELSE Int64(0) END), sum(all_nations.volume)]
- UnresolvedShuffleExec: stage=18, partitioning: Hash([o_year@0], 16)
+ UnresolvedShuffleExec: stage=15, partitioning: Hash([o_year@0], 16)
-=== Stage 20 ===
+=== Stage 17 ===
ShuffleWriterExec: partitioning: None
SortPreservingMergeExec: [o_year@0 ASC NULLS LAST]
- UnresolvedShuffleExec: stage=19, partitioning: Hash([o_year@0], 16)
+ UnresolvedShuffleExec: stage=16, partitioning: Hash([o_year@0], 16)
diff --git a/ballista/scheduler/tests/tpch_plan_stability/approved/q9.txt
b/ballista/scheduler/tests/tpch_plan_stability/approved/q9.txt
index 0f0fa2d63..48b15bca3 100644
--- a/ballista/scheduler/tests/tpch_plan_stability/approved/q9.txt
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q9.txt
@@ -1,85 +1,82 @@
=== Stage 1 ===
-SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
- StatsExec: rows=25
-
-=== Stage 2 ===
-ShuffleWriterExec: partitioning: None
- UnresolvedShuffleExec: stage=1, partitioning: Hash([n_nationkey@0], 16)
-
-=== Stage 3 ===
SortShuffleWriterExec: partitioning=Hash([p_partkey@0], 16)
FilterExec: p_name@1 LIKE %green%, projection=[p_partkey@0]
StatsExec: rows=20000000
-=== Stage 4 ===
+=== Stage 2 ===
SortShuffleWriterExec: partitioning=Hash([l_partkey@1], 16)
StatsExec: rows=600037902
-=== Stage 5 ===
+=== Stage 3 ===
SortShuffleWriterExec: partitioning=Hash([l_suppkey@2], 16)
ProjectionExec: expr=[l_orderkey@1 as l_orderkey, l_partkey@2 as l_partkey,
l_suppkey@3 as l_suppkey, l_quantity@4 as l_quantity, l_extendedprice@5 as
l_extendedprice, l_discount@6 as l_discount]
SortMergeJoinExec: join_type=Inner, on=[(p_partkey@0, l_partkey@1)]
SortExec: expr=[p_partkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=3, partitioning: Hash([p_partkey@0], 16)
+ UnresolvedShuffleExec: stage=1, partitioning: Hash([p_partkey@0], 16)
SortExec: expr=[l_partkey@1 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=4, partitioning: Hash([l_partkey@1], 16)
+ UnresolvedShuffleExec: stage=2, partitioning: Hash([l_partkey@1], 16)
-=== Stage 6 ===
+=== Stage 4 ===
SortShuffleWriterExec: partitioning=Hash([s_suppkey@0], 16)
StatsExec: rows=1000000
-=== Stage 7 ===
+=== Stage 5 ===
SortShuffleWriterExec: partitioning=Hash([l_suppkey@2, l_partkey@1], 16)
ProjectionExec: expr=[l_orderkey@0 as l_orderkey, l_partkey@1 as l_partkey,
l_suppkey@2 as l_suppkey, l_quantity@3 as l_quantity, l_extendedprice@4 as
l_extendedprice, l_discount@5 as l_discount, s_nationkey@7 as s_nationkey]
SortMergeJoinExec: join_type=Inner, on=[(l_suppkey@2, s_suppkey@0)]
SortExec: expr=[l_suppkey@2 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=5, partitioning: Hash([l_suppkey@2], 16)
+ UnresolvedShuffleExec: stage=3, partitioning: Hash([l_suppkey@2], 16)
SortExec: expr=[s_suppkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=6, partitioning: Hash([s_suppkey@0], 16)
+ UnresolvedShuffleExec: stage=4, partitioning: Hash([s_suppkey@0], 16)
-=== Stage 8 ===
+=== Stage 6 ===
SortShuffleWriterExec: partitioning=Hash([ps_suppkey@1, ps_partkey@0], 16)
StatsExec: rows=80000000
-=== Stage 9 ===
+=== Stage 7 ===
SortShuffleWriterExec: partitioning=Hash([l_orderkey@0], 16)
ProjectionExec: expr=[l_orderkey@0 as l_orderkey, l_quantity@3 as
l_quantity, l_extendedprice@4 as l_extendedprice, l_discount@5 as l_discount,
s_nationkey@6 as s_nationkey, ps_supplycost@9 as ps_supplycost]
SortMergeJoinExec: join_type=Inner, on=[(l_suppkey@2, ps_suppkey@1),
(l_partkey@1, ps_partkey@0)]
SortExec: expr=[l_suppkey@2 ASC, l_partkey@1 ASC],
preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=7, partitioning: Hash([l_suppkey@2,
l_partkey@1], 16)
+ UnresolvedShuffleExec: stage=5, partitioning: Hash([l_suppkey@2,
l_partkey@1], 16)
SortExec: expr=[ps_suppkey@1 ASC, ps_partkey@0 ASC],
preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=8, partitioning: Hash([ps_suppkey@1,
ps_partkey@0], 16)
+ UnresolvedShuffleExec: stage=6, partitioning: Hash([ps_suppkey@1,
ps_partkey@0], 16)
-=== Stage 10 ===
+=== Stage 8 ===
SortShuffleWriterExec: partitioning=Hash([o_orderkey@0], 16)
StatsExec: rows=150000000
-=== Stage 11 ===
+=== Stage 9 ===
SortShuffleWriterExec: partitioning=Hash([s_nationkey@3], 16)
ProjectionExec: expr=[l_quantity@1 as l_quantity, l_extendedprice@2 as
l_extendedprice, l_discount@3 as l_discount, s_nationkey@4 as s_nationkey,
ps_supplycost@5 as ps_supplycost, o_orderdate@7 as o_orderdate]
SortMergeJoinExec: join_type=Inner, on=[(l_orderkey@0, o_orderkey@0)]
SortExec: expr=[l_orderkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=9, partitioning: Hash([l_orderkey@0], 16)
+ UnresolvedShuffleExec: stage=7, partitioning: Hash([l_orderkey@0], 16)
SortExec: expr=[o_orderkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: stage=10, partitioning: Hash([o_orderkey@0], 16)
+ UnresolvedShuffleExec: stage=8, partitioning: Hash([o_orderkey@0], 16)
-=== Stage 12 ===
+=== Stage 10 ===
+SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
+ StatsExec: rows=25
+
+=== Stage 11 ===
SortShuffleWriterExec: partitioning=Hash([nation@0, o_year@1], 16)
AggregateExec: mode=Partial, gby=[nation@0 as nation, o_year@1 as o_year],
aggr=[sum(profit.amount)]
ProjectionExec: expr=[n_name@7 as nation, date_part(YEAR, o_orderdate@5)
as o_year, l_extendedprice@1 * (Some(1),20,0 - l_discount@2) - ps_supplycost@4
* l_quantity@0 as amount]
- ProjectionExec: expr=[l_quantity@2 as l_quantity, l_extendedprice@3 as
l_extendedprice, l_discount@4 as l_discount, s_nationkey@5 as s_nationkey,
ps_supplycost@6 as ps_supplycost, o_orderdate@7 as o_orderdate, n_nationkey@0
as n_nationkey, n_name@1 as n_name]
- HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(n_nationkey@0,
s_nationkey@3)]
- UnresolvedShuffleExec: stage=2, broadcast=true, upstream_partitions:
16
- UnresolvedShuffleExec: stage=11, partitioning: Hash([s_nationkey@3],
16)
+ SortMergeJoinExec: join_type=Inner, on=[(s_nationkey@3, n_nationkey@0)]
+ SortExec: expr=[s_nationkey@3 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: stage=9, partitioning: Hash([s_nationkey@3],
16)
+ SortExec: expr=[n_nationkey@0 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: stage=10, partitioning: Hash([n_nationkey@0],
16)
-=== Stage 13 ===
+=== Stage 12 ===
ShuffleWriterExec: partitioning: None
SortExec: expr=[nation@0 ASC NULLS LAST, o_year@1 DESC],
preserve_partitioning=[true]
ProjectionExec: expr=[nation@0 as nation, o_year@1 as o_year,
sum(profit.amount)@2 as sum_profit]
AggregateExec: mode=FinalPartitioned, gby=[nation@0 as nation, o_year@1
as o_year], aggr=[sum(profit.amount)]
- UnresolvedShuffleExec: stage=12, partitioning: Hash([nation@0,
o_year@1], 16)
+ UnresolvedShuffleExec: stage=11, partitioning: Hash([nation@0,
o_year@1], 16)
-=== Stage 14 ===
+=== Stage 13 ===
ShuffleWriterExec: partitioning: None
SortPreservingMergeExec: [nation@0 ASC NULLS LAST, o_year@1 DESC]
- UnresolvedShuffleExec: stage=13, partitioning: Hash([nation@0, o_year@1],
16)
+ UnresolvedShuffleExec: stage=12, partitioning: Hash([nation@0, o_year@1],
16)
diff --git a/docs/source/upgrading/54.0.0.md b/docs/source/upgrading/54.0.0.md
index 0b3a28249..03f8a3b4a 100644
--- a/docs/source/upgrading/54.0.0.md
+++ b/docs/source/upgrading/54.0.0.md
@@ -92,7 +92,14 @@ static `DefaultDistributedPlanner` now converts a
`SortMergeJoinExec` whose
smaller side fits under `ballista.optimizer.broadcast_join_threshold_bytes`
(default `10 MB`) into a broadcast `CollectLeft` hash join. This conversion is
governed by the new `ballista.optimizer.broadcast_sort_merge_join_enabled`
-config key, which defaults to `true`.
+config key.
+
+```{warning}
+That key defaulted to `true` in `54.0.0` and `54.1.0`. The conversion can
+produce **incorrect results**, so the default is `false` from `54.2.0` onward
+and the conversion is opt-in. If you are on `54.0.0` or `54.1.0`, set
+`ballista.optimizer.broadcast_sort_merge_join_enabled = false` explicitly.
+```
The consequence is that join plans — and therefore performance and shuffle
behavior — change on upgrade **even with AQE off**. Broadcast is applied only
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
