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 4f1d7da52 Add TPC-H distributed plan-stability test suite (#1966)
4f1d7da52 is described below
commit 4f1d7da525624f1642efc408bdc241ea6b8999a6
Author: Andy Grove <[email protected]>
AuthorDate: Wed Jul 8 14:00:30 2026 -0600
Add TPC-H distributed plan-stability test suite (#1966)
* test: add dataless stats-injecting table provider for plan stability
* test: add TPC-H fixtures and distributed staged-plan text helper
Add the 8 TPC-H schemas (copied verbatim from benchmarks/src/bin/tpch.rs),
SF100 row-count constants, and staged_plan_text(query_name) which loads a
TPC-H query's SQL, registers stats-only tables, builds the physical plan
with the Ballista-configured planner (target_partitions=16), splits it into
distributed query stages via DefaultDistributedPlanner, and renders the
stages to normalized text. Copy the 22 TPC-H query SQL files and add
helper tests covering a single-statement query (q1) and a multi-statement
query with view DDL (q15).
* test: add TPC-H distributed plan-stability suite with approved plans
* ci: add regenerate script and README for TPC-H plan-stability suite
Add dev/update-tpch-plan-stability.sh to regenerate approved golden
plans, and document the suite's scope, usage, and existing CI
coverage. No workflow changes needed: rust.yml already runs
workspace-wide cargo test/clippy/fmt jobs that pick up the new
[[test]] target automatically.
* style: apply cargo fmt to TPC-H plan-stability test files
Fix outstanding cargo fmt --check deltas in fixtures.rs, stats_table.rs,
and the plan_stability_test! macro invocation in main.rs.
* ci: exclude plan-stability goldens from RAT and add license header
* test: read TPC-H queries from benchmarks/queries instead of copying
---
Cargo.lock | 1 +
ballista/scheduler/Cargo.toml | 5 +
.../scheduler/tests/tpch_plan_stability/README.md | 59 ++++++
.../tests/tpch_plan_stability/approved/q1.txt | 18 ++
.../tests/tpch_plan_stability/approved/q10.txt | 60 ++++++
.../tests/tpch_plan_stability/approved/q11.txt | 87 ++++++++
.../tests/tpch_plan_stability/approved/q12.txt | 30 +++
.../tests/tpch_plan_stability/approved/q13.txt | 32 +++
.../tests/tpch_plan_stability/approved/q14.txt | 25 +++
.../tests/tpch_plan_stability/approved/q15.txt | 57 +++++
.../tests/tpch_plan_stability/approved/q16.txt | 46 +++++
.../tests/tpch_plan_stability/approved/q17.txt | 36 ++++
.../tests/tpch_plan_stability/approved/q18.txt | 46 +++++
.../tests/tpch_plan_stability/approved/q19.txt | 26 +++
.../tests/tpch_plan_stability/approved/q2.txt | 137 ++++++++++++
.../tests/tpch_plan_stability/approved/q20.txt | 69 +++++++
.../tests/tpch_plan_stability/approved/q21.txt | 82 ++++++++
.../tests/tpch_plan_stability/approved/q22.txt | 35 ++++
.../tests/tpch_plan_stability/approved/q3.txt | 40 ++++
.../tests/tpch_plan_stability/approved/q4.txt | 31 +++
.../tests/tpch_plan_stability/approved/q5.txt | 89 ++++++++
.../tests/tpch_plan_stability/approved/q6.txt | 6 +
.../tests/tpch_plan_stability/approved/q7.txt | 89 ++++++++
.../tests/tpch_plan_stability/approved/q8.txt | 118 +++++++++++
.../tests/tpch_plan_stability/approved/q9.txt | 85 ++++++++
.../tests/tpch_plan_stability/fixtures.rs | 229 +++++++++++++++++++++
.../scheduler/tests/tpch_plan_stability/main.rs | 112 ++++++++++
.../tests/tpch_plan_stability/stats_table.rs | 152 ++++++++++++++
dev/release/rat_exclude_files.txt | 1 +
dev/update-tpch-plan-stability.sh | 25 +++
30 files changed, 1828 insertions(+)
diff --git a/Cargo.lock b/Cargo.lock
index 50ee692b8..4d0b4e1d4 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1195,6 +1195,7 @@ dependencies = [
"prost",
"prost-types",
"rand 0.10.1",
+ "regex",
"rstest",
"serde",
"tokio",
diff --git a/ballista/scheduler/Cargo.toml b/ballista/scheduler/Cargo.toml
index f882d26f3..92339a5c5 100644
--- a/ballista/scheduler/Cargo.toml
+++ b/ballista/scheduler/Cargo.toml
@@ -79,7 +79,12 @@ tracing-appender = { workspace = true, optional = true }
tracing-subscriber = { workspace = true, optional = true }
uuid = { workspace = true }
+[[test]]
+name = "tpch_plan_stability"
+path = "tests/tpch_plan_stability/main.rs"
+
[dev-dependencies]
+regex = "1"
rstest = { workspace = true }
[build-dependencies]
diff --git a/ballista/scheduler/tests/tpch_plan_stability/README.md
b/ballista/scheduler/tests/tpch_plan_stability/README.md
new file mode 100644
index 000000000..5b62c7f9a
--- /dev/null
+++ b/ballista/scheduler/tests/tpch_plan_stability/README.md
@@ -0,0 +1,59 @@
+<!---
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+-->
+
+# TPC-H plan-stability suite
+
+Freezes each TPC-H query's **distributed staged plan** (static planner, SF100
+table statistics, `target_partitions=16`) as an approved text file under
+`approved/`. The suite fails if a code change alters plan shape — join
strategy,
+shuffle/stage boundaries, or broadcast decisions.
+
+- Run: `cargo test -p ballista-scheduler --test tpch_plan_stability`
+- Regenerate after an intended change: `dev/update-tpch-plan-stability.sh`
+ (or `BALLISTA_GENERATE_GOLDEN=1 cargo test -p ballista-scheduler --test
tpch_plan_stability`),
+ then review the diff under `approved/`.
+
+Scope: TPC-H only, static planner, Ballista default config (SortMergeJoin).
+Tables are dataless providers with injected SF100 cardinalities
(`fixtures.rs`).
+
+Query SQL is read directly from the canonical `benchmarks/queries/` at test
time
+(not copied), so a change to a benchmark query flows into the planned plan and
+surfaces as a golden diff, prompting a deliberate regeneration.
+
+## CI coverage
+
+This suite is registered as a `[[test]]` target in
`ballista/scheduler/Cargo.toml`
+(`tpch_plan_stability`), so it already runs wherever CI exercises the
workspace's
+default cargo tests — no dedicated job was added:
+
+- `.github/workflows/rust.yml` → `linux-test` (`cargo test --profile ci
--features=testcontainers`) and `macos-test` (`cargo test --profile ci
--locked`) both run from the workspace root without `-p`/`--workspace`
+ scoping. Since the root `Cargo.toml` sets no `default-members`, this tests
+ every workspace member, including `ballista-scheduler`, which picks up this
+ target automatically.
+- `.github/workflows/rust.yml` → `clippy` already runs `cargo clippy
--all-targets --package ballista-scheduler --all-features -- -D warnings`,
+ which lints this test target too.
+- `.github/workflows/rust.yml` → `lint` runs `cargo fmt --all -- --check`,
+ which covers these files as well.
+
+The generated `approved/*.txt` golden plans carry no license header (the test
+compares their exact bytes), so they are excluded from the Apache RAT license
+check in `.github/workflows/dev.yml` via
+`ballista/scheduler/tests/tpch_plan_stability/approved/*` in
+`dev/release/rat_exclude_files.txt` — mirroring the existing
+`ballista/scheduler/testdata/*` exclusion.
diff --git a/ballista/scheduler/tests/tpch_plan_stability/approved/q1.txt
b/ballista/scheduler/tests/tpch_plan_stability/approved/q1.txt
new file mode 100644
index 000000000..3f26e2a58
--- /dev/null
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q1.txt
@@ -0,0 +1,18 @@
+=== Stage 1 ===
+SortShuffleWriterExec: partitioning=Hash([l_returnflag@0, l_linestatus@1], 16)
+ AggregateExec: mode=Partial, gby=[l_returnflag@5 as l_returnflag,
l_linestatus@6 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_discount), count(Int64(1))]
+ ProjectionExec: expr=[l_extendedprice@0 * (Some(1),20,0 - l_discount@1) as
__common_expr_1, l_quantity@2 as l_quantity, l_extendedprice@0 as
l_extendedprice, l_discount@1 as l_discount, l_tax@3 as l_tax, l_returnflag@4
as l_returnflag, l_linestatus@5 as l_linestatus]
+ FilterExec: l_shipdate@6 <= 1998-09-02, projection=[l_extendedprice@1,
l_discount@2, l_quantity@0, l_tax@3, l_returnflag@4, l_linestatus@5]
+ StatsExec: rows=600037902
+
+=== Stage 2 ===
+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)
+
+=== 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)
diff --git a/ballista/scheduler/tests/tpch_plan_stability/approved/q10.txt
b/ballista/scheduler/tests/tpch_plan_stability/approved/q10.txt
new file mode 100644
index 000000000..ef52dce9c
--- /dev/null
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q10.txt
@@ -0,0 +1,60 @@
+=== Stage 1 ===
+SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
+ StatsExec: rows=25
+
+=== Stage 2 ===
+ShuffleWriterExec: partitioning: None
+ UnresolvedShuffleExec: partitioning: Hash([n_nationkey@0], 16)
+
+=== Stage 3 ===
+SortShuffleWriterExec: partitioning=Hash([c_custkey@0], 16)
+ StatsExec: rows=15000000
+
+=== Stage 4 ===
+SortShuffleWriterExec: partitioning=Hash([o_custkey@1], 16)
+ FilterExec: o_orderdate@2 >= 1993-10-01 AND o_orderdate@2 < 1994-01-01,
projection=[o_orderkey@0, o_custkey@1]
+ StatsExec: rows=150000000
+
+=== Stage 5 ===
+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)
+ SortExec: expr=[o_custkey@1 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: partitioning: Hash([o_custkey@1], 16)
+
+=== Stage 6 ===
+SortShuffleWriterExec: partitioning=Hash([l_orderkey@0], 16)
+ FilterExec: l_returnflag@3 = R, projection=[l_orderkey@0, l_extendedprice@1,
l_discount@2]
+ StatsExec: rows=600037902
+
+=== Stage 7 ===
+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)
+ SortExec: expr=[l_orderkey@0 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: 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)
+ AggregateExec: mode=Partial, gby=[c_custkey@0 as c_custkey, c_name@1 as
c_name, c_acctbal@4 as c_acctbal, c_phone@3 as c_phone, n_name@8 as n_name,
c_address@2 as c_address, c_comment@5 as c_comment],
aggr=[sum(lineitem.l_extendedprice * Some(1),20,0 - lineitem.l_discount) as
sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)]
+ ProjectionExec: expr=[c_custkey@0 as c_custkey, c_name@1 as c_name,
c_address@2 as c_address, c_phone@4 as c_phone, c_acctbal@5 as c_acctbal,
c_comment@6 as c_comment, l_extendedprice@7 as l_extendedprice, l_discount@8 as
l_discount, n_name@10 as n_name]
+ ProjectionExec: expr=[c_custkey@2 as c_custkey, c_name@3 as c_name,
c_address@4 as c_address, c_nationkey@5 as c_nationkey, c_phone@6 as c_phone,
c_acctbal@7 as c_acctbal, c_comment@8 as c_comment, l_extendedprice@9 as
l_extendedprice, l_discount@10 as l_discount, n_nationkey@0 as n_nationkey,
n_name@1 as n_name]
+ HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(n_nationkey@0,
c_nationkey@3)]
+ UnresolvedShuffleExec: broadcast=true, upstream_partitions: 16
+ UnresolvedShuffleExec: 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)
+
+=== 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)
diff --git a/ballista/scheduler/tests/tpch_plan_stability/approved/q11.txt
b/ballista/scheduler/tests/tpch_plan_stability/approved/q11.txt
new file mode 100644
index 000000000..059c25242
--- /dev/null
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q11.txt
@@ -0,0 +1,87 @@
+=== Stage 1 ===
+SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
+ FilterExec: n_name@1 = GERMANY, projection=[n_nationkey@0]
+ StatsExec: rows=25
+
+=== Stage 2 ===
+ShuffleWriterExec: partitioning: None
+ UnresolvedShuffleExec: partitioning: Hash([n_nationkey@0], 16)
+
+=== Stage 3 ===
+SortShuffleWriterExec: partitioning=Hash([ps_suppkey@0], 16)
+ StatsExec: rows=80000000
+
+=== Stage 4 ===
+SortShuffleWriterExec: partitioning=Hash([s_suppkey@0], 16)
+ StatsExec: rows=1000000
+
+=== Stage 5 ===
+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)
+ SortExec: expr=[s_suppkey@0 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: partitioning: Hash([s_suppkey@0], 16)
+
+=== Stage 6 ===
+ShuffleWriterExec: partitioning: None
+ AggregateExec: mode=Partial, gby=[], aggr=[sum(partsupp.ps_supplycost *
partsupp.ps_availqty)]
+ ProjectionExec: expr=[ps_availqty@0 as ps_availqty, ps_supplycost@1 as
ps_supplycost]
+ ProjectionExec: expr=[ps_availqty@1 as ps_availqty, ps_supplycost@2 as
ps_supplycost, s_nationkey@3 as s_nationkey, n_nationkey@0 as n_nationkey]
+ HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(n_nationkey@0,
s_nationkey@2)]
+ UnresolvedShuffleExec: broadcast=true, upstream_partitions: 16
+ UnresolvedShuffleExec: partitioning: Hash([s_nationkey@2], 16)
+
+=== Stage 7 ===
+SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
+ FilterExec: n_name@1 = GERMANY, projection=[n_nationkey@0]
+ StatsExec: rows=25
+
+=== Stage 8 ===
+ShuffleWriterExec: partitioning: None
+ UnresolvedShuffleExec: partitioning: Hash([n_nationkey@0], 16)
+
+=== Stage 9 ===
+SortShuffleWriterExec: partitioning=Hash([ps_suppkey@1], 16)
+ StatsExec: rows=80000000
+
+=== Stage 10 ===
+SortShuffleWriterExec: partitioning=Hash([s_suppkey@0], 16)
+ StatsExec: rows=1000000
+
+=== Stage 11 ===
+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)
+ SortExec: expr=[s_suppkey@0 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: partitioning: Hash([s_suppkey@0], 16)
+
+=== Stage 12 ===
+SortShuffleWriterExec: partitioning=Hash([ps_partkey@0], 16)
+ AggregateExec: mode=Partial, gby=[ps_partkey@0 as ps_partkey],
aggr=[sum(partsupp.ps_supplycost * partsupp.ps_availqty)]
+ ProjectionExec: expr=[ps_partkey@0 as ps_partkey, ps_availqty@1 as
ps_availqty, ps_supplycost@2 as ps_supplycost]
+ ProjectionExec: expr=[ps_partkey@1 as ps_partkey, ps_availqty@2 as
ps_availqty, ps_supplycost@3 as ps_supplycost, s_nationkey@4 as s_nationkey,
n_nationkey@0 as n_nationkey]
+ HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(n_nationkey@0,
s_nationkey@3)]
+ UnresolvedShuffleExec: broadcast=true, upstream_partitions: 16
+ UnresolvedShuffleExec: partitioning: Hash([s_nationkey@3], 16)
+
+=== Stage 13 ===
+ShuffleWriterExec: partitioning: None
+ SortExec: expr=[value@1 DESC], preserve_partitioning=[true]
+ ProjectionExec: expr=[ps_partkey@1 as ps_partkey,
sum(partsupp.ps_supplycost * partsupp.ps_availqty)@2 as value]
+ 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]
+ 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)
+ 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)
+
+=== Stage 14 ===
+ShuffleWriterExec: partitioning: None
+ SortPreservingMergeExec: [value@1 DESC]
+ UnresolvedShuffleExec: 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
new file mode 100644
index 000000000..9802dd3a7
--- /dev/null
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q12.txt
@@ -0,0 +1,30 @@
+=== Stage 1 ===
+SortShuffleWriterExec: partitioning=Hash([l_orderkey@0], 16)
+ FilterExec: (l_shipmode@4 = MAIL OR l_shipmode@4 = SHIP) AND l_receiptdate@3
> l_commitdate@2 AND l_shipdate@1 < l_commitdate@2 AND l_receiptdate@3 >=
1994-01-01 AND l_receiptdate@3 < 1995-01-01, projection=[l_orderkey@0,
l_shipmode@4]
+ StatsExec: rows=600037902
+
+=== Stage 2 ===
+SortShuffleWriterExec: partitioning=Hash([o_orderkey@0], 16)
+ StatsExec: rows=150000000
+
+=== Stage 3 ===
+SortShuffleWriterExec: partitioning=Hash([l_shipmode@0], 16)
+ 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)]
+ 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)
+ SortExec: expr=[o_orderkey@0 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: 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)
+
+=== Stage 5 ===
+ShuffleWriterExec: partitioning: None
+ SortPreservingMergeExec: [l_shipmode@0 ASC NULLS LAST]
+ UnresolvedShuffleExec: 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
new file mode 100644
index 000000000..252b08f08
--- /dev/null
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q13.txt
@@ -0,0 +1,32 @@
+=== Stage 1 ===
+SortShuffleWriterExec: partitioning=Hash([c_custkey@0], 16)
+ StatsExec: rows=15000000
+
+=== Stage 2 ===
+SortShuffleWriterExec: partitioning=Hash([o_custkey@1], 16)
+ FilterExec: o_comment@2 NOT LIKE %special%requests%,
projection=[o_orderkey@0, o_custkey@1]
+ StatsExec: rows=150000000
+
+=== Stage 3 ===
+SortShuffleWriterExec: partitioning=Hash([c_count@0], 16)
+ AggregateExec: mode=Partial, gby=[c_count@0 as c_count],
aggr=[count(Int64(1))]
+ ProjectionExec: expr=[count(orders.o_orderkey)@1 as c_count]
+ AggregateExec: mode=SinglePartitioned, gby=[c_custkey@0 as c_custkey],
aggr=[count(orders.o_orderkey)], ordering_mode=Sorted
+ 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)
+ SortExec: expr=[o_custkey@1 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: 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)
+
+=== Stage 5 ===
+ShuffleWriterExec: partitioning: None
+ SortPreservingMergeExec: [custdist@1 DESC, c_count@0 DESC]
+ UnresolvedShuffleExec: 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
new file mode 100644
index 000000000..172add80c
--- /dev/null
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q14.txt
@@ -0,0 +1,25 @@
+=== Stage 1 ===
+SortShuffleWriterExec: partitioning=Hash([l_partkey@0], 16)
+ FilterExec: l_shipdate@3 >= 1995-09-01 AND l_shipdate@3 < 1995-10-01,
projection=[l_partkey@0, l_extendedprice@1, l_discount@2]
+ StatsExec: rows=600037902
+
+=== Stage 2 ===
+SortShuffleWriterExec: partitioning=Hash([p_partkey@0], 16)
+ StatsExec: rows=20000000
+
+=== Stage 3 ===
+ShuffleWriterExec: partitioning: None
+ AggregateExec: mode=Partial, 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)]
+ 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)
+ SortExec: expr=[p_partkey@0 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: 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)
diff --git a/ballista/scheduler/tests/tpch_plan_stability/approved/q15.txt
b/ballista/scheduler/tests/tpch_plan_stability/approved/q15.txt
new file mode 100644
index 000000000..2a357f08d
--- /dev/null
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q15.txt
@@ -0,0 +1,57 @@
+=== Stage 1 ===
+SortShuffleWriterExec: partitioning=Hash([l_suppkey@0], 16)
+ AggregateExec: mode=Partial, gby=[l_suppkey@0 as l_suppkey],
aggr=[sum(lineitem.l_extendedprice * Some(1),20,0 - lineitem.l_discount) as
sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)]
+ FilterExec: l_shipdate@3 >= 1996-01-01 AND l_shipdate@3 < 1996-04-01,
projection=[l_suppkey@0, l_extendedprice@1, l_discount@2]
+ StatsExec: rows=600037902
+
+=== Stage 2 ===
+ShuffleWriterExec: partitioning: None
+ AggregateExec: mode=Partial, gby=[], aggr=[max(revenue0.total_revenue)]
+ ProjectionExec: expr=[sum(lineitem.l_extendedprice * Int64(1) -
lineitem.l_discount)@1 as total_revenue]
+ AggregateExec: mode=FinalPartitioned, gby=[l_suppkey@0 as l_suppkey],
aggr=[sum(lineitem.l_extendedprice * Some(1),20,0 - lineitem.l_discount) as
sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)]
+ UnresolvedShuffleExec: 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)
+
+=== Stage 4 ===
+ShuffleWriterExec: partitioning: None
+ UnresolvedShuffleExec: partitioning: Hash([max(revenue0.total_revenue)@0],
16)
+
+=== Stage 5 ===
+SortShuffleWriterExec: partitioning=Hash([s_suppkey@0], 16)
+ StatsExec: rows=1000000
+
+=== Stage 6 ===
+SortShuffleWriterExec: partitioning=Hash([l_suppkey@0], 16)
+ AggregateExec: mode=Partial, gby=[l_suppkey@0 as l_suppkey],
aggr=[sum(lineitem.l_extendedprice * Some(1),20,0 - lineitem.l_discount) as
sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)]
+ FilterExec: l_shipdate@3 >= 1996-01-01 AND l_shipdate@3 < 1996-04-01,
projection=[l_suppkey@0, l_extendedprice@1, l_discount@2]
+ StatsExec: rows=600037902
+
+=== Stage 7 ===
+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)
+ 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)
+
+=== Stage 8 ===
+ShuffleWriterExec: partitioning: None
+ SortExec: expr=[s_suppkey@0 ASC NULLS LAST], preserve_partitioning=[true]
+ ProjectionExec: expr=[s_suppkey@0 as s_suppkey, s_name@1 as s_name,
s_address@2 as s_address, s_phone@3 as s_phone, total_revenue@4 as
total_revenue]
+ ProjectionExec: expr=[s_suppkey@1 as s_suppkey, s_name@2 as s_name,
s_address@3 as s_address, s_phone@4 as s_phone, total_revenue@5 as
total_revenue, max(revenue0.total_revenue)@0 as max(revenue0.total_revenue)]
+ HashJoinExec: mode=CollectLeft, join_type=Inner,
on=[(max(revenue0.total_revenue)@0, total_revenue@4)]
+ UnresolvedShuffleExec: broadcast=true, upstream_partitions: 16
+ UnresolvedShuffleExec: 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)
diff --git a/ballista/scheduler/tests/tpch_plan_stability/approved/q16.txt
b/ballista/scheduler/tests/tpch_plan_stability/approved/q16.txt
new file mode 100644
index 000000000..4403c3587
--- /dev/null
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q16.txt
@@ -0,0 +1,46 @@
+=== Stage 1 ===
+SortShuffleWriterExec: partitioning=Hash([s_suppkey@0], 16)
+ FilterExec: s_comment@1 LIKE %Customer%Complaints%, projection=[s_suppkey@0]
+ StatsExec: rows=1000000
+
+=== Stage 2 ===
+ShuffleWriterExec: partitioning: None
+ UnresolvedShuffleExec: partitioning: Hash([s_suppkey@0], 16)
+
+=== Stage 3 ===
+SortShuffleWriterExec: partitioning=Hash([ps_partkey@0], 16)
+ StatsExec: rows=80000000
+
+=== Stage 4 ===
+SortShuffleWriterExec: partitioning=Hash([p_partkey@0], 16)
+ FilterExec: p_brand@1 != Brand#45 AND p_type@2 NOT LIKE MEDIUM POLISHED% AND
p_size@3 IN (SET) ([49, 14, 23, 45, 19, 3, 36, 9])
+ StatsExec: rows=20000000
+
+=== Stage 5 ===
+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)
+ SortExec: expr=[p_partkey@0 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: 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)
+
+=== 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)
+
+=== 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)
diff --git a/ballista/scheduler/tests/tpch_plan_stability/approved/q17.txt
b/ballista/scheduler/tests/tpch_plan_stability/approved/q17.txt
new file mode 100644
index 000000000..e3b6744bb
--- /dev/null
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q17.txt
@@ -0,0 +1,36 @@
+=== Stage 1 ===
+SortShuffleWriterExec: partitioning=Hash([l_partkey@0], 16)
+ StatsExec: rows=600037902
+
+=== Stage 2 ===
+SortShuffleWriterExec: partitioning=Hash([p_partkey@0], 16)
+ FilterExec: p_brand@1 = Brand#23 AND p_container@2 = MED BOX,
projection=[p_partkey@0]
+ StatsExec: rows=20000000
+
+=== Stage 3 ===
+SortShuffleWriterExec: partitioning=Hash([l_partkey@0], 16)
+ AggregateExec: mode=Partial, gby=[l_partkey@0 as l_partkey],
aggr=[avg(lineitem.l_quantity)]
+ StatsExec: rows=600037902
+
+=== Stage 4 ===
+ShuffleWriterExec: partitioning: None
+ AggregateExec: mode=Partial, gby=[], aggr=[sum(lineitem.l_extendedprice)]
+ ProjectionExec: expr=[l_extendedprice@1 as l_extendedprice]
+ SortMergeJoinExec: join_type=Inner, on=[(p_partkey@2, l_partkey@1)],
filter=CAST(l_quantity@0 AS Decimal128(30, 15)) < Float64(0.2) *
avg(lineitem.l_quantity)@1
+ 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)
+ SortExec: expr=[p_partkey@0 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: 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)
+
+=== 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)
diff --git a/ballista/scheduler/tests/tpch_plan_stability/approved/q18.txt
b/ballista/scheduler/tests/tpch_plan_stability/approved/q18.txt
new file mode 100644
index 000000000..190add99b
--- /dev/null
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q18.txt
@@ -0,0 +1,46 @@
+=== Stage 1 ===
+SortShuffleWriterExec: partitioning=Hash([c_custkey@0], 16)
+ StatsExec: rows=15000000
+
+=== Stage 2 ===
+SortShuffleWriterExec: partitioning=Hash([o_custkey@1], 16)
+ StatsExec: rows=150000000
+
+=== Stage 3 ===
+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)
+ SortExec: expr=[o_custkey@1 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: partitioning: Hash([o_custkey@1], 16)
+
+=== Stage 4 ===
+SortShuffleWriterExec: partitioning=Hash([l_orderkey@0], 16)
+ StatsExec: rows=600037902
+
+=== Stage 5 ===
+SortShuffleWriterExec: partitioning=Hash([l_orderkey@0], 16)
+ AggregateExec: mode=Partial, gby=[l_orderkey@0 as l_orderkey],
aggr=[sum(lineitem.l_quantity)]
+ StatsExec: rows=600037902
+
+=== Stage 6 ===
+ShuffleWriterExec: partitioning: None
+ 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)]
+ 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)
+ SortExec: expr=[l_orderkey@0 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: 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)
+
+=== 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)
diff --git a/ballista/scheduler/tests/tpch_plan_stability/approved/q19.txt
b/ballista/scheduler/tests/tpch_plan_stability/approved/q19.txt
new file mode 100644
index 000000000..a3960ed6e
--- /dev/null
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q19.txt
@@ -0,0 +1,26 @@
+=== Stage 1 ===
+SortShuffleWriterExec: partitioning=Hash([l_partkey@0], 16)
+ FilterExec: (l_shipmode@5 = AIR OR l_shipmode@5 = AIR REG) AND
l_shipinstruct@4 = DELIVER IN PERSON AND (l_quantity@1 >= Some(100),15,2 AND
l_quantity@1 <= Some(1100),15,2 OR l_quantity@1 >= Some(1000),15,2 AND
l_quantity@1 <= Some(2000),15,2 OR l_quantity@1 >= Some(2000),15,2 AND
l_quantity@1 <= Some(3000),15,2), projection=[l_partkey@0, l_quantity@1,
l_extendedprice@2, l_discount@3]
+ StatsExec: rows=600037902
+
+=== Stage 2 ===
+SortShuffleWriterExec: partitioning=Hash([p_partkey@0], 16)
+ FilterExec: p_size@2 >= 1 AND (p_brand@1 = Brand#12 AND p_container@3 IN
(SET) ([SM CASE, SM BOX, SM PACK, SM PKG]) 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
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 p_size@2 <= 15)
+ StatsExec: rows=20000000
+
+=== Stage 3 ===
+ShuffleWriterExec: partitioning: None
+ AggregateExec: mode=Partial, gby=[], aggr=[sum(lineitem.l_extendedprice *
Some(1),20,0 - lineitem.l_discount) as sum(lineitem.l_extendedprice * Int64(1)
- lineitem.l_discount)]
+ ProjectionExec: expr=[l_extendedprice@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)
+ SortExec: expr=[p_partkey@0 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: 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)
diff --git a/ballista/scheduler/tests/tpch_plan_stability/approved/q2.txt
b/ballista/scheduler/tests/tpch_plan_stability/approved/q2.txt
new file mode 100644
index 000000000..5d3dba459
--- /dev/null
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q2.txt
@@ -0,0 +1,137 @@
+=== Stage 1 ===
+SortShuffleWriterExec: partitioning=Hash([r_regionkey@0], 16)
+ FilterExec: r_name@1 = EUROPE, projection=[r_regionkey@0]
+ StatsExec: rows=5
+
+=== Stage 2 ===
+ShuffleWriterExec: partitioning: None
+ UnresolvedShuffleExec: partitioning: Hash([r_regionkey@0], 16)
+
+=== Stage 3 ===
+SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
+ StatsExec: rows=25
+
+=== Stage 4 ===
+ShuffleWriterExec: partitioning: None
+ UnresolvedShuffleExec: partitioning: Hash([n_nationkey@0], 16)
+
+=== Stage 5 ===
+SortShuffleWriterExec: partitioning=Hash([p_partkey@0], 16)
+ FilterExec: p_size@3 = 15 AND p_type@2 LIKE %BRASS, projection=[p_partkey@0,
p_mfgr@1]
+ StatsExec: rows=20000000
+
+=== Stage 6 ===
+SortShuffleWriterExec: partitioning=Hash([ps_partkey@0], 16)
+ StatsExec: rows=80000000
+
+=== Stage 7 ===
+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)
+ SortExec: expr=[ps_partkey@0 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: partitioning: Hash([ps_partkey@0], 16)
+
+=== Stage 8 ===
+SortShuffleWriterExec: partitioning=Hash([s_suppkey@0], 16)
+ StatsExec: rows=1000000
+
+=== Stage 9 ===
+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)
+ SortExec: expr=[s_suppkey@0 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: 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)
+
+=== 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)
+
+=== Stage 12 ===
+ShuffleWriterExec: partitioning: None
+ UnresolvedShuffleExec: partitioning: Hash([p_partkey@0, ps_supplycost@7], 16)
+
+=== Stage 13 ===
+SortShuffleWriterExec: partitioning=Hash([r_regionkey@0], 16)
+ FilterExec: r_name@1 = EUROPE, projection=[r_regionkey@0]
+ StatsExec: rows=5
+
+=== Stage 14 ===
+ShuffleWriterExec: partitioning: None
+ UnresolvedShuffleExec: partitioning: Hash([r_regionkey@0], 16)
+
+=== Stage 15 ===
+SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
+ StatsExec: rows=25
+
+=== Stage 16 ===
+ShuffleWriterExec: partitioning: None
+ UnresolvedShuffleExec: partitioning: Hash([n_nationkey@0], 16)
+
+=== Stage 17 ===
+SortShuffleWriterExec: partitioning=Hash([ps_suppkey@1], 16)
+ StatsExec: rows=80000000
+
+=== Stage 18 ===
+SortShuffleWriterExec: partitioning=Hash([s_suppkey@0], 16)
+ StatsExec: rows=1000000
+
+=== Stage 19 ===
+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)
+ SortExec: expr=[s_suppkey@0 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: 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)
+
+=== Stage 21 ===
+SortShuffleWriterExec: partitioning=Hash([ps_partkey@0], 16)
+ AggregateExec: mode=Partial, gby=[ps_partkey@0 as ps_partkey],
aggr=[min(partsupp.ps_supplycost)]
+ ProjectionExec: expr=[ps_partkey@0 as ps_partkey, ps_supplycost@1 as
ps_supplycost]
+ ProjectionExec: expr=[ps_partkey@1 as ps_partkey, ps_supplycost@2 as
ps_supplycost, n_regionkey@3 as n_regionkey, r_regionkey@0 as r_regionkey]
+ HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(r_regionkey@0,
n_regionkey@2)]
+ UnresolvedShuffleExec: broadcast=true, upstream_partitions: 16
+ UnresolvedShuffleExec: 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)
+
+=== 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)
+
+=== 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)
diff --git a/ballista/scheduler/tests/tpch_plan_stability/approved/q20.txt
b/ballista/scheduler/tests/tpch_plan_stability/approved/q20.txt
new file mode 100644
index 000000000..a684c1a4d
--- /dev/null
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q20.txt
@@ -0,0 +1,69 @@
+=== Stage 1 ===
+SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
+ FilterExec: n_name@1 = CANADA, projection=[n_nationkey@0]
+ StatsExec: rows=25
+
+=== Stage 2 ===
+ShuffleWriterExec: partitioning: None
+ UnresolvedShuffleExec: partitioning: Hash([n_nationkey@0], 16)
+
+=== Stage 3 ===
+SortShuffleWriterExec: partitioning=Hash([s_nationkey@3], 16)
+ StatsExec: rows=1000000
+
+=== Stage 4 ===
+SortShuffleWriterExec: partitioning=Hash([s_suppkey@0], 16)
+ ProjectionExec: expr=[s_suppkey@0 as s_suppkey, s_name@1 as s_name,
s_address@2 as s_address]
+ ProjectionExec: expr=[s_suppkey@1 as s_suppkey, s_name@2 as s_name,
s_address@3 as s_address, s_nationkey@4 as s_nationkey, n_nationkey@0 as
n_nationkey]
+ HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(n_nationkey@0,
s_nationkey@3)]
+ UnresolvedShuffleExec: broadcast=true, upstream_partitions: 16
+ UnresolvedShuffleExec: partitioning: Hash([s_nationkey@3], 16)
+
+=== Stage 5 ===
+SortShuffleWriterExec: partitioning=Hash([ps_partkey@0], 16)
+ StatsExec: rows=80000000
+
+=== Stage 6 ===
+SortShuffleWriterExec: partitioning=Hash([p_partkey@0], 16)
+ FilterExec: p_name@1 LIKE forest%, projection=[p_partkey@0]
+ StatsExec: rows=20000000
+
+=== Stage 7 ===
+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)
+ SortExec: expr=[p_partkey@0 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: partitioning: Hash([p_partkey@0], 16)
+
+=== Stage 8 ===
+SortShuffleWriterExec: partitioning=Hash([l_partkey@0, l_suppkey@1], 16)
+ AggregateExec: mode=Partial, gby=[l_partkey@0 as l_partkey, l_suppkey@1 as
l_suppkey], aggr=[sum(lineitem.l_quantity)]
+ FilterExec: l_shipdate@3 >= 1994-01-01 AND l_shipdate@3 < 1995-01-01,
projection=[l_partkey@0, l_suppkey@1, l_quantity@2]
+ StatsExec: rows=600037902
+
+=== Stage 9 ===
+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)
+ 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)
+
+=== Stage 10 ===
+ShuffleWriterExec: partitioning: None
+ SortExec: expr=[s_name@0 ASC NULLS LAST], preserve_partitioning=[true]
+ ProjectionExec: expr=[s_name@1 as s_name, s_address@2 as s_address]
+ SortMergeJoinExec: join_type=LeftSemi, on=[(s_suppkey@0, ps_suppkey@0)]
+ SortExec: expr=[s_suppkey@0 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: partitioning: Hash([s_suppkey@0], 16)
+ SortExec: expr=[ps_suppkey@0 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: 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)
diff --git a/ballista/scheduler/tests/tpch_plan_stability/approved/q21.txt
b/ballista/scheduler/tests/tpch_plan_stability/approved/q21.txt
new file mode 100644
index 000000000..693152cd6
--- /dev/null
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q21.txt
@@ -0,0 +1,82 @@
+=== Stage 1 ===
+SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
+ FilterExec: n_name@1 = SAUDI ARABIA, projection=[n_nationkey@0]
+ StatsExec: rows=25
+
+=== Stage 2 ===
+ShuffleWriterExec: partitioning: None
+ UnresolvedShuffleExec: partitioning: Hash([n_nationkey@0], 16)
+
+=== Stage 3 ===
+SortShuffleWriterExec: partitioning=Hash([s_suppkey@0], 16)
+ StatsExec: rows=1000000
+
+=== Stage 4 ===
+SortShuffleWriterExec: partitioning=Hash([l_suppkey@1], 16)
+ FilterExec: l_receiptdate@3 > l_commitdate@2, projection=[l_orderkey@0,
l_suppkey@1]
+ StatsExec: rows=600037902
+
+=== Stage 5 ===
+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)
+ SortExec: expr=[l_suppkey@1 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: partitioning: Hash([l_suppkey@1], 16)
+
+=== Stage 6 ===
+SortShuffleWriterExec: partitioning=Hash([o_orderkey@0], 16)
+ FilterExec: o_orderstatus@1 = F, projection=[o_orderkey@0]
+ StatsExec: rows=150000000
+
+=== Stage 7 ===
+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)
+ SortExec: expr=[o_orderkey@0 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: 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)
+
+=== Stage 9 ===
+SortShuffleWriterExec: partitioning=Hash([l_orderkey@0], 16)
+ StatsExec: rows=600037902
+
+=== Stage 10 ===
+SortShuffleWriterExec: partitioning=Hash([l_orderkey@0], 16)
+ FilterExec: l_receiptdate@3 > l_commitdate@2, projection=[l_orderkey@0,
l_suppkey@1]
+ StatsExec: rows=600037902
+
+=== Stage 11 ===
+SortShuffleWriterExec: partitioning=Hash([s_name@0], 16)
+ AggregateExec: mode=Partial, gby=[s_name@0 as s_name], aggr=[count(Int64(1))]
+ ProjectionExec: expr=[s_name@0 as s_name]
+ SortMergeJoinExec: join_type=LeftAnti, on=[(l_orderkey@1,
l_orderkey@0)], filter=l_suppkey@1 != l_suppkey@0
+ SortMergeJoinExec: join_type=LeftSemi, on=[(l_orderkey@1,
l_orderkey@0)], filter=l_suppkey@1 != l_suppkey@0
+ SortExec: expr=[l_orderkey@1 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: partitioning: Hash([l_orderkey@1], 16)
+ SortExec: expr=[l_orderkey@0 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: partitioning: Hash([l_orderkey@0], 16)
+ SortExec: expr=[l_orderkey@0 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: 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)
+
+=== Stage 13 ===
+ShuffleWriterExec: partitioning: None
+ SortPreservingMergeExec: [numwait@1 DESC, s_name@0 ASC NULLS LAST], fetch=100
+ UnresolvedShuffleExec: 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
new file mode 100644
index 000000000..4dd609587
--- /dev/null
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q22.txt
@@ -0,0 +1,35 @@
+=== Stage 1 ===
+SortShuffleWriterExec: partitioning=Hash([c_custkey@0], 16)
+ FilterExec: substr(c_phone@1, 1, 2) IN (SET) ([13, 31, 23, 29, 30, 18, 17])
+ StatsExec: rows=15000000
+
+=== Stage 2 ===
+SortShuffleWriterExec: partitioning=Hash([o_custkey@0], 16)
+ StatsExec: rows=150000000
+
+=== Stage 3 ===
+SortShuffleWriterExec: partitioning=Hash([cntrycode@0], 16)
+ AggregateExec: mode=Partial, gby=[cntrycode@0 as cntrycode],
aggr=[count(Int64(1)), sum(custsale.c_acctbal)]
+ ProjectionExec: expr=[substr(c_phone@1, 1, 2) as cntrycode, c_acctbal@2 as
c_acctbal]
+ NestedLoopJoinExec: join_type=Inner, filter=join_proj_push_down_1@1 >
avg(customer.c_acctbal)@0, projection=[avg(customer.c_acctbal)@0, c_phone@1,
c_acctbal@2]
+ AggregateExec: mode=Single, gby=[], aggr=[avg(customer.c_acctbal)]
+ FilterExec: c_acctbal@1 > Some(0),15,2 AND substr(c_phone@0, 1, 2)
IN (SET) ([13, 31, 23, 29, 30, 18, 17]), projection=[c_acctbal@1]
+ StatsExec: rows=15000000
+ 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)
+ SortExec: expr=[o_custkey@0 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: 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)
+
+=== Stage 5 ===
+ShuffleWriterExec: partitioning: None
+ SortPreservingMergeExec: [cntrycode@0 ASC NULLS LAST]
+ UnresolvedShuffleExec: 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
new file mode 100644
index 000000000..a8d755ca9
--- /dev/null
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q3.txt
@@ -0,0 +1,40 @@
+=== Stage 1 ===
+SortShuffleWriterExec: partitioning=Hash([c_custkey@0], 16)
+ FilterExec: c_mktsegment@1 = BUILDING, projection=[c_custkey@0]
+ StatsExec: rows=15000000
+
+=== Stage 2 ===
+SortShuffleWriterExec: partitioning=Hash([o_custkey@1], 16)
+ FilterExec: o_orderdate@2 < 1995-03-15
+ StatsExec: rows=150000000
+
+=== Stage 3 ===
+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)
+ SortExec: expr=[o_custkey@1 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: partitioning: Hash([o_custkey@1], 16)
+
+=== Stage 4 ===
+SortShuffleWriterExec: partitioning=Hash([l_orderkey@0], 16)
+ FilterExec: l_shipdate@3 > 1995-03-15, projection=[l_orderkey@0,
l_extendedprice@1, l_discount@2]
+ StatsExec: rows=600037902
+
+=== Stage 5 ===
+ShuffleWriterExec: partitioning: None
+ SortExec: TopK(fetch=10), expr=[revenue@1 DESC, o_orderdate@2 ASC NULLS
LAST], preserve_partitioning=[true]
+ 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]
+ 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 * Some(1),20,0 - lineitem.l_discount) as
sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)],
ordering_mode=PartiallySorted([0])
+ 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)
+ SortExec: expr=[l_orderkey@0 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: 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)
diff --git a/ballista/scheduler/tests/tpch_plan_stability/approved/q4.txt
b/ballista/scheduler/tests/tpch_plan_stability/approved/q4.txt
new file mode 100644
index 000000000..bbb97d34b
--- /dev/null
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q4.txt
@@ -0,0 +1,31 @@
+=== Stage 1 ===
+SortShuffleWriterExec: partitioning=Hash([o_orderkey@0], 16)
+ FilterExec: o_orderdate@1 >= 1993-07-01 AND o_orderdate@1 < 1993-10-01,
projection=[o_orderkey@0, o_orderpriority@2]
+ StatsExec: rows=150000000
+
+=== Stage 2 ===
+SortShuffleWriterExec: partitioning=Hash([l_orderkey@0], 16)
+ FilterExec: l_receiptdate@2 > l_commitdate@1, projection=[l_orderkey@0]
+ StatsExec: rows=600037902
+
+=== Stage 3 ===
+SortShuffleWriterExec: partitioning=Hash([o_orderpriority@0], 16)
+ AggregateExec: mode=Partial, gby=[o_orderpriority@0 as o_orderpriority],
aggr=[count(Int64(1))]
+ 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)
+ SortExec: expr=[l_orderkey@0 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: 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)
+
+=== Stage 5 ===
+ShuffleWriterExec: partitioning: None
+ SortPreservingMergeExec: [o_orderpriority@0 ASC NULLS LAST]
+ UnresolvedShuffleExec: 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
new file mode 100644
index 000000000..c159882e4
--- /dev/null
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q5.txt
@@ -0,0 +1,89 @@
+=== Stage 1 ===
+SortShuffleWriterExec: partitioning=Hash([r_regionkey@0], 16)
+ FilterExec: r_name@1 = ASIA, projection=[r_regionkey@0]
+ StatsExec: rows=5
+
+=== Stage 2 ===
+ShuffleWriterExec: partitioning: None
+ UnresolvedShuffleExec: partitioning: Hash([r_regionkey@0], 16)
+
+=== Stage 3 ===
+SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
+ StatsExec: rows=25
+
+=== Stage 4 ===
+ShuffleWriterExec: partitioning: None
+ UnresolvedShuffleExec: partitioning: Hash([n_nationkey@0], 16)
+
+=== Stage 5 ===
+SortShuffleWriterExec: partitioning=Hash([c_custkey@0], 16)
+ StatsExec: rows=15000000
+
+=== Stage 6 ===
+SortShuffleWriterExec: partitioning=Hash([o_custkey@1], 16)
+ FilterExec: o_orderdate@2 >= 1994-01-01 AND o_orderdate@2 < 1995-01-01,
projection=[o_orderkey@0, o_custkey@1]
+ StatsExec: rows=150000000
+
+=== Stage 7 ===
+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)
+ SortExec: expr=[o_custkey@1 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: partitioning: Hash([o_custkey@1], 16)
+
+=== Stage 8 ===
+SortShuffleWriterExec: partitioning=Hash([l_orderkey@0], 16)
+ StatsExec: rows=600037902
+
+=== Stage 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)
+ SortExec: expr=[l_orderkey@0 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: partitioning: Hash([l_orderkey@0], 16)
+
+=== Stage 10 ===
+SortShuffleWriterExec: partitioning=Hash([s_suppkey@0, s_nationkey@1], 16)
+ StatsExec: rows=1000000
+
+=== Stage 11 ===
+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)
+ SortExec: expr=[s_suppkey@0 ASC, s_nationkey@1 ASC],
preserve_partitioning=[true]
+ UnresolvedShuffleExec: 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)
+
+=== Stage 13 ===
+SortShuffleWriterExec: partitioning=Hash([n_name@0], 16)
+ AggregateExec: mode=Partial, gby=[n_name@2 as n_name],
aggr=[sum(lineitem.l_extendedprice * Some(1),20,0 - lineitem.l_discount) as
sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)]
+ ProjectionExec: expr=[l_extendedprice@0 as l_extendedprice, l_discount@1
as l_discount, n_name@2 as n_name]
+ ProjectionExec: expr=[l_extendedprice@1 as l_extendedprice, l_discount@2
as l_discount, n_name@3 as n_name, n_regionkey@4 as n_regionkey, r_regionkey@0
as r_regionkey]
+ HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(r_regionkey@0,
n_regionkey@3)]
+ UnresolvedShuffleExec: broadcast=true, upstream_partitions: 16
+ UnresolvedShuffleExec: 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)
+
+=== Stage 15 ===
+ShuffleWriterExec: partitioning: None
+ SortPreservingMergeExec: [revenue@1 DESC]
+ UnresolvedShuffleExec: 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
new file mode 100644
index 000000000..1da12a0cb
--- /dev/null
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q6.txt
@@ -0,0 +1,6 @@
+=== Stage 1 ===
+ShuffleWriterExec: partitioning: None
+ 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 >= Some(5),15,2 AND l_discount@2 <= Some(7),15,2 AND l_quantity@0
< Some(2400),15,2, projection=[l_extendedprice@1, l_discount@2]
+ StatsExec: rows=600037902
diff --git a/ballista/scheduler/tests/tpch_plan_stability/approved/q7.txt
b/ballista/scheduler/tests/tpch_plan_stability/approved/q7.txt
new file mode 100644
index 000000000..77cbc87a8
--- /dev/null
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q7.txt
@@ -0,0 +1,89 @@
+=== Stage 1 ===
+SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
+ FilterExec: n_name@1 = FRANCE OR n_name@1 = GERMANY
+ StatsExec: rows=25
+
+=== Stage 2 ===
+ShuffleWriterExec: partitioning: None
+ UnresolvedShuffleExec: partitioning: Hash([n_nationkey@0], 16)
+
+=== Stage 3 ===
+SortShuffleWriterExec: partitioning=Hash([s_suppkey@0], 16)
+ StatsExec: rows=1000000
+
+=== Stage 4 ===
+SortShuffleWriterExec: partitioning=Hash([l_suppkey@1], 16)
+ FilterExec: l_shipdate@4 >= 1995-01-01 AND l_shipdate@4 <= 1996-12-31
+ StatsExec: rows=600037902
+
+=== Stage 5 ===
+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)
+ SortExec: expr=[l_suppkey@1 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: partitioning: Hash([l_suppkey@1], 16)
+
+=== Stage 6 ===
+SortShuffleWriterExec: partitioning=Hash([o_orderkey@0], 16)
+ StatsExec: rows=150000000
+
+=== Stage 7 ===
+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)
+ SortExec: expr=[o_orderkey@0 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: partitioning: Hash([o_orderkey@0], 16)
+
+=== Stage 8 ===
+SortShuffleWriterExec: partitioning=Hash([c_custkey@0], 16)
+ StatsExec: rows=15000000
+
+=== Stage 9 ===
+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)
+ SortExec: expr=[c_custkey@0 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: 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)
+
+=== Stage 11 ===
+ShuffleWriterExec: partitioning: None
+ UnresolvedShuffleExec: partitioning: Hash([c_nationkey@3], 16)
+
+=== Stage 12 ===
+SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
+ FilterExec: n_name@1 = GERMANY OR n_name@1 = FRANCE
+ StatsExec: rows=25
+
+=== Stage 13 ===
+SortShuffleWriterExec: partitioning=Hash([supp_nation@0, cust_nation@1,
l_year@2], 16)
+ AggregateExec: mode=Partial, gby=[supp_nation@0 as supp_nation,
cust_nation@1 as cust_nation, l_year@2 as l_year], aggr=[sum(shipping.volume)]
+ ProjectionExec: expr=[n_name@4 as supp_nation, n_name@6 as cust_nation,
date_part(YEAR, l_shipdate@2) as l_year, l_extendedprice@0 * (Some(1),20,0 -
l_discount@1) as volume]
+ HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(c_nationkey@3,
n_nationkey@0)], filter=n_name@0 = FRANCE AND n_name@1 = GERMANY OR n_name@0 =
GERMANY AND n_name@1 = FRANCE
+ UnresolvedShuffleExec: broadcast=true, upstream_partitions: 16
+ UnresolvedShuffleExec: 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)
+
+=== 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)
diff --git a/ballista/scheduler/tests/tpch_plan_stability/approved/q8.txt
b/ballista/scheduler/tests/tpch_plan_stability/approved/q8.txt
new file mode 100644
index 000000000..d70be863e
--- /dev/null
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q8.txt
@@ -0,0 +1,118 @@
+=== Stage 1 ===
+SortShuffleWriterExec: partitioning=Hash([r_regionkey@0], 16)
+ FilterExec: r_name@1 = AMERICA, projection=[r_regionkey@0]
+ StatsExec: rows=5
+
+=== Stage 2 ===
+ShuffleWriterExec: partitioning: None
+ UnresolvedShuffleExec: partitioning: Hash([r_regionkey@0], 16)
+
+=== Stage 3 ===
+SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
+ StatsExec: rows=25
+
+=== Stage 4 ===
+ShuffleWriterExec: partitioning: None
+ UnresolvedShuffleExec: partitioning: Hash([n_nationkey@0], 16)
+
+=== Stage 5 ===
+SortShuffleWriterExec: partitioning=Hash([p_partkey@0], 16)
+ FilterExec: p_type@1 = ECONOMY ANODIZED STEEL, projection=[p_partkey@0]
+ StatsExec: rows=20000000
+
+=== Stage 6 ===
+SortShuffleWriterExec: partitioning=Hash([l_partkey@1], 16)
+ StatsExec: rows=600037902
+
+=== Stage 7 ===
+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)
+ SortExec: expr=[l_partkey@1 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: partitioning: Hash([l_partkey@1], 16)
+
+=== Stage 8 ===
+SortShuffleWriterExec: partitioning=Hash([s_suppkey@0], 16)
+ StatsExec: rows=1000000
+
+=== Stage 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)
+ SortExec: expr=[s_suppkey@0 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: partitioning: Hash([s_suppkey@0], 16)
+
+=== Stage 10 ===
+SortShuffleWriterExec: partitioning=Hash([o_orderkey@0], 16)
+ FilterExec: o_orderdate@2 >= 1995-01-01 AND o_orderdate@2 <= 1996-12-31
+ StatsExec: rows=150000000
+
+=== Stage 11 ===
+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)
+ SortExec: expr=[o_orderkey@0 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: partitioning: Hash([o_orderkey@0], 16)
+
+=== Stage 12 ===
+SortShuffleWriterExec: partitioning=Hash([c_custkey@0], 16)
+ StatsExec: rows=15000000
+
+=== Stage 13 ===
+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)
+ SortExec: expr=[c_custkey@0 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: 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)
+
+=== Stage 15 ===
+ShuffleWriterExec: partitioning: None
+ UnresolvedShuffleExec: partitioning: Hash([s_nationkey@2], 16)
+
+=== Stage 16 ===
+SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
+ StatsExec: rows=25
+
+=== Stage 17 ===
+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)
+
+=== Stage 18 ===
+SortShuffleWriterExec: partitioning=Hash([o_year@0], 16)
+ AggregateExec: mode=Partial, gby=[o_year@0 as o_year], aggr=[sum(CASE WHEN
all_nations.nation = BRAZIL THEN all_nations.volume ELSE Some(0),38,4 END) as
sum(CASE WHEN all_nations.nation = Utf8("BRAZIL") THEN all_nations.volume ELSE
Int64(0) END), sum(all_nations.volume)]
+ ProjectionExec: expr=[date_part(YEAR, o_orderdate@2) as o_year,
l_extendedprice@0 * (Some(1),20,0 - l_discount@1) as volume, n_name@4 as nation]
+ ProjectionExec: expr=[l_extendedprice@1 as l_extendedprice, l_discount@2
as l_discount, o_orderdate@3 as o_orderdate, n_regionkey@4 as n_regionkey,
n_name@5 as n_name, r_regionkey@0 as r_regionkey]
+ HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(r_regionkey@0,
n_regionkey@3)]
+ UnresolvedShuffleExec: broadcast=true, upstream_partitions: 16
+ UnresolvedShuffleExec: 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)
+
+=== Stage 20 ===
+ShuffleWriterExec: partitioning: None
+ SortPreservingMergeExec: [o_year@0 ASC NULLS LAST]
+ UnresolvedShuffleExec: 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
new file mode 100644
index 000000000..3d91e1dd2
--- /dev/null
+++ b/ballista/scheduler/tests/tpch_plan_stability/approved/q9.txt
@@ -0,0 +1,85 @@
+=== Stage 1 ===
+SortShuffleWriterExec: partitioning=Hash([n_nationkey@0], 16)
+ StatsExec: rows=25
+
+=== Stage 2 ===
+ShuffleWriterExec: partitioning: None
+ UnresolvedShuffleExec: partitioning: Hash([n_nationkey@0], 16)
+
+=== Stage 3 ===
+SortShuffleWriterExec: partitioning=Hash([p_partkey@0], 16)
+ FilterExec: p_name@1 LIKE %green%, projection=[p_partkey@0]
+ StatsExec: rows=20000000
+
+=== Stage 4 ===
+SortShuffleWriterExec: partitioning=Hash([l_partkey@1], 16)
+ StatsExec: rows=600037902
+
+=== Stage 5 ===
+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)
+ SortExec: expr=[l_partkey@1 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: partitioning: Hash([l_partkey@1], 16)
+
+=== Stage 6 ===
+SortShuffleWriterExec: partitioning=Hash([s_suppkey@0], 16)
+ StatsExec: rows=1000000
+
+=== Stage 7 ===
+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)
+ SortExec: expr=[s_suppkey@0 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: partitioning: Hash([s_suppkey@0], 16)
+
+=== Stage 8 ===
+SortShuffleWriterExec: partitioning=Hash([ps_suppkey@1, ps_partkey@0], 16)
+ StatsExec: rows=80000000
+
+=== Stage 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)
+ SortExec: expr=[ps_suppkey@1 ASC, ps_partkey@0 ASC],
preserve_partitioning=[true]
+ UnresolvedShuffleExec: partitioning: Hash([ps_suppkey@1,
ps_partkey@0], 16)
+
+=== Stage 10 ===
+SortShuffleWriterExec: partitioning=Hash([o_orderkey@0], 16)
+ StatsExec: rows=150000000
+
+=== Stage 11 ===
+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)
+ SortExec: expr=[o_orderkey@0 ASC], preserve_partitioning=[true]
+ UnresolvedShuffleExec: partitioning: Hash([o_orderkey@0], 16)
+
+=== Stage 12 ===
+SortShuffleWriterExec: partitioning=Hash([nation@0, o_year@1], 16)
+ AggregateExec: mode=Partial, gby=[nation@0 as nation, o_year@1 as o_year],
aggr=[sum(profit.amount)]
+ ProjectionExec: expr=[n_name@7 as nation, date_part(YEAR, o_orderdate@5)
as o_year, l_extendedprice@1 * (Some(1),20,0 - l_discount@2) - ps_supplycost@4
* l_quantity@0 as amount]
+ ProjectionExec: expr=[l_quantity@2 as l_quantity, l_extendedprice@3 as
l_extendedprice, l_discount@4 as l_discount, s_nationkey@5 as s_nationkey,
ps_supplycost@6 as ps_supplycost, o_orderdate@7 as o_orderdate, n_nationkey@0
as n_nationkey, n_name@1 as n_name]
+ HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(n_nationkey@0,
s_nationkey@3)]
+ UnresolvedShuffleExec: broadcast=true, upstream_partitions: 16
+ UnresolvedShuffleExec: 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)
+
+=== Stage 14 ===
+ShuffleWriterExec: partitioning: None
+ SortPreservingMergeExec: [nation@0 ASC NULLS LAST, o_year@1 DESC]
+ UnresolvedShuffleExec: partitioning: Hash([nation@0, o_year@1], 16)
diff --git a/ballista/scheduler/tests/tpch_plan_stability/fixtures.rs
b/ballista/scheduler/tests/tpch_plan_stability/fixtures.rs
new file mode 100644
index 000000000..811370875
--- /dev/null
+++ b/ballista/scheduler/tests/tpch_plan_stability/fixtures.rs
@@ -0,0 +1,229 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! TPC-H schema/fixture helpers and the distributed staged-plan text helper.
+//!
+//! `staged_plan_text` loads a TPC-H query's SQL, registers the 8 TPC-H tables
+//! (via [`TpchStatsTable`]) with SF100 row-count statistics, builds the
+//! physical plan with the static (Ballista) planner, breaks it into
+//! distributed query stages with [`DefaultDistributedPlanner`], and renders
+//! the stages to a normalized text representation suitable for snapshotting.
+
+use std::sync::Arc;
+
+use ballista_core::JobId;
+use ballista_core::extension::SessionConfigExt;
+use ballista_scheduler::planner::{DefaultDistributedPlanner,
DistributedPlanner};
+use datafusion::arrow::datatypes::{DataType, Field, Schema};
+use datafusion::physical_plan::display::DisplayableExecutionPlan;
+use datafusion::prelude::{SessionConfig, SessionContext};
+
+use crate::stats_table::TpchStatsTable;
+
+const TARGET_PARTITIONS: usize = 16;
+const JOB_ID: &str = "plan_stability";
+
+/// The 8 TPC-H table names and their SF100 row counts.
+pub const SF100_ROWS: &[(&str, usize)] = &[
+ ("region", 5),
+ ("nation", 25),
+ ("supplier", 1_000_000),
+ ("customer", 15_000_000),
+ ("part", 20_000_000),
+ ("partsupp", 80_000_000),
+ ("orders", 150_000_000),
+ ("lineitem", 600_037_902),
+];
+
+/// Returns the schema for a TPC-H table, copied verbatim from
+/// `benchmarks/src/bin/tpch.rs::get_schema` (that function lives in a binary
+/// and cannot be imported directly).
+pub fn tpch_schema(table: &str) -> Schema {
+ match table {
+ "part" => Schema::new(vec![
+ Field::new("p_partkey", DataType::Int64, false),
+ Field::new("p_name", DataType::Utf8, false),
+ Field::new("p_mfgr", DataType::Utf8, false),
+ Field::new("p_brand", DataType::Utf8, false),
+ Field::new("p_type", DataType::Utf8, false),
+ Field::new("p_size", DataType::Int32, false),
+ Field::new("p_container", DataType::Utf8, false),
+ Field::new("p_retailprice", DataType::Decimal128(15, 2), false),
+ Field::new("p_comment", DataType::Utf8, false),
+ ]),
+
+ "supplier" => Schema::new(vec![
+ Field::new("s_suppkey", DataType::Int64, false),
+ Field::new("s_name", DataType::Utf8, false),
+ Field::new("s_address", DataType::Utf8, false),
+ Field::new("s_nationkey", DataType::Int64, false),
+ Field::new("s_phone", DataType::Utf8, false),
+ Field::new("s_acctbal", DataType::Decimal128(15, 2), false),
+ Field::new("s_comment", DataType::Utf8, false),
+ ]),
+
+ "partsupp" => Schema::new(vec![
+ Field::new("ps_partkey", DataType::Int64, false),
+ Field::new("ps_suppkey", DataType::Int64, false),
+ Field::new("ps_availqty", DataType::Int32, false),
+ Field::new("ps_supplycost", DataType::Decimal128(15, 2), false),
+ Field::new("ps_comment", DataType::Utf8, false),
+ ]),
+
+ "customer" => Schema::new(vec![
+ Field::new("c_custkey", DataType::Int64, false),
+ Field::new("c_name", DataType::Utf8, false),
+ Field::new("c_address", DataType::Utf8, false),
+ Field::new("c_nationkey", DataType::Int64, false),
+ Field::new("c_phone", DataType::Utf8, false),
+ Field::new("c_acctbal", DataType::Decimal128(15, 2), false),
+ Field::new("c_mktsegment", DataType::Utf8, false),
+ Field::new("c_comment", DataType::Utf8, false),
+ ]),
+
+ "orders" => Schema::new(vec![
+ Field::new("o_orderkey", DataType::Int64, false),
+ Field::new("o_custkey", DataType::Int64, false),
+ Field::new("o_orderstatus", DataType::Utf8, false),
+ Field::new("o_totalprice", DataType::Decimal128(15, 2), false),
+ Field::new("o_orderdate", DataType::Date32, false),
+ Field::new("o_orderpriority", DataType::Utf8, false),
+ Field::new("o_clerk", DataType::Utf8, false),
+ Field::new("o_shippriority", DataType::Int32, false),
+ Field::new("o_comment", DataType::Utf8, false),
+ ]),
+
+ "lineitem" => Schema::new(vec![
+ Field::new("l_orderkey", DataType::Int64, false),
+ Field::new("l_partkey", DataType::Int64, false),
+ Field::new("l_suppkey", DataType::Int64, false),
+ Field::new("l_linenumber", DataType::Int32, false),
+ Field::new("l_quantity", DataType::Decimal128(15, 2), false),
+ Field::new("l_extendedprice", DataType::Decimal128(15, 2), false),
+ Field::new("l_discount", DataType::Decimal128(15, 2), false),
+ Field::new("l_tax", DataType::Decimal128(15, 2), false),
+ Field::new("l_returnflag", DataType::Utf8, false),
+ Field::new("l_linestatus", DataType::Utf8, false),
+ Field::new("l_shipdate", DataType::Date32, false),
+ Field::new("l_commitdate", DataType::Date32, false),
+ Field::new("l_receiptdate", DataType::Date32, false),
+ Field::new("l_shipinstruct", DataType::Utf8, false),
+ Field::new("l_shipmode", DataType::Utf8, false),
+ Field::new("l_comment", DataType::Utf8, false),
+ ]),
+
+ "nation" => Schema::new(vec![
+ Field::new("n_nationkey", DataType::Int64, false),
+ Field::new("n_name", DataType::Utf8, false),
+ Field::new("n_regionkey", DataType::Int64, false),
+ Field::new("n_comment", DataType::Utf8, false),
+ ]),
+
+ "region" => Schema::new(vec![
+ Field::new("r_regionkey", DataType::Int64, false),
+ Field::new("r_name", DataType::Utf8, false),
+ Field::new("r_comment", DataType::Utf8, false),
+ ]),
+
+ other => panic!("unknown tpch table {other}"),
+ }
+}
+
+fn make_ctx() -> SessionContext {
+ let config =
+
SessionConfig::new_with_ballista().with_target_partitions(TARGET_PARTITIONS);
+ let ctx = SessionContext::new_with_config(config);
+ for (table, rows) in SF100_ROWS {
+ let schema = Arc::new(tpch_schema(table));
+ ctx.register_table(*table, Arc::new(TpchStatsTable::new(schema,
*rows)))
+ .unwrap();
+ }
+ ctx
+}
+
+fn is_query_stmt(stmt: &str) -> bool {
+ let u = stmt.trim_start().to_uppercase();
+ u.starts_with("SELECT") || u.starts_with("WITH")
+}
+
+/// Produce the normalized distributed staged-plan text for a TPC-H query.
+pub async fn staged_plan_text(query_name: &str) -> String {
+ // Read the query SQL directly from the canonical benchmark location rather
+ // than a copy, so a change to a benchmark query surfaces as a golden diff.
+ let sql_path = format!(
+ "{}/../../benchmarks/queries/{query_name}.sql",
+ env!("CARGO_MANIFEST_DIR")
+ );
+ let sql = std::fs::read_to_string(&sql_path)
+ .unwrap_or_else(|e| panic!("read {sql_path}: {e}"));
+
+ let ctx = make_ctx();
+
+ // Split into statements; execute DDL, capture the physical plan of the
answer
+ // (last SELECT/WITH) statement. Single-statement queries take the one
statement.
+ let stmts: Vec<&str> = sql
+ .split(';')
+ .map(str::trim)
+ .filter(|s| !s.is_empty())
+ .collect();
+ let answer_idx = stmts
+ .iter()
+ .rposition(|s| is_query_stmt(s))
+ .expect("no SELECT/WITH statement in query");
+
+ let mut physical = None;
+ for (i, stmt) in stmts.iter().enumerate() {
+ if i == answer_idx {
+ physical = Some(
+ ctx.sql(stmt)
+ .await
+ .unwrap()
+ .create_physical_plan()
+ .await
+ .unwrap(),
+ );
+ } else {
+ // DDL such as CREATE VIEW / DROP VIEW (q15) — apply it.
+ ctx.sql(stmt).await.unwrap().collect().await.unwrap();
+ }
+ }
+ let physical = physical.unwrap();
+
+ let mut planner = DefaultDistributedPlanner::new();
+ let state = ctx.state();
+ let job_id: JobId = JOB_ID.into();
+ let stages = planner
+ .plan_query_stages(&job_id, physical, state.config().options())
+ .unwrap();
+
+ let mut out = String::new();
+ for stage in &stages {
+ out.push_str(&format!("=== Stage {} ===\n", stage.stage_id()));
+ let ep: &dyn datafusion::physical_plan::ExecutionPlan = stage.as_ref();
+
out.push_str(&DisplayableExecutionPlan::new(ep).indent(false).to_string());
+ out.push('\n');
+ }
+ normalize(&out)
+}
+
+fn normalize(plan: &str) -> String {
+ // Strip the fixed job id and any hex addresses so output is byte-stable.
+ let s = plan.replace(JOB_ID, "<job_id>");
+ // remove 0x… addresses if any appear
+ let re = regex::Regex::new(r"0x[0-9a-fA-F]+").unwrap();
+ re.replace_all(&s, "0x<addr>").into_owned()
+}
diff --git a/ballista/scheduler/tests/tpch_plan_stability/main.rs
b/ballista/scheduler/tests/tpch_plan_stability/main.rs
new file mode 100644
index 000000000..38d6d5f2a
--- /dev/null
+++ b/ballista/scheduler/tests/tpch_plan_stability/main.rs
@@ -0,0 +1,112 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+mod fixtures;
+mod stats_table;
+
+use std::path::PathBuf;
+use std::sync::Arc;
+
+use datafusion::arrow::datatypes::{DataType, Field, Schema};
+use datafusion::common::stats::Precision;
+use datafusion::prelude::SessionContext;
+
+use stats_table::TpchStatsTable;
+
+#[tokio::test]
+async fn stats_table_reports_injected_rows() {
+ let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int64,
false)]));
+ let ctx = SessionContext::new();
+ ctx.register_table(
+ "t",
+ Arc::new(TpchStatsTable::new(Arc::clone(&schema), 12_345)),
+ )
+ .unwrap();
+
+ let plan = ctx
+ .sql("SELECT a FROM t")
+ .await
+ .unwrap()
+ .create_physical_plan()
+ .await
+ .unwrap();
+
+ let stats = plan.partition_statistics(None).unwrap();
+ assert_eq!(stats.num_rows, Precision::Inexact(12_345));
+}
+
+#[tokio::test]
+async fn staged_plan_text_is_nonempty_and_shuffled() {
+ let text = fixtures::staged_plan_text("q1").await;
+ assert!(
+ text.contains("ShuffleWriterExec"),
+ "q1 plan should contain shuffle stages:\n{text}"
+ );
+ assert!(
+ !text.contains("plan_stability"),
+ "job id should be normalized out"
+ );
+ assert!(text.contains("=== Stage"), "stage banners present");
+}
+
+#[tokio::test]
+async fn multi_statement_q15_plans() {
+ let text = fixtures::staged_plan_text("q15").await;
+ assert!(
+ text.contains("=== Stage"),
+ "q15 (create/select/drop view) should plan:\n{text}"
+ );
+}
+
+fn golden_path(name: &str) -> PathBuf {
+ PathBuf::from(env!("CARGO_MANIFEST_DIR"))
+ .join("tests/tpch_plan_stability/approved")
+ .join(format!("{name}.txt"))
+}
+
+async fn check_query(name: &str) {
+ let actual = fixtures::staged_plan_text(name).await;
+ let path = golden_path(name);
+ if std::env::var("BALLISTA_GENERATE_GOLDEN").is_ok() {
+ std::fs::create_dir_all(path.parent().unwrap()).unwrap();
+ std::fs::write(&path, actual.trim_end().to_string() + "\n").unwrap();
+ return;
+ }
+ let expected = std::fs::read_to_string(&path).unwrap_or_else(|_| {
+ panic!("missing golden {path:?}; regenerate with
BALLISTA_GENERATE_GOLDEN=1")
+ });
+ assert_eq!(
+ actual.trim_end(),
+ expected.trim_end(),
+ "distributed plan drift for {name}. Review the change; if intended,
regenerate with \
+ BALLISTA_GENERATE_GOLDEN=1 cargo test -p ballista-scheduler --test
tpch_plan_stability"
+ );
+}
+
+macro_rules! plan_stability_test {
+ ($($name:ident),+ $(,)?) => {
+ $(
+ #[tokio::test]
+ async fn $name() { check_query(stringify!($name)).await; }
+ )+
+ };
+}
+
+plan_stability_test!(
+ q1, q2, q3, q4, q5, q6, q7, q8, q9, q10, q11, q12, q13, q14, q15, q16,
q17, q18, q19,
+ q20, q21, q22
+);
diff --git a/ballista/scheduler/tests/tpch_plan_stability/stats_table.rs
b/ballista/scheduler/tests/tpch_plan_stability/stats_table.rs
new file mode 100644
index 000000000..2d8d7b1b5
--- /dev/null
+++ b/ballista/scheduler/tests/tpch_plan_stability/stats_table.rs
@@ -0,0 +1,152 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::fmt;
+use std::sync::Arc;
+
+use datafusion::arrow::datatypes::SchemaRef;
+use datafusion::catalog::{Session, TableProvider};
+use datafusion::common::stats::Precision;
+use datafusion::common::{Result, Statistics};
+use datafusion::execution::TaskContext;
+use datafusion::logical_expr::{Expr, TableType};
+use datafusion::physical_expr::EquivalenceProperties;
+use datafusion::physical_plan::execution_plan::{
+ Boundedness, EmissionType, SchedulingType,
+};
+use datafusion::physical_plan::memory::MemoryStream;
+use datafusion::physical_plan::{
+ DisplayAs, DisplayFormatType, ExecutionPlan, Partitioning, PlanProperties,
+ SendableRecordBatchStream,
+};
+
+use async_trait::async_trait;
+
+/// A leaf execution plan that produces no rows but reports a fixed row-count
+/// statistic, so the physical optimizer makes realistic join/broadcast choices
+/// without any data on disk.
+#[derive(Debug)]
+pub struct StatsExec {
+ schema: SchemaRef,
+ num_rows: usize,
+ cache: Arc<PlanProperties>,
+}
+
+impl StatsExec {
+ pub fn new(schema: SchemaRef, num_rows: usize) -> Self {
+ let cache = PlanProperties::new(
+ EquivalenceProperties::new(Arc::clone(&schema)),
+ Partitioning::UnknownPartitioning(1),
+ EmissionType::Incremental,
+ Boundedness::Bounded,
+ )
+ .with_scheduling_type(SchedulingType::Cooperative);
+ Self {
+ schema,
+ num_rows,
+ cache: Arc::new(cache),
+ }
+ }
+}
+
+impl DisplayAs for StatsExec {
+ fn fmt_as(&self, t: DisplayFormatType, f: &mut fmt::Formatter) ->
fmt::Result {
+ match t {
+ DisplayFormatType::Default | DisplayFormatType::Verbose => {
+ write!(f, "StatsExec: rows={}", self.num_rows)
+ }
+ DisplayFormatType::TreeRender => write!(f, ""),
+ }
+ }
+}
+
+impl ExecutionPlan for StatsExec {
+ fn name(&self) -> &'static str {
+ "StatsExec"
+ }
+ fn properties(&self) -> &Arc<PlanProperties> {
+ &self.cache
+ }
+ fn children(&self) -> Vec<&Arc<dyn ExecutionPlan>> {
+ vec![]
+ }
+ fn with_new_children(
+ self: Arc<Self>,
+ _: Vec<Arc<dyn ExecutionPlan>>,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ Ok(self)
+ }
+ fn execute(
+ &self,
+ _partition: usize,
+ _context: Arc<TaskContext>,
+ ) -> Result<SendableRecordBatchStream> {
+ Ok(Box::pin(MemoryStream::try_new(
+ vec![],
+ Arc::clone(&self.schema),
+ None,
+ )?))
+ }
+ fn partition_statistics(&self, _partition: Option<usize>) ->
Result<Arc<Statistics>> {
+ Ok(Arc::new(
+ Statistics::new_unknown(&self.schema)
+ .with_num_rows(Precision::Inexact(self.num_rows)),
+ ))
+ }
+}
+
+/// A dataless `TableProvider` with a real schema and a fixed row-count
statistic.
+#[derive(Debug)]
+pub struct TpchStatsTable {
+ schema: SchemaRef,
+ num_rows: usize,
+}
+
+impl TpchStatsTable {
+ pub fn new(schema: SchemaRef, num_rows: usize) -> Self {
+ Self { schema, num_rows }
+ }
+}
+
+#[async_trait]
+impl TableProvider for TpchStatsTable {
+ fn schema(&self) -> SchemaRef {
+ Arc::clone(&self.schema)
+ }
+ fn table_type(&self) -> TableType {
+ TableType::Base
+ }
+ async fn scan(
+ &self,
+ _state: &dyn Session,
+ projection: Option<&Vec<usize>>,
+ _filters: &[Expr],
+ _limit: Option<usize>,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ let schema = match projection {
+ Some(p) => Arc::new(self.schema.project(p)?),
+ None => Arc::clone(&self.schema),
+ };
+ Ok(Arc::new(StatsExec::new(schema, self.num_rows)))
+ }
+ fn statistics(&self) -> Option<Statistics> {
+ Some(
+ Statistics::new_unknown(&self.schema)
+ .with_num_rows(Precision::Inexact(self.num_rows)),
+ )
+ }
+}
diff --git a/dev/release/rat_exclude_files.txt
b/dev/release/rat_exclude_files.txt
index 47a5ee2eb..83d47821c 100644
--- a/dev/release/rat_exclude_files.txt
+++ b/dev/release/rat_exclude_files.txt
@@ -34,6 +34,7 @@ requirements.txt
.gitattributes
benchmarks/queries/q*.sql
ballista/scheduler/testdata/*
+ballista/scheduler/tests/tpch_plan_stability/approved/*
**/yarn.lock
python/requirements*.txt
**/testdata/*
diff --git a/dev/update-tpch-plan-stability.sh
b/dev/update-tpch-plan-stability.sh
new file mode 100755
index 000000000..072e2a1b6
--- /dev/null
+++ b/dev/update-tpch-plan-stability.sh
@@ -0,0 +1,25 @@
+#!/usr/bin/env bash
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+# Regenerate the approved TPC-H distributed plans for the plan-stability suite.
+# Run after an intended planner/plan-shape change, then review the diff.
+set -euo pipefail
+cd "$(dirname "$0")/.."
+BALLISTA_GENERATE_GOLDEN=1 cargo test -p ballista-scheduler --test
tpch_plan_stability
+echo "Regenerated. Review changes under
ballista/scheduler/tests/tpch_plan_stability/approved/"
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
