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

[jayzhan211]

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:

// 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 perf: reuse cached dictionary value hashes in vectorized_append #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:

 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::replaces 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 perf: reuse cached dictionary value hashes in vectorized_append #25185, which introduced sync_value_cache and the val_to_inner / val_hashes invariants this would build on. The prior scalar-path cache is perf: cache dictionary arc pointer #24418.
• Benchmark exercising this path: bench: cover dictionary group keys with a shared values array #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 perf: reuse cached dictionary value hashes in vectorized_append #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).