Skip to content

feat(bench): time TPC-DS queries and share the runner code with TPC-H - #2565

Merged
andygrove merged 3 commits into
apache:mainfrom
andygrove:tpcds-benchmark-runner
Oct 12, 2026
Merged

andygrove merged 3 commits into
apache:mainfrom
andygrove:tpcds-benchmark-runner

Conversation

@andygrove

@andygrove andygrove commented Oct 7, 2026 •

Copy link
Copy Markdown
Member

Which issue does this PR close?

No issue was filed.

Rationale for this change

The tpcds binary is a correctness gate. It runs each query once, prints the time, and writes nothing to disk, so it can't be used to benchmark TPC-DS the way tpch benchmark ballista benchmarks TPC-H. It also doesn't fit TPC-DS data generated by other tools:

  • It always rewrites three column names to the ones tpcgen-cli uses (ib_income_band_id, r_reason_description, cr_return_amount_inc_tax). Data from dsdgen or Spark uses the spec names that the queries already use, so the rewrite breaks q64, q81, q84, q85 and q93 on that data.
  • It registers tables without their Hive partition columns. Data written with Spark's partitionBy only stores the fact tables' date keys (ss_sold_date_sk and so on) in the directory names, so those columns would be missing.

Much of what a benchmark run needs already exists in tpch.rs: iterations, the JSON summary written after every query, per-query error recording, config overrides, and partition-aware Parquet registration. This PR moves those pieces into the ballista-benchmarks library and builds the TPC-DS benchmark on them, so the two runners share the code rather than each having a copy.

