stuhood opened a new issue, #25439: URL: https://github.com/apache/datafusion/issues/25439
### Is your feature request related to a problem or challenge? Range partitioning landed in DataFusion 55 (#22395), establishing the foundation for range-partitioned physical execution plans. Continued optimizations are underway (epic #25421) covering: * Symmetric co-partitioned joins with zero shuffle * Asymmetric stream adaptation (adapting an unpartitioned stream against a range-partitioned reference table) * Partition elasticity and stage scaling (such as #24712 down-sampling split points across pipeline stages) * Multi-way joins and hybrid pipelines preserving range metadata across broadcast and partitioned operators * Partition-aware dynamic filtering (#23817) However, DataFusion currently lacks an end-to-end macro benchmark in `benchmarks/` to measure and validate these range-partitioned execution paths against traditional hash-partitioned and broadcast pipelines. While #24095 tracks a specific dynamic filter A/B benchmark, and `datafusion/physical-plan/benches/range_repartition.rs` micro-benchmarks the isolated CPU routing loop of `BatchPartitioner`, there is no end-to-end benchmark suite exercising the full vertical stack (table scan -> physical optimizer -> multi-threaded Tokio execution -> memory pools). An end-to-end benchmark is needed to verify that the physical optimizer reliably avoids shuffles for pre-partitioned data, track peak memory and throughput gains across commits to prevent regressions, and provide reproducible numbers for the upcoming Range Partitioning blog post (#24667). ### Describe the solution you'd like Add an end-to-end macro benchmark suite in `benchmarks/` (e.g. extending `dfbench` or adding `benchmarks/src/range_join.rs`) that models the multi-table range join and stage scaling scenarios explored in `datafusion-distributed#734` (https://github.com/paradedb/datafusion-distributed/blob/adf49d1df76111e436007ba87497a4f812539167/tests/range_partitioning.rs), while sharing harness architecture and query shapes with `benchmarks/src/hj.rs`. #### 1. Three-Table Dataset Shape To model multi-way joins, stream adaptation, and broadcast interactions, the benchmark should generate and register three tables (for example, inspired by the schema in `tests/range_partitioning.rs` in `datafusion-distributed`): * `fact`: Large streaming fact table (e.g. key, timestamp, value, payload), range-partitioned by key (or composite key) across N partitions. * `dim`: Dimension table (e.g. key, attributes), configurable as either range-partitioned with split points matching `fact`, or unpartitioned for stream adaptation. * `lookup` / `services`: Small dimension table (e.g. category, label), suitable for broadcast joins or secondary hash shuffles. #### 2. Benchmark Scenarios to Cover Model the core range join scenarios derived from `datafusion-distributed#734`: 1. Symmetric 2-way range join: `fact` (range) JOIN `dim` (range) on matching split points, measuring task-local execution (0-shuffle) vs. baseline hash repartitioning (`Hash([key], N)`). 2. Asymmetric stream adaptation: `fact` (range) JOIN `dim` (unpartitioned stream), measuring 1-sided stream adaptation against 2-sided repartitioning. 3. Concurrency mismatch and stage scaling: `fact` and `dim` partitioned into N splits executed with worker concurrency < N (under-parallelism) and > N (over-parallelism), evaluating sample down-sampling (#24712) versus re-shuffling. 4. 3-way join with Range -> Hash transition: `fact` JOIN `dim` on range key, followed by JOIN `lookup` on a secondary hash key, measuring pipeline transition when range partitioning cannot be preserved. 5. 3-way hybrid join with broadcast dimension: `fact` JOIN `dim` on range key with `lookup` broadcast, verifying that range partitioning guarantees are preserved across the multi-way join pipeline. 6. Dynamic filter pushdown: Measuring probe-side Parquet scan pruning when dynamic min/max bounds are targeted to matching range partitions. #### 3. Code and Pattern Sharing with `benchmarks/src/hj.rs` Rather than creating an isolated runner from scratch, attempt to share common execution and measurement infrastructure with `benchmarks/src/hj.rs`: * Harness options: Reuse `CommonOpt` (iterations, batch size, target_partitions, thread runtime configuration). * Metrics collection: Track elapsed time, throughput (rows/sec), and peak memory usage via `PeakRecordingPool` from `ctx.runtime_env().memory_pool`. * Plan verification / isolation: Adopt the pattern in `hj.rs` of setting `hash_join_single_partition_threshold = 0` to isolate partitioned join execution and assert expected physical plan structures. * Query parameterization: Adopt `hj.rs`'s parameterization of key density, hit rates, and fanout across the range join query suite. #### 4. Tooling and Data Generation Integration * Add a data generator step in `benchmarks/bench.sh` (e.g., `bench.sh data range_join`) that writes pre-partitioned Parquet files for the three tables. * Add a runner command in `benchmarks/bench.sh` (e.g., `bench.sh run range_join`) outputting results in standard JSON format compatible with `benchmarks/compare.py` for comparing against `main`. ### Describe alternatives you've considered * Relying solely on Criterion micro-benchmarks: While Criterion is well-suited for isolated CPU algorithms like `BatchPartitioner` in `datafusion/physical-plan/benches/range_repartition.rs` or split-point quantile derivation in #24712, it constructs physical plans directly and cannot test optimizer decisions, memory pool interactions, Parquet I/O pruning, or multi-threaded Tokio execution pipelines. * Relying exclusively on downstream tests and benchmarks in `datafusion-distributed#734`: While `datafusion-distributed` exercises these queries and has its own distributed benchmarks, having first-class macro benchmarks in Apache DataFusion ensures regression detection directly in core CI/nightly runs and supports publication of official benchmarks for #24667. ### Additional context * Part of epic #25421. * References integration patterns in datafusion-contrib/datafusion-distributed#734. * Complements #24095 and supports evidence for #24667. -- 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]
