This is an automated email from the ASF dual-hosted git repository.
Rachelint pushed a commit to branch improve-compare-in-view-map
in repository https://gitbox.apache.org/repos/asf/datafusion.git
The following commit(s) were added to refs/heads/improve-compare-in-view-map by
this push:
new 8643ebe4e1 Cache byte view map batch lookups
8643ebe4e1 is described below
commit 8643ebe4e13142f9ca41f54b8e91b018fb5aeebb
Author: kamille <[email protected]>
AuthorDate: Thu Jul 9 06:52:41 2026 +0800
Cache byte view map batch lookups
---
.../physical-expr-common/src/binary_view_map.rs | 89 +++++++---------------
1 file changed, 29 insertions(+), 60 deletions(-)
diff --git a/datafusion/physical-expr-common/src/binary_view_map.rs
b/datafusion/physical-expr-common/src/binary_view_map.rs
index 4cbf827e38..2c54998330 100644
--- a/datafusion/physical-expr-common/src/binary_view_map.rs
+++ b/datafusion/physical-expr-common/src/binary_view_map.rs
@@ -119,49 +119,6 @@ impl ArrowBytesViewSet {
/// Max size of the in-progress buffer before flushing to completed buffers
const BYTE_VIEW_MAX_BLOCK_SIZE: usize = 2 * 1024 * 1024;
-#[derive(Debug)]
-struct BatchNonInlineCompareStats<V>
-where
- V: Debug + PartialEq + Eq + Clone + Copy + Default,
-{
- num_non_inline_compares: usize,
- target_group_counts: Vec<(V, usize)>,
-}
-
-impl<V> BatchNonInlineCompareStats<V>
-where
- V: Debug + PartialEq + Eq + Clone + Copy + Default,
-{
- fn new() -> Self {
- Self {
- num_non_inline_compares: 0,
- target_group_counts: Vec::new(),
- }
- }
-
- fn record(&mut self, target_group_id: V) {
- self.num_non_inline_compares += 1;
- if let Some((_, count)) = self
- .target_group_counts
- .iter_mut()
- .find(|(group_id, _)| *group_id == target_group_id)
- {
- *count += 1;
- } else {
- self.target_group_counts.push((target_group_id, 1));
- }
- }
-
- fn num_target_groups(&self) -> usize {
- self.target_group_counts.len()
- }
-
- fn num_redundant_compares(&self) -> usize {
- self.num_non_inline_compares
- .saturating_sub(self.num_target_groups())
- }
-}
-
pub struct ArrowBytesViewMap<V>
where
V: Debug + PartialEq + Eq + Clone + Copy + Default,
@@ -316,7 +273,8 @@ where
assert_eq!(values.len(), self.hashes_buffer.len());
let input_has_buffers = !values.data_buffers().is_empty();
- let mut batch_non_inline_compare_stats =
BatchNonInlineCompareStats::new();
+ let mut input_view_to_payload =
+ hashbrown::hash_table::HashTable::<(u128, V)>::new();
for i in 0..values.len() {
let view_u128 = input_views[i];
let hash = self.hashes_buffer[i];
@@ -340,6 +298,20 @@ where
// Extract length from the view (first 4 bytes of u128 in
little-endian)
let len = view_u128 as u32;
+ let cached_payload = if input_has_buffers && len > 12 {
+ let view_hash = Self::hash_input_view(view_u128);
+ input_view_to_payload
+ .find(view_hash, |(cached_view, _)| *cached_view ==
view_u128)
+ .map(|(_, payload)| *payload)
+ } else {
+ None
+ };
+
+ if let Some(payload) = cached_payload {
+ observe_payload_fn(payload);
+ continue;
+ }
+
// Check if value already exists
let maybe_payload = {
let completed = &self.completed;
@@ -354,22 +326,18 @@ where
if input_has_buffers {
Self::view_equal_to_input::<B, true>(
header.view,
- header.payload,
completed,
in_progress,
values,
i,
- &mut batch_non_inline_compare_stats,
)
} else {
Self::view_equal_to_input::<B, false>(
header.view,
- header.payload,
completed,
in_progress,
values,
i,
- &mut batch_non_inline_compare_stats,
)
}
})
@@ -412,28 +380,31 @@ where
.insert_accounted(new_header, |h| h.hash, &mut
self.map_size);
payload
};
+ if input_has_buffers && len > 12 {
+ let view_hash = Self::hash_input_view(view_u128);
+ input_view_to_payload.insert_unique(
+ view_hash,
+ (view_u128, payload),
+ |(cached_view, _)| Self::hash_input_view(*cached_view),
+ );
+ }
+
observe_payload_fn(payload);
}
+ }
- dbg!((
- "ArrowBytesViewMap::insert_if_new_inner",
- values.len(),
- batch_non_inline_compare_stats.num_non_inline_compares,
- batch_non_inline_compare_stats.num_target_groups(),
- batch_non_inline_compare_stats.num_redundant_compares(),
- &batch_non_inline_compare_stats.target_group_counts,
- ));
+ #[inline(always)]
+ fn hash_input_view(view: u128) -> u64 {
+ (view as u64) ^ ((view >> 64) as u64)
}
#[inline(always)]
fn view_equal_to_input<B: ByteViewType, const HAS_BUFFERS: bool>(
exist_view: u128,
- target_group_id: V,
completed: &[Buffer],
in_progress: &[u8],
array: &GenericByteViewArray<B>,
rhs_row: usize,
- batch_non_inline_compare_stats: &mut BatchNonInlineCompareStats<V>,
) -> bool {
// SAFETY: caller ensures `rhs_row` is valid.
let input_view = unsafe { *array.views().get_unchecked(rhs_row) };
@@ -460,8 +431,6 @@ where
return false;
}
- batch_non_inline_compare_stats.record(target_group_id);
-
let exist_full = {
let byte_view = ByteView::from(exist_view);
let buffer_index = byte_view.buffer_index as usize;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