What changes are included in this PR?

  • Shared code (benchmarks/src/lib.rs, new benchmarks/src/summary.rs):
    • summary::{BenchmarkRun, QueryRun, QueryResult}, which are unchanged, plus run_suite and time_query.
      • run_suite runs the queries in order, records each one's iterations or error, writes <name>-<start_time>.json atomically after every query, prints the suite total and a list of failures, and returns an error if any query failed. tpch benchmark ballista, tpch benchmark datafusion and tpcds all use it.
      • time_query is the timed iteration loop that was run_local_query in tpch.rs. tpch benchmark datafusion and tpcds use it. tpch benchmark ballista keeps its own loop, which also plans each statement and checks --expected, so its timings don't change.
    • benchmark_session_config builds the session config: S3 support, target partitions, batch size, statistics collection and -c key=value overrides. ballista_context connects a session with that config and a job name to the scheduler. local_context builds a single-process session that can read S3.
    • parquet_table_layout moves out of tpch.rs. It now takes a PathColumnType function that gives the type of a partition column that only exists in the directory names. For TPC-H that type comes from the TPC-H schema, as before.
    • ParquetTable holds a table's path, file schema and partition columns. infer_parquet_tables infers them for several tables concurrently, and register_parquet_tables registers them on a session without reading any files. register_parquet_table (infer, then register) is what tpch uses.
    • Hive and Spark write rows whose partition key is NULL to a <column>=__HIVE_DEFAULT_PARTITION__/ directory, which DataFusion can only read as a string (apache/datafusion#18083). ParquetTable finds such columns when it infers the layout. It then registers the files as <table>_raw with those columns as strings, and <table> as a view that maps that directory to NULL and casts the other values back. Filters on the column still skip directories. Tables without that directory are registered as before.
    • read_query_file finds <dir>/qN.sql relative to the current directory or the repository root. Both runners use it.
  • tpcds runner:
    • New --iterations, --output and --no-partition-cols options, which mean what they mean for tpch. The summary uses the TPC-H format, so tpch compare reads it.
    • The 24 table layouts are inferred once at startup, concurrently, instead of each query's session reading every file footer again. That matters at scale on object storage. Each query's session and the --verify oracle register the cached layouts.
    • Column renames now apply only when the tables have the tpcgen-cli name and not the spec name. That is decided once, from the inferred layouts. CI's tpcgen-cli data still gets all three.
    • A partition column that only exists in the directory names is read as Int32 if it is a surrogate key (*_sk), because TPC-DS partitions its fact tables by date key. Rows with a NULL date key, which Spark writes to ss_sold_date_sk=__HIVE_DEFAULT_PARTITION__/ and so on, read as NULL.
    • Uses mimalloc, like tpch.
    • The CLI is otherwise unchanged, so the CI invocation still works.
  • tpch runner: no change in behaviour, with one exception. A query file that can't be loaded is now recorded as that query's failure instead of ending the run.
  • Tests: the QueryRun tests move with the struct. New tests cover run_suite (failures are recorded and the suite carries on, the summary file name and contents, and no output directory), which renames apply to tpcgen-cli and to spec column names, a Spark-style store_sales/ss_sold_date_sk=N/ table that registers the key as Int32 and scans only the matching partition, and the same table with a __HIVE_DEFAULT_PARTITION__ directory, whose rows read back with a NULL key while a filter on the key still skips that directory.
  • Docs: the TPC-DS section of benchmarks/README.md describes the benchmark mode, partition handling and the conditional renames.

Are there any user-facing changes?

tpcds gains --iterations, --output and --no-partition-cols. It also declares Hive partition columns by default, and it no longer rewrites column names on data that uses the spec names. Ballista's own APIs don't change.

@avantgardnerio avantgardnerio left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think this is sensible, and low risk. My LLM did flag something that may trip you up later though:

Spark's null-partition directory makes the scan fail
  - Spark's partitionBy writes null keys to ss_sold_date_sk=__HIVE_DEFAULT_PARTITION__/. TPC-DS fact tables do have null *_sold_date_sk rows, so data written by Spark will contain
    that directory.
  - tpcds_path_column_type declares any *_sk column as Int32.
  - In DataFusion 55.1, try_into_partitioned_file (datafusion-catalog-listing/src/helpers.rs:370) calls ScalarValue::try_from_string. That calls cast_to, which uses
    DEFAULT_CAST_OPTIONS { safe: false }.
  - So casting "__HIVE_DEFAULT_PARTITION__" to Int32 fails, and every query that scans a Spark-written fact table errors. That's exactly the data the PR is meant to support.
  - The new test path_only_date_sk_partition_is_an_integer_and_prunes only writes non-null partitions, so it can't catch this.
  - Possible fixes:
    a. Map that directory to NULL in Ballista.
    b. Push a fix upstream to DataFusion.
    c. At minimum, document it and add a test that shows the failure.

Spark writes rows whose partition key is NULL to a
`<column>=__HIVE_DEFAULT_PARTITION__/` directory, and TPC-DS fact tables
have rows with a NULL date key. DataFusion casts each directory's value to
the partition column's type, so with `ss_sold_date_sk` declared `Int32`
every query that scanned such a table failed with a cast error
(apache/datafusion#18083).

When a partition column has that directory, register the files as
`<table>_raw` with the column as a string, and register `<table>` as a
view that maps the directory to NULL and casts the other values back.
Filters on the column still skip directories. Tables without that
directory are registered as before.
@andygrove

Copy link
Copy Markdown
Member Author

Good catch. At SF1, 130,093 of the 2,880,404 store_sales rows have a NULL ss_sold_date_sk, and four other fact tables have NULL date keys too. So Spark-written data will always have that directory.

I went with option a in bdd253c. When a partition column has a __HIVE_DEFAULT_PARTITION__ directory, the runner registers the files as <table>_raw with that column as a string. <table> is then a view that maps the directory to NULL and casts the other values back to Int32. Filters on the key still prune directories, and DataFusion pushes the cast into the scan's projection, so the plan barely changes. Tables without that directory register exactly as before.

null_date_sk_partition_reads_as_null covers it. I also partitioned SF1 the way Spark does, with the NULL rows in __HIVE_DEFAULT_PARTITION__, and all 99 queries return the same results as on the original data. The real fix belongs upstream in apache/datafusion#18083, so the view can go once that lands.

@andygrove
andygrove merged commit da31f66 into apache:main Oct 12, 2026
24 checks passed
@andygrove
andygrove deleted the tpcds-benchmark-runner branch October 12, 2026 12:05
@andygrove

Copy link
Copy Markdown
Member Author

Thanks @avantgardnerio. That was a good catch. I filed some follow on issues to improve Hive support for users coming over from Spark.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants