andygrove opened a new pull request, #2553: URL: https://github.com/apache/datafusion-ballista/pull/2553
# Which issue does this PR close? No issue was filed. The gap is called out in the SF1000 [benchmarking notes](https://github.com/apache/datafusion-ballista/blob/main/docs/source/contributors-guide/benchmarking.md#environment): the data is Hive-partitioned, but the `tpch` runner registers the tables without partition columns. # Rationale for this change The SF1000 data on S3 is partitioned by date (`lineitem/l_shipdate=YYYY-MM-DD/`, `orders/o_orderdate=YYYY-MM-DD/`). Spark, Comet and Trino declare those partition columns, so a date filter skips whole directories. The `tpch` runner registers each table with `register_parquet(.., ParquetReadOptions::default())`, which declares none. Every job therefore lists every file, the scheduler reads every footer to collect statistics, and every file is assigned to a task. Only the executors' statistics pruning skips the files that don't match. Ballista already supports partition pruning. The client sends the partition columns in the logical plan, and the scheduler's `ListingTable::scan` drops partitions that don't match before it collects statistics. Nothing tested that path, though, so this PR adds tests for it. I ran all 22 queries locally at SF10, with data laid out like the SF1000 data: one directory and one file per date for `lineitem` and `orders`, with the date columns also stored in the files. `customer` is partitioned by `c_mktsegment` and `part` by `p_brand`. The cluster was one scheduler and one executor with 16 vcores, and the runner used `--partitions 16`. Times are the fastest of 3 iterations, averaged over two runs in opposite orders. Every query was verified against DataFusion in both modes. | Query | Without partition columns (s) | With partition columns (s) | Change | | ----: | ----------------------------: | -------------------------: | -----: | | 1 | 0.383 | 0.360 | -6% | | 2 | 0.141 | 0.144 | +2% | | 3 | 0.789 | 0.329 | -58% | | 4 | 0.613 | 0.452 | -26% | | 5 | 0.574 | 0.458 | -20% | | 6 | 0.228 | 0.097 | -58% | | 7 | 0.955 | 0.756 | -21% | | 8 | 0.606 | 0.491 | -19% | | 9 | 0.948 | 0.962 | +2% | | 10 | 2.471 | 2.144 | -13% | | 11 | 0.064 | 0.065 | +1% | | 12 | 0.494 | 0.511 | +3% | | 13 | 0.413 | 0.427 | +3% | | 14 | 0.245 | 0.075 | -70% | | 15 | 0.567 | 0.168 | -70% | | 16 | 0.103 | 0.106 | +2% | | 17 | 1.066 | 1.007 | -6% | | 18 | 2.434 | 2.450 | +1% | | 19 | 0.403 | 0.409 | +1% | | 20 | 0.487 | 0.345 | -29% | | 21 | 1.924 | 2.064 | +7% | | 22 | 0.383 | 0.393 | +3% | | **Total** | **16.29** | **14.21** | **-13%** | Queries that filter on `l_shipdate` or `o_orderdate` are faster. Q12 isn't, because it filters on `l_receiptdate`. The other queries move by a few percent either way between runs, except that planning takes about 4.5 ms longer per job (73.8 ms against 69.3 ms). DataFusion parses and checks the partition values of every file, and there are about 5,000 files here. A local disk understates the benefit, because on S3 each footer read the scheduler skips is a round trip. I'll refresh the SF1000 results in a separate PR. The local runs used 128 MiB gRPC message limits, as the SF1000 setup does. With the 16 MiB defaults, Q10 fails on this many-file layout with or without this change, because its last stage writes about 26 MB for 20 result rows. # What changes are included in this PR? - **`tpch` runner:** `benchmark ballista`, `benchmark datafusion`, the `--verify` oracle and `loadtest ballista-load` now register the Hive partition columns of Parquet tables. They are found from the `col=value` directory names with `ListingOptions::infer_partitions`. - A partition column that the files also store keeps the files' type and is removed from the file schema. `ListingTable` appends partition columns itself and rejects duplicate names. - A column that only exists in the directory names, as Spark's `partitionBy` writes it, takes its type from the TPC-H schema. That keeps `l_shipdate` a `DATE`. - Flat data, such as `tpchgen-cli` output, registers as before. - `--no-partition-cols` restores the old registration for A/B runs. - **Tests:** - New `ballista/client/tests/partition_pruning.rs` runs end to end on standalone and remote contexts, with AQE on and off. One partition holds a file that isn't Parquet, so a job that reads it fails. The cases are a date range, an equality on the partition column plus a filter on a data column, and a filter that matches no partition (the aggregate must still return its row). A guard case checks that an unpruned scan fails. With filter pushdown turned off, every pruning case fails with `Invalid Parquet file. Corrupt footer` while the scheduler plans the job. - Unit tests for the runner cover both layouts, a string partition column, flat data, the flag, and planning only the matching partition's files. - **Docs:** `benchmarks/README.md` gets a "Hive-partitioned data" section. The SF1000 notes now say that the published results predate partition columns. # Are there any user-facing changes? On Hive-partitioned Parquet data, the `tpch` runner now declares the partition columns. Date-filtered queries get faster, and the partition columns move to the end of those tables' columns. `--no-partition-cols` keeps the old behaviour. Ballista's own APIs don't change. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
