Add A/B benchmark for dynamic filtering on compatible Range-partitioned hash joins
- Dominant language
- Rust
- Stars
- 9.3k
- Forks
- 2.4k
- Avg merge
- 3d 7h
- Merged PRs (30d)
- 344
Description
Part of #22395; follow-up to #23376 and #23854.
### Is your feature request related to a problem or challenge?
#23376 and #23854 add partition-local dynamic filtering for compatible Range-partitioned joins, but DataFusion lacks reproducible performance evidence. Dynamic filtering should help selective joins by reducing probe-side scan work, while non-selective joins may expose filter construction and evaluation overhead.
Stock TPC-H plans are not sufficient unless the measured join actually has compatible Range-partitioned inputs.
### Describe the solution you'd like
Add a reproducible A/B benchmark for a compatible Range-partitioned `HashJoinExec: mode=Partitioned`.
Compare the same data and physical plan with only this setting changed:
1. `enable_join_dynamic_filter_pushdown = true`
2. `enable_join_dynamic_filter_pushdown = false`
The benchmark should:
- use a fixed partition count and fixed `N - 1` split points;
- reuse the identical Range ordering and split vector for both join sides and both A/B arms;
- derive split points before the timed section and report that cost separately if they are not fixed;
- verify the plan contains compatible Range children and no intervening Hash repartition;
- verify both arms return identical results;
- include a selective case where build keys eliminate most probe rows;
- include a low-selectivity or non-selective case to measure overhead;
- include narrow and wide probe projections;
- include a Parquet-backed case with filter pushdown enabled so scan pruning is observable;
- hold partition count, batch size, row-group size, cache policy, and `preserve_file_partitions` constant;
- report median latency and spread, throughput, probe rows/bytes scanned, row groups pruned, and output rows.
The benchmark should characterize the break-even point rather than require every case to improve. A Hash-partitioned join using the same data may be included as a reference baseline.
### Describe alternatives you've considered
Correctness tests establish behavior but do not measure performance. A selective-only microbenchmark hides worst-case overhead.
TPC-H can be added as a follow-up, but each measured join must explicitly configure a compatible, key-specific Range layout. Ordinary `UnknownPartitioning` TPC-H runs do not exercise this feature.
Automatic split-point discovery, DynamicRange planning, and distributed coordination are outside this benchmark's scope.
### Additional context
- Parent epic: #22395
- Design issue: #23376
- Implementation PR: #23854
- Review request: https://github.com/apache/datafusion/pull/23854#pullrequestreview-4857747002
Contributor guide
Research direction
Start by locating the benchmark harness and the partition-local dynamic filtering implementation around HashJoinExec with mode=Partitioned. Build the A/B cases with identical Range ordering and split vectors, then verify compatible Range children and identical results before timing. Done means selective, non-selective, narrow, wide, and Parquet-backed cases report median latency, spread, scan work, pruning, and output rows.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- rust
- Domain
- data-engineering, performance, testing
- Issue type
- Feature
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Quiet
- Clarity
- Mostly clear
- Newbie friendliness
- 48/100