Repository navigation
feat(bench): time TPC-DS queries and share the runner code with TPC-H - #2565
Conversation
avantgardnerio
left a comment
There was a problem hiding this comment.
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.
|
Good catch. At SF1, 130,093 of the 2,880,404 I went with option a in bdd253c. When a partition column has a
|
|
Thanks @avantgardnerio. That was a good catch. I filed some follow on issues to improve Hive support for users coming over from Spark. |
Which issue does this PR close?
No issue was filed.
Rationale for this change
The
tpcdsbinary 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 waytpch benchmark ballistabenchmarks TPC-H. It also doesn't fit TPC-DS data generated by other tools:tpcgen-cliuses (ib_income_band_id,r_reason_description,cr_return_amount_inc_tax). Data fromdsdgenor Spark uses the spec names that the queries already use, so the rewrite breaks q64, q81, q84, q85 and q93 on that data.partitionByonly stores the fact tables' date keys (ss_sold_date_skand 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 theballista-benchmarkslibrary 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?
benchmarks/src/lib.rs, newbenchmarks/src/summary.rs):summary::{BenchmarkRun, QueryRun, QueryResult}, which are unchanged, plusrun_suiteandtime_query.run_suiteruns the queries in order, records each one's iterations or error, writes<name>-<start_time>.jsonatomically 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 datafusionandtpcdsall use it.time_queryis the timed iteration loop that wasrun_local_queryintpch.rs.tpch benchmark datafusionandtpcdsuse it.tpch benchmark ballistakeeps its own loop, which also plans each statement and checks--expected, so its timings don't change.benchmark_session_configbuilds the session config: S3 support, target partitions, batch size, statistics collection and-c key=valueoverrides.ballista_contextconnects a session with that config and a job name to the scheduler.local_contextbuilds a single-process session that can read S3.parquet_table_layoutmoves out oftpch.rs. It now takes aPathColumnTypefunction 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.ParquetTableholds a table's path, file schema and partition columns.infer_parquet_tablesinfers them for several tables concurrently, andregister_parquet_tablesregisters them on a session without reading any files.register_parquet_table(infer, then register) is whattpchuses.<column>=__HIVE_DEFAULT_PARTITION__/directory, which DataFusion can only read as a string (apache/datafusion#18083).ParquetTablefinds such columns when it infers the layout. It then registers the files as<table>_rawwith 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_filefinds<dir>/qN.sqlrelative to the current directory or the repository root. Both runners use it.tpcdsrunner:--iterations,--outputand--no-partition-colsoptions, which mean what they mean fortpch. The summary uses the TPC-H format, sotpch comparereads it.--verifyoracle register the cached layouts.tpcgen-cliname and not the spec name. That is decided once, from the inferred layouts. CI'stpcgen-clidata still gets all three.Int32if 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 toss_sold_date_sk=__HIVE_DEFAULT_PARTITION__/and so on, read as NULL.tpch.tpchrunner: 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.QueryRuntests move with the struct. New tests coverrun_suite(failures are recorded and the suite carries on, the summary file name and contents, and no output directory), which renames apply totpcgen-cliand to spec column names, a Spark-stylestore_sales/ss_sold_date_sk=N/table that registers the key asInt32and 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.benchmarks/README.mddescribes the benchmark mode, partition handling and the conditional renames.Are there any user-facing changes?
tpcdsgains--iterations,--outputand--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.