From 63d89aa07e6334dc2b08621c4c4ed12d260f6845 Mon Sep 17 00:00:00 2001
From: RIchard Baah <137434454+Rich-T-kid@users.noreply.github.com>
Date: Wed, 26 Aug 2026 11:29:06 +0000
Subject: [PATCH] perf: cache dictionary arc pointer (#24418)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

## Which issue does this PR close?

<!--
We generally require a GitHub issue to be filed for all bug fixes and
enhancements and this helps us generate change logs for our releases.
You can link an issue to this PR using the GitHub syntax. For example
`Closes #123` indicates that this PR will close issue #123.
-->

- Closes #23921.
- this PR was created on top of #24416, so once that is merged in we can
rebase this branch

## Rationale for this change

append_val is called once per new group on the scalar (streaming) code
path. Previously it hashed a single-element slice of the dictionary
values array on every call, paying create_hashes fixed overhead 65k
times for a high-cardinality batch. Caching the full values array hash
keyed on Arc::ptr_eq collapses that to one vectorized hash pass per
batch, yielding a 2× speedup on the all-unique case with no measurable
regression elsewhere.

<!--
Why are you proposing this change? If this is already explained clearly
in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand
your changes and offer better suggestions for fixes.

Please explain the problem you are trying to solve in terms of the
user-visible
behavior, rather than the implementation.

For example, "The code in `foo.rs` doesn't handle nulls" is a symptom of
the
implementation. "COUNT(DISTINCT) returns wrong results when the column
contains
nulls" is the user-visible problem.
-->

## What changes are included in this PR?

- DictionaryGroupValuesColumn gains a cached_values: Option<ArrayRef>
field
- append_val: on a cache miss (Arc::ptr_eq fails), hashes the entire
dict.values() array into val_hashes and stores the Arc; on a hit, reuses
val_hashes[val_idx] directly — eliminating the per-call slice(val_idx,
1) allocation and create_hashes call

<!--
There is no need to duplicate the description in the issue here, but it
is sometimes worth providing a summary of the individual changes in this
PR.
-->

## Are these changes tested?
nothing logically changed.
<!--
We typically require tests for all PRs in order to:
1. Prevent the code from being accidentally broken by subsequent changes
2. Serve as another way to document the expected behavior of the code

If tests are not included in your PR, please explain why (for example,
are they covered by existing tests)?
-->

## Are there any user-facing changes?
no
<!--
If there are user-facing changes then we may require documentation to be
updated before approving the PR.

If there are any breaking changes to public APIs, please add the `api
change` label.
-->

---------

Co-authored-by: Andrew Lamb <andrew@nerdnetworks.org>
(cherry picked from commit 66a901f68ddf7ce9c114ee6cb27b3801d4c334e3)
---
 .../group_values/multi_group_by/dictionary.rs | 27 ++++++++++++++++---
 1 file changed, 24 insertions(+), 3 deletions(-)

diff --git a/datafusion/physical-plan/src/aggregates/group_values/multi_group_by/dictionary.rs b/datafusion/physical-plan/src/aggregates/group_values/multi_group_by/dictionary.rs
index 501b13d0cd183..d0c167e1d2d59 100644
--- a/datafusion/physical-plan/src/aggregates/group_values/multi_group_by/dictionary.rs
+++ b/datafusion/physical-plan/src/aggregates/group_values/multi_group_by/dictionary.rs
@@ -57,6 +57,9 @@ pub struct DictionaryGroupValuesColumn<K: ArrowDictionaryKeyType + Send + Sync>
     val_to_inner: Vec<usize>,
     /// Reusable hash buffer for the dictionary values array.
     val_hashes: Vec<u64>,
+    /// The last `dict.values()` Arc hashed in `append_val`. When the incoming
+    /// values array is `ptr_eq` to this, `val_hashes` can be reused directly.
+    cached_values: Option<ArrayRef>,
     _phantom: PhantomData<K>,
 }
 
@@ -73,6 +76,7 @@ impl<K: ArrowDictionaryKeyType + Send + Sync> DictionaryGroupValuesColumn<K> {
             random_state: AGGREGATION_HASH_SEED,
             val_to_inner: Vec::default(),
             val_hashes: Vec::default(),
+            cached_values: None,
             _phantom: PhantomData,
         }
     }
@@ -159,6 +163,7 @@ impl<K: ArrowDictionaryKeyType + Send + Sync> DictionaryGroupValuesColumn<K> {
     }
 
     fn hash_values(&mut self, values: &ArrayRef) {
+        self.cached_values = None;
         self.val_hashes.clear();
         self.val_hashes.resize(values.len(), 0);
         create_hashes(
@@ -296,9 +301,25 @@ impl<K: ArrowDictionaryKeyType + Send + Sync> GroupColumn
             }
             Some(val_idx) => {
                 let dict_values = dict.values();
-                let single = dict_values.slice(val_idx, 1);
-                self.hash_values(&single);
-                self.find_or_insert_value(dict_values, val_idx, self.val_hashes[0])?
+                // check if the dictionary values array we are hashing was already seen.
+                // if its arc was already stored we dont need to rehash the entire array again
+                // if its new hash the entire array and store an arc ptr for future use
+                let cache_hit = self
+                    .cached_values
+                    .as_ref()
+                    .is_some_and(|c| Arc::ptr_eq(c, dict_values));
+                if !cache_hit {
+                    self.val_hashes.clear();
+                    self.val_hashes.resize(dict_values.len(), 0);
+                    create_hashes(
+                        std::slice::from_ref(dict_values),
+                        &self.random_state,
+                        &mut self.val_hashes,
+                    )
+                    .unwrap();
+                    self.cached_values = Some(Arc::clone(dict_values));
+                }
+                self.find_or_insert_value(dict_values, val_idx, self.val_hashes[val_idx])?
             }
         };
         self.group_to_inner.push(inner_slot);