ci: sweep TPC-H/TPC-DS verification over partition counts and empty tables

.github/workflows/tpcds.yml

@@ -61,21 +61,33 @@ jobs:
     # sitting for hours with no useful signal.
     timeout-minutes: 45
     strategy:
-      # One red leg must not cancel the other; the two task-packing modes are
-      # independent signals.
+      # One red leg must not cancel the others; the legs are independent signals.
       fail-fast: false
       matrix:
         include:
           - label: "1 partition per task"
             task_args: ""
             slug: "mpt1"
+            partitions: "16"
           # 4 rather than a larger cap because the scheduler clamps the slice to
           # the executor's free vcores (`budget.vcores.min(cap)`), and the
           # executor below runs `--concurrent-tasks 4`. A higher cap would be
           # unreachable and the label would overstate what is covered.
           - label: "4 partitions per task"
             task_args: "-c ballista.scheduler.max_partitions_per_task=4"
             slug: "mpt4"
+            partitions: "16"
+          # Low-partition legs: target_partitions=1 exercises the degenerate
+          # single-task path; =2 exercises low-partition task packing
+          # (the #2186 bug class).
+          - label: "target_partitions=1"
+            task_args: ""
+            slug: "tp1"
+            partitions: "1"
+          - label: "target_partitions=2, 4 partitions per task"
+            task_args: "-c ballista.scheduler.max_partitions_per_task=4"
+            slug: "tp2-mpt4"
+            partitions: "2"
     steps:
       - name: Install dependencies
         run: |
@@ -180,15 +192,15 @@ jobs:
           # This matrix leg runs the suite under the default (static) planner at
           # one task-packing setting. The tpcds binary internally loops all
           # non-skipped queries and exits non-zero on any failure, so a single
-          # invocation covers the whole suite. The other leg runs in parallel.
+          # invocation covers the whole suite. The other legs run in parallel.
           #
           # Coverage for the adaptive planner (AQE on) is being added
           # separately; it currently fails on pre-existing bugs.
           echo "::group::[static planner, ${{ matrix.label }}] TPC-DS suite"
           ./target/tpch-ci/tpcds \
             --host 127.0.0.1 --port 50050 \
             --path "$DATA_DIR" \
-            --partitions 16 \
+            --partitions ${{ matrix.partitions }} \
             --verify \
             -c datafusion.optimizer.prefer_hash_join=false \
             ${{ matrix.task_args }}

.github/workflows/tpch.yml

@@ -15,7 +15,7 @@
 # specific language governing permissions and limitations
 # under the License.
 
-name: TPC-H SF10
+name: TPC-H
 
 permissions:
   contents: read
@@ -52,23 +52,61 @@ on:
 
 jobs:
   tpch-sf10:
-    name: TPC-H SF10 (${{ matrix.label }})
+    name: TPC-H (${{ matrix.label }})
     runs-on: ubuntu-latest
     container:
       image: amd64/rust
+    # Cap the job well under the 6-hour default so a hung query fails fast
+    # and frees the runner rather than sitting for hours with no useful
+    # signal.
+    timeout-minutes: 60
     strategy:
       fail-fast: false
       matrix:
         include:
-          - label: "AQE off"
+          - label: "SF10, AQE off"
             planner_args: ""
             slug: "aqe-off"
-          - label: "AQE on"
+            scale_factor: "10"
+            partitions: "16"
+            empty_rotation: false
+          - label: "SF10, AQE on"
             planner_args: "-c ballista.planner.adaptive.enabled=true"
             slug: "aqe-on"
-          - label: "AQE on, multi-partition tasks"
+            scale_factor: "10"
+            partitions: "16"
+            empty_rotation: false
+          - label: "SF10, AQE on, multi-partition tasks"
             planner_args: "-c ballista.planner.adaptive.enabled=true -c ballista.scheduler.max_partitions_per_task=0"
             slug: "aqe-on-mpt"
