apache / apache/datafusion

Add A/B benchmark for dynamic filtering on compatible Range-partitioned hash joins

Open
#24,095 0 comments 0 reactions 0 assignees View on GitHub
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.