This is an automated email from the ASF dual-hosted git repository.
andygrove pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion-ballista.git
The following commit(s) were added to refs/heads/main by this push:
new 5298a9f1f feat: display input stage id on UnresolvedShuffleExec
(#1967) (#1969)
5298a9f1f is described below
commit 5298a9f1ff0e57a20b2b6f23695e930b41167de8
Author: Andy Grove <[email protected]>
AuthorDate: Wed Jul 8 22:19:12 2026 -0600
feat: display input stage id on UnresolvedShuffleExec (#1967) (#1969)
---
ballista/client/tests/context_checks.rs | 2 +-
.../core/src/execution_plans/unresolved_shuffle.rs | 22 +++++++++--
ballista/scheduler/src/planner.rs | 20 +++++-----
.../tests/tpch_plan_stability/approved/q1.txt | 4 +-
.../tests/tpch_plan_stability/approved/q10.txt | 18 ++++-----
.../tests/tpch_plan_stability/approved/q11.txt | 26 ++++++------
.../tests/tpch_plan_stability/approved/q12.txt | 8 ++--
.../tests/tpch_plan_stability/approved/q13.txt | 8 ++--
.../tests/tpch_plan_stability/approved/q14.txt | 6 +--
.../tests/tpch_plan_stability/approved/q15.txt | 16 ++++----
.../tests/tpch_plan_stability/approved/q16.txt | 14 +++----
.../tests/tpch_plan_stability/approved/q17.txt | 8 ++--
.../tests/tpch_plan_stability/approved/q18.txt | 12 +++---
.../tests/tpch_plan_stability/approved/q19.txt | 6 +--
.../tests/tpch_plan_stability/approved/q2.txt | 46 +++++++++++-----------
.../tests/tpch_plan_stability/approved/q20.txt | 20 +++++-----
.../tests/tpch_plan_stability/approved/q21.txt | 24 +++++------
.../tests/tpch_plan_stability/approved/q22.txt | 8 ++--
.../tests/tpch_plan_stability/approved/q3.txt | 10 ++---
.../tests/tpch_plan_stability/approved/q4.txt | 8 ++--
.../tests/tpch_plan_stability/approved/q5.txt | 28 ++++++-------
.../tests/tpch_plan_stability/approved/q7.txt | 28 ++++++-------
.../tests/tpch_plan_stability/approved/q8.txt | 38 +++++++++---------
.../tests/tpch_plan_stability/approved/q9.txt | 26 ++++++------
24 files changed, 211 insertions(+), 195 deletions(-)
diff --git a/ballista/client/tests/context_checks.rs
b/ballista/client/tests/context_checks.rs
index dc0576d9d..272abc94b 100644
--- a/ballista/client/tests/context_checks.rs
+++ b/ballista/client/tests/context_checks.rs
@@ -1120,7 +1120,7 @@ mod supported {
"| | ShuffleWriterExec: partitioning: None
|",
"| | ProjectionExec: expr=[count(Int64(1))@1 as
count(*), id@0 as id]
|",
"| | AggregateExec: mode=FinalPartitioned,
gby=[id@0 as id], aggr=[count(Int64(1))]
|",
- "| | UnresolvedShuffleExec: partitioning:
Hash([id@0], 16)
|",
+ "| | UnresolvedShuffleExec: stage=1,
partitioning: Hash([id@0], 16)
|",
"| |
|",
"| |
|",
"+------------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+",
diff --git a/ballista/core/src/execution_plans/unresolved_shuffle.rs
b/ballista/core/src/execution_plans/unresolved_shuffle.rs
index 64bfd8fec..f544f0b0a 100644
--- a/ballista/core/src/execution_plans/unresolved_shuffle.rs
+++ b/ballista/core/src/execution_plans/unresolved_shuffle.rs
@@ -155,13 +155,14 @@ impl DisplayAs for UnresolvedShuffleExec {
if self.broadcast {
write!(
f,
- "UnresolvedShuffleExec: broadcast=true,
upstream_partitions: {}",
- self.upstream_partition_count,
+ "UnresolvedShuffleExec: stage={}, broadcast=true,
upstream_partitions: {}",
+ self.stage_id, self.upstream_partition_count,
)
} else {
write!(
f,
- "UnresolvedShuffleExec: partitioning: {}",
+ "UnresolvedShuffleExec: stage={}, partitioning: {}",
+ self.stage_id,
self.properties().output_partitioning()
)?;
if let Some(c) = &self.coalesce {
@@ -266,4 +267,19 @@ mod tests {
"expected ', coalesce: 3 of 8' annotation; got: {s}"
);
}
+
+ #[test]
+ fn display_includes_stage_id() {
+ let schema = Arc::new(Schema::new(vec![Field::new("a",
DataType::Int32, false)]));
+ let exec =
+ UnresolvedShuffleExec::new(3, schema,
Partitioning::UnknownPartitioning(4));
+ let s = format!(
+ "{}",
+ datafusion::physical_plan::displayable(&exec).indent(false)
+ );
+ assert!(
+ s.contains("stage=3"),
+ "expected stage id in display, got: {s}"
+ );
+ }
}
diff --git a/ballista/scheduler/src/planner.rs
b/ballista/scheduler/src/planner.rs
index d5602f57b..97cc1a7ac 100644
--- a/ballista/scheduler/src/planner.rs
+++ b/ballista/scheduler/src/planner.rs
@@ -800,11 +800,11 @@ mod test {
SortExec: expr=[l_returnflag@0 ASC NULLS LAST],
preserve_partitioning=[true]
ProjectionExec: expr=[l_returnflag@0 as l_returnflag,
sum(lineitem.l_extendedprice * Int64(1))@1 as sum_disc_price]
AggregateExec: mode=FinalPartitioned, gby=[l_returnflag@0 as
l_returnflag], aggr=[sum(lineitem.l_extendedprice * Int64(1))]
- UnresolvedShuffleExec: partitioning: Hash([l_returnflag@0], 2)
+ UnresolvedShuffleExec: stage=1, partitioning:
Hash([l_returnflag@0], 2)
ShuffleWriterExec: partitioning: None
SortPreservingMergeExec: [l_returnflag@0 ASC NULLS LAST]
- UnresolvedShuffleExec: partitioning: Hash([l_returnflag@0], 2)
+ UnresolvedShuffleExec: stage=2, partitioning:
Hash([l_returnflag@0], 2)
*/
assert_eq!(3, stages.len());
@@ -918,18 +918,18 @@ order by
ShuffleWriterExec: partitioning: Hash([l_shipmode@0], 2)
AggregateExec: mode=Partial, gby=[l_shipmode@0 as l_shipmode],
aggr=[sum(CASE WHEN orders.o_orderpriority = Utf8("1-URGENT") OR
orders.o_orderpriority = Utf8("2-HIGH") THEN Int64(1) ELSE Int64(0) END),
sum(CASE WHEN orders.o_orderpriority != Utf8("1-URGENT") AND
orders.o_orderpriority != Utf8("2-HIGH") THEN Int64(1) ELSE Int64(0) END)]
HashJoinExec: mode=Partitioned, join_type=Inner,
on=[(l_orderkey@0, o_orderkey@0)], projection=[l_shipmode@1, o_orderpriority@3]
- UnresolvedShuffleExec: partitioning: Hash([l_orderkey@0], 2)
- UnresolvedShuffleExec: partitioning: Hash([o_orderkey@0], 2)
+ UnresolvedShuffleExec: stage=1, partitioning:
Hash([l_orderkey@0], 2)
+ UnresolvedShuffleExec: stage=2, partitioning:
Hash([o_orderkey@0], 2)
ShuffleWriterExec: partitioning: None
SortExec: expr=[l_shipmode@0 ASC NULLS LAST],
preserve_partitioning=[true]
ProjectionExec: expr=[l_shipmode@0 as l_shipmode, sum(CASE WHEN
orders.o_orderpriority = Utf8("1-URGENT") OR orders.o_orderpriority =
Utf8("2-HIGH") THEN Int64(1) ELSE Int64(0) END)@1 as high_line_count, sum(CASE
WHEN orders.o_orderpriority != Utf8("1-URGENT") AND orders.o_orderpriority !=
Utf8("2-HIGH") THEN Int64(1) ELSE Int64(0) END)@2 as low_line_count]
AggregateExec: mode=FinalPartitioned, gby=[l_shipmode@0 as
l_shipmode], aggr=[sum(CASE WHEN orders.o_orderpriority = Utf8("1-URGENT") OR
orders.o_orderpriority = Utf8("2-HIGH") THEN Int64(1) ELSE Int64(0) END),
sum(CASE WHEN orders.o_orderpriority != Utf8("1-URGENT") AND
orders.o_orderpriority != Utf8("2-HIGH") THEN Int64(1) ELSE Int64(0) END)]
- UnresolvedShuffleExec: partitioning: Hash([l_shipmode@0], 2)
+ UnresolvedShuffleExec: stage=3, partitioning:
Hash([l_shipmode@0], 2)
ShuffleWriterExec: partitioning: None
SortPreservingMergeExec: [l_shipmode@0 ASC NULLS LAST]
- UnresolvedShuffleExec: partitioning: Hash([l_shipmode@0], 2)
+ UnresolvedShuffleExec: stage=4, partitioning: Hash([l_shipmode@0],
2)
*/
assert_eq!(5, stages.len());
@@ -1099,13 +1099,13 @@ order by
// Stage 1 holds the join: a broadcast CollectLeft hash join, no
// SortMergeJoinExec and no SortExec.
- assert_plan!(stages[1].as_ref(), @r"
+ assert_plan!(stages[1].as_ref(), @"
ShuffleWriterExec: partitioning: None
AggregateExec: mode=Partial, gby=[], aggr=[count(Int64(1))]
ProjectionExec: expr=[]
ProjectionExec: expr=[k@1 as k, k@0 as k]
HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(k@0,
k@0)]
- UnresolvedShuffleExec: broadcast=true, upstream_partitions: 1
+ UnresolvedShuffleExec: stage=1, broadcast=true,
upstream_partitions: 1
DataSourceExec: partitions=1, partition_sizes=[1]
");
@@ -1647,12 +1647,12 @@ order by
FilterExec: rank() PARTITION BY [lineitem.l_shipmode] ORDER BY
[lineitem.l_shipdate DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND
CURRENT ROW@2 <= 100
BoundedWindowAggExec: wdw=[rank() PARTITION BY
[lineitem.l_shipmode] ORDER BY [lineitem.l_shipdate DESC NULLS FIRST] RANGE
BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Field { "rank() PARTITION BY
[lineitem.l_shipmode] ORDER BY [lineitem.l_shipdate DESC NULLS FIRST] RANGE
BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW": UInt64 }, frame: RANGE BETWEEN
UNBOUNDED PRECEDING AND CURRENT ROW], mode=[Sorted]
SortExec: expr=[l_shipmode@1 ASC NULLS LAST, l_shipdate@0
DESC], preserve_partitioning=[true]
- UnresolvedShuffleExec: partitioning:
Hash([l_shipmode@1], 2)
+ UnresolvedShuffleExec: stage=1, partitioning:
Hash([l_shipmode@1], 2)
Stage 2:
ShuffleWriterExec: partitioning: None
SortPreservingMergeExec: [l_shipdate@1 ASC NULLS LAST, rk@2 ASC
NULLS LAST]
- UnresolvedShuffleExec: partitioning: Hash([l_shipmode@0], 2)
+ UnresolvedShuffleExec: stage=2, partitioning:
Hash([l_shipmode@0], 2)
*/
diff --git a/ballista/scheduler/tests/tpch_plan_stability/approved/q1.txt
b/ballista/scheduler/tests/tpch_plan_stability/approved/q1.txt
index 3f26e2a58..b01d7f2af 100644
--- a/ballista/scheduler/tests/tpch_plan_stability/approved/q1.txt
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q1.txt
@@ -10,9 +10,9 @@ ShuffleWriterExec: partitioning: None
SortExec: expr=[l_returnflag@0 ASC NULLS LAST, l_linestatus@1 ASC NULLS
LAST], preserve_partitioning=[true]
ProjectionExec: expr=[l_returnflag@0 as l_returnflag, l_linestatus@1 as
l_linestatus, sum(lineitem.l_quantity)@2 as sum_qty,
sum(lineitem.l_extendedprice)@3 as sum_base_price, sum(lineitem.l_extendedprice
* Int64(1) - lineitem.l_discount)@4 as sum_disc_price,
sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount * Int64(1) +
lineitem.l_tax)@5 as sum_charge, avg(lineitem.l_quantity)@6 as avg_qty,
avg(lineitem.l_extendedprice)@7 as avg_price, avg(lineitem.l_discount)@8 as
avg_d [...]
AggregateExec: mode=FinalPartitioned, gby=[l_returnflag@0 as
l_returnflag, l_linestatus@1 as l_linestatus], aggr=[sum(lineitem.l_quantity),
sum(lineitem.l_extendedprice), sum(__common_expr_1) as
sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount),
sum(__common_expr_1 * Some(1),20,0 + lineitem.l_tax) as
sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount * Int64(1) +
lineitem.l_tax), avg(lineitem.l_quantity), avg(lineitem.l_extendedprice),
avg(lineitem.l_discoun [...]
- UnresolvedShuffleExec: partitioning: Hash([l_returnflag@0,
l_linestatus@1], 16)
+ UnresolvedShuffleExec: stage=1, partitioning: Hash([l_returnflag@0,
l_linestatus@1], 16)
=== Stage 3 ===
ShuffleWriterExec: partitioning: None
SortPreservingMergeExec: [l_returnflag@0 ASC NULLS LAST, l_linestatus@1 ASC
NULLS LAST]
- UnresolvedShuffleExec: partitioning: Hash([l_returnflag@0,
l_linestatus@1], 16)
+ UnresolvedShuffleExec: stage=2, partitioning: Hash([l_returnflag@0,
l_linestatus@1], 16)
diff --git a/ballista/scheduler/tests/tpch_plan_stability/approved/q10.txt
b/ballista/scheduler/tests/tpch_plan_stability/approved/q10.txt
index ef52dce9c..cf63a2940 100644
--- a/ballista/scheduler/tests/tpch_plan_stability/approved/q10.txt
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q10.txt
@@ -4,7 +4,7 @@ SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
=== Stage 2 ===
ShuffleWriterExec: partitioning: None
- UnresolvedShuffleExec: partitioning: Hash([n_nationkey@0], 16)
+ UnresolvedShuffleExec: stage=1, partitioning: Hash([n_nationkey@0], 16)
=== Stage 3 ===
SortShuffleWriterExec: partitioning=Hash([c_custkey@0], 16)
@@ -20,9 +20,9 @@ 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: partitioning: Hash([c_custkey@0], 16)
+ UnresolvedShuffleExec: stage=3, partitioning: Hash([c_custkey@0], 16)
SortExec: expr=[o_custkey@1 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: partitioning: Hash([o_custkey@1], 16)
+ UnresolvedShuffleExec: stage=4, partitioning: Hash([o_custkey@1], 16)
=== Stage 6 ===
SortShuffleWriterExec: partitioning=Hash([l_orderkey@0], 16)
@@ -34,9 +34,9 @@ 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: partitioning: Hash([o_orderkey@7], 16)
+ UnresolvedShuffleExec: stage=5, partitioning: Hash([o_orderkey@7], 16)
SortExec: expr=[l_orderkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: partitioning: Hash([l_orderkey@0], 16)
+ UnresolvedShuffleExec: stage=6, partitioning: Hash([l_orderkey@0], 16)
=== Stage 8 ===
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)
@@ -44,17 +44,17 @@ SortShuffleWriterExec: partitioning=Hash([c_custkey@0,
c_name@1, c_acctbal@2, c_
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: broadcast=true, upstream_partitions: 16
- UnresolvedShuffleExec: partitioning: Hash([c_nationkey@3], 16)
+ UnresolvedShuffleExec: stage=2, broadcast=true, upstream_partitions:
16
+ UnresolvedShuffleExec: stage=7, partitioning: Hash([c_nationkey@3],
16)
=== Stage 9 ===
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: 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=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)
=== Stage 10 ===
ShuffleWriterExec: partitioning: None
SortPreservingMergeExec: [revenue@2 DESC], fetch=20
- UnresolvedShuffleExec: 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=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)
diff --git a/ballista/scheduler/tests/tpch_plan_stability/approved/q11.txt
b/ballista/scheduler/tests/tpch_plan_stability/approved/q11.txt
index 059c25242..9a68df95d 100644
--- a/ballista/scheduler/tests/tpch_plan_stability/approved/q11.txt
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q11.txt
@@ -5,7 +5,7 @@ SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
=== Stage 2 ===
ShuffleWriterExec: partitioning: None
- UnresolvedShuffleExec: partitioning: Hash([n_nationkey@0], 16)
+ UnresolvedShuffleExec: stage=1, partitioning: Hash([n_nationkey@0], 16)
=== Stage 3 ===
SortShuffleWriterExec: partitioning=Hash([ps_suppkey@0], 16)
@@ -20,9 +20,9 @@ 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: partitioning: Hash([ps_suppkey@0], 16)
+ UnresolvedShuffleExec: stage=3, partitioning: Hash([ps_suppkey@0], 16)
SortExec: expr=[s_suppkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: partitioning: Hash([s_suppkey@0], 16)
+ UnresolvedShuffleExec: stage=4, partitioning: Hash([s_suppkey@0], 16)
=== Stage 6 ===
ShuffleWriterExec: partitioning: None
@@ -30,8 +30,8 @@ ShuffleWriterExec: partitioning: None
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: broadcast=true, upstream_partitions: 16
- UnresolvedShuffleExec: partitioning: Hash([s_nationkey@2], 16)
+ UnresolvedShuffleExec: stage=2, broadcast=true, upstream_partitions:
16
+ UnresolvedShuffleExec: stage=5, partitioning: Hash([s_nationkey@2],
16)
=== Stage 7 ===
SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
@@ -40,7 +40,7 @@ SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
=== Stage 8 ===
ShuffleWriterExec: partitioning: None
- UnresolvedShuffleExec: partitioning: Hash([n_nationkey@0], 16)
+ UnresolvedShuffleExec: stage=7, partitioning: Hash([n_nationkey@0], 16)
=== Stage 9 ===
SortShuffleWriterExec: partitioning=Hash([ps_suppkey@1], 16)
@@ -55,9 +55,9 @@ 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: partitioning: Hash([ps_suppkey@1], 16)
+ UnresolvedShuffleExec: stage=9, partitioning: Hash([ps_suppkey@1], 16)
SortExec: expr=[s_suppkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: partitioning: Hash([s_suppkey@0], 16)
+ UnresolvedShuffleExec: stage=10, partitioning: Hash([s_suppkey@0], 16)
=== Stage 12 ===
SortShuffleWriterExec: partitioning=Hash([ps_partkey@0], 16)
@@ -65,8 +65,8 @@ SortShuffleWriterExec: partitioning=Hash([ps_partkey@0], 16)
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: broadcast=true, upstream_partitions: 16
- UnresolvedShuffleExec: partitioning: Hash([s_nationkey@3], 16)
+ UnresolvedShuffleExec: stage=8, broadcast=true, upstream_partitions:
16
+ UnresolvedShuffleExec: stage=11, partitioning: Hash([s_nationkey@3],
16)
=== Stage 13 ===
ShuffleWriterExec: partitioning: None
@@ -76,12 +76,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: partitioning: Hash([s_nationkey@2], 16)
+ UnresolvedShuffleExec: stage=6, partitioning:
Hash([s_nationkey@2], 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: partitioning: Hash([ps_partkey@0], 16)
+ UnresolvedShuffleExec: stage=12, partitioning:
Hash([ps_partkey@0], 16)
=== Stage 14 ===
ShuffleWriterExec: partitioning: None
SortPreservingMergeExec: [value@1 DESC]
- UnresolvedShuffleExec: partitioning: Hash([ps_partkey@0], 16)
+ UnresolvedShuffleExec: stage=13, partitioning: Hash([ps_partkey@0], 16)
diff --git a/ballista/scheduler/tests/tpch_plan_stability/approved/q12.txt
b/ballista/scheduler/tests/tpch_plan_stability/approved/q12.txt
index 9802dd3a7..3b86494e3 100644
--- a/ballista/scheduler/tests/tpch_plan_stability/approved/q12.txt
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q12.txt
@@ -13,18 +13,18 @@ SortShuffleWriterExec: partitioning=Hash([l_shipmode@0], 16)
ProjectionExec: expr=[l_shipmode@1 as l_shipmode, o_orderpriority@3 as
o_orderpriority]
SortMergeJoinExec: join_type=Inner, on=[(l_orderkey@0, o_orderkey@0)]
SortExec: expr=[l_orderkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: partitioning: Hash([l_orderkey@0], 16)
+ UnresolvedShuffleExec: stage=1, partitioning: Hash([l_orderkey@0],
16)
SortExec: expr=[o_orderkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: partitioning: Hash([o_orderkey@0], 16)
+ UnresolvedShuffleExec: stage=2, partitioning: Hash([o_orderkey@0],
16)
=== Stage 4 ===
ShuffleWriterExec: partitioning: None
SortExec: expr=[l_shipmode@0 ASC NULLS LAST], preserve_partitioning=[true]
ProjectionExec: expr=[l_shipmode@0 as l_shipmode, sum(CASE WHEN
orders.o_orderpriority = Utf8("1-URGENT") OR orders.o_orderpriority =
Utf8("2-HIGH") THEN Int64(1) ELSE Int64(0) END)@1 as high_line_count, sum(CASE
WHEN orders.o_orderpriority != Utf8("1-URGENT") AND orders.o_orderpriority !=
Utf8("2-HIGH") THEN Int64(1) ELSE Int64(0) END)@2 as low_line_count]
AggregateExec: mode=FinalPartitioned, gby=[l_shipmode@0 as l_shipmode],
aggr=[sum(CASE WHEN orders.o_orderpriority = Utf8("1-URGENT") OR
orders.o_orderpriority = Utf8("2-HIGH") THEN Int64(1) ELSE Int64(0) END),
sum(CASE WHEN orders.o_orderpriority != Utf8("1-URGENT") AND
orders.o_orderpriority != Utf8("2-HIGH") THEN Int64(1) ELSE Int64(0) END)]
- UnresolvedShuffleExec: partitioning: Hash([l_shipmode@0], 16)
+ UnresolvedShuffleExec: stage=3, partitioning: Hash([l_shipmode@0], 16)
=== Stage 5 ===
ShuffleWriterExec: partitioning: None
SortPreservingMergeExec: [l_shipmode@0 ASC NULLS LAST]
- UnresolvedShuffleExec: partitioning: Hash([l_shipmode@0], 16)
+ UnresolvedShuffleExec: stage=4, partitioning: Hash([l_shipmode@0], 16)
diff --git a/ballista/scheduler/tests/tpch_plan_stability/approved/q13.txt
b/ballista/scheduler/tests/tpch_plan_stability/approved/q13.txt
index 252b08f08..10bc255ae 100644
--- a/ballista/scheduler/tests/tpch_plan_stability/approved/q13.txt
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q13.txt
@@ -15,18 +15,18 @@ SortShuffleWriterExec: partitioning=Hash([c_count@0], 16)
ProjectionExec: expr=[c_custkey@0 as c_custkey, o_orderkey@1 as
o_orderkey]
SortMergeJoinExec: join_type=Left, on=[(c_custkey@0, o_custkey@1)]
SortExec: expr=[c_custkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: 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: partitioning: Hash([o_custkey@1], 16)
+ UnresolvedShuffleExec: stage=2, partitioning:
Hash([o_custkey@1], 16)
=== Stage 4 ===
ShuffleWriterExec: partitioning: None
SortExec: expr=[custdist@1 DESC, c_count@0 DESC],
preserve_partitioning=[true]
ProjectionExec: expr=[c_count@0 as c_count, count(Int64(1))@1 as custdist]
AggregateExec: mode=FinalPartitioned, gby=[c_count@0 as c_count],
aggr=[count(Int64(1))]
- UnresolvedShuffleExec: partitioning: Hash([c_count@0], 16)
+ UnresolvedShuffleExec: stage=3, partitioning: Hash([c_count@0], 16)
=== Stage 5 ===
ShuffleWriterExec: partitioning: None
SortPreservingMergeExec: [custdist@1 DESC, c_count@0 DESC]
- UnresolvedShuffleExec: partitioning: Hash([c_count@0], 16)
+ UnresolvedShuffleExec: stage=4, partitioning: Hash([c_count@0], 16)
diff --git a/ballista/scheduler/tests/tpch_plan_stability/approved/q14.txt
b/ballista/scheduler/tests/tpch_plan_stability/approved/q14.txt
index 172add80c..5f8721f42 100644
--- a/ballista/scheduler/tests/tpch_plan_stability/approved/q14.txt
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q14.txt
@@ -13,13 +13,13 @@ ShuffleWriterExec: partitioning: None
ProjectionExec: expr=[l_extendedprice@1 * (Some(1),20,0 - l_discount@2) as
__common_expr_1, p_type@4 as p_type]
SortMergeJoinExec: join_type=Inner, on=[(l_partkey@0, p_partkey@0)]
SortExec: expr=[l_partkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: partitioning: Hash([l_partkey@0], 16)
+ UnresolvedShuffleExec: stage=1, partitioning: Hash([l_partkey@0], 16)
SortExec: expr=[p_partkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: partitioning: Hash([p_partkey@0], 16)
+ UnresolvedShuffleExec: stage=2, partitioning: Hash([p_partkey@0], 16)
=== Stage 4 ===
ShuffleWriterExec: partitioning: None
ProjectionExec: expr=[100 * CAST(sum(CASE WHEN part.p_type LIKE
Utf8("PROMO%") THEN lineitem.l_extendedprice * Int64(1) - lineitem.l_discount
ELSE Int64(0) END)@0 AS Float64) / CAST(sum(lineitem.l_extendedprice * Int64(1)
- lineitem.l_discount)@1 AS Float64) as promo_revenue]
AggregateExec: mode=Final, gby=[], aggr=[sum(CASE WHEN part.p_type LIKE
PROMO% THEN __common_expr_1 ELSE Some(0),38,4 END) as sum(CASE WHEN part.p_type
LIKE Utf8("PROMO%") THEN lineitem.l_extendedprice * Int64(1) -
lineitem.l_discount ELSE Int64(0) END), sum(__common_expr_1) as
sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)]
CoalescePartitionsExec
- UnresolvedShuffleExec: partitioning: Hash([p_partkey@3], 16)
+ UnresolvedShuffleExec: stage=3, partitioning: Hash([p_partkey@3], 16)
diff --git a/ballista/scheduler/tests/tpch_plan_stability/approved/q15.txt
b/ballista/scheduler/tests/tpch_plan_stability/approved/q15.txt
index 2a357f08d..3863e7bc8 100644
--- a/ballista/scheduler/tests/tpch_plan_stability/approved/q15.txt
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q15.txt
@@ -9,17 +9,17 @@ 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: partitioning: Hash([l_suppkey@0], 16)
+ 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: partitioning: Hash([l_suppkey@0], 16)
+ UnresolvedShuffleExec: stage=2, partitioning: Hash([l_suppkey@0], 16)
=== Stage 4 ===
ShuffleWriterExec: partitioning: None
- UnresolvedShuffleExec: partitioning: Hash([max(revenue0.total_revenue)@0],
16)
+ UnresolvedShuffleExec: stage=3, partitioning:
Hash([max(revenue0.total_revenue)@0], 16)
=== Stage 5 ===
SortShuffleWriterExec: partitioning=Hash([s_suppkey@0], 16)
@@ -36,11 +36,11 @@ 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: partitioning: Hash([s_suppkey@0], 16)
+ UnresolvedShuffleExec: stage=5, 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: partitioning: Hash([l_suppkey@0], 16)
+ UnresolvedShuffleExec: stage=6, partitioning: Hash([l_suppkey@0],
16)
=== Stage 8 ===
ShuffleWriterExec: partitioning: None
@@ -48,10 +48,10 @@ ShuffleWriterExec: partitioning: None
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: broadcast=true, upstream_partitions: 16
- UnresolvedShuffleExec: partitioning: Hash([total_revenue@4], 16)
+ UnresolvedShuffleExec: stage=4, broadcast=true, upstream_partitions:
16
+ UnresolvedShuffleExec: stage=7, partitioning:
Hash([total_revenue@4], 16)
=== Stage 9 ===
ShuffleWriterExec: partitioning: None
SortPreservingMergeExec: [s_suppkey@0 ASC NULLS LAST]
- UnresolvedShuffleExec: partitioning: Hash([total_revenue@4], 16)
+ UnresolvedShuffleExec: stage=8, 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 4403c3587..2947632db 100644
--- a/ballista/scheduler/tests/tpch_plan_stability/approved/q16.txt
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q16.txt
@@ -5,7 +5,7 @@ SortShuffleWriterExec: partitioning=Hash([s_suppkey@0], 16)
=== Stage 2 ===
ShuffleWriterExec: partitioning: None
- UnresolvedShuffleExec: partitioning: Hash([s_suppkey@0], 16)
+ UnresolvedShuffleExec: stage=1, partitioning: Hash([s_suppkey@0], 16)
=== Stage 3 ===
SortShuffleWriterExec: partitioning=Hash([ps_partkey@0], 16)
@@ -21,26 +21,26 @@ 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: partitioning: Hash([ps_partkey@0], 16)
+ UnresolvedShuffleExec: stage=3, partitioning: Hash([ps_partkey@0], 16)
SortExec: expr=[p_partkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: partitioning: Hash([p_partkey@0], 16)
+ UnresolvedShuffleExec: stage=4, partitioning: Hash([p_partkey@0], 16)
=== Stage 6 ===
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: broadcast=true, upstream_partitions: 16
- UnresolvedShuffleExec: partitioning: Hash([ps_suppkey@0], 16)
+ UnresolvedShuffleExec: stage=2, broadcast=true, upstream_partitions: 16
+ UnresolvedShuffleExec: stage=5, partitioning: Hash([ps_suppkey@0], 16)
=== Stage 7 ===
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: 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)
=== Stage 8 ===
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: partitioning: Hash([p_brand@0, p_type@1, p_size@2],
16)
+ UnresolvedShuffleExec: stage=7, partitioning: Hash([p_brand@0, p_type@1,
p_size@2], 16)
diff --git a/ballista/scheduler/tests/tpch_plan_stability/approved/q17.txt
b/ballista/scheduler/tests/tpch_plan_stability/approved/q17.txt
index e3b6744bb..a74197f02 100644
--- a/ballista/scheduler/tests/tpch_plan_stability/approved/q17.txt
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q17.txt
@@ -20,17 +20,17 @@ ShuffleWriterExec: partitioning: None
ProjectionExec: expr=[l_quantity@1 as l_quantity, l_extendedprice@2 as
l_extendedprice, p_partkey@3 as p_partkey]
SortMergeJoinExec: join_type=Inner, on=[(l_partkey@0, p_partkey@0)]
SortExec: expr=[l_partkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: partitioning: Hash([l_partkey@0], 16)
+ UnresolvedShuffleExec: stage=1, partitioning:
Hash([l_partkey@0], 16)
SortExec: expr=[p_partkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: partitioning: Hash([p_partkey@0], 16)
+ UnresolvedShuffleExec: stage=2, partitioning:
Hash([p_partkey@0], 16)
SortExec: expr=[l_partkey@1 ASC], preserve_partitioning=[true]
ProjectionExec: expr=[CAST(0.2 * CAST(avg(lineitem.l_quantity)@1 AS
Float64) AS Decimal128(30, 15)) as Float64(0.2) * avg(lineitem.l_quantity),
l_partkey@0 as l_partkey]
AggregateExec: mode=FinalPartitioned, gby=[l_partkey@0 as
l_partkey], aggr=[avg(lineitem.l_quantity)]
- UnresolvedShuffleExec: partitioning: Hash([l_partkey@0], 16)
+ UnresolvedShuffleExec: stage=3, partitioning:
Hash([l_partkey@0], 16)
=== Stage 5 ===
ShuffleWriterExec: partitioning: None
ProjectionExec: expr=[CAST(sum(lineitem.l_extendedprice)@0 AS Float64) / 7
as avg_yearly]
AggregateExec: mode=Final, gby=[], aggr=[sum(lineitem.l_extendedprice)]
CoalescePartitionsExec
- UnresolvedShuffleExec: partitioning: Hash([l_partkey@4], 16)
+ UnresolvedShuffleExec: stage=4, partitioning: Hash([l_partkey@4], 16)
diff --git a/ballista/scheduler/tests/tpch_plan_stability/approved/q18.txt
b/ballista/scheduler/tests/tpch_plan_stability/approved/q18.txt
index 190add99b..7cfabb07e 100644
--- a/ballista/scheduler/tests/tpch_plan_stability/approved/q18.txt
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q18.txt
@@ -11,9 +11,9 @@ SortShuffleWriterExec: partitioning=Hash([o_orderkey@2], 16)
ProjectionExec: expr=[c_custkey@0 as c_custkey, c_name@1 as c_name,
o_orderkey@2 as o_orderkey, o_totalprice@4 as o_totalprice, o_orderdate@5 as
o_orderdate]
SortMergeJoinExec: join_type=Inner, on=[(c_custkey@0, o_custkey@1)]
SortExec: expr=[c_custkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: 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: partitioning: Hash([o_custkey@1], 16)
+ UnresolvedShuffleExec: stage=2, partitioning: Hash([o_custkey@1], 16)
=== Stage 4 ===
SortShuffleWriterExec: partitioning=Hash([l_orderkey@0], 16)
@@ -32,15 +32,15 @@ ShuffleWriterExec: partitioning: None
ProjectionExec: expr=[c_custkey@0 as c_custkey, c_name@1 as c_name,
o_orderkey@2 as o_orderkey, o_totalprice@3 as o_totalprice, o_orderdate@4 as
o_orderdate, l_quantity@6 as l_quantity]
SortMergeJoinExec: join_type=Inner, on=[(o_orderkey@2, l_orderkey@0)]
SortExec: expr=[o_orderkey@2 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: partitioning: Hash([o_orderkey@2], 16)
+ UnresolvedShuffleExec: stage=3, partitioning:
Hash([o_orderkey@2], 16)
SortExec: expr=[l_orderkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: partitioning: Hash([l_orderkey@0], 16)
+ UnresolvedShuffleExec: stage=4, partitioning:
Hash([l_orderkey@0], 16)
SortExec: expr=[l_orderkey@0 ASC], preserve_partitioning=[true]
FilterExec: sum(lineitem.l_quantity)@1 > Some(30000),25,2,
projection=[l_orderkey@0]
AggregateExec: mode=FinalPartitioned, gby=[l_orderkey@0 as
l_orderkey], aggr=[sum(lineitem.l_quantity)]
- UnresolvedShuffleExec: partitioning: Hash([l_orderkey@0], 16)
+ UnresolvedShuffleExec: stage=5, partitioning:
Hash([l_orderkey@0], 16)
=== Stage 7 ===
ShuffleWriterExec: partitioning: None
SortPreservingMergeExec: [o_totalprice@4 DESC, o_orderdate@3 ASC NULLS
LAST], fetch=100
- UnresolvedShuffleExec: partitioning: Hash([o_orderkey@2], 16)
+ UnresolvedShuffleExec: stage=6, partitioning: Hash([o_orderkey@2], 16)
diff --git a/ballista/scheduler/tests/tpch_plan_stability/approved/q19.txt
b/ballista/scheduler/tests/tpch_plan_stability/approved/q19.txt
index a3960ed6e..acd9d8cf1 100644
--- a/ballista/scheduler/tests/tpch_plan_stability/approved/q19.txt
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q19.txt
@@ -14,13 +14,13 @@ ShuffleWriterExec: partitioning: None
ProjectionExec: expr=[l_extendedprice@2 as l_extendedprice, l_discount@3
as l_discount]
SortMergeJoinExec: join_type=Inner, on=[(l_partkey@0, p_partkey@0)],
filter=p_brand@1 = Brand#12 AND p_container@3 IN (SET) ([SM CASE, SM BOX, SM
PACK, SM PKG]) AND l_quantity@0 >= Some(100),15,2 AND l_quantity@0 <=
Some(1100),15,2 AND p_size@2 <= 5 OR p_brand@1 = Brand#23 AND p_container@3 IN
(SET) ([MED BAG, MED BOX, MED PKG, MED PACK]) AND l_quantity@0 >=
Some(1000),15,2 AND l_quantity@0 <= Some(2000),15,2 AND p_size@2 <= 10 OR
p_brand@1 = Brand#34 AND p_container@3 IN (SET) ([L [...]
SortExec: expr=[l_partkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: partitioning: Hash([l_partkey@0], 16)
+ UnresolvedShuffleExec: stage=1, partitioning: Hash([l_partkey@0], 16)
SortExec: expr=[p_partkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: partitioning: Hash([p_partkey@0], 16)
+ UnresolvedShuffleExec: stage=2, partitioning: Hash([p_partkey@0], 16)
=== Stage 4 ===
ShuffleWriterExec: partitioning: None
ProjectionExec: expr=[sum(lineitem.l_extendedprice * Int64(1) -
lineitem.l_discount)@0 as revenue]
AggregateExec: mode=Final, gby=[], aggr=[sum(lineitem.l_extendedprice *
Some(1),20,0 - lineitem.l_discount) as sum(lineitem.l_extendedprice * Int64(1)
- lineitem.l_discount)]
CoalescePartitionsExec
- UnresolvedShuffleExec: partitioning: Hash([p_partkey@4], 16)
+ UnresolvedShuffleExec: stage=3, partitioning: Hash([p_partkey@4], 16)
diff --git a/ballista/scheduler/tests/tpch_plan_stability/approved/q2.txt
b/ballista/scheduler/tests/tpch_plan_stability/approved/q2.txt
index 5d3dba459..a3448fbdf 100644
--- a/ballista/scheduler/tests/tpch_plan_stability/approved/q2.txt
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q2.txt
@@ -5,7 +5,7 @@ SortShuffleWriterExec: partitioning=Hash([r_regionkey@0], 16)
=== Stage 2 ===
ShuffleWriterExec: partitioning: None
- UnresolvedShuffleExec: partitioning: Hash([r_regionkey@0], 16)
+ UnresolvedShuffleExec: stage=1, partitioning: Hash([r_regionkey@0], 16)
=== Stage 3 ===
SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
@@ -13,7 +13,7 @@ SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
=== Stage 4 ===
ShuffleWriterExec: partitioning: None
- UnresolvedShuffleExec: partitioning: Hash([n_nationkey@0], 16)
+ UnresolvedShuffleExec: stage=3, partitioning: Hash([n_nationkey@0], 16)
=== Stage 5 ===
SortShuffleWriterExec: partitioning=Hash([p_partkey@0], 16)
@@ -29,9 +29,9 @@ 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: partitioning: Hash([p_partkey@0], 16)
+ UnresolvedShuffleExec: stage=5, partitioning: Hash([p_partkey@0], 16)
SortExec: expr=[ps_partkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: partitioning: Hash([ps_partkey@0], 16)
+ UnresolvedShuffleExec: stage=6, partitioning: Hash([ps_partkey@0], 16)
=== Stage 8 ===
SortShuffleWriterExec: partitioning=Hash([s_suppkey@0], 16)
@@ -42,29 +42,29 @@ 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: partitioning: Hash([ps_suppkey@2], 16)
+ UnresolvedShuffleExec: stage=7, partitioning: Hash([ps_suppkey@2], 16)
SortExec: expr=[s_suppkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: partitioning: Hash([s_suppkey@0], 16)
+ UnresolvedShuffleExec: stage=8, partitioning: Hash([s_suppkey@0], 16)
=== Stage 10 ===
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: broadcast=true, upstream_partitions: 16
- UnresolvedShuffleExec: partitioning: Hash([s_nationkey@4], 16)
+ 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: broadcast=true, upstream_partitions: 16
- UnresolvedShuffleExec: partitioning: Hash([n_regionkey@9], 16)
+ UnresolvedShuffleExec: stage=2, broadcast=true, upstream_partitions: 16
+ UnresolvedShuffleExec: stage=10, partitioning: Hash([n_regionkey@9],
16)
=== Stage 12 ===
ShuffleWriterExec: partitioning: None
- UnresolvedShuffleExec: partitioning: Hash([p_partkey@0, ps_supplycost@7], 16)
+ UnresolvedShuffleExec: stage=11, partitioning: Hash([p_partkey@0,
ps_supplycost@7], 16)
=== Stage 13 ===
SortShuffleWriterExec: partitioning=Hash([r_regionkey@0], 16)
@@ -73,7 +73,7 @@ SortShuffleWriterExec: partitioning=Hash([r_regionkey@0], 16)
=== Stage 14 ===
ShuffleWriterExec: partitioning: None
- UnresolvedShuffleExec: partitioning: Hash([r_regionkey@0], 16)
+ UnresolvedShuffleExec: stage=13, partitioning: Hash([r_regionkey@0], 16)
=== Stage 15 ===
SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
@@ -81,7 +81,7 @@ SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
=== Stage 16 ===
ShuffleWriterExec: partitioning: None
- UnresolvedShuffleExec: partitioning: Hash([n_nationkey@0], 16)
+ UnresolvedShuffleExec: stage=15, partitioning: Hash([n_nationkey@0], 16)
=== Stage 17 ===
SortShuffleWriterExec: partitioning=Hash([ps_suppkey@1], 16)
@@ -96,17 +96,17 @@ 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: partitioning: Hash([ps_suppkey@1], 16)
+ UnresolvedShuffleExec: stage=17, partitioning: Hash([ps_suppkey@1], 16)
SortExec: expr=[s_suppkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: partitioning: Hash([s_suppkey@0], 16)
+ UnresolvedShuffleExec: stage=18, partitioning: Hash([s_suppkey@0], 16)
=== Stage 20 ===
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: broadcast=true, upstream_partitions: 16
- UnresolvedShuffleExec: partitioning: Hash([s_nationkey@2], 16)
+ UnresolvedShuffleExec: stage=16, broadcast=true, upstream_partitions:
16
+ UnresolvedShuffleExec: stage=19, partitioning: Hash([s_nationkey@2],
16)
=== Stage 21 ===
SortShuffleWriterExec: partitioning=Hash([ps_partkey@0], 16)
@@ -114,24 +114,24 @@ SortShuffleWriterExec: partitioning=Hash([ps_partkey@0],
16)
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: broadcast=true, upstream_partitions: 16
- UnresolvedShuffleExec: partitioning: Hash([n_regionkey@2], 16)
+ UnresolvedShuffleExec: stage=14, broadcast=true,
upstream_partitions: 16
+ UnresolvedShuffleExec: stage=20, partitioning: Hash([n_regionkey@2],
16)
=== Stage 22 ===
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: partitioning: Hash([ps_partkey@0], 16)
+ UnresolvedShuffleExec: stage=21, partitioning: Hash([ps_partkey@0], 16)
=== Stage 23 ===
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: broadcast=true, upstream_partitions: 16
- UnresolvedShuffleExec: partitioning: Hash([ps_partkey@1,
min(partsupp.ps_supplycost)@0], 16)
+ UnresolvedShuffleExec: stage=12, broadcast=true, upstream_partitions:
16
+ UnresolvedShuffleExec: stage=22, partitioning: Hash([ps_partkey@1,
min(partsupp.ps_supplycost)@0], 16)
=== Stage 24 ===
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: partitioning: Hash([p_partkey@3,
min(partsupp.ps_supplycost)@9], 16)
+ UnresolvedShuffleExec: stage=23, 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 a684c1a4d..0066e4843 100644
--- a/ballista/scheduler/tests/tpch_plan_stability/approved/q20.txt
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q20.txt
@@ -5,7 +5,7 @@ SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
=== Stage 2 ===
ShuffleWriterExec: partitioning: None
- UnresolvedShuffleExec: partitioning: Hash([n_nationkey@0], 16)
+ UnresolvedShuffleExec: stage=1, partitioning: Hash([n_nationkey@0], 16)
=== Stage 3 ===
SortShuffleWriterExec: partitioning=Hash([s_nationkey@3], 16)
@@ -16,8 +16,8 @@ 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: broadcast=true, upstream_partitions: 16
- UnresolvedShuffleExec: partitioning: Hash([s_nationkey@3], 16)
+ UnresolvedShuffleExec: stage=2, broadcast=true, upstream_partitions: 16
+ UnresolvedShuffleExec: stage=3, partitioning: Hash([s_nationkey@3], 16)
=== Stage 5 ===
SortShuffleWriterExec: partitioning=Hash([ps_partkey@0], 16)
@@ -32,9 +32,9 @@ SortShuffleWriterExec: partitioning=Hash([p_partkey@0], 16)
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: partitioning: Hash([ps_partkey@0], 16)
+ UnresolvedShuffleExec: stage=5, partitioning: Hash([ps_partkey@0], 16)
SortExec: expr=[p_partkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: partitioning: Hash([p_partkey@0], 16)
+ UnresolvedShuffleExec: stage=6, partitioning: Hash([p_partkey@0], 16)
=== Stage 8 ===
SortShuffleWriterExec: partitioning=Hash([l_partkey@0, l_suppkey@1], 16)
@@ -47,11 +47,11 @@ 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: partitioning: Hash([ps_partkey@0,
ps_suppkey@1], 16)
+ UnresolvedShuffleExec: stage=7, 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: partitioning: Hash([l_partkey@0,
l_suppkey@1], 16)
+ UnresolvedShuffleExec: stage=8, partitioning: Hash([l_partkey@0,
l_suppkey@1], 16)
=== Stage 10 ===
ShuffleWriterExec: partitioning: None
@@ -59,11 +59,11 @@ ShuffleWriterExec: partitioning: None
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: partitioning: Hash([s_suppkey@0], 16)
+ UnresolvedShuffleExec: stage=4, partitioning: Hash([s_suppkey@0], 16)
SortExec: expr=[ps_suppkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: partitioning: Hash([ps_suppkey@0], 16)
+ UnresolvedShuffleExec: stage=9, partitioning: Hash([ps_suppkey@0],
16)
=== Stage 11 ===
ShuffleWriterExec: partitioning: None
SortPreservingMergeExec: [s_name@0 ASC NULLS LAST]
- UnresolvedShuffleExec: partitioning: Hash([s_suppkey@0], 16)
+ UnresolvedShuffleExec: stage=10, 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 693152cd6..bd49954af 100644
--- a/ballista/scheduler/tests/tpch_plan_stability/approved/q21.txt
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q21.txt
@@ -5,7 +5,7 @@ SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
=== Stage 2 ===
ShuffleWriterExec: partitioning: None
- UnresolvedShuffleExec: partitioning: Hash([n_nationkey@0], 16)
+ UnresolvedShuffleExec: stage=1, partitioning: Hash([n_nationkey@0], 16)
=== Stage 3 ===
SortShuffleWriterExec: partitioning=Hash([s_suppkey@0], 16)
@@ -21,9 +21,9 @@ 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: partitioning: Hash([s_suppkey@0], 16)
+ UnresolvedShuffleExec: stage=3, partitioning: Hash([s_suppkey@0], 16)
SortExec: expr=[l_suppkey@1 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: partitioning: Hash([l_suppkey@1], 16)
+ UnresolvedShuffleExec: stage=4, partitioning: Hash([l_suppkey@1], 16)
=== Stage 6 ===
SortShuffleWriterExec: partitioning=Hash([o_orderkey@0], 16)
@@ -35,17 +35,17 @@ 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: partitioning: Hash([l_orderkey@2], 16)
+ UnresolvedShuffleExec: stage=5, partitioning: Hash([l_orderkey@2], 16)
SortExec: expr=[o_orderkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: partitioning: Hash([o_orderkey@0], 16)
+ UnresolvedShuffleExec: stage=6, partitioning: Hash([o_orderkey@0], 16)
=== Stage 8 ===
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: broadcast=true, upstream_partitions: 16
- UnresolvedShuffleExec: partitioning: Hash([s_nationkey@1], 16)
+ UnresolvedShuffleExec: stage=2, broadcast=true, upstream_partitions: 16
+ UnresolvedShuffleExec: stage=7, partitioning: Hash([s_nationkey@1], 16)
=== Stage 9 ===
SortShuffleWriterExec: partitioning=Hash([l_orderkey@0], 16)
@@ -63,20 +63,20 @@ SortShuffleWriterExec: partitioning=Hash([s_name@0], 16)
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: partitioning: Hash([l_orderkey@1], 16)
+ UnresolvedShuffleExec: stage=8, partitioning: Hash([l_orderkey@1],
16)
SortExec: expr=[l_orderkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: partitioning: Hash([l_orderkey@0], 16)
+ UnresolvedShuffleExec: stage=9, partitioning: Hash([l_orderkey@0],
16)
SortExec: expr=[l_orderkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: partitioning: Hash([l_orderkey@0], 16)
+ UnresolvedShuffleExec: stage=10, partitioning: Hash([l_orderkey@0],
16)
=== Stage 12 ===
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: partitioning: Hash([s_name@0], 16)
+ UnresolvedShuffleExec: stage=11, partitioning: Hash([s_name@0], 16)
=== Stage 13 ===
ShuffleWriterExec: partitioning: None
SortPreservingMergeExec: [numwait@1 DESC, s_name@0 ASC NULLS LAST], fetch=100
- UnresolvedShuffleExec: partitioning: Hash([s_name@0], 16)
+ UnresolvedShuffleExec: stage=12, partitioning: Hash([s_name@0], 16)
diff --git a/ballista/scheduler/tests/tpch_plan_stability/approved/q22.txt
b/ballista/scheduler/tests/tpch_plan_stability/approved/q22.txt
index 4dd609587..8255afee9 100644
--- a/ballista/scheduler/tests/tpch_plan_stability/approved/q22.txt
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q22.txt
@@ -18,18 +18,18 @@ SortShuffleWriterExec: partitioning=Hash([cntrycode@0], 16)
ProjectionExec: expr=[c_phone@1 as c_phone, c_acctbal@2 as c_acctbal,
CAST(c_acctbal@2 AS Decimal128(19, 6)) as join_proj_push_down_1]
SortMergeJoinExec: join_type=LeftAnti, on=[(c_custkey@0,
o_custkey@0)]
SortExec: expr=[c_custkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: partitioning: Hash([c_custkey@0], 16)
+ UnresolvedShuffleExec: stage=1, partitioning:
Hash([c_custkey@0], 16)
SortExec: expr=[o_custkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: partitioning: Hash([o_custkey@0], 16)
+ UnresolvedShuffleExec: stage=2, partitioning:
Hash([o_custkey@0], 16)
=== Stage 4 ===
ShuffleWriterExec: partitioning: None
SortExec: expr=[cntrycode@0 ASC NULLS LAST], preserve_partitioning=[true]
ProjectionExec: expr=[cntrycode@0 as cntrycode, count(Int64(1))@1 as
numcust, sum(custsale.c_acctbal)@2 as totacctbal]
AggregateExec: mode=FinalPartitioned, gby=[cntrycode@0 as cntrycode],
aggr=[count(Int64(1)), sum(custsale.c_acctbal)]
- UnresolvedShuffleExec: partitioning: Hash([cntrycode@0], 16)
+ UnresolvedShuffleExec: stage=3, partitioning: Hash([cntrycode@0], 16)
=== Stage 5 ===
ShuffleWriterExec: partitioning: None
SortPreservingMergeExec: [cntrycode@0 ASC NULLS LAST]
- UnresolvedShuffleExec: partitioning: Hash([cntrycode@0], 16)
+ UnresolvedShuffleExec: stage=4, partitioning: Hash([cntrycode@0], 16)
diff --git a/ballista/scheduler/tests/tpch_plan_stability/approved/q3.txt
b/ballista/scheduler/tests/tpch_plan_stability/approved/q3.txt
index a8d755ca9..ea275d184 100644
--- a/ballista/scheduler/tests/tpch_plan_stability/approved/q3.txt
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q3.txt
@@ -13,9 +13,9 @@ SortShuffleWriterExec: partitioning=Hash([o_orderkey@0], 16)
ProjectionExec: expr=[o_orderkey@1 as o_orderkey, o_orderdate@3 as
o_orderdate, o_shippriority@4 as o_shippriority]
SortMergeJoinExec: join_type=Inner, on=[(c_custkey@0, o_custkey@1)]
SortExec: expr=[c_custkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: 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: partitioning: Hash([o_custkey@1], 16)
+ UnresolvedShuffleExec: stage=2, partitioning: Hash([o_custkey@1], 16)
=== Stage 4 ===
SortShuffleWriterExec: partitioning=Hash([l_orderkey@0], 16)
@@ -30,11 +30,11 @@ ShuffleWriterExec: partitioning: None
ProjectionExec: expr=[o_orderdate@1 as o_orderdate, o_shippriority@2
as o_shippriority, l_orderkey@3 as l_orderkey, l_extendedprice@4 as
l_extendedprice, l_discount@5 as l_discount]
SortMergeJoinExec: join_type=Inner, on=[(o_orderkey@0, l_orderkey@0)]
SortExec: expr=[o_orderkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: partitioning: Hash([o_orderkey@0], 16)
+ UnresolvedShuffleExec: stage=3, partitioning:
Hash([o_orderkey@0], 16)
SortExec: expr=[l_orderkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: partitioning: Hash([l_orderkey@0], 16)
+ UnresolvedShuffleExec: stage=4, partitioning:
Hash([l_orderkey@0], 16)
=== Stage 6 ===
ShuffleWriterExec: partitioning: None
SortPreservingMergeExec: [revenue@1 DESC, o_orderdate@2 ASC NULLS LAST],
fetch=10
- UnresolvedShuffleExec: partitioning: Hash([l_orderkey@0], 16)
+ UnresolvedShuffleExec: stage=5, partitioning: Hash([l_orderkey@0], 16)
diff --git a/ballista/scheduler/tests/tpch_plan_stability/approved/q4.txt
b/ballista/scheduler/tests/tpch_plan_stability/approved/q4.txt
index bbb97d34b..5e05011fe 100644
--- a/ballista/scheduler/tests/tpch_plan_stability/approved/q4.txt
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q4.txt
@@ -14,18 +14,18 @@ SortShuffleWriterExec:
partitioning=Hash([o_orderpriority@0], 16)
ProjectionExec: expr=[o_orderpriority@1 as o_orderpriority]
SortMergeJoinExec: join_type=LeftSemi, on=[(o_orderkey@0, l_orderkey@0)]
SortExec: expr=[o_orderkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: partitioning: Hash([o_orderkey@0], 16)
+ UnresolvedShuffleExec: stage=1, partitioning: Hash([o_orderkey@0],
16)
SortExec: expr=[l_orderkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: partitioning: Hash([l_orderkey@0], 16)
+ UnresolvedShuffleExec: stage=2, partitioning: Hash([l_orderkey@0],
16)
=== Stage 4 ===
ShuffleWriterExec: partitioning: None
SortExec: expr=[o_orderpriority@0 ASC NULLS LAST],
preserve_partitioning=[true]
ProjectionExec: expr=[o_orderpriority@0 as o_orderpriority,
count(Int64(1))@1 as order_count]
AggregateExec: mode=FinalPartitioned, gby=[o_orderpriority@0 as
o_orderpriority], aggr=[count(Int64(1))]
- UnresolvedShuffleExec: partitioning: Hash([o_orderpriority@0], 16)
+ UnresolvedShuffleExec: stage=3, partitioning:
Hash([o_orderpriority@0], 16)
=== Stage 5 ===
ShuffleWriterExec: partitioning: None
SortPreservingMergeExec: [o_orderpriority@0 ASC NULLS LAST]
- UnresolvedShuffleExec: partitioning: Hash([o_orderpriority@0], 16)
+ UnresolvedShuffleExec: stage=4, partitioning: Hash([o_orderpriority@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 c159882e4..78c140e22 100644
--- a/ballista/scheduler/tests/tpch_plan_stability/approved/q5.txt
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q5.txt
@@ -5,7 +5,7 @@ SortShuffleWriterExec: partitioning=Hash([r_regionkey@0], 16)
=== Stage 2 ===
ShuffleWriterExec: partitioning: None
- UnresolvedShuffleExec: partitioning: Hash([r_regionkey@0], 16)
+ UnresolvedShuffleExec: stage=1, partitioning: Hash([r_regionkey@0], 16)
=== Stage 3 ===
SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
@@ -13,7 +13,7 @@ SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
=== Stage 4 ===
ShuffleWriterExec: partitioning: None
- UnresolvedShuffleExec: partitioning: Hash([n_nationkey@0], 16)
+ UnresolvedShuffleExec: stage=3, partitioning: Hash([n_nationkey@0], 16)
=== Stage 5 ===
SortShuffleWriterExec: partitioning=Hash([c_custkey@0], 16)
@@ -29,9 +29,9 @@ 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: partitioning: Hash([c_custkey@0], 16)
+ UnresolvedShuffleExec: stage=5, partitioning: Hash([c_custkey@0], 16)
SortExec: expr=[o_custkey@1 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: partitioning: Hash([o_custkey@1], 16)
+ UnresolvedShuffleExec: stage=6, partitioning: Hash([o_custkey@1], 16)
=== Stage 8 ===
SortShuffleWriterExec: partitioning=Hash([l_orderkey@0], 16)
@@ -42,9 +42,9 @@ 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: partitioning: Hash([o_orderkey@1], 16)
+ UnresolvedShuffleExec: stage=7, partitioning: Hash([o_orderkey@1], 16)
SortExec: expr=[l_orderkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: partitioning: Hash([l_orderkey@0], 16)
+ UnresolvedShuffleExec: stage=8, partitioning: Hash([l_orderkey@0], 16)
=== Stage 10 ===
SortShuffleWriterExec: partitioning=Hash([s_suppkey@0, s_nationkey@1], 16)
@@ -55,17 +55,17 @@ 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: partitioning: Hash([l_suppkey@1,
c_nationkey@0], 16)
+ UnresolvedShuffleExec: stage=9, partitioning: Hash([l_suppkey@1,
c_nationkey@0], 16)
SortExec: expr=[s_suppkey@0 ASC, s_nationkey@1 ASC],
preserve_partitioning=[true]
- UnresolvedShuffleExec: partitioning: Hash([s_suppkey@0,
s_nationkey@1], 16)
+ UnresolvedShuffleExec: stage=10, partitioning: Hash([s_suppkey@0,
s_nationkey@1], 16)
=== Stage 12 ===
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: broadcast=true, upstream_partitions: 16
- UnresolvedShuffleExec: partitioning: Hash([s_nationkey@2], 16)
+ UnresolvedShuffleExec: stage=4, broadcast=true, upstream_partitions: 16
+ UnresolvedShuffleExec: stage=11, partitioning: Hash([s_nationkey@2],
16)
=== Stage 13 ===
SortShuffleWriterExec: partitioning=Hash([n_name@0], 16)
@@ -73,17 +73,17 @@ SortShuffleWriterExec: partitioning=Hash([n_name@0], 16)
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: broadcast=true, upstream_partitions: 16
- UnresolvedShuffleExec: partitioning: Hash([n_regionkey@3], 16)
+ UnresolvedShuffleExec: stage=2, broadcast=true, upstream_partitions:
16
+ UnresolvedShuffleExec: stage=12, partitioning: Hash([n_regionkey@3],
16)
=== Stage 14 ===
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: partitioning: Hash([n_name@0], 16)
+ UnresolvedShuffleExec: stage=13, partitioning: Hash([n_name@0], 16)
=== Stage 15 ===
ShuffleWriterExec: partitioning: None
SortPreservingMergeExec: [revenue@1 DESC]
- UnresolvedShuffleExec: partitioning: Hash([n_name@0], 16)
+ UnresolvedShuffleExec: stage=14, 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 77cbc87a8..abfcad9fb 100644
--- a/ballista/scheduler/tests/tpch_plan_stability/approved/q7.txt
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q7.txt
@@ -5,7 +5,7 @@ SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
=== Stage 2 ===
ShuffleWriterExec: partitioning: None
- UnresolvedShuffleExec: partitioning: Hash([n_nationkey@0], 16)
+ UnresolvedShuffleExec: stage=1, partitioning: Hash([n_nationkey@0], 16)
=== Stage 3 ===
SortShuffleWriterExec: partitioning=Hash([s_suppkey@0], 16)
@@ -21,9 +21,9 @@ 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: partitioning: Hash([s_suppkey@0], 16)
+ UnresolvedShuffleExec: stage=3, partitioning: Hash([s_suppkey@0], 16)
SortExec: expr=[l_suppkey@1 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: partitioning: Hash([l_suppkey@1], 16)
+ UnresolvedShuffleExec: stage=4, partitioning: Hash([l_suppkey@1], 16)
=== Stage 6 ===
SortShuffleWriterExec: partitioning=Hash([o_orderkey@0], 16)
@@ -34,9 +34,9 @@ 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: partitioning: Hash([l_orderkey@1], 16)
+ UnresolvedShuffleExec: stage=5, partitioning: Hash([l_orderkey@1], 16)
SortExec: expr=[o_orderkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: partitioning: Hash([o_orderkey@0], 16)
+ UnresolvedShuffleExec: stage=6, partitioning: Hash([o_orderkey@0], 16)
=== Stage 8 ===
SortShuffleWriterExec: partitioning=Hash([c_custkey@0], 16)
@@ -47,21 +47,21 @@ 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: partitioning: Hash([o_custkey@4], 16)
+ UnresolvedShuffleExec: stage=7, partitioning: Hash([o_custkey@4], 16)
SortExec: expr=[c_custkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: partitioning: Hash([c_custkey@0], 16)
+ UnresolvedShuffleExec: stage=8, partitioning: Hash([c_custkey@0], 16)
=== Stage 10 ===
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: broadcast=true, upstream_partitions: 16
- UnresolvedShuffleExec: partitioning: Hash([s_nationkey@0], 16)
+ UnresolvedShuffleExec: stage=2, broadcast=true, upstream_partitions: 16
+ UnresolvedShuffleExec: stage=9, partitioning: Hash([s_nationkey@0], 16)
=== Stage 11 ===
ShuffleWriterExec: partitioning: None
- UnresolvedShuffleExec: partitioning: Hash([c_nationkey@3], 16)
+ UnresolvedShuffleExec: stage=10, partitioning: Hash([c_nationkey@3], 16)
=== Stage 12 ===
SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
@@ -73,17 +73,17 @@ SortShuffleWriterExec: partitioning=Hash([supp_nation@0,
cust_nation@1, l_year@2
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: broadcast=true, upstream_partitions: 16
- UnresolvedShuffleExec: partitioning: Hash([n_nationkey@0], 16)
+ UnresolvedShuffleExec: stage=11, broadcast=true, upstream_partitions:
16
+ UnresolvedShuffleExec: stage=12, partitioning: Hash([n_nationkey@0],
16)
=== Stage 14 ===
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: partitioning: Hash([supp_nation@0,
cust_nation@1, l_year@2], 16)
+ UnresolvedShuffleExec: stage=13, partitioning: Hash([supp_nation@0,
cust_nation@1, l_year@2], 16)
=== Stage 15 ===
ShuffleWriterExec: partitioning: None
SortPreservingMergeExec: [supp_nation@0 ASC NULLS LAST, cust_nation@1 ASC
NULLS LAST, l_year@2 ASC NULLS LAST]
- UnresolvedShuffleExec: partitioning: Hash([supp_nation@0, cust_nation@1,
l_year@2], 16)
+ UnresolvedShuffleExec: stage=14, 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 d70be863e..912a82560 100644
--- a/ballista/scheduler/tests/tpch_plan_stability/approved/q8.txt
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q8.txt
@@ -5,7 +5,7 @@ SortShuffleWriterExec: partitioning=Hash([r_regionkey@0], 16)
=== Stage 2 ===
ShuffleWriterExec: partitioning: None
- UnresolvedShuffleExec: partitioning: Hash([r_regionkey@0], 16)
+ UnresolvedShuffleExec: stage=1, partitioning: Hash([r_regionkey@0], 16)
=== Stage 3 ===
SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
@@ -13,7 +13,7 @@ SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
=== Stage 4 ===
ShuffleWriterExec: partitioning: None
- UnresolvedShuffleExec: partitioning: Hash([n_nationkey@0], 16)
+ UnresolvedShuffleExec: stage=3, partitioning: Hash([n_nationkey@0], 16)
=== Stage 5 ===
SortShuffleWriterExec: partitioning=Hash([p_partkey@0], 16)
@@ -29,9 +29,9 @@ 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: partitioning: Hash([p_partkey@0], 16)
+ UnresolvedShuffleExec: stage=5, partitioning: Hash([p_partkey@0], 16)
SortExec: expr=[l_partkey@1 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: partitioning: Hash([l_partkey@1], 16)
+ UnresolvedShuffleExec: stage=6, partitioning: Hash([l_partkey@1], 16)
=== Stage 8 ===
SortShuffleWriterExec: partitioning=Hash([s_suppkey@0], 16)
@@ -42,9 +42,9 @@ 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: partitioning: Hash([l_suppkey@1], 16)
+ UnresolvedShuffleExec: stage=7, partitioning: Hash([l_suppkey@1], 16)
SortExec: expr=[s_suppkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: partitioning: Hash([s_suppkey@0], 16)
+ UnresolvedShuffleExec: stage=8, partitioning: Hash([s_suppkey@0], 16)
=== Stage 10 ===
SortShuffleWriterExec: partitioning=Hash([o_orderkey@0], 16)
@@ -56,9 +56,9 @@ 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: partitioning: Hash([l_orderkey@0], 16)
+ UnresolvedShuffleExec: stage=9, partitioning: Hash([l_orderkey@0], 16)
SortExec: expr=[o_orderkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: partitioning: Hash([o_orderkey@0], 16)
+ UnresolvedShuffleExec: stage=10, partitioning: Hash([o_orderkey@0], 16)
=== Stage 12 ===
SortShuffleWriterExec: partitioning=Hash([c_custkey@0], 16)
@@ -69,21 +69,21 @@ 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: partitioning: Hash([o_custkey@3], 16)
+ UnresolvedShuffleExec: stage=11, partitioning: Hash([o_custkey@3], 16)
SortExec: expr=[c_custkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: partitioning: Hash([c_custkey@0], 16)
+ UnresolvedShuffleExec: stage=12, partitioning: Hash([c_custkey@0], 16)
=== Stage 14 ===
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: broadcast=true, upstream_partitions: 16
- UnresolvedShuffleExec: partitioning: Hash([c_nationkey@4], 16)
+ UnresolvedShuffleExec: stage=4, broadcast=true, upstream_partitions: 16
+ UnresolvedShuffleExec: stage=13, partitioning: Hash([c_nationkey@4],
16)
=== Stage 15 ===
ShuffleWriterExec: partitioning: None
- UnresolvedShuffleExec: partitioning: Hash([s_nationkey@2], 16)
+ UnresolvedShuffleExec: stage=14, partitioning: Hash([s_nationkey@2], 16)
=== Stage 16 ===
SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
@@ -93,8 +93,8 @@ SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
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: broadcast=true, upstream_partitions: 16
- UnresolvedShuffleExec: partitioning: Hash([n_nationkey@0], 16)
+ UnresolvedShuffleExec: stage=15, broadcast=true, upstream_partitions: 16
+ UnresolvedShuffleExec: stage=16, partitioning: Hash([n_nationkey@0], 16)
=== Stage 18 ===
SortShuffleWriterExec: partitioning=Hash([o_year@0], 16)
@@ -102,17 +102,17 @@ SortShuffleWriterExec: partitioning=Hash([o_year@0], 16)
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: broadcast=true, upstream_partitions: 16
- UnresolvedShuffleExec: partitioning: Hash([n_regionkey@3], 16)
+ UnresolvedShuffleExec: stage=2, broadcast=true, upstream_partitions:
16
+ UnresolvedShuffleExec: stage=17, partitioning: Hash([n_regionkey@3],
16)
=== Stage 19 ===
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: partitioning: Hash([o_year@0], 16)
+ UnresolvedShuffleExec: stage=18, partitioning: Hash([o_year@0], 16)
=== Stage 20 ===
ShuffleWriterExec: partitioning: None
SortPreservingMergeExec: [o_year@0 ASC NULLS LAST]
- UnresolvedShuffleExec: partitioning: Hash([o_year@0], 16)
+ UnresolvedShuffleExec: stage=19, 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 3d91e1dd2..0f0fa2d63 100644
--- a/ballista/scheduler/tests/tpch_plan_stability/approved/q9.txt
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q9.txt
@@ -4,7 +4,7 @@ SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
=== Stage 2 ===
ShuffleWriterExec: partitioning: None
- UnresolvedShuffleExec: partitioning: Hash([n_nationkey@0], 16)
+ UnresolvedShuffleExec: stage=1, partitioning: Hash([n_nationkey@0], 16)
=== Stage 3 ===
SortShuffleWriterExec: partitioning=Hash([p_partkey@0], 16)
@@ -20,9 +20,9 @@ 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: partitioning: Hash([p_partkey@0], 16)
+ UnresolvedShuffleExec: stage=3, partitioning: Hash([p_partkey@0], 16)
SortExec: expr=[l_partkey@1 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: partitioning: Hash([l_partkey@1], 16)
+ UnresolvedShuffleExec: stage=4, partitioning: Hash([l_partkey@1], 16)
=== Stage 6 ===
SortShuffleWriterExec: partitioning=Hash([s_suppkey@0], 16)
@@ -33,9 +33,9 @@ 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: partitioning: Hash([l_suppkey@2], 16)
+ UnresolvedShuffleExec: stage=5, partitioning: Hash([l_suppkey@2], 16)
SortExec: expr=[s_suppkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: partitioning: Hash([s_suppkey@0], 16)
+ UnresolvedShuffleExec: stage=6, partitioning: Hash([s_suppkey@0], 16)
=== Stage 8 ===
SortShuffleWriterExec: partitioning=Hash([ps_suppkey@1, ps_partkey@0], 16)
@@ -46,9 +46,9 @@ 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: partitioning: Hash([l_suppkey@2, l_partkey@1],
16)
+ UnresolvedShuffleExec: stage=7, partitioning: Hash([l_suppkey@2,
l_partkey@1], 16)
SortExec: expr=[ps_suppkey@1 ASC, ps_partkey@0 ASC],
preserve_partitioning=[true]
- UnresolvedShuffleExec: partitioning: Hash([ps_suppkey@1,
ps_partkey@0], 16)
+ UnresolvedShuffleExec: stage=8, partitioning: Hash([ps_suppkey@1,
ps_partkey@0], 16)
=== Stage 10 ===
SortShuffleWriterExec: partitioning=Hash([o_orderkey@0], 16)
@@ -59,9 +59,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: partitioning: Hash([l_orderkey@0], 16)
+ UnresolvedShuffleExec: stage=9, partitioning: Hash([l_orderkey@0], 16)
SortExec: expr=[o_orderkey@0 ASC], preserve_partitioning=[true]
- UnresolvedShuffleExec: partitioning: Hash([o_orderkey@0], 16)
+ UnresolvedShuffleExec: stage=10, partitioning: Hash([o_orderkey@0], 16)
=== Stage 12 ===
SortShuffleWriterExec: partitioning=Hash([nation@0, o_year@1], 16)
@@ -69,17 +69,17 @@ SortShuffleWriterExec: partitioning=Hash([nation@0,
o_year@1], 16)
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: broadcast=true, upstream_partitions: 16
- UnresolvedShuffleExec: partitioning: Hash([s_nationkey@3], 16)
+ UnresolvedShuffleExec: stage=2, broadcast=true, upstream_partitions:
16
+ UnresolvedShuffleExec: stage=11, partitioning: Hash([s_nationkey@3],
16)
=== Stage 13 ===
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: partitioning: Hash([nation@0, o_year@1], 16)
+ UnresolvedShuffleExec: stage=12, partitioning: Hash([nation@0,
o_year@1], 16)
=== Stage 14 ===
ShuffleWriterExec: partitioning: None
SortPreservingMergeExec: [nation@0 ASC NULLS LAST, o_year@1 DESC]
- UnresolvedShuffleExec: partitioning: Hash([nation@0, o_year@1], 16)
+ UnresolvedShuffleExec: stage=13, partitioning: Hash([nation@0, o_year@1],
16)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
