adriangb commented on PR #24086: URL: https://github.com/apache/datafusion/pull/24086#issuecomment-5185461558
# Benchmark summary All GKE runs are adriangbot on `c4a-highmem-16`, compared against the PR's merge-base, 100MB budget. `SIMULATE_LATENCY` injects per-request object-store latency; without it the bot reads local NVMe. Wall times are reported in 5s steps, so on a 50s benchmark one bucket is ±10% — I re-ran rather than interpret single-bucket deltas (one apparent tpcds regression turned out to be exactly that). ## With simulated latency Commit `7f3e448` (final shape — no scheduler-side coalescing): | suite | baseline | `streaming` | speedup | peak memory | per-query | |---|---|---|---|---|---| | tpch_sf10 | 635.1s | **100.0s** | **6.35x** | 4.1 → 4.3 GiB | 22F / 0S / 0N | | tpch_sf1 | 110.0s | **65.0s** | **1.69x** | 858 → 805 MiB | 21F / 0S / 1N | | tpcds_sf1 | 380.1s | 380.1s | 1.00x | 1.0 GiB → **836 MiB** | 11F / 5S / 83N | | clickbench_partitioned | 450.1s | 445.1s | 1.01x | flat | 2F / 0S / 41N | `pipelined`, measured earlier at `5ff7af6`: | suite | baseline | `pipelined` | speedup | |---|---|---|---| | tpch_sf10 | 635.1s | 130.0s | 4.9x | | tpch_sf1 | 115.0s | 80.0s | 1.44x | | clickbench_partitioned | 445.1s | **335.1s** | **1.33x** | | tpcds_sf1 | 380.1s | 380.1s | 1.00x | Neither policy dominates. `streaming` is far ahead on TPC-H and is the only one that reduces memory; `pipelined` is the only one that helps clickbench (see below for why). Both are >= baseline everywhere. ## Without simulated latency (local NVMe) Every suite neutral on wall time for both policies — per-batch scheduling costs nothing on fast storage — with memory consistently lower for `streaming`: tpcds peak 2.1 → 1.4 GiB, tpch10 peak 4.7 → 3.9 GiB, tpch avg −18%. ## ClickBench: the files have no page index clickbench reported *all 43 queries* as "no change", which is the signature of a code path not executing rather than executing without benefit. It isn't: the published `hits_*.parquet` files from `datasets.clickhouse.com` (what `bench.sh` downloads) contain **no page index at all** — 0 of 105 columns in row group 0 carry an offset index or column index. Verified with parquet-rs itself (`ArrowReaderMetadata::load` with `with_page_index(true)`, then `metadata.offset_index()`), by range-fetching just the file tail. Note pyarrow's `ColumnChunkMetaData.offset_index_offset` is *not* reliable for this — it reported 0 even for files that demonstrably have one. So page-granular planning cannot run there by construction, and only row-group-granular prefetch can help — which is exactly why `pipelined` gets 1.33x on clickbench and `streaming` gets nothing. It also means page-index pruning is inert on this dataset. **Adding a page index changes the answer completely.** I downloaded 3 of the 100 files, rewrote them with `write_page_index=True` (same row-group structure — 2 row groups, 450,560 rows in RG0; file sizes within 1.5%), and ran all 43 queries against both copies at 50ms latency, 8 partitions, medians of 2: | dataset | `off` | `pipelined` | `streaming` | |---|---|---|---| | as published (no page index) | 5449 ms | 5422 ms (1.00x) | 5419 ms (1.01x) | | **same data + page index** | 5471 ms | — | **3549 ms (1.54x)** | Per query on the page-index copy: **35 faster, 0 slower, 1 neutral**. The `off` baseline is within 1% across the two copies, so the rewrite did not materially change decode cost — the whole difference is what `streaming` can do once an offset index exists. The mechanism is *not* fetching less. For most queries the GET count and bytes fetched are identical between `off` and `streaming` (q20 and q33: 10 GETs and 81.3MB either way, yet 200 → 145ms and 230 → 186ms). The win is intra-row-group overlap: ClickBench row groups hold 450k rows (~52MB compressed), so row-group-granular readiness means waiting for all of it before decoding starts, while batch-granular readiness starts decoding as soon as the first pages land. Caveats: 3 of 100 files, so absolute numbers do not extrapolate; `pipelined` shows no benefit in this local setup (1.01x) unlike the bot's 1.33x, because 3 files across 8 partitions with 2 row groups each leaves it almost no cross-row-group work to overlap; and q36 fails on both copies for an unrelated pre-existing cast issue. ## Synthetic: where the ceiling is A 300MB file written as a single 8M-row row group — the case row-group-granular policies cannot help at all (50ms latency, 100MB window): | policy | scan total | time-to-first-batch | peak staged | |---|---|---|---| | off / batched / pipelined | ~845ms | ~700ms | 283.5MB | | **streaming** | **708ms** | **112ms** | **99.9MB** | Same effect the ClickBench page-index experiment later reproduced on real data. On a 4 × 71.5MB row-group file, `streaming` at a 20MB window is the only policy that respects a budget smaller than one row group (19.7MB peak) — at a throughput cost, which is the dial. ## Hypotheses that measurement killed Recorded because the wrong ones cost real time: - **"clickbench's gap is scattered GETs."** Added gap-merging; it collapsed 193 GETs → 15 on a local filtered scan but moved clickbench zero. Falsified. - **"clickbench regresses because streaming bypasses TopK dynamic pruning."** That would show *some* queries slower; all 43 were unchanged. Falsified. - **"The +10% bytes is gap bytes being discarded then re-fetched."** Implemented the fix; it recovered 7MB of 157MB (~0.4%). The real cause was the deliberate over-fetch of coalescing itself. Mostly falsified — the fix is right in principle, but it was not the cause. - **"A 4MB coalesce gap is a reasonable default."** It merged away 59 requests on clickbench but pulled 157MB of unprojected columns with them and ran *slower* (q22: 204ms/236MB vs 170ms/212MB at 1MB). Removing scheduler-side coalescing entirely beat every setting of it — 1.54x vs 1.53x at 1MB vs 1.51x at 4MB — and also fixed a memory regression it was causing (tpch peak went from +12% over baseline to −6%). That last one has a layering lesson worth stating: `ObjectStore::get_ranges` already coalesces within 1MB for every store using the default implementation (S3, GCS, Azure), while `LocalFileSystem` and the other overriding stores coalesce not at all. A scheduler-side merge can therefore only *raise* the effective threshold, never lower it, and cannot express "leave this alone" on a real object store. If a scheduler-side merge is wanted later it should be driven by characteristics the store reports — round-trip cost, bandwidth, useful concurrency — rather than a constant, so the two layers stop duplicating a decision neither can make alone. ## Reproducing Local experiments used a benchmark harness with an `ObjectStore` wrapper injecting per-request latency plus a per-connection bandwidth cost, reporting wall time, time-to-first-batch, GET count, bytes fetched and peak staged bytes. Happy to share it if useful. -- 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]
