jayzhan211 commented on code in PR #25652:
URL: https://github.com/apache/datafusion/pull/25652#discussion_r4115458038
##########
datafusion/physical-plan/src/aggregates/group_values/multi_group_by/dictionary.rs:
##########
@@ -348,59 +318,39 @@ impl<K: ArrowDictionaryKeyType + Send + Sync> GroupColumn
let dict = array.as_dictionary::<K>();
let dict_keys = dict.keys();
let dict_values = dict.values();
- let num_distinct = dict_values.len();
-
- // The fallback is in a separate #[cold] function so its code does not
- // appear inline here and cannot prevent LLVM from pipelining /
unrolling
- // the hot lookup-table loops below.
- if rhs_rows.len() < num_distinct {
- self.equal_to_per_row(
- lhs_rows,
- dict_values,
- dict,
- rhs_rows,
- equal_to_results,
- );
- return;
- }
-
- 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);
- let group_to_inner = self.group_to_inner.as_slice();
+ self.sync_value_cache(dict_values);
+ let raw_keys = dict_keys.values();
if dict_keys.null_count() == 0 {
// No null keys : skip the get_bit guard: we only ever write false,
// so overwriting an already-false bit is a no-op.
- let raw_keys = dict_keys.values();
for (idx, (&lhs_row, &rhs_row)) in
lhs_rows.iter().zip(rhs_rows.iter()).enumerate()
Review Comment:
Calling `lookup_inner_slot(&mut self, ..)` on every row makes the hot loop
slower than the old `lookup[..]` slice index. The existing
`dictionary_group_values` bench regresses: `dict_intern_emit` 1.21–1.71x at
card ≤ 1000, `dict_repeated_intern_emit` 1.08–1.52x. In `intern_emit` the
values `Arc` is already cached by `vectorized_append`, so this is the loop
itself, not the removed fallback. Reading the cache inline and taking the
resolver only on a miss brought `intern_emit` back to 0.96–1.05x in my run:
```diff
- let val_idx = raw_keys[rhs_row].as_usize();
- let rhs_slot = self.lookup_inner_slot(dict_values, val_idx);
+ let val_idx = raw_keys[rhs_row].as_usize();
+ let rhs_slot = match self.val_to_inner[val_idx] {
+ usize::MAX => self.lookup_inner_slot(dict_values,
val_idx),
+ slot => slot,
+ };
```
```diff
+ #[cold]
+ #[inline(never)]
fn lookup_inner_slot(&mut self, dict_values: &ArrayRef, val_idx: usize)
-> usize {
```
Apply the same change to the nullable-key loop and keep its `get_bit` guard.
A merged single loop still left `repeated_intern_emit` (10% null keys) at
1.06–1.19x.
##########
datafusion/physical-plan/src/aggregates/group_values/multi_group_by/dictionary.rs:
##########
@@ -348,59 +318,39 @@ impl<K: ArrowDictionaryKeyType + Send + Sync> GroupColumn
let dict = array.as_dictionary::<K>();
let dict_keys = dict.keys();
let dict_values = dict.values();
- let num_distinct = dict_values.len();
-
- // The fallback is in a separate #[cold] function so its code does not
- // appear inline here and cannot prevent LLVM from pipelining /
unrolling
- // the hot lookup-table loops below.
- if rhs_rows.len() < num_distinct {
- self.equal_to_per_row(
- lhs_rows,
- dict_values,
- dict,
- rhs_rows,
- equal_to_results,
- );
- return;
- }
-
- 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);
- let group_to_inner = self.group_to_inner.as_slice();
+ self.sync_value_cache(dict_values);
Review Comment:
Dropping the `rhs_rows.len() < num_distinct` fallback regresses the
steady-state case with a new values array per batch: every row hits an existing
group, so `vectorized_append` isn't called for the column. `sync_value_cache`
now hashes all D values per batch where base did per-row `equal_to`. That
happens with concatenated/coalesced dictionaries and at parquet row-group
boundaries. With 8192 rows/batch and D = 100000, I measured 1.19–1.48x vs base.
Keep the fallback, but only when the `Arc` isn't already cached (0.89–0.95x in
my run):
```rs
let cached = self
.cached_values
.as_ref()
.is_some_and(|c| Arc::ptr_eq(c, dict_values));
if !cached && rhs_rows.len() < dict_values.len() {
self.equal_to_per_row(lhs_rows, dict_values, dict, rhs_rows,
equal_to_results);
return;
}
self.sync_value_cache(dict_values);
```
Please also add before/after bench numbers to the PR description, since this
is a perf PR.
--
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]