+            scale_factor: "10"
+            partitions: "16"
+            empty_rotation: false
+          # Low-partition legs run at SF1 to keep CI runtime bounded; low
+          # partition counts are what triggered the union-restriction and
+          # single-task-path bug class (#2186).
+          - label: "SF1, AQE on, 1 partition"
+            planner_args: "-c ballista.planner.adaptive.enabled=true"
+            slug: "aqe-on-sf1-p1"
+            scale_factor: "1"
+            partitions: "1"
+            empty_rotation: false
+          - label: "SF1, AQE on, 2 partitions"
+            planner_args: "-c ballista.planner.adaptive.enabled=true"
+            slug: "aqe-on-sf1-p2"
+            scale_factor: "1"
+            partitions: "2"
+            empty_rotation: false
+          # Empty-stage leg: a rotating set of tables is registered zero-row,
+          # forcing empty intermediate stages through PropagateEmptyExecRule
+          # (the #2185/#2194 bug class). Rotation is keyed on day-of-week in
+          # the run step below.
+          - label: "SF1, AQE on, empty tables"
+            planner_args: "-c ballista.planner.adaptive.enabled=true"
+            slug: "aqe-on-sf1-empty"
+            scale_factor: "1"
+            partitions: "16"
+            empty_rotation: true
     steps:
       - name: Install dependencies
         run: |
@@ -88,7 +126,7 @@ jobs:
         with:
           # Share the build cache across all matrix legs — the compiled
           # binaries are identical; only the per-suite CLI args differ.
-          shared-key: tpch-sf10
+          shared-key: tpch
 
       - name: Build Ballista binaries
         run: |
@@ -102,16 +140,19 @@ jobs:
         with:
           tool: tpchgen-cli@2.0.2
 
-      - name: Generate TPC-H SF10 data
+      - name: Generate TPC-H data
         run: |
           mkdir -p "$RUNNER_TEMP/tpch-data"
           tpchgen-cli \
-            --scale-factor 10 \
+            --scale-factor ${{ matrix.scale_factor }} \
             --parts 16 \
             --format=parquet \
             --output-dir "$RUNNER_TEMP/tpch-data"
 
       - name: Run TPC-H queries against Ballista cluster
+        # Primary timeout, on the step rather than the job, so that a hang
+        # still runs the log-upload step below.
+        timeout-minutes: 45
         env:
           DATA_DIR: ${{ runner.temp }}/tpch-data
           WORK_DIR: ${{ runner.temp }}/work
@@ -168,26 +209,47 @@ jobs:
           done
           nc -z 127.0.0.1 50051 || { echo "executor did not start"; exit 1; }
 
+          # Empty-tables rotation: keyed on day-of-week so different
+          # query/empty-table combinations get coverage over time while any
+          # single run stays cheap. Deterministic within a day.
+          EMPTY_ARGS=""
+          if [ "${{ matrix.empty_rotation }}" = "true" ]; then
+            case $(( $(date +%u) % 3 )) in
+              0) EMPTY_TABLES="lineitem" ;;
+              1) EMPTY_TABLES="orders" ;;
+              2) EMPTY_TABLES="supplier,nation" ;;
+            esac
+            echo "Empty tables for this run: $EMPTY_TABLES"
+            EMPTY_ARGS="--empty-tables $EMPTY_TABLES"
+          fi
+
           # This matrix leg runs one planner configuration against the whole
-          # SF10 dataset. The other legs run in parallel jobs.
+          # generated dataset. The other legs run in parallel jobs.
           # q16 omitted: still unsupported (matches benchmarks/run.sh).
           for q in 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 17 18 19 20 21 22; do
             echo "::group::[${{ matrix.label }}] Query $q"
+            # shellcheck disable=SC2086 # EMPTY_ARGS is intentionally unquoted:
+            # it expands to zero words or a flag plus value, never a single
+            # word containing spaces.
             ./target/tpch-ci/tpch benchmark ballista \
               --host 127.0.0.1 --port 50050 \
               --query "$q" \
               --path "$DATA_DIR" \
               --format parquet \
-              --partitions 16 \
+              --partitions ${{ matrix.partitions }} \
               --iterations 1 \
               --verify \
+              $EMPTY_ARGS \
               -c datafusion.optimizer.prefer_hash_join=false \
               ${{ matrix.planner_args }}
             echo "::endgroup::"
           done
 
       - name: Upload cluster logs on failure
-        if: failure()
+        # `!success()` rather than `failure()` so a timed-out or cancelled
+        # run still surfaces its scheduler/executor logs — the hang case is
+        # exactly when they are most worth having.
+        if: ${{ !success() }}
         uses: actions/upload-artifact@v7
         with:
           name: tpch-sf10-cluster-logs-${{ matrix.slug }}

benchmarks/README.md

@@ -400,4 +400,12 @@ scheduler + executor on every push/PR touching `ballista/**` or
 `EmptyExec invalid partition` assertion (issue #2047); re-enable an AQE-on run
 once that is fixed.
 
+### Forcing empty intermediate stages
+
+`--empty-tables lineitem,orders` registers the named tables as zero-row
+Parquet tables (schema preserved) in both the Ballista session and the
+`--verify` oracle. Every query touching an emptied table then produces empty
+intermediate stages, exercising the scheduler's empty-stage plan rewrites.
+Requires `--format parquet` (TPC-H) and is incompatible with `--expected`.
+
 [1]: http://www.tpc.org/tpch/

benchmarks/src/bin/tpcds.rs

@@ -19,13 +19,14 @@ use ballista::extension::SessionConfigExt;
 use ballista::prelude::SessionContextExt;
 use ballista_benchmarks::{
     answer_statement_index, compare_results, execute_query_capturing_answer,
-    register_parquet_tables,
+    parse_empty_tables, prepare_empty_tables, register_parquet_tables,
 };
 use ballista_core::object_store::{
     session_config_with_s3_support, session_state_with_s3_support,
 };
 use datafusion::error::{DataFusionError, Result};
 use datafusion::prelude::{SessionConfig, SessionContext};
+use std::collections::HashMap;
 use std::fs;
 use std::time::Instant;
 use structopt::StructOpt;
@@ -132,6 +133,12 @@ struct Opt {
     /// Verify each Ballista result against single-process DataFusion.
     #[structopt(long = "verify")]
     verify: bool,
+
+    /// Comma-separated table names to register as zero-row tables (schema
+    /// preserved) in both the Ballista and oracle contexts, forcing empty
+    /// intermediate stages through the distributed planner.
+    #[structopt(long = "empty-tables")]
+    empty_tables: Option<String>,
 }
 
 /// Split a query file into statements, dropping full-line `--` comments and
@@ -222,6 +229,7 @@ async fn run_one_query(
     opt: &Opt,
     address: &str,
     oracle_ctx: Option<&SessionContext>,
+    overrides: &HashMap<String, String>,
     query: usize,
 ) -> Result<()> {
     let sqls = get_query_sql(query)
@@ -247,7 +255,7 @@ async fn run_one_query(
     let ctx = SessionContext::remote_with_state(address, state)
         .await
         .map_err(|e| DataFusionError::Execution(format!("connect: {e}")))?;
-    register_parquet_tables(&ctx, TABLES, opt.path.as_str(), opt.debug)
+    register_parquet_tables(&ctx, TABLES, opt.path.as_str(), overrides, opt.debug)
         .await
         .map_err(|e| DataFusionError::Execution(format!("register-tables: {e}")))?;
 
@@ -309,14 +317,33 @@ async fn main() -> Result<()> {
     let opt = Opt::from_args();
     let address = format!("df://{}:{}", opt.host, opt.port);
 
+    // Zero-row mirrors are built once and shared by every query's Ballista
+    // session and by the oracle, so both sides see identical empty tables.
+    let empty_overrides = match &opt.empty_tables {
+        Some(spec) => {
+            let names = parse_empty_tables(spec, TABLES)?;
+            let dir = std::env::temp_dir()
+                .join(format!("tpcds-empty-tables-{}", std::process::id()));
+            prepare_empty_tables(&opt.path, &names, &dir).await?
+        }
+        None => std::collections::HashMap::new(),
+    };
+
     // Oracle context (single-process DataFusion), built once when verifying.
     // Not per-query, so a failure here is genuinely fatal to the whole run.
     let oracle_ctx = if opt.verify {
         let cfg = SessionConfig::new()
             .with_target_partitions(opt.partitions)
             .with_batch_size(opt.batch_size);
         let ctx = SessionContext::new_with_config(cfg);
-        register_parquet_tables(&ctx, TABLES, opt.path.as_str(), opt.debug).await?;
+        register_parquet_tables(
+            &ctx,
+            TABLES,
+            opt.path.as_str(),
+            &empty_overrides,
+            opt.debug,
+        )
+        .await?;
         Some(ctx)
     } else {
         None
@@ -325,7 +352,10 @@ async fn main() -> Result<()> {
     let mut failures: Vec<(usize, String)> = vec![];
 
     for query in selected_queries(opt.query, SKIP) {
-        if let Err(e) = run_one_query(&opt, &address, oracle_ctx.as_ref(), query).await {
+        if let Err(e) =
+            run_one_query(&opt, &address, oracle_ctx.as_ref(), &empty_overrides, query)
+                .await
+        {
             eprintln!("Query {query} FAILED: {e}");
             failures.push((query, e.to_string()));
         }