zhuqi-lucas opened a new pull request, #23986: URL: https://github.com/apache/datafusion/pull/23986
## Which issue does this PR close? - Part of #23937 (the adaptive-disable follow-up listed there is superseded by the per-batch hash gate for the multi-column path; a persistent-state variant for the single-column paths can be a follow-up if benchmarks warrant it). ## Rationale for this change `GroupValues::intern` pays the full hash + hash-table-probe cost for every input row, even when a row's group key equals the previous row's. For high-cardinality keys the table is far too large to stay cache-resident, so each probe is typically an L3/DRAM round trip. Real-world data very often arrives with **runs of identical keys** (time-ordered logs, clustered writes). Measured on ClickBench `hits.parquet` in file order — which is what the Partial aggregation phase sees, since it runs before repartitioning: | column | avg run length | consecutive-keys hit rate | |---|---:|---:| | CounterID | 58,693 | ~100% | | OS | 15.6 | 93.6% | | SearchPhrase | 11.7 | 91.5% | | RegionID | 11.0 | 90.9% | | UserID | 10.1 | 90.1% | | URL | 1.4 | 30.2% | Remembering the previous row's key and group index lets a run reuse the answer: a hit replaces a random memory access with one or two well-predicted compares against L1-resident state; a miss costs a single compare. Break-even hit rate is under 1%. This is the same optimization ClickHouse ships as its *consecutive keys optimization* (`LastElementCache` in `ColumnsHashing`, refined in [ClickHouse#57872](https://github.com/ClickHouse/ClickHouse/pull/57872)). The win scales with `hit rate × probe cost`: Q33's URL table is the largest string hash table, so even a 30% hit rate pays the most per skipped probe, while Q27's ~6K-entry CounterID table is L1/L2-resident and gains little at ~100% hit rate. ## What changes are included in this PR? Four implementations, one per intern path: - **`GroupValuesPrimitive`** (single primitive key): cache `(last_key, last_group)` fields; the guard compares the canonicalized value itself, so a hit skips the hash as well as the probe. Invalidated in `emit` / `clear_shrink` since both reassign group indices. - **`ArrowBytesMap::insert_if_new_inner`** (single `Utf8` / `LargeUtf8` / `Binary` key via `GroupValuesBytes`; also used by distinct-aggregate accumulators, which benefit for free): compare the current row's bytes against the previous input row's — adjacent rows in the same values buffer, so the comparison is cache-hot. Per-call state, no invalidation concerns. - **`ArrowBytesViewMap::insert_if_new_inner`** (single `Utf8View` / `BinaryView` key via `GroupValuesBytesView`): for inline views (len ≤ 12) the u128 view comparison is a complete equality check; longer values get length + 4-byte-prefix guards from the views before touching bytes. - **`GroupValuesColumn`** (multi-column keys, both the vectorized and streaming interns): a batch-level adjacent-equality mask (`not_distinct` per column, AND-ed — `null == null` counts as equal, matching GROUP BY semantics), guarded by a two-tier gate: a sequential scan over the already-computed hashes counts adjacent hash matches, and only run-heavy batches (≥ ~6%) build the precise mask. Shuffled inputs — e.g. the Final phase after hash repartitioning — fail the gate and pay nothing beyond that one scan. Correctness never rests on the hashes; they only decide whether the mask is worth building. Nested-type columns (handled by `RowsGroupColumn`) make the helper return `None` and fall back to the normal path. Integration is deliberately non-invasive to the vectorized machinery: masked rows are skipped in `collect_vectorized_process_context` (they enter none of the append / equal-to / remaining lists), and a final ascending fill pass assigns each one its predecessor's group index. Intervening nulls do not invalidate any of the caches (a value's group never changes within an accumulation), and the make-payload-once / observe-per-row-in-order contract of the byte maps is preserved. ## Are these changes tested? - `consecutive_keys_runs_and_emit_invalidation` (primitive): runs with interleaved nulls, plus the `emit(First(n))` index-shift trap — a stale cache would assign shifted group ids. - `string_map_consecutive_keys_runs` / `view_map_consecutive_keys_runs`: payload-assignment equality between run and scattered occurrences, interleaved nulls, inline and non-inline string lengths, `make_payload_fn` called exactly once per distinct. - `test_intern_consecutive_keys_multi_column`: identical group assignments on both intern paths, including the "one column runs while the other breaks the run" case and `null == null` runs; `test_intern_consecutive_keys_no_runs` exercises the gate fallback. - Probe-verified (temporary `eprintln`) that the fast path fires for exactly the designed rows on both multi-column paths and that the gate rejects the no-runs batch. - Full `aggregates::` suite (158), `datafusion-physical-expr-common` suite (77), and workspace clippy with `-D warnings` all pass. ## Are there any user-facing changes? No API changes. Local ClickBench numbers (hits.parquet, 5–8 iterations, `release-nonlto`, macOS ARM, both binaries built from this branch's base commit) — CI benchmarks to confirm: | query | GROUP BY | delta | |---|---|---| | Q33 | URL | **−18.0%** | | Q18 | UserID, m, SearchPhrase | **−10.8%** | | Q15 | UserID | **−10.7%** | | Q32 | WatchID, ClientIP | −10.2% | | Q12 | SearchPhrase | −8.2% | | Q7 | AdvEngineID | −6.2% | | Q17 | UserID, SearchPhrase | −5.4% | | Q27 | CounterID | −4.8% | | Q16 | UserID, SearchPhrase | −3.0% | | Q13 | (COUNT DISTINCT rewrite, untouched path) | noise ±3% | No regressions observed; queries whose plans don't touch these paths are unchanged. -- 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]
