adriangb opened a new pull request, #26442:
URL: https://github.com/apache/datafusion/pull/26442
## Which issue does this PR close?
- Closes #26169.
Depends on apache/arrow-rs#11641 (new `Sbbf` APIs). Until arrow-rs releases
them, this PR patches the workspace to that branch with `[patch.crates-io]` git
dependencies. **The patch must be removed before merge.**
## Rationale for this change
The dynamic filter of a partitioned hash join routes each probe row to its
partition with a `CASE` on the repartition hash. Profiles show that the `CASE`
machinery costs about 6 / 50 / 150 ns per probe row at 1 / 16 / 128 partitions,
and it is most of the query CPU time for join-heavy queries.
Before (16 partitions, `EXPLAIN`):
```text
predicate=DynamicFilter [ CASE hash_repartition % 16
WHEN 0 THEN k@0 >= 3 AND k@0 <= 97 AND hash_lookup
WHEN 1 THEN k@0 >= 1 AND k@0 <= 99 AND hash_lookup
... ELSE false END ]
```
After:
```text
predicate=DynamicFilter [ k@0 >= 1 AND k@0 <= 99 AND bloom_filter ]
```
TPC-H SF100, process CPU time, ratio to `main` (lower is better):
| Query | 16 partitions | 128 partitions |
| --- | --- | --- |
| Q17 | 0.35 | 0.22 |
| Q9 | 0.48 | 0.26 |
| Q5 | ~0.65 | 0.42 |
| Q18 | - | ~0.39 |
- No regressions with `pushdown_filters = false` (the bloom filter is never
built then).
- Synthetic joins: the bloom filter is faster than the exact hash table
lookup on string keys and on joins that are not selective.
Methodology: dedicated GCP `c4a-standard-16` VMs (Arm, 64 GB), TPC-H SF100
parquet. Each config runs 5 A/B rounds with alternating order, plus an A/A
control of `main` against itself in the same run. Ratios are paired per round,
with a bootstrap 95% CI. Results come from the prototype of this design. Final
confirmation runs of this branch are in progress; I will add the numbers here.
## What changes are included in this PR?
One global filter for all probe rows, with no routing:
```text
[key IS NULL OR] key >= global_min AND key <= global_max AND membership(key)
membership = key IN (SET) (all build keys) if total distinct keys <=
hash_join_inlist_pushdown_max_distinct_values
| bloom_filter otherwise (one parquet Sbbf per
key column)
| (none) if the bloom filter is saturated
```
- The filter is true for every build key whatever the partitioning, so it is
used for hash partitioned, range partitioned and `CollectLeft` joins.
- `IS NULL OR` is kept for null-aware (`NOT IN`) and null-equal joins, as
before.
- Fallback to the existing routed `CASE` filter:
- when a build partition was canceled (its keys are unknown, so the routed
filter keeps every row of that route);
- when a key type has no bloom filter support (floats, booleans,
dictionaries, nested types).
- Bloom filter (`BloomFilterExpr`, new):
- Uses the parquet `Sbbf` with parquet hashing (xxHash64 of the plain
encoding, `ArrowWriter` type mapping). A variant with the join hash was within
2-4% end to end, so the parquet-compatible hash is the default.
- Built lazily on the first `evaluate()` (`OnceLock`). A scan that uses
the filter only for statistics pruning never builds it.
- Sized at the barrier for the sum of the per-partition distinct key
counts (exact, because partitions hold disjoint keys), capped at
`hash_join_bloom_filter_max_bytes`, filled from many threads with
`Sbbf::shards_mut` (no atomics, no locks), then folded with
`fold_to_target_fpp(fpp)`.
- Not used when `estimated_fpp() > hash_join_bloom_filter_max_fpp`
(saturated); then only the bounds are tested.
- Serializes to proto as `lit(true)`, like `HashTableLookupExpr` (it holds
runtime build state).
- Pruning: the `KeyRangeBitmap`s of all partitions (#25602) are merged with
`KeyRangeBitmap::try_union` and attached to the bloom filter, so parquet keeps
row group and page pruning, also with `pushdown_filters = false`.
- Merged bounds ignore NULL bounds of partitions that hold only NULL keys.
New options:
| Option | Default | Meaning |
| --- | --- | --- |
| `datafusion.optimizer.hash_join_bloom_filter_fpp` | `0.01` | Target false
positive probability |
| `datafusion.optimizer.hash_join_bloom_filter_max_bytes` | `134217728` (128
MiB) | Maximum filter size per key column |
| `datafusion.optimizer.hash_join_bloom_filter_max_fpp` | `0.15` | Do not
use a saturated filter |
`hash_join_inlist_pushdown_max_distinct_values` (default 150) is now also
the limit for the global InList (sum over all partitions).
Other changes:
- `datafusion-physical-plan` now depends on `parquet` (for `Sbbf`).
- `unnest.rs`: `value_offset` → `value_offset_at` (deprecated on arrow-rs
`main`).
## What is the testing strategy for this PR?
- `bloom_filter.rs` unit tests: no false negatives, few false positives,
NULL probe keys, multi-column keys, parallel fill equals serial fill, saturated
filter passes every row, proto encodes as `lit(true)`.
- `shared_bounds.rs` unit tests: global bloom filter shape and results,
InList vs bloom at the global limit, range partitioning uses the global filter,
canceled partition and unsupported key type fall back to the routed filter,
merged pruning bitmap, merged bounds ignore NULL bounds. Existing null-aware /
null-equal / canceled tests pass.
- `key_range_bitmap.rs`: `try_union` never drops a key.
- `dynamic_row_group_pruning.slt`: new `EXPLAIN ANALYZE` of a partitioned
join with `pushdown_filters = false`: the global filter, and 2 of 4 row groups
pruned by the merged bitmap.
- Updated snapshots: `filter_pushdown.rs`, `push_down_filter_parquet.slt`,
`range_partitioning.slt`, `dynamic_row_group_pruning.slt`,
`information_schema.slt`.
Known failures caused by the arrow-rs `main` patch, not by this change (they
go away when the patch goes away): `datetime/timestamps_timezone.slt` (DST
handling), `spill::tests::test_gc_*`, and `clippy::mutable_key_type` in
`datafusion-optimizer`.
## Are there any user-facing changes?
- `EXPLAIN` output of hash join dynamic filters changes (no `CASE
hash_repartition`, `bloom_filter` instead of `hash_lookup`).
- Three new configuration options (see above). No breaking API changes.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
--
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]
