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 cb33e65de fix(core): report the real partitioning from
ShuffleWriterExec (#2307)
cb33e65de is described below
commit cb33e65def309f05a5b1588e3b1d069c92f16cca
Author: Andy Grove <[email protected]>
AuthorDate: Sat Aug 15 12:01:31 2026 -0700
fix(core): report the real partitioning from ShuffleWriterExec (#2307)
* fix(core): report the real partitioning from ShuffleWriterExec
`ShuffleWriterExec`'s `DisplayAs` wrote the string literal
"partitioning: None". It used to print a value, until #2106 removed the
`shuffle_output_partitioning` field along with the hash-partitioning
writer; the token was then frozen as a literal so the TPC-H plan-stability
goldens would not churn in the same diff.
The result carried no information and was kept because it carried none: a
constant cannot drift, so it cannot break a golden. It was also misleading,
since the `None` refers to the absence of a repartitioning scheme, not to
the writer's output partitioning. This writer preserves its input's, and
`try_new` already sets its `PlanProperties` accordingly.
Report `self.properties().output_partitioning()`, matching how
`SortShuffleWriterExec` renders its own. Stage plans now show either the
inherited hash partitioning or the collapsed single partition, both
verifiable against the child plan directly below.
Goldens regenerated with BALLISTA_GENERATE_GOLDEN=1.
* test: update plan expectations for the real ShuffleWriterExec partitioning
The display change left three sets of expectations still asserting the old
"partitioning: None" literal:
- `context_checks.rs` EXPLAIN / EXPLAIN ANALYZE tables, which now show the
hash partitioning the final stage actually carries
- eight `assert_plan!` inline snapshots in the scheduler (planner and AQE
tests), regenerated with `cargo insta`
- stale `/* Expected result: */` comment blocks in `planner.rs` and the
sample stage plan in `dev/bin/showplan.sh`
CI only reported the client failures because `cargo test` stops at the
first failing test binary, so the scheduler snapshots never ran.
---
ballista/client/tests/context_checks.rs | 4 +-
.../core/src/execution_plans/shuffle_writer.rs | 46 ++++++++++++++++++++--
ballista/scheduler/src/planner.rs | 14 +++----
.../scheduler/src/state/aqe/test/alter_stages.rs | 8 ++--
.../scheduler/src/state/aqe/test/coalesce_rule.rs | 2 +-
.../scheduler/src/state/aqe/test/join_selection.rs | 2 +-
.../scheduler/src/state/aqe/test/plan_to_stages.rs | 2 +-
.../tests/tpch_plan_stability/approved/q1.txt | 4 +-
.../tests/tpch_plan_stability/approved/q10.txt | 4 +-
.../tests/tpch_plan_stability/approved/q11.txt | 6 +--
.../tests/tpch_plan_stability/approved/q12.txt | 4 +-
.../tests/tpch_plan_stability/approved/q13.txt | 4 +-
.../tests/tpch_plan_stability/approved/q14.txt | 4 +-
.../tests/tpch_plan_stability/approved/q15.txt | 6 +--
.../tests/tpch_plan_stability/approved/q16.txt | 4 +-
.../tests/tpch_plan_stability/approved/q17.txt | 4 +-
.../tests/tpch_plan_stability/approved/q18.txt | 4 +-
.../tests/tpch_plan_stability/approved/q19.txt | 4 +-
.../tests/tpch_plan_stability/approved/q2.txt | 4 +-
.../tests/tpch_plan_stability/approved/q20.txt | 4 +-
.../tests/tpch_plan_stability/approved/q21.txt | 4 +-
.../tests/tpch_plan_stability/approved/q22.txt | 4 +-
.../tests/tpch_plan_stability/approved/q3.txt | 4 +-
.../tests/tpch_plan_stability/approved/q4.txt | 4 +-
.../tests/tpch_plan_stability/approved/q5.txt | 4 +-
.../tests/tpch_plan_stability/approved/q6.txt | 2 +-
.../tests/tpch_plan_stability/approved/q7.txt | 4 +-
.../tests/tpch_plan_stability/approved/q8.txt | 4 +-
.../tests/tpch_plan_stability/approved/q9.txt | 4 +-
dev/bin/showplan.sh | 2 +-
30 files changed, 104 insertions(+), 66 deletions(-)
diff --git a/ballista/client/tests/context_checks.rs
b/ballista/client/tests/context_checks.rs
index f980ad96b..45a03b78b 100644
--- a/ballista/client/tests/context_checks.rs
+++ b/ballista/client/tests/context_checks.rs
@@ -1111,7 +1111,7 @@ mod supported {
"| |
|",
"| | =========UnResolvedStage[stage_id=2.0,
children=1]=========
|",
"| | Inputs{1: StageOutput { partition_locations:
{}, complete: false }}
|",
- "| | ShuffleWriterExec: partitioning: None
|",
+ "| | ShuffleWriterExec: partitioning:
Hash([id@1], 16)
|",
"| | 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: stage=1,
partitioning: Hash([id@0], 16)
|",
@@ -1194,7 +1194,7 @@ mod supported {
"| | PlaceholderRowExec, metrics=[...]
|",
"| |
|",
"| | =========SuccessfulStage[stage_id=2,
partitions=16]=========
|",
- "| | ShuffleWriterExec: partitioning: None,
metrics=[output_rows=..., input_rows=..., write_time=...]
|",
+ "| | ShuffleWriterExec: partitioning:
Hash([id@1], 16), metrics=[output_rows=..., input_rows=..., write_time=...]
|",
"| | ProjectionExec: expr=[count(Int64(1))@1
as count(*), id@0 as id], metrics=[output_rows=..., elapsed_compute=...,
output_bytes=..., output_batches=..., expr_0_eval_time=...,
expr_1_eval_time=...]
|",
"| | AggregateExec: mode=FinalPartitioned,
gby=[id@0 as id], aggr=[count(Int64(1))], metrics=[output_rows=...,
elapsed_compute=..., output_bytes=..., output_batches=..., spill_count=...,
spilled_bytes=..., spilled_rows=..., aggregate_arguments_time=...,
aggregation_time=..., emitting_time=..., time_calculating_group_ids=...]
|",
"| | ShuffleReaderExec: upstream_stage: 1,
partitioning: Hash([id@0], 16), metrics=[output_rows=..., elapsed_compute=...,
output_bytes=..., output_batches=..., decoded_bytes=..., fetch_requests=...,
fetch_retries=..., local_partitions=..., remote_partitions=..., fetch_time=...,
local_read_time=..., permit_wait_time=...]
|",
diff --git a/ballista/core/src/execution_plans/shuffle_writer.rs
b/ballista/core/src/execution_plans/shuffle_writer.rs
index 1cabe6e14..1e2736c1f 100644
--- a/ballista/core/src/execution_plans/shuffle_writer.rs
+++ b/ballista/core/src/execution_plans/shuffle_writer.rs
@@ -627,14 +627,16 @@ impl DisplayAs for ShuffleWriterExec {
t: DisplayFormatType,
f: &mut std::fmt::Formatter,
) -> std::fmt::Result {
+ // This writer never repartitions, so its output partitioning is its
+ // input's. `shuffle_output_partitioning()` is the *repartitioning
+ // scheme*, always None here, which says nothing a reader can use.
+ let partitioning = self.properties().output_partitioning();
match t {
- // "None" is retained for plan-shape stability: this writer never
- // repartitions, so the value can only ever be None.
DisplayFormatType::Default | DisplayFormatType::Verbose => {
- write!(f, "ShuffleWriterExec: partitioning: None")
+ write!(f, "ShuffleWriterExec: partitioning: {partitioning}")
}
DisplayFormatType::TreeRender => {
- write!(f, "partitioning=None")
+ write!(f, "partitioning={partitioning}")
}
}
}
@@ -961,6 +963,7 @@ mod tests {
use datafusion::datasource::memory::MemorySourceConfig;
use datafusion::datasource::source::DataSourceExec;
use datafusion::physical_plan::coalesce_partitions::CoalescePartitionsExec;
+ use datafusion::physical_plan::display::DefaultDisplay;
use datafusion::physical_plan::expressions::Column;
use datafusion::prelude::SessionContext;
use tempfile::TempDir;
@@ -1123,6 +1126,41 @@ mod tests {
Ok(())
}
+ /// The rendered partitioning has to track the input plan, not be a
constant.
+ /// Two plans with different partitioning must render differently.
+ #[tokio::test]
+ async fn display_as_reports_real_partitioning() -> Result<()> {
+ fn render(input: Arc<dyn ExecutionPlan>) -> Result<String> {
+ let work_dir = TempDir::new()?;
+ let writer = ShuffleWriterExec::try_new(
+ JobId::new("jobPartitioning"),
+ 1,
+ input,
+ work_dir.path().to_str().unwrap().to_owned(),
+ )?;
+ Ok(format!("{}", DefaultDisplay(writer)))
+ }
+
+ // the passthrough case: the writer inherits its input's partitioning
+ let passthrough = render(create_input_plan()?)?;
+ assert!(
+ passthrough.contains("UnknownPartitioning(2)"),
+ "expected the input plan's 2 partitions:\n{passthrough}"
+ );
+
+ // a hash-partitioned input has to come through as such, exprs and all
+ let hashed = render(Arc::new(RepartitionExec::try_new(
+ create_input_plan()?,
+ Partitioning::Hash(vec![Arc::new(Column::new("a", 0))], 4),
+ )?))?;
+ assert!(
+ hashed.contains("Hash([a@0], 4)"),
+ "expected the input plan's hash partitioning:\n{hashed}"
+ );
+
+ Ok(())
+ }
+
#[tokio::test]
async fn write_failure_is_shared_to_all_output_partitions() -> Result<()> {
let session_ctx = SessionContext::new();
diff --git a/ballista/scheduler/src/planner.rs
b/ballista/scheduler/src/planner.rs
index bb3c80c60..7a5400d66 100644
--- a/ballista/scheduler/src/planner.rs
+++ b/ballista/scheduler/src/planner.rs
@@ -1096,13 +1096,13 @@ mod test {
AggregateExec: mode=Partial, gby=[l_returnflag@1 as l_returnflag],
aggr=[sum(lineitem.l_extendedprice * Int64(1))]
DataSourceExec: file_groups={2 groups:
[[ballista/scheduler/testdata/lineitem/partition0.tbl],
[ballista/scheduler/testdata/lineitem/partition1.tbl]]},
projection=[l_extendedprice, l_returnflag], file_type=csv, has_header=false
- ShuffleWriterExec: partitioning: None
+ ShuffleWriterExec: partitioning: Hash([l_returnflag@0], 2)
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: stage=1, partitioning:
Hash([l_returnflag@0], 2)
- ShuffleWriterExec: partitioning: None
+ ShuffleWriterExec: partitioning: UnknownPartitioning(1)
SortPreservingMergeExec: [l_returnflag@0 ASC NULLS LAST]
UnresolvedShuffleExec: stage=2, partitioning:
Hash([l_returnflag@0], 2)
*/
@@ -1221,13 +1221,13 @@ order by
UnresolvedShuffleExec: stage=1, partitioning:
Hash([l_orderkey@0], 2)
UnresolvedShuffleExec: stage=2, partitioning:
Hash([o_orderkey@0], 2)
- ShuffleWriterExec: partitioning: None
+ ShuffleWriterExec: partitioning: Hash([l_shipmode@0], 2)
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: stage=3, partitioning:
Hash([l_shipmode@0], 2)
- ShuffleWriterExec: partitioning: None
+ ShuffleWriterExec: partitioning: UnknownPartitioning(1)
SortPreservingMergeExec: [l_shipmode@0 ASC NULLS LAST]
UnresolvedShuffleExec: stage=4, partitioning: Hash([l_shipmode@0],
2)
*/
@@ -1530,7 +1530,7 @@ order by
// Stage 0 holds the join, still a SortMergeJoinExec over sorted inputs
// (no HashJoinExec, no broadcast UnresolvedShuffleExec).
assert_plan!(stages[0].as_ref(), @r"
- ShuffleWriterExec: partitioning: None
+ ShuffleWriterExec: partitioning: RoundRobinBatch(2)
AggregateExec: mode=Partial, gby=[], aggr=[count(Int64(1))]
RepartitionExec: partitioning=RoundRobinBatch(2),
input_partitions=1
ProjectionExec: expr=[]
@@ -2151,7 +2151,7 @@ order by
DataSourceExec: file_groups={2 groups:
[[ballista/scheduler/testdata/lineitem/partition0.tbl],
[ballista/scheduler/testdata/lineitem/partition1.tbl]]},
projection=[l_shipdate, l_shipmode], file_type=csv, has_header=false
Stage 1:
- ShuffleWriterExec: partitioning: None
+ ShuffleWriterExec: partitioning: Hash([l_shipmode@0], 2)
SortExec: expr=[l_shipdate@1 ASC NULLS LAST, rk@2 ASC NULLS
LAST], preserve_partitioning=[true]
ProjectionExec: expr=[l_shipmode@1 as l_shipmode, l_shipdate@0
as l_shipdate, rank() PARTITION BY [lineitem.l_shipmode] ORDER BY
[lineitem.l_shipdate DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND
CURRENT ROW@2 as rk]
FilterExec: rank() PARTITION BY [lineitem.l_shipmode] ORDER BY
[lineitem.l_shipdate DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND
CURRENT ROW@2 <= 100
@@ -2160,7 +2160,7 @@ order by
UnresolvedShuffleExec: stage=1, partitioning:
Hash([l_shipmode@1], 2)
Stage 2:
- ShuffleWriterExec: partitioning: None
+ ShuffleWriterExec: partitioning: UnknownPartitioning(1)
SortPreservingMergeExec: [l_shipdate@1 ASC NULLS LAST, rk@2 ASC
NULLS LAST]
UnresolvedShuffleExec: stage=2, partitioning:
Hash([l_shipmode@0], 2)
diff --git a/ballista/scheduler/src/state/aqe/test/alter_stages.rs
b/ballista/scheduler/src/state/aqe/test/alter_stages.rs
index 387f1c58b..2ec0a50b1 100644
--- a/ballista/scheduler/src/state/aqe/test/alter_stages.rs
+++ b/ballista/scheduler/src/state/aqe/test/alter_stages.rs
@@ -72,7 +72,7 @@ async fn should_propagate_empty_stage() ->
datafusion::error::Result<()> {
let stages = planner.runnable_stages()?.unwrap();
assert_eq!(1, stages.len());
assert_plan!(stages.first().unwrap().plan.as_ref(), @ r"
- ShuffleWriterExec: partitioning: None
+ ShuffleWriterExec: partitioning: RoundRobinBatch(2)
RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
EmptyExec
");
@@ -147,7 +147,7 @@ async fn should_propagate_empty_stage_and_remove() ->
datafusion::error::Result<
let stages = planner.runnable_stages()?.unwrap();
assert_eq!(1, stages.len());
assert_plan!(stages.first().unwrap().plan.as_ref(), @ r"
- ShuffleWriterExec: partitioning: None
+ ShuffleWriterExec: partitioning: RoundRobinBatch(2)
RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
EmptyExec
");
@@ -302,7 +302,7 @@ async fn should_support_cross_join() ->
datafusion::error::Result<()> {
assert_eq!(1, stages.len());
assert_plan!(stages[0].plan.as_ref(), @ r"
- ShuffleWriterExec: partitioning: None
+ ShuffleWriterExec: partitioning: UnknownPartitioning(2)
MockPartitionedScan: num_partitions=2, statistics=[Rows=Exact(1024),
Bytes=Exact(8192), [(Col[0]:)]]
");
@@ -326,7 +326,7 @@ async fn should_support_cross_join() ->
datafusion::error::Result<()> {
assert_eq!(1, stages.len());
assert_plan!(stages[0].plan.as_ref(), @ r"
- ShuffleWriterExec: partitioning: None
+ ShuffleWriterExec: partitioning: UnknownPartitioning(2)
ProjectionExec: expr=[big_col@1 as big_col, big_col@0 as big_col]
CrossJoinExec
CoalescePartitionsExec
diff --git a/ballista/scheduler/src/state/aqe/test/coalesce_rule.rs
b/ballista/scheduler/src/state/aqe/test/coalesce_rule.rs
index a3da8bb8b..ab132af49 100644
--- a/ballista/scheduler/src/state/aqe/test/coalesce_rule.rs
+++ b/ballista/scheduler/src/state/aqe/test/coalesce_rule.rs
@@ -494,7 +494,7 @@ async fn shuffle_reader_uses_coalesced_k_when_rule_fires()
-> datafusion::error:
let stages = planner.runnable_stages()?.unwrap();
assert_eq!(1, stages.len());
assert_plan!(stages[0].plan.as_ref(), @ r"
- ShuffleWriterExec: partitioning: None
+ ShuffleWriterExec: partitioning: Hash([c2@2], 2)
ProjectionExec: expr=[min(t.a)@1 as c0, max(t.b)@2 as c1, c@0 as c2]
AggregateExec: mode=FinalPartitioned, gby=[c@0 as c], aggr=[min(t.a),
max(t.b)]
ShuffleReaderExec: upstream_stage: 0, partitioning: Hash([c@0], 2),
coalesce: 2 of 8
diff --git a/ballista/scheduler/src/state/aqe/test/join_selection.rs
b/ballista/scheduler/src/state/aqe/test/join_selection.rs
index 6243d7010..dcbd85bf0 100644
--- a/ballista/scheduler/src/state/aqe/test/join_selection.rs
+++ b/ballista/scheduler/src/state/aqe/test/join_selection.rs
@@ -373,7 +373,7 @@ async fn test_hash_join_three_tables_collect_left() ->
datafusion::common::Resul
// plan for first stage
assert_plan!(stage.plan.as_ref(), @ r"
- ShuffleWriterExec: partitioning: None
+ ShuffleWriterExec: partitioning: UnknownPartitioning(4)
DataSourceExec: partitions=4, partition_sizes=[1, 1, 1, 1]
");
diff --git a/ballista/scheduler/src/state/aqe/test/plan_to_stages.rs
b/ballista/scheduler/src/state/aqe/test/plan_to_stages.rs
index 3a18933e7..f7dea1870 100644
--- a/ballista/scheduler/src/state/aqe/test/plan_to_stages.rs
+++ b/ballista/scheduler/src/state/aqe/test/plan_to_stages.rs
@@ -145,7 +145,7 @@ async fn should_split_plan_into_stages() ->
datafusion::error::Result<()> {
let stages = planner.runnable_stages()?.unwrap();
assert_eq!(1, stages.len());
assert_plan!(stages.first().unwrap().plan.as_ref(), @ r"
- ShuffleWriterExec: partitioning: None
+ ShuffleWriterExec: partitioning: Hash([c2@2], 2)
ProjectionExec: expr=[min(t.a)@1 as c0, max(t.b)@2 as c1, c@0 as c2]
AggregateExec: mode=FinalPartitioned, gby=[c@0 as c], aggr=[min(t.a),
max(t.b)]
ShuffleReaderExec: upstream_stage: 0, partitioning: Hash([c@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 29efbf376..42774e174 100644
--- a/ballista/scheduler/tests/tpch_plan_stability/approved/q1.txt
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q1.txt
@@ -6,13 +6,13 @@ SortShuffleWriterExec: partitioning=Hash([l_returnflag@0,
l_linestatus@1], 16)
StatsExec: rows=600037902
=== Stage 2 ===
-ShuffleWriterExec: partitioning: None
+ShuffleWriterExec: partitioning: Hash([l_returnflag@0, l_linestatus@1], 16)
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_dis [...]
SortExec: expr=[l_returnflag@0 ASC NULLS LAST, l_linestatus@1 ASC NULLS
LAST], preserve_partitioning=[true]
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 * 1 + 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_discount), count(I [...]
UnresolvedShuffleExec: stage=1, partitioning: Hash([l_returnflag@0,
l_linestatus@1], 16)
=== Stage 3 ===
-ShuffleWriterExec: partitioning: None
+ShuffleWriterExec: partitioning: UnknownPartitioning(1)
SortPreservingMergeExec: [l_returnflag@0 ASC NULLS LAST, l_linestatus@1 ASC
NULLS LAST]
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 b59c2f0c2..76b86520a 100644
--- a/ballista/scheduler/tests/tpch_plan_stability/approved/q10.txt
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q10.txt
@@ -45,13 +45,13 @@ SortShuffleWriterExec: partitioning=Hash([c_custkey@0,
c_name@1, c_acctbal@2, c_
UnresolvedShuffleExec: stage=6, partitioning: Hash([n_nationkey@0],
16)
=== Stage 8 ===
-ShuffleWriterExec: partitioning: None
+ShuffleWriterExec: partitioning: Hash([c_custkey@0, c_name@1, c_acctbal@3,
c_phone@6, n_name@4, c_address@5, c_comment@7], 16)
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]
SortExec: TopK(fetch=20), expr=[sum(lineitem.l_extendedprice * Int64(1) -
lineitem.l_discount)@7 DESC], preserve_partitioning=[true]
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 * 1 - lineitem.l_discount) as
sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)]
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 9 ===
-ShuffleWriterExec: partitioning: None
+ShuffleWriterExec: partitioning: UnknownPartitioning(1)
SortPreservingMergeExec: [revenue@2 DESC], fetch=20
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 736c88e26..8a214b60b 100644
--- a/ballista/scheduler/tests/tpch_plan_stability/approved/q11.txt
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q11.txt
@@ -21,7 +21,7 @@ SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
StatsExec: rows=25
=== Stage 5 ===
-ShuffleWriterExec: partitioning: None
+ShuffleWriterExec: partitioning: Hash([n_nationkey@3], 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)]
@@ -63,7 +63,7 @@ SortShuffleWriterExec: partitioning=Hash([ps_partkey@0], 16)
UnresolvedShuffleExec: stage=9, partitioning: Hash([n_nationkey@0],
16)
=== Stage 11 ===
-ShuffleWriterExec: partitioning: None
+ShuffleWriterExec: partitioning: Hash([ps_partkey@0], 16)
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]
NestedLoopJoinExec: join_type=Inner, filter=join_proj_push_down_1@1 >
sum(partsupp.ps_supplycost * partsupp.ps_availqty) * Float64(0.0001)@0,
projection=[sum(partsupp.ps_supplycost * partsupp.ps_availqty) *
Float64(0.0001)@0, ps_partkey@1, sum(partsupp.ps_supplycost *
partsupp.ps_availqty)@2]
@@ -76,6 +76,6 @@ ShuffleWriterExec: partitioning: None
UnresolvedShuffleExec: stage=10, partitioning:
Hash([ps_partkey@0], 16)
=== Stage 12 ===
-ShuffleWriterExec: partitioning: None
+ShuffleWriterExec: partitioning: UnknownPartitioning(1)
SortPreservingMergeExec: [value@1 DESC]
UnresolvedShuffleExec: stage=11, 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 770753636..59368e79b 100644
--- a/ballista/scheduler/tests/tpch_plan_stability/approved/q12.txt
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q12.txt
@@ -18,13 +18,13 @@ SortShuffleWriterExec: partitioning=Hash([l_shipmode@0], 16)
UnresolvedShuffleExec: stage=2, partitioning: Hash([o_orderkey@0],
16)
=== Stage 4 ===
-ShuffleWriterExec: partitioning: None
+ShuffleWriterExec: partitioning: Hash([l_shipmode@0], 16)
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]
SortExec: expr=[l_shipmode@0 ASC NULLS LAST], preserve_partitioning=[true]
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: stage=3, partitioning: Hash([l_shipmode@0], 16)
=== Stage 5 ===
-ShuffleWriterExec: partitioning: None
+ShuffleWriterExec: partitioning: UnknownPartitioning(1)
SortPreservingMergeExec: [l_shipmode@0 ASC NULLS LAST]
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 ca52111ed..eb72a2c4e 100644
--- a/ballista/scheduler/tests/tpch_plan_stability/approved/q13.txt
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q13.txt
@@ -20,13 +20,13 @@ SortShuffleWriterExec: partitioning=Hash([c_count@0], 16)
UnresolvedShuffleExec: stage=2, partitioning:
Hash([o_custkey@1], 16)
=== Stage 4 ===
-ShuffleWriterExec: partitioning: None
+ShuffleWriterExec: partitioning: Hash([c_count@0], 16)
ProjectionExec: expr=[c_count@0 as c_count, count(Int64(1))@1 as custdist]
SortExec: expr=[count(Int64(1))@1 DESC, c_count@0 DESC],
preserve_partitioning=[true]
AggregateExec: mode=FinalPartitioned, gby=[c_count@0 as c_count],
aggr=[count(Int64(1))]
UnresolvedShuffleExec: stage=3, partitioning: Hash([c_count@0], 16)
=== Stage 5 ===
-ShuffleWriterExec: partitioning: None
+ShuffleWriterExec: partitioning: UnknownPartitioning(1)
SortPreservingMergeExec: [custdist@1 DESC, c_count@0 DESC]
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 beae64429..cc409446e 100644
--- a/ballista/scheduler/tests/tpch_plan_stability/approved/q14.txt
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q14.txt
@@ -8,7 +8,7 @@ SortShuffleWriterExec: partitioning=Hash([p_partkey@0], 16)
StatsExec: rows=20000000
=== Stage 3 ===
-ShuffleWriterExec: partitioning: None
+ShuffleWriterExec: partitioning: Hash([p_partkey@3], 16)
AggregateExec: mode=Partial, gby=[], aggr=[sum(CASE WHEN part.p_type LIKE
PROMO% THEN __common_expr_1 ELSE 0.0000 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)]
ProjectionExec: expr=[l_extendedprice@1 * (1 - l_discount@2) as
__common_expr_1, p_type@4 as p_type]
SortMergeJoinExec: join_type=Inner, on=[(l_partkey@0, p_partkey@0)]
@@ -18,7 +18,7 @@ ShuffleWriterExec: partitioning: None
UnresolvedShuffleExec: stage=2, partitioning: Hash([p_partkey@0], 16)
=== Stage 4 ===
-ShuffleWriterExec: partitioning: None
+ShuffleWriterExec: partitioning: UnknownPartitioning(1)
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 0.0000 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
diff --git a/ballista/scheduler/tests/tpch_plan_stability/approved/q15.txt
b/ballista/scheduler/tests/tpch_plan_stability/approved/q15.txt
index c82ed09ab..52423c15a 100644
--- a/ballista/scheduler/tests/tpch_plan_stability/approved/q15.txt
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q15.txt
@@ -26,7 +26,7 @@ SortShuffleWriterExec: partitioning=Hash([l_suppkey@0], 16)
StatsExec: rows=600037902
=== Stage 5 ===
-ShuffleWriterExec: partitioning: None
+ShuffleWriterExec: partitioning: Hash([l_suppkey@0], 16)
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 * 1 - lineitem.l_discount) as
sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)]
@@ -39,7 +39,7 @@ SortShuffleWriterExec:
partitioning=Hash([max(revenue0.total_revenue)@0], 16)
UnresolvedShuffleExec: stage=5, partitioning: Hash([l_suppkey@0], 16)
=== Stage 7 ===
-ShuffleWriterExec: partitioning: None
+ShuffleWriterExec: partitioning: Hash([total_revenue@4], 16)
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]
SortMergeJoinExec: join_type=Inner, on=[(total_revenue@4,
max(revenue0.total_revenue)@0)]
@@ -48,6 +48,6 @@ ShuffleWriterExec: partitioning: None
UnresolvedShuffleExec: stage=6, partitioning:
Hash([max(revenue0.total_revenue)@0], 16)
=== Stage 8 ===
-ShuffleWriterExec: partitioning: None
+ShuffleWriterExec: partitioning: UnknownPartitioning(1)
SortPreservingMergeExec: [s_suppkey@0 ASC NULLS LAST]
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 bf85900ab..8d23a14f5 100644
--- a/ballista/scheduler/tests/tpch_plan_stability/approved/q16.txt
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q16.txt
@@ -32,13 +32,13 @@ SortShuffleWriterExec: partitioning=Hash([p_brand@0,
p_type@1, p_size@2], 16)
UnresolvedShuffleExec: stage=4, partitioning: Hash([s_suppkey@0], 16)
=== Stage 6 ===
-ShuffleWriterExec: partitioning: None
+ShuffleWriterExec: partitioning: Hash([p_brand@0, p_type@1, p_size@2], 16)
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]
SortExec: expr=[count(alias1)@3 DESC, p_brand@0 ASC NULLS LAST, p_type@1
ASC NULLS LAST, p_size@2 ASC NULLS LAST], preserve_partitioning=[true]
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=5, partitioning: Hash([p_brand@0,
p_type@1, p_size@2], 16)
=== Stage 7 ===
-ShuffleWriterExec: partitioning: None
+ShuffleWriterExec: partitioning: UnknownPartitioning(1)
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=6, 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 20e64a196..a73d43084 100644
--- a/ballista/scheduler/tests/tpch_plan_stability/approved/q17.txt
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q17.txt
@@ -13,7 +13,7 @@ SortShuffleWriterExec: partitioning=Hash([l_partkey@0], 16)
StatsExec: rows=600037902
=== Stage 4 ===
-ShuffleWriterExec: partitioning: None
+ShuffleWriterExec: partitioning: Hash([p_partkey@2], 16)
AggregateExec: mode=Partial, gby=[], aggr=[sum(lineitem.l_extendedprice)]
ProjectionExec: expr=[l_extendedprice@1 as l_extendedprice]
SortMergeJoinExec: join_type=LeftSemi, on=[(p_partkey@2, l_partkey@1)],
filter=CAST(l_quantity@0 AS Decimal128(30, 15)) < Float64(0.2) *
avg(lineitem.l_quantity)@1
@@ -29,7 +29,7 @@ ShuffleWriterExec: partitioning: None
UnresolvedShuffleExec: stage=3, partitioning:
Hash([l_partkey@0], 16)
=== Stage 5 ===
-ShuffleWriterExec: partitioning: None
+ShuffleWriterExec: partitioning: UnknownPartitioning(1)
ProjectionExec: expr=[CAST(sum(lineitem.l_extendedprice)@0 AS Float64) / 7
as avg_yearly]
AggregateExec: mode=Final, gby=[], aggr=[sum(lineitem.l_extendedprice)]
CoalescePartitionsExec
diff --git a/ballista/scheduler/tests/tpch_plan_stability/approved/q18.txt
b/ballista/scheduler/tests/tpch_plan_stability/approved/q18.txt
index aca5a06eb..3aa63682b 100644
--- a/ballista/scheduler/tests/tpch_plan_stability/approved/q18.txt
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q18.txt
@@ -25,7 +25,7 @@ SortShuffleWriterExec: partitioning=Hash([l_orderkey@0], 16)
StatsExec: rows=600037902
=== Stage 6 ===
-ShuffleWriterExec: partitioning: None
+ShuffleWriterExec: partitioning: Hash([o_orderkey@2], 16)
SortExec: TopK(fetch=100), expr=[o_totalprice@4 DESC, o_orderdate@3 ASC
NULLS LAST], preserve_partitioning=[true]
AggregateExec: mode=SinglePartitioned, gby=[c_name@1 as c_name,
c_custkey@0 as c_custkey, o_orderkey@2 as o_orderkey, o_orderdate@4 as
o_orderdate, o_totalprice@3 as o_totalprice], aggr=[sum(lineitem.l_quantity)],
ordering_mode=PartiallySorted([2])
SortMergeJoinExec: join_type=LeftSemi, on=[(o_orderkey@2, l_orderkey@0)]
@@ -41,6 +41,6 @@ ShuffleWriterExec: partitioning: None
UnresolvedShuffleExec: stage=5, partitioning:
Hash([l_orderkey@0], 16)
=== Stage 7 ===
-ShuffleWriterExec: partitioning: None
+ShuffleWriterExec: partitioning: UnknownPartitioning(1)
SortPreservingMergeExec: [o_totalprice@4 DESC, o_orderdate@3 ASC NULLS
LAST], fetch=100
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 fefda7d02..7188e0874 100644
--- a/ballista/scheduler/tests/tpch_plan_stability/approved/q19.txt
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q19.txt
@@ -9,7 +9,7 @@ SortShuffleWriterExec: partitioning=Hash([p_partkey@0], 16)
StatsExec: rows=20000000
=== Stage 3 ===
-ShuffleWriterExec: partitioning: None
+ShuffleWriterExec: partitioning: Hash([p_partkey@4], 16)
AggregateExec: mode=Partial, gby=[], aggr=[sum(lineitem.l_extendedprice * 1
- lineitem.l_discount) as sum(lineitem.l_extendedprice * Int64(1) -
lineitem.l_discount)]
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 >= 1.00 AND l_quantity@0 <= 11.00 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 >= 10.00 AND l_quantity@0 <= 20.00 AND
p_size@2 <= 10 OR p_brand@1 = Brand#34 AND p_container@3 IN (SET) ([LG CASE, LG
BOX, LG PACK, LG PKG]) AND l_ [...]
@@ -19,7 +19,7 @@ ShuffleWriterExec: partitioning: None
UnresolvedShuffleExec: stage=2, partitioning: Hash([p_partkey@0], 16)
=== Stage 4 ===
-ShuffleWriterExec: partitioning: None
+ShuffleWriterExec: partitioning: UnknownPartitioning(1)
ProjectionExec: expr=[sum(lineitem.l_extendedprice * Int64(1) -
lineitem.l_discount)@0 as revenue]
AggregateExec: mode=Final, gby=[], aggr=[sum(lineitem.l_extendedprice * 1
- lineitem.l_discount) as sum(lineitem.l_extendedprice * Int64(1) -
lineitem.l_discount)]
CoalescePartitionsExec
diff --git a/ballista/scheduler/tests/tpch_plan_stability/approved/q2.txt
b/ballista/scheduler/tests/tpch_plan_stability/approved/q2.txt
index 0490884f6..d64f7012a 100644
--- a/ballista/scheduler/tests/tpch_plan_stability/approved/q2.txt
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q2.txt
@@ -108,7 +108,7 @@ SortShuffleWriterExec: partitioning=Hash([ps_partkey@1,
min(partsupp.ps_supplyco
UnresolvedShuffleExec: stage=16, partitioning: Hash([ps_partkey@0], 16)
=== Stage 18 ===
-ShuffleWriterExec: partitioning: None
+ShuffleWriterExec: partitioning: Hash([p_partkey@3, ps_supplycost@7], 16)
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]
SortMergeJoinExec: join_type=LeftSemi, on=[(p_partkey@0, ps_partkey@1),
(ps_supplycost@7, min(partsupp.ps_supplycost)@0)]
@@ -118,6 +118,6 @@ ShuffleWriterExec: partitioning: None
UnresolvedShuffleExec: stage=17, partitioning: Hash([ps_partkey@1,
min(partsupp.ps_supplycost)@0], 16)
=== Stage 19 ===
-ShuffleWriterExec: partitioning: None
+ShuffleWriterExec: partitioning: UnknownPartitioning(1)
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=18, partitioning: Hash([p_partkey@3,
ps_supplycost@7], 16)
diff --git a/ballista/scheduler/tests/tpch_plan_stability/approved/q20.txt
b/ballista/scheduler/tests/tpch_plan_stability/approved/q20.txt
index 8d5118be1..b56ba3f00 100644
--- a/ballista/scheduler/tests/tpch_plan_stability/approved/q20.txt
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q20.txt
@@ -51,7 +51,7 @@ SortShuffleWriterExec: partitioning=Hash([ps_suppkey@0], 16)
UnresolvedShuffleExec: stage=7, partitioning: Hash([l_partkey@0,
l_suppkey@1], 16)
=== Stage 9 ===
-ShuffleWriterExec: partitioning: None
+ShuffleWriterExec: partitioning: Hash([s_suppkey@0], 16)
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)]
@@ -61,6 +61,6 @@ ShuffleWriterExec: partitioning: None
UnresolvedShuffleExec: stage=8, partitioning: Hash([ps_suppkey@0],
16)
=== Stage 10 ===
-ShuffleWriterExec: partitioning: None
+ShuffleWriterExec: partitioning: UnknownPartitioning(1)
SortPreservingMergeExec: [s_name@0 ASC NULLS LAST]
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 6d03dbe5e..da77d3e85 100644
--- a/ballista/scheduler/tests/tpch_plan_stability/approved/q21.txt
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q21.txt
@@ -67,13 +67,13 @@ SortShuffleWriterExec: partitioning=Hash([s_name@0], 16)
UnresolvedShuffleExec: stage=9, partitioning: Hash([l_orderkey@0],
16)
=== Stage 11 ===
-ShuffleWriterExec: partitioning: None
+ShuffleWriterExec: partitioning: Hash([s_name@0], 16)
ProjectionExec: expr=[s_name@0 as s_name, count(Int64(1))@1 as numwait]
SortExec: TopK(fetch=100), expr=[count(Int64(1))@1 DESC, s_name@0 ASC
NULLS LAST], preserve_partitioning=[true]
AggregateExec: mode=FinalPartitioned, gby=[s_name@0 as s_name],
aggr=[count(Int64(1))]
UnresolvedShuffleExec: stage=10, partitioning: Hash([s_name@0], 16)
=== Stage 12 ===
-ShuffleWriterExec: partitioning: None
+ShuffleWriterExec: partitioning: UnknownPartitioning(1)
SortPreservingMergeExec: [numwait@1 DESC, s_name@0 ASC NULLS LAST], fetch=100
UnresolvedShuffleExec: stage=11, 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 3d01e97a4..11dec2c5d 100644
--- a/ballista/scheduler/tests/tpch_plan_stability/approved/q22.txt
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q22.txt
@@ -23,13 +23,13 @@ SortShuffleWriterExec: partitioning=Hash([cntrycode@0], 16)
UnresolvedShuffleExec: stage=2, partitioning:
Hash([o_custkey@0], 16)
=== Stage 4 ===
-ShuffleWriterExec: partitioning: None
+ShuffleWriterExec: partitioning: Hash([cntrycode@0], 16)
ProjectionExec: expr=[cntrycode@0 as cntrycode, count(Int64(1))@1 as
numcust, sum(custsale.c_acctbal)@2 as totacctbal]
SortExec: expr=[cntrycode@0 ASC NULLS LAST], preserve_partitioning=[true]
AggregateExec: mode=FinalPartitioned, gby=[cntrycode@0 as cntrycode],
aggr=[count(Int64(1)), sum(custsale.c_acctbal)]
UnresolvedShuffleExec: stage=3, partitioning: Hash([cntrycode@0], 16)
=== Stage 5 ===
-ShuffleWriterExec: partitioning: None
+ShuffleWriterExec: partitioning: UnknownPartitioning(1)
SortPreservingMergeExec: [cntrycode@0 ASC NULLS LAST]
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 f8049b08c..0e9cd4329 100644
--- a/ballista/scheduler/tests/tpch_plan_stability/approved/q3.txt
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q3.txt
@@ -23,7 +23,7 @@ SortShuffleWriterExec: partitioning=Hash([l_orderkey@0], 16)
StatsExec: rows=600037902
=== Stage 5 ===
-ShuffleWriterExec: partitioning: None
+ShuffleWriterExec: partitioning: Hash([l_orderkey@0], 16)
ProjectionExec: expr=[l_orderkey@0 as l_orderkey,
sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)@3 as revenue,
o_orderdate@1 as o_orderdate, o_shippriority@2 as o_shippriority]
SortExec: TopK(fetch=10), expr=[sum(lineitem.l_extendedprice * Int64(1) -
lineitem.l_discount)@3 DESC, o_orderdate@1 ASC NULLS LAST],
preserve_partitioning=[true]
AggregateExec: mode=SinglePartitioned, gby=[l_orderkey@2 as l_orderkey,
o_orderdate@0 as o_orderdate, o_shippriority@1 as o_shippriority],
aggr=[sum(lineitem.l_extendedprice * 1 - lineitem.l_discount) as
sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)],
ordering_mode=PartiallySorted([0])
@@ -35,6 +35,6 @@ ShuffleWriterExec: partitioning: None
UnresolvedShuffleExec: stage=4, partitioning:
Hash([l_orderkey@0], 16)
=== Stage 6 ===
-ShuffleWriterExec: partitioning: None
+ShuffleWriterExec: partitioning: UnknownPartitioning(1)
SortPreservingMergeExec: [revenue@1 DESC, o_orderdate@2 ASC NULLS LAST],
fetch=10
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 4df80b934..43b378d87 100644
--- a/ballista/scheduler/tests/tpch_plan_stability/approved/q4.txt
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q4.txt
@@ -19,13 +19,13 @@ SortShuffleWriterExec:
partitioning=Hash([o_orderpriority@0], 16)
UnresolvedShuffleExec: stage=2, partitioning: Hash([l_orderkey@0],
16)
=== Stage 4 ===
-ShuffleWriterExec: partitioning: None
+ShuffleWriterExec: partitioning: Hash([o_orderpriority@0], 16)
ProjectionExec: expr=[o_orderpriority@0 as o_orderpriority,
count(Int64(1))@1 as order_count]
SortExec: expr=[o_orderpriority@0 ASC NULLS LAST],
preserve_partitioning=[true]
AggregateExec: mode=FinalPartitioned, gby=[o_orderpriority@0 as
o_orderpriority], aggr=[count(Int64(1))]
UnresolvedShuffleExec: stage=3, partitioning:
Hash([o_orderpriority@0], 16)
=== Stage 5 ===
-ShuffleWriterExec: partitioning: None
+ShuffleWriterExec: partitioning: UnknownPartitioning(1)
SortPreservingMergeExec: [o_orderpriority@0 ASC NULLS LAST]
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 c3dcae5ec..c58e8852c 100644
--- a/ballista/scheduler/tests/tpch_plan_stability/approved/q5.txt
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q5.txt
@@ -71,13 +71,13 @@ SortShuffleWriterExec: partitioning=Hash([n_name@0], 16)
UnresolvedShuffleExec: stage=10, partitioning: Hash([r_regionkey@0],
16)
=== Stage 12 ===
-ShuffleWriterExec: partitioning: None
+ShuffleWriterExec: partitioning: Hash([n_name@0], 16)
ProjectionExec: expr=[n_name@0 as n_name, sum(lineitem.l_extendedprice *
Int64(1) - lineitem.l_discount)@1 as revenue]
SortExec: expr=[sum(lineitem.l_extendedprice * Int64(1) -
lineitem.l_discount)@1 DESC], preserve_partitioning=[true]
AggregateExec: mode=FinalPartitioned, gby=[n_name@0 as n_name],
aggr=[sum(lineitem.l_extendedprice * 1 - lineitem.l_discount) as
sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)]
UnresolvedShuffleExec: stage=11, partitioning: Hash([n_name@0], 16)
=== Stage 13 ===
-ShuffleWriterExec: partitioning: None
+ShuffleWriterExec: partitioning: UnknownPartitioning(1)
SortPreservingMergeExec: [revenue@1 DESC]
UnresolvedShuffleExec: stage=12, partitioning: Hash([n_name@0], 16)
diff --git a/ballista/scheduler/tests/tpch_plan_stability/approved/q6.txt
b/ballista/scheduler/tests/tpch_plan_stability/approved/q6.txt
index 3a70454e3..c8b330649 100644
--- a/ballista/scheduler/tests/tpch_plan_stability/approved/q6.txt
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q6.txt
@@ -1,5 +1,5 @@
=== Stage 1 ===
-ShuffleWriterExec: partitioning: None
+ShuffleWriterExec: partitioning: UnknownPartitioning(1)
ProjectionExec: expr=[sum(lineitem.l_extendedprice * lineitem.l_discount)@0
as revenue]
AggregateExec: mode=Single, gby=[], aggr=[sum(lineitem.l_extendedprice *
lineitem.l_discount)]
FilterExec: l_shipdate@3 >= 1994-01-01 AND l_shipdate@3 < 1995-01-01 AND
l_discount@2 >= 0.05 AND l_discount@2 <= 0.07 AND l_quantity@0 < 24.00,
projection=[l_extendedprice@1, l_discount@2]
diff --git a/ballista/scheduler/tests/tpch_plan_stability/approved/q7.txt
b/ballista/scheduler/tests/tpch_plan_stability/approved/q7.txt
index 8f877e6d4..d7c831be5 100644
--- a/ballista/scheduler/tests/tpch_plan_stability/approved/q7.txt
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q7.txt
@@ -72,13 +72,13 @@ SortShuffleWriterExec: partitioning=Hash([supp_nation@0,
cust_nation@1, l_year@2
UnresolvedShuffleExec: stage=10, partitioning: Hash([n_nationkey@0],
16)
=== Stage 12 ===
-ShuffleWriterExec: partitioning: None
+ShuffleWriterExec: partitioning: Hash([supp_nation@0, cust_nation@1,
l_year@2], 16)
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]
SortExec: expr=[supp_nation@0 ASC NULLS LAST, cust_nation@1 ASC NULLS
LAST, l_year@2 ASC NULLS LAST], preserve_partitioning=[true]
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=11, partitioning: Hash([supp_nation@0,
cust_nation@1, l_year@2], 16)
=== Stage 13 ===
-ShuffleWriterExec: partitioning: None
+ShuffleWriterExec: partitioning: UnknownPartitioning(1)
SortPreservingMergeExec: [supp_nation@0 ASC NULLS LAST, cust_nation@1 ASC
NULLS LAST, l_year@2 ASC NULLS LAST]
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 a595e9fd1..61e445e64 100644
--- a/ballista/scheduler/tests/tpch_plan_stability/approved/q8.txt
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q8.txt
@@ -98,13 +98,13 @@ SortShuffleWriterExec: partitioning=Hash([o_year@0], 16)
UnresolvedShuffleExec: stage=14, partitioning: Hash([r_regionkey@0],
16)
=== Stage 16 ===
-ShuffleWriterExec: partitioning: None
+ShuffleWriterExec: partitioning: Hash([o_year@0], 16)
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
0.0000 END) as sum(CASE WHEN all_nations.nation = Utf8("BRAZIL") THEN
all_nations.volume ELSE Int64(0) END), sum(all_nations.volume)]
UnresolvedShuffleExec: stage=15, partitioning: Hash([o_year@0], 16)
=== Stage 17 ===
-ShuffleWriterExec: partitioning: None
+ShuffleWriterExec: partitioning: UnknownPartitioning(1)
SortPreservingMergeExec: [o_year@0 ASC NULLS LAST]
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 95dd57e66..f630057b2 100644
--- a/ballista/scheduler/tests/tpch_plan_stability/approved/q9.txt
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q9.txt
@@ -70,13 +70,13 @@ SortShuffleWriterExec: partitioning=Hash([nation@0,
o_year@1], 16)
UnresolvedShuffleExec: stage=10, partitioning: Hash([n_nationkey@0],
16)
=== Stage 12 ===
-ShuffleWriterExec: partitioning: None
+ShuffleWriterExec: partitioning: Hash([nation@0, o_year@1], 16)
ProjectionExec: expr=[nation@0 as nation, o_year@1 as o_year,
sum(profit.amount)@2 as sum_profit]
SortExec: expr=[nation@0 ASC NULLS LAST, o_year@1 DESC],
preserve_partitioning=[true]
AggregateExec: mode=FinalPartitioned, gby=[nation@0 as nation, o_year@1
as o_year], aggr=[sum(profit.amount)]
UnresolvedShuffleExec: stage=11, partitioning: Hash([nation@0,
o_year@1], 16)
=== Stage 13 ===
-ShuffleWriterExec: partitioning: None
+ShuffleWriterExec: partitioning: UnknownPartitioning(1)
SortPreservingMergeExec: [nation@0 ASC NULLS LAST, o_year@1 DESC]
UnresolvedShuffleExec: stage=12, partitioning: Hash([nation@0, o_year@1],
16)
diff --git a/dev/bin/showplan.sh b/dev/bin/showplan.sh
index bff9b2423..eb0c43fb6 100755
--- a/dev/bin/showplan.sh
+++ b/dev/bin/showplan.sh
@@ -79,7 +79,7 @@
# "input_rows": 1,
# "output_rows": 2,
# "elapsed_compute": "6.00ms",
-# "stage_plan": "ShuffleWriterExec: partitioning: None\n
ProjectionExec: expr=[1 as Int64(1)]\n PlaceholderRowExec\n",
+# "stage_plan": "ShuffleWriterExec: partitioning:
UnknownPartitioning(1)\n ProjectionExec: expr=[1 as Int64(1)]\n
PlaceholderRowExec\n",
# "task_duration_percentiles": {
# "min": 6,
# "p25": 6,
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
