zhuqi-lucas opened a new pull request, #23690: URL: https://github.com/apache/datafusion/pull/23690
## Which issue does this PR close? None yet — filing as **draft for discussion** before opening a tracking issue. Happy to open one if reviewers want to consolidate follow-up work behind it. ## Rationale for this change `LimitedBatchCoalescer` currently hardcodes the bypass threshold at `target_batch_size / 2`. Batches at or below this size get merged into `target_batch_size` chunks via `arrow::compute::BatchCoalescer`; only batches larger than half the target skip coalescing. The SIGMOD 2025 paper [_Data Chunk Compaction in Vectorized Execution_](https://doi.org/10.1145/3709676) by Qiao and Zhang (Tsinghua) analyzes this trade-off across five compaction strategies and explicitly identifies the **Full Compaction** approach — which the authors attribute to Apache DataFusion — as the worst-case strategy for pipelines with high Chunk-Reducing Factor or wide payloads, because the fixed and per-tuple memcpy cost outweighs the interpretation savings for downstream operators. Their **Binary Compaction** alternative uses a much lower threshold (`alpha = 128` in DuckDB). The paper reports end-to-end wins on DuckDB of **+11.8%** on the Join Order Benchmark, **+6.1%** on TPC-DS, and **+4.6%** on TPC-H, with individual queries reaching **up to +63%** when the CRF is high. TPC-H alone shows the smallest average lift because its join pipelines are shallow; deep-join workloads benefit most. ## What changes are included in this PR Minimal API surface. `LimitedBatchCoalescer` gains a second constructor: ```rust pub fn new_with_bypass_threshold( schema: SchemaRef, target_batch_size: usize, fetch: Option<usize>, bypass_threshold: Option<usize>, // None => target_batch_size / 2 (legacy) ) -> Self ``` Passing `None` reproduces the current behavior byte-for-byte. Passing a small value (e.g. `Some(128)`) enables Binary Compaction semantics — mid-sized batches pass through instead of being memcpy'd into a larger buffer. No existing call sites change. A `ConfigOptions` surface can be added in a follow-up once the community agrees on the config shape. ## Microbenchmark A new `coalesce_threshold` bench feeds 5 000 batches of a given size through the coalescer at four threshold settings. Wall time on x86_64 laptop, narrow (2-column int32) input: | input rows / batch | default (target/2 = 4096) | target/8 = 1024 | target/32 = 256 | target/64 = 128 | | ---: | ---: | ---: | ---: | ---: | | 64 | 725 μs | +2.0% | +1.5% | +1.1% | | 256 | 999 μs | +0.3% | −1.5% | **−17.5%** | | 1024 | 1 697 μs | +2.1% | **−33.5%** | **−33.6%** | | 4096 | 5 032 μs | **−42.6%** | **−42.9%** | **−44.4%** | For wider inputs (11 columns, mixed Int32 + Utf8) the effect shrinks to single-digit percentages but stays non-negative — at 1024 rows/batch the `target/64` case is **6.1% faster**, at 4096 rows/batch **5.0% faster**, with no scenario regressing more than 3%. The 4096-row case shows the largest win because 4096 sits exactly at the current default threshold, so every input batch hits the copy path. Lowering the threshold lets those medium batches bypass entirely. ## Expected impact on end-to-end benchmarks Based on the paper's DuckDB numbers, ordered by expected win: - **Join Order Benchmark**: +10–30% average, individual queries up to +60%. Deep join pipelines with high CRF. - **TPC-DS**: +5–10% average, larger on Q64-style chained hash joins. - **TPC-H**: +3–8% average — smaller because join pipelines are shallow. - **ClickBench**: near-zero or micro-regression. Almost no hash joins; the only plausible beneficiary is Q23 (`URL LIKE '%google%'` + ORDER BY LIMIT) where post-filter batches are tiny, and even there the pending [Adaptive Filter Scheduling work](https://github.com/apache/datafusion/issues/20324) is likely to prevent the small batches from ever appearing. I would love to have community members run TPC-H / JOB with `Some(128)` vs `None` and post numbers — I do not have production hardware for a fair end-to-end comparison. ## Are these changes tested? Yes. - Added `test_coalesce_low_bypass_threshold_passthrough` to verify that mid-sized input batches bypass the coalescer when the new argument is `Some(16)`. - Added `test_coalesce_default_bypass_threshold_merges` to lock in legacy behavior — same input merges into a single output when the new argument is `None`. - Existing coalesce tests continue to pass. - `cargo fmt` and `cargo clippy` clean. ## Explicit non-goals / follow-ups 1. **ConfigOptions surface** — I did not wire `datafusion.execution.coalesce_bypass_threshold` yet, to keep the diff small and let reviewers steer the config shape (single global? per-operator? computed from `target_batch_size`?). 2. **Multi-Armed-Bandit adaptive alpha** — the paper's UCB-based per-operator learner would be a larger, separate change. This PR is deliberately a threshold *plumbing* change, not the adaptive algorithm. 3. **Logical Compaction for hash join** — the paper's second contribution (extended selection vectors to compact join output without data copy) is a hash-join-operator change, separate scope. 4. **arrow-rs default** — `arrow::compute::BatchCoalescer::with_biggest_coalesce_batch_size` already takes an `Option<usize>`; only DataFusion's wrapper picks `target/2`. This PR does not touch arrow-rs. ## Are there any user-facing changes? No behavior change unless callers pass a non-`None` bypass threshold to the new constructor. Legacy `new()` behavior is byte-identical. Once a follow-up wires a `ConfigOptions` value, users would be able to tune the threshold via a session config for their workload. ## References - SIGMOD 2025 paper: [Data Chunk Compaction in Vectorized Execution](https://doi.org/10.1145/3709676) (Qiao & Zhang, Tsinghua) - DuckDB implementation reference: [duckdb/duckdb#14956](https://github.com/duckdb/duckdb/pull/14956) - Adjacent DataFusion work on adaptive filter scheduling: [apache/datafusion#20324](https://github.com/apache/datafusion/issues/20324) -- 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]
