jayzhan211 opened a new issue, #25219:
URL: https://github.com/apache/datafusion/issues/25219
### Is your feature request related to a problem or challenge?
#25185 added a `Arc::ptr_eq`-keyed cache of `val_hashes` / `val_to_inner` to
`DictionaryGroupValuesColumn`, so `vectorized_append` no longer re-hashes the
whole dictionary values array on every batch. That removed the one *unbounded*
cost on this path: `vectorized_append` is called with only the **new group**
rows, so hashing all `D` values to resolve a handful of rows was O(dictionary
cardinality) work unrelated to the batch's actual workload.
The other consumer of the same information, `vectorized_equal_to`, was left
untouched and still rebuilds everything from scratch on every batch:
```rust
// dictionary.rs, vectorized_equal_to
let mut val_hashes = vec![0u64; dict_values.len()];
create_hashes(std::slice::from_ref(dict_values), &self.random_state, &mut
val_hashes).unwrap();
let lookup = self.build_lookup_table(dict_values, &val_hashes);
```
`build_lookup_table` produces exactly the `val_idx -> inner_slot` map that
`val_to_inner` now caches — it is thrown away and rebuilt per batch, at a cost
of `D` hashes plus `D` `value_dedup` probes, each with a full `inner.equal_to`
comparison.
Two things make this the dominant remaining cost rather than a rounding
error:
1. `GroupValuesColumns::vectorized_append` early-returns when
`append_row_indices` is empty (`multi_group_by/mod.rs`). Once the group set
converges — the steady state for most aggregations — the cache added in #25185
is never consulted again, while `vectorized_equal_to` keeps running on every
batch.
2. It is bounded but with a heavy constant. The `rhs_rows.len() <
num_distinct` guard diverts to the per-row fallback when `D` exceeds the row
count, so the table path is O(rows), not unbounded. But at, say, `D = 4000` and
`rows = 8192` we pay 4000 hashes + 4000 hash-table probes + 4000 byte
comparisons per batch to answer questions a cached array index would answer
directly.
### Describe the solution you'd like
Let `val_to_inner` serve both paths, so the `val_idx -> inner_slot` map is
built once per distinct dictionary values array instead of once per batch per
path.
This requires widening the trait method to take `&mut self`:
```diff
pub trait GroupColumn: Send + Sync {
fn vectorized_equal_to(
- &self,
+ &mut self,
lhs_rows: &[usize],
array: &ArrayRef,
rhs_rows: &[usize],
equal_to_results: &mut BooleanBufferBuilder,
);
}
```
Scope of that change:
- 8 impls of `vectorized_equal_to` (`boolean`, `bytes`, `bytes_view`,
`fixed_size_binary`, `list`, `primitive`, `row_backed`, `dictionary`); 7 are a
mechanical signature change with no body edit.
- `GroupValuesColumns::vectorized_equal_to` switches
`self.group_values.iter()` to `iter_mut()`. It already `mem::replace`s
`equal_to_results` to work around the split borrow of `self`;
`equal_to_group_indices` / `equal_to_row_indices` need the same treatment.
Then in `DictionaryGroupValuesColumn::vectorized_equal_to`:
- call `sync_value_cache(dict_values)` and index `val_to_inner` directly,
filling misses lazily, instead of calling `build_lookup_table`;
- delete `build_lookup_table`;
- drop the `rhs_rows.len() < num_distinct` heuristic and, most likely,
`equal_to_per_row` with it. That crossover only exists because the lookup table
is rebuilt every batch; once it is cached there is nothing to amortize and the
table path should always win.
Net effect: hashing drops to once per distinct values `Arc` across *both*
paths, and per-batch work in `vectorized_equal_to` becomes O(rows) with an
array-index inner loop. It is also a net deletion of code.
One detail that makes `&mut self` necessary rather than merely convenient:
`vectorized_append` runs **before** `vectorized_equal_to` (`mod.rs`, steps 2
and 3 of `intern`) but is skipped entirely when there are no new groups. In the
steady state `cached_values` therefore points at a stale array, so a
`&self`-only reuse would miss precisely when it matters most.
`vectorized_equal_to` has to be able to *populate* the cache, not just read it.
### Describe alternatives you've considered
- **Keep `&self` and reuse `self.val_hashes` under a `ptr_eq` guard.** Cheap
and safe (when `cached_values` is `Some` and `ptr_eq` holds, `val_hashes.len()
== dict_values.len()` by construction), but per the note above it misses in the
steady state, and it still leaves the `D` hash-table probes in
`build_lookup_table`. Saves the smaller half.
- **Hash only the referenced `val_idx` on a cache miss** instead of the
whole values array. Attractive when a fresh values array arrives every batch,
but there is no generic single-row hash for a `dyn GroupColumn` — it would mean
a `take` into a scratch array first, which has its own constant. Low value
while misses are rare; worth revisiting only if a workload shows per-batch
dictionary churn.
- **Content fingerprint instead of `Arc::ptr_eq`.** Self-defeating:
computing it is O(D), which is the cost we are trying to avoid. `ptr_eq` is
correct here because `take` / `filter` / repartition preserve the values `Arc`,
and holding the `Arc` in `cached_values` rules out address reuse.
- **Multi-entry LRU instead of the single cache slot.** No benefit: one
partition has one upstream, so batches within a partition carry one values
array.
### Additional context
- Follow-up to #25185, which introduced `sync_value_cache` and the
`val_to_inner` / `val_hashes` invariants this would build on. The prior
scalar-path cache is #24418.
- Benchmark exercising this path: #25198. Note that `take_n` clears the
cache and `EmitTo::First(n)` is what partial aggregates use for early emission
— a benchmark where early emit fires per batch rebuilds the cache every batch
and will show flat results for both #25185 and this follow-up. Worth isolating
the group build-up phase from the steady state when measuring.
- The cache invariant to preserve: `val_to_inner` entries are only ever
filled in, never invalidated, because `inner` slot indices are stable under
append. `take_n` is the only method that remaps them, and it drops the cache
(currently as a side effect of `hash_values` setting `cached_values = None`,
pinned by `take_n_invalidates_value_cache`).
--
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]
