apache / apache/datafusion

Reuse the cached dictionary value→slot map in vectorized_equal_to instead of rebuilding it per batch

Open
#25,219 1 comment 0 reactions 1 assignee Claimed by @CuteChuanChuan View on GitHub
enhancement performance
Dominant language
Rust
Stars
9.3k
Forks
2.4k
Avg merge
3d 7h
Merged PRs (30d)
344

Description

### 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`).

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.