jayzhan211 opened a new pull request, #26703:
URL: https://github.com/apache/datafusion/pull/26703
## Which issue does this PR close?
- Part of #26696.
## Rationale for this change
Batches often carry a dictionary with far more values than rows: each
`batch_size` slice of a GROUP BY's output keeps every group's value, and
repartitioning and filtering keep the whole dictionary too. Code that processes
every dictionary value then does work proportional to the dictionary, not the
batch, so queries slow down as the group count grows.
On ClickBench `hits_partitioned` with
`datafusion.execution.parquet.enable_rle_to_dictionary = true`:
```sql
SELECT lower(s), c FROM (
SELECT "SearchPhrase" s, COUNT(*) c FROM hits WHERE "SearchPhrase" <> ''
GROUP BY s
) ORDER BY c DESC LIMIT 10;
```
takes 12581 ms on main and 1019 ms with this PR: `lower` runs over every
group's value once for each 8192-row output slice. Hashing has the same
problem: the hash repartition after a partial aggregate hashes every group's
value for each slice (#26696).
## What changes are included in this PR?
Code that processes every value of a dictionary now handles at most
`MAX_DICTIONARY_VALUES_PER_ROW` (8) values per row. Past that, it first takes
the values the rows reference (`take(values, keys)` with keys `0..len`), which
costs O(rows).
- `datafusion_common::utils::map_dictionary_values(array, f)` applies `f` to
a dictionary's values within that bound. The string functions that keep
dictionaries (`ascii`, `btrim`, `ltrim`, `rtrim`, `lower`/`upper`,
`character_length`, `initcap`, `reverse`), Spark `hex`, and `array_agg`'s
payload compaction use it instead of each calling
`dictionary.with_values(f(values))`.
- Dictionary hashing (`hash_dictionary`,
`hash_dictionary_with_child_hashing`) applies the same bound.
- `clippy.toml` disallows calling `with_values` directly, so new code goes
through the helper. `get_field` keeps a direct call because it only picks a
child column.
The threshold comes from the `with_hashes` benchmark. Just above 8 values
per row, compacting before hashing is already faster than hashing every value:
0.56x for short strings, 0.60x for `Int64` and 0.30x for 64-byte strings. At 16
values per row it is 0.13x to 0.31x. Below the threshold the code path is
unchanged.
#24588 proposes evaluating opt-in functions once per distinct value in
`ScalarFunctionExpr` and keeps the per-function dictionary arms. This PR bounds
the cost of those arms and of hashing, so the two don't conflict.
## Benchmarks
ClickBench `hits_partitioned`, `enable_rle_to_dictionary = true`, 12
partitions, `release-nonlto`, minimum over 3 to 6 interleaved rounds (ms), same
results on both:
| query | main | this PR |
|---|---|---|
| `lower` over a dictionary GROUP BY (the query above) | 12581 | 1019 |
| GROUP BY `"SearchPhrase"` (#26696) | 1194 | 867 |
| GROUP BY `arrow_cast("SearchPhrase", 'Utf8')` | 463 | 476 |
| `MAX(lower("SearchPhrase"))` | 634 | 631 |
| GROUP BY `"MobilePhoneModel"` (low cardinality) | 123 | 125 |
| `MAX(lower("MobilePhoneModel"))` | 160 | 159 |
The rest of the #26696 gap (867 ms vs 463 ms for `Utf8` keys) is in arrow's
`concat_dictionaries`, which the repartition coalescer uses, and in the general
cost of dictionary group keys (#24822).
`with_hashes`, 8192-row slices of a dictionary with one value per row of the
full array, median of 5 interleaved runs (µs). Only `dictionary_sliced` is
added to the benchmark; the threshold cases were run locally:
| case | main | this PR |
|---|---|---|
| `dictionary_sliced`: 10 values per row | 70.5 | 39.7 |
| short strings, 8 values per row (unchanged path) | 67.7 | 67.9 |
| short strings, just over 8 values per row | 64.9 | 36.1 |
| `Int64`, just over 8 values per row | 19.3 | 11.5 |
| 64-byte strings, just over 8 values per row | 180.2 | 53.6 |
| short strings, 16 values per row | 116.4 | 35.4 |
| `Int64`, 16 values per row | 34.8 | 10.7 |
| 64-byte strings, 16 values per row | 362.5 | 48.2 |
The existing `dictionary_utf8_int32` cases (100 values per batch) take the
unchanged path and stay within noise.
## What is the testing strategy for this PR?
- `dictionary.slt`: a GROUP BY with `batch_size = 1` makes every output
slice carry many more dictionary values than rows. The test compares each
function above on the grouped dictionary with the same function on the decoded
value, including a NULL group, and checks the result stays a dictionary. It
also goes through the hash repartition of the partial aggregate's slices.
- Unit tests: `create_hashes_for_dict_slices_with_more_values_than_rows`
(single and multi-column hashes of slices with null keys and null values match
the unsliced hashes) and `map_dictionary_values_bounds_values_per_row`.
- Benchmark: the `dictionary_sliced` case in
`datafusion/common/benches/with_hashes.rs`.
## Are there any user-facing changes?
No behavior changes. `datafusion_common::utils` gains
`map_dictionary_values` and `MAX_DICTIONARY_VALUES_PER_ROW`.
--
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]
