adriangb opened a new pull request, #24236:
URL: https://github.com/apache/datafusion/pull/24236
## Which issue does this PR close?
<!-- No existing issue; this is a small standalone perf fix found while
reading the hash join InList pushdown path. Happy to file one if maintainers
prefer. -->
- N/A
## Rationale for this change
The hash join InList filter pushdown is gated on the build side's
**distinct** key count:
```rust
map.num_of_distinct_key() >
config.optimizer.hash_join_inlist_pushdown_max_distinct_values // default 150
```
but the values it actually ships are the raw build-side key arrays — **one
entry per build row**, never deduplicated. So a partition that passes the
150-distinct-value gate can still emit a list with thousands of entries,
bounded only by `hash_join_inlist_pushdown_max_size` (default 128 KiB).
On TPC-H q17 at SF=1 (12 partitions) the pushed-down filter on the
`lineitem` scan looked like this before the change:
```
DynamicFilter [ CASE hash_repartition % 12
WHEN 0 THEN l_partkey >= 33299 AND l_partkey <= 197255 AND
l_partkey IN (SET) ([ ...551 values... ])
WHEN 1 THEN ... ([ ...351 values... ])
... ]
```
**6088 list entries in total across the 12 partitions, for 204 distinct
keys.**
That length is paid for repeatedly and mostly discarded:
* `InListExpr::try_new_from_array` materialises one `ScalarValue` + one
`Literal` expression per entry, so the `list` field is ~30× larger than it
needs to be.
* The list becomes a `LiteralGuarantee`, which `PruningPredicate::prune`
re-materialises **per row group** wherever the pushed-down filter reaches a
pruning predicate. `RowGroupPruningStatistics::contained()` in
`datafusion/datasource-parquet/src/row_group_filter.rs` returns `None`
unconditionally, so that guarantee can never prune anything — the work is
provably discarded.
* Separately, `max_in_list_size` (default 20) blocks the InList→OR rewrite,
so an inflated list yields no pruning benefit on that path either.
The duplicates carry no information: `IN` is a set membership test.
## What changes are included in this PR?
`build_struct_inlist_values` now deduplicates the value array before it is
wrapped in `PushdownStrategy::InList`.
Deduplication is done on the arrow array itself: encode it once with
`RowConverter` and keep first occurrences via a `HashSet<Row>`, then a single
`take`. This is one pass and two allocations, rather than one
`ScalarValue::try_from_array` per row, and it handles both shapes the pushdown
produces — a single key column (possibly dictionary-encoded) and a multi-column
`StructArray` — without special-casing either.
Semantics are preserved exactly, and the transform is a **set-preserving
no-op**:
* Deduplication is on the whole tuple, so multi-column keys whose individual
columns repeat but whose tuples are distinct are untouched.
* NULLs compare equal under the row format, so a list with many NULLs
collapses to a single NULL. Under three-valued logic one NULL in the haystack
already produces the same result as a thousand, so `IN`/`NOT IN` results are
unchanged.
* First-occurrence order is preserved, `take` preserves dictionary encoding,
and the array is returned untouched when it has <2 rows, contains no
duplicates, or has a type `RowConverter` cannot encode.
## Are these changes tested?
Yes.
* New unit tests in `inlist_builder.rs` cover: single-column dedup, an
already-distinct array (identity), NULL collapsing, multi-column tuple dedup,
distinct tuples built from individually-duplicated columns, and dictionary
arrays.
* A new sqllogictest in `push_down_filter_parquet.slt` covers it end to end:
a build side with 6 rows and 2 distinct keys now pushes `IN (SET) ([11, 22])`
with `required_guarantees=[id in (11, 22)]`, where `main` pushed all 6 rows. It
also asserts the join output is unchanged (still one row per matching build
row).
* All existing `IN (SET)` snapshots in `push_down_filter_parquet.slt` and
`filter_pushdown.rs` are byte-identical — their build sides were already
distinct, which is the expected no-op.
`cargo test -p datafusion-physical-plan --lib`, the full sqllogictest suite,
`cargo clippy --all-targets -- -D warnings` (CI feature set) and `cargo doc`
are all green.
### Measured effect
Counts first, since they are exact rather than timings. TPC-H SF=1, parquet,
from `dfbench --debug`, summing the `IN (SET)` list lengths across the 12
partitions of the q17 dynamic filter:
| query | InList entries (main) | InList entries (this PR) |
|---|---|---|
| q17 | 6088 | 204 |
| q18 | 57 | 57 |
| q3 | no InList pushdown | no InList pushdown |
| q5 | 26 | 26 |
| q9 | 25 | 25 |
Only q17 has a build side with duplicate join keys at SF=1; everywhere else
the build keys are already distinct (group-by outputs, `nation`/`region` keys)
and the change is a literal no-op. That is the expected shape: this only helps
when the build side repeats keys.
Timings: both binaries built from the **same worktree and target dir**,
differing only by this commit. 20 rounds × 6 iterations, A/B order
counterbalanced per round (A,B on odd rounds; B,A on even), per-round median
over iterations 1–5 (cold iteration dropped), then median across rounds. TPC-H
SF=1 parquet on a machine that was **not** idle.
| query | main (ms) | this PR (ms) | delta | odd rounds (A,B) | even rounds
(B,A) | rounds favouring PR |
|---|---|---|---|---|---|---|
| q17 | 62.25 | 59.20 | **−3.8%** | −4.2% | −3.3% | 18/20 (sign test
p≈0.0004) |
| q18 | 58.25 | 59.45 | −2.3% | −2.3% | −2.9% | 12/20 (p≈0.50) |
| **q1 (control)** | 38.20 | 39.10 | **+2.4%** | +2.6% | +1.2% | 6/20 |
| **q6 (control)** | 14.80 | 14.80 | **−0.0%** | −0.0% | +0.6% | 10/20 |
**Honest reading of that table.** q1 and q6 have no joins, so this change
cannot touch them; whatever they show is the noise floor, and here that floor
is up to ±2.4%. q17's −3.8% is only just clear of it. The reason I believe it
is real rather than noise is the sign test, not the magnitude: 18 of 20
counterbalanced rounds favoured the change, both orderings agree in sign and
size, and the controls sit at 6/20 and 10/20 as they should. q18's −2.3% is
inside the floor and 12/20 is a coin flip — and its list length does not change
at all, so that column is noise by construction, which is a useful internal
check on the method.
Per-round spread is wide (individual q17 rounds ranged −44% to +22%), which
is why only the 20-round medians and the sign test are quoted.
A first, lower-powered pass (12 rounds × 5 iterations over q17/q18/q3/q5/q9
+ q1/q6) produced a ±9% control floor and showed nothing above it in either
direction. It is reported here for completeness, not as evidence.
I would also not over-claim the pruning half of the mechanism from this
benchmark: at SF=1 each table is a single file and the dynamic filter arrives
after row-group pruning has already run (`row_groups_pruned_dynamic_filter=0`,
`statistics_eval_time=24ns` on that scan), so on this dataset the discarded
`LiteralGuarantee` work is not actually being paid. The saving I measured at
SF=1 is the ~5900 `ScalarValue`/`Literal` allocations that no longer happen.
The guarantee argument should matter more on datasets with many row groups per
scan, where the filter is live before pruning — I have not measured that here.
If reviewers consider −3.8% on one query too thin to justify the change on
performance grounds alone, the correctness-of-effort argument still stands on
its own: the deduplicated list is semantically identical, strictly smaller, and
the work it removes is work that could never have produced a result.
## Are there any user-facing changes?
No API changes and no behaviour changes. The only visible difference is in
`EXPLAIN ANALYZE` output, where the pushed-down `IN (SET) (...)` list and the
`required_guarantees` derived from it no longer repeat duplicate build-side
keys.
--
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]
