Add TPC-H distributed plan-stability test suite

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]

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.

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_disc, count(Int64(1))@9 as count_order]
+      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_discount), count(Int64(1))]
+        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)

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)

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)

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)

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)