Rich-T-kid commented on code in PR #26816:
URL: https://github.com/apache/datafusion/pull/26816#discussion_r4502520050
##########
datafusion/common/src/hash_utils.rs:
##########
@@ -28,6 +32,134 @@ use std::hash::{BuildHasher, Hash, Hasher};
pub type RandomState = FixedState;
pub type QualityRandomState = foldhash::quality::FixedState;
+type DictionaryHashEntry = (Weak<dyn Array>, Vec<u64>);
+
+/// Hashes arrays, reusing dictionary value hashes by column when possible.
+#[derive(Debug)]
+pub struct CachingArrayHasher {
+ /// Stored with the cache so value hashes cannot cross hash seeds.
+ random_state: RandomState,
+ /// One entry per column. Weak identity avoids retaining dictionary values.
+ dictionaries: Vec<Option<DictionaryHashEntry>>,
+}
+
+impl CachingArrayHasher {
+ /// Creates one cache slot per column.
+ ///
+ /// For each dictionary column, the cache tracks the values array with a
+ /// weak pointer, so it does not keep the values alive, and owns a reusable
+ /// value-hash buffer. [`Self::create_hashes`] reuses those hashes while
the
+ /// values `Arc` is unchanged, recomputes them when it changes, and
scatters
+ /// them through the dictionary keys into the row hashes. Non-dictionary
+ /// columns use the regular array hashing path.
+ pub fn new(num_columns: usize, random_state: RandomState) -> Self {
+ Self {
+ random_state,
+ dictionaries: vec![None; num_columns],
+ }
+ }
+
+ /// Hashes `arrays`, reusing dictionary value hashes while each column's
+ /// values array is unchanged.
+ pub fn create_hashes<'a, I, T>(
+ &mut self,
+ arrays: I,
+ hashes_buffer: &'a mut [u64],
+ ) -> Result<&'a mut [u64]>
+ where
+ I: IntoIterator<Item = T>,
+ I::IntoIter: ExactSizeIterator,
+ T: AsDynArray,
+ {
+ let arrays = arrays.into_iter();
+ if arrays.len() != self.dictionaries.len() {
+ return _internal_err!(
+ "CachingArrayHasher expected {} arrays, got {}",
+ self.dictionaries.len(),
+ arrays.len()
+ );
+ }
+ for (index, array) in arrays.enumerate() {
+ self.hash_array(index, array.as_dyn_array(), hashes_buffer, index
!= 0)?;
+ }
+ Ok(hashes_buffer)
+ }
+
+ fn hash_array(
+ &mut self,
+ index: usize,
+ array: &dyn Array,
+ hashes_buffer: &mut [u64],
+ rehash: bool,
+ ) -> Result<()> {
+ #[cfg(feature = "force_hash_collisions")]
+ {
+ let _ = index;
+ hash_single_array(array, &self.random_state, hashes_buffer,
rehash)?;
+ }
+
+ #[cfg(not(feature = "force_hash_collisions"))]
+ {
+ let dictionary = &mut self.dictionaries[index];
+ if !matches!(array.data_type(), DataType::Dictionary(_, _)) {
+ *dictionary = None;
+ hash_single_array(array, &self.random_state, hashes_buffer,
rehash)?;
+ return Ok(());
+ }
+
+ downcast_dictionary_array! {
+ array => {
+ let values = array.values();
+ if !dictionary.as_ref().is_some_and(|(cached, _)| {
+ std::ptr::addr_eq(cached.as_ptr(), Arc::as_ptr(values))
+ }) {
+ let mut hashes = dictionary
+ .take()
+ .map(|(_, hashes)| hashes)
+ .unwrap_or_default();
+ hashes.resize(values.len(), 0);
+ // Fresh hashing leaves null value slots unchanged.
+ hashes.fill(0);
+ hash_single_array(
+ values.as_ref(),
+ &self.random_state,
+ &mut hashes,
+ false,
+ )?;
+ *dictionary = Some((Arc::downgrade(values), hashes));
+ }
+ dispatch_dictionary_scatter(
+ array,
+ &dictionary.as_ref().unwrap().1,
+ hashes_buffer,
+ rehash,
+ );
+ }
+ _ => unreachable!(),
+ };
+ }
+
+ Ok(())
+ }
+
+ /// Returns the current dictionary values and hashes for `index`, if any.
+ pub fn dictionary_hashes(&self, index: usize) -> Option<(ArrayRef,
&[u64])> {
+ let (values, hashes) = self.dictionaries.get(index)?.as_ref()?;
+ Some((values.upgrade()?, hashes))
+ }
+
+ /// Returns the allocated size of the cache.
+ pub fn size(&self) -> usize {
Review Comment:
Thanks for the correction and for checking the tight-memory boundary.
Agreed, keeping the best-effort reservation and dropping the cache on refusal
is the right behavior.
--
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]
