You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by ka...@apache.org on 2026/07/09 13:18:14 UTC
(datafusion) branch improve-compare-in-view-map-v2 updated (ea16efb605 -> 14f735ab73)
This is an automated email from the ASF dual-hosted git repository.
Rachelint pushed a change to branch improve-compare-in-view-map-v2
in repository https://gitbox.apache.org/repos/asf/datafusion.git
discard ea16efb605 fast path test.
new 14f735ab73 Buffer pending byte views before append
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (ea16efb605)
\
N -- N -- N refs/heads/improve-compare-in-view-map-v2 (14f735ab73)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../physical-expr-common/src/binary_view_map.rs | 272 ++++++++++-----------
1 file changed, 124 insertions(+), 148 deletions(-)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion) 01/01: Buffer pending byte views before append
Posted by ka...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
Rachelint pushed a commit to branch improve-compare-in-view-map-v2
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 14f735ab73a30e9a783e3b8eb35724593b7c0b3f
Author: kamille <ka...@apache.org>
AuthorDate: Thu Jul 9 21:15:34 2026 +0800
Buffer pending byte views before append
---
.../physical-expr-common/src/binary_view_map.rs | 163 ++++++++++++++-------
1 file changed, 108 insertions(+), 55 deletions(-)
diff --git a/datafusion/physical-expr-common/src/binary_view_map.rs b/datafusion/physical-expr-common/src/binary_view_map.rs
index 53b84e7dea..10d4930d1d 100644
--- a/datafusion/physical-expr-common/src/binary_view_map.rs
+++ b/datafusion/physical-expr-common/src/binary_view_map.rs
@@ -168,6 +168,16 @@ where
null: Option<(V, usize)>,
}
+/// Pending group values created during a single `insert_if_new` call.
+#[derive(Debug, Clone, Copy)]
+enum PendingViewBufferEntry<V>
+where
+ V: ByteViewMapPayload,
+{
+ Null { payload: V },
+ NonNull { payload: V, view: u128, row: usize },
+}
+
/// The size, in number of entries, of the initial hash table
const INITIAL_MAP_CAPACITY: usize = 512;
@@ -273,50 +283,45 @@ where
OP: FnMut(V),
B: ByteViewType,
{
- // step 1: compute hashes
let batch_hashes = &mut self.hashes_buffer;
batch_hashes.clear();
batch_hashes.resize(values.len(), 0);
- create_hashes([values], &self.random_state, batch_hashes)
- // hash is supported for all types and create_hashes only
- // returns errors for unsupported types
- .unwrap();
+ create_hashes([values], &self.random_state, batch_hashes).unwrap();
- // step 2: insert each value into the set, if not already present
let values = values.as_byte_view::<B>();
-
- // Get raw views buffer for direct comparison
let input_views = values.views();
-
- // Ensure lengths are equivalent
assert_eq!(values.len(), self.hashes_buffer.len());
+ let persisted_num_group_values = self.views.len();
+ let mut view_buffers: Vec<PendingViewBufferEntry<V>> = Vec::new();
+
for i in 0..values.len() {
let view_u128 = input_views[i];
let hash = self.hashes_buffer[i];
- // handle null value via validity bitmap check
if values.is_null(i) {
let payload = if let Some(&(payload, _offset)) = self.null.as_ref() {
payload
} else {
+ view_buffers.push(PendingViewBufferEntry::Null {
+ payload: V::default(),
+ });
let payload = make_payload_fn(None);
- let null_index = self.views.len();
- self.views.push(0);
- self.nulls.append_null();
- self.null = Some((payload, null_index));
+ let view_buffer_index = view_buffers.len() - 1;
+ view_buffers[view_buffer_index] =
+ PendingViewBufferEntry::Null { payload };
+ debug_assert_eq!(
+ payload.group_index(),
+ persisted_num_group_values + view_buffer_index
+ );
payload
};
observe_payload_fn(payload);
continue;
}
- // Extract length from the view (first 4 bytes of u128 in little-endian)
let len = view_u128 as u32;
-
- // Check if value already exists
let maybe_payload = {
- // Borrow completed and in_progress for comparison
let completed = &self.completed;
let in_progress = &self.in_progress;
@@ -326,34 +331,59 @@ where
return false;
}
- let stored_view = self.views[header.payload.group_index()];
+ let group_index = header.payload.group_index();
+ if group_index < persisted_num_group_values {
+ let stored_view = self.views[group_index];
+
+ if len <= 12 {
+ return stored_view == view_u128;
+ }
+
+ let stored_prefix = (stored_view >> 32) as u32;
+ let input_prefix = (view_u128 >> 32) as u32;
+ if stored_prefix != input_prefix {
+ return false;
+ }
+
+ let byte_view = ByteView::from(stored_view);
+ let stored_len = byte_view.length as usize;
+ let buffer_index = byte_view.buffer_index as usize;
+ let offset = byte_view.offset as usize;
+
+ let stored_value = if buffer_index < completed.len() {
+ &completed[buffer_index].as_slice()
+ [offset..offset + stored_len]
+ } else {
+ &in_progress[offset..offset + stored_len]
+ };
+ let input_value: &[u8] = values.value(i).as_ref();
+ return stored_value == input_value;
+ }
+
+ let view_buffer_index = group_index - persisted_num_group_values;
+ let pending_entry = &view_buffers[view_buffer_index];
+ let PendingViewBufferEntry::NonNull {
+ view: buffered_view,
+ row: buffered_row,
+ ..
+ } = pending_entry
+ else {
+ return false;
+ };
- // Fast path: inline strings can be compared directly
if len <= 12 {
- return stored_view == view_u128;
+ return *buffered_view == view_u128;
}
- // For larger strings: first compare the 4-byte prefix
- let stored_prefix = (stored_view >> 32) as u32;
+ let buffered_prefix = (*buffered_view >> 32) as u32;
let input_prefix = (view_u128 >> 32) as u32;
- if stored_prefix != input_prefix {
+ if buffered_prefix != input_prefix {
return false;
}
- // Prefix matched - compare full bytes
- let byte_view = ByteView::from(stored_view);
- let stored_len = byte_view.length as usize;
- let buffer_index = byte_view.buffer_index as usize;
- let offset = byte_view.offset as usize;
-
- let stored_value = if buffer_index < completed.len() {
- &completed[buffer_index].as_slice()
- [offset..offset + stored_len]
- } else {
- &in_progress[offset..offset + stored_len]
- };
+ let buffered_value: &[u8] = values.value(*buffered_row).as_ref();
let input_value: &[u8] = values.value(i).as_ref();
- stored_value == input_value
+ buffered_value == input_value
})
.map(|entry| entry.payload)
};
@@ -361,37 +391,60 @@ where
let payload = if let Some(payload) = maybe_payload {
payload
} else {
- // no existing value, make a new one
+ view_buffers.push(PendingViewBufferEntry::NonNull {
+ payload: V::default(),
+ view: view_u128,
+ row: i,
+ });
+
let payload = if len <= 12 {
- // Inline path: bytes are already packed in view_u128.
- // The inline ByteView format is [len:u32 LE][data:12 bytes zero-padded],
- // so extracting bytes from the u128 avoids a round-trip through
- // values.value(i) (which reads the views buffer and returns the same slice).
let view_bytes = view_u128.to_le_bytes();
let value = &view_bytes[4..4 + len as usize];
- let payload = make_payload_fn(Some(value));
- // For inline strings, the stored view is identical to the input view:
- // make_view(value, 0, 0) produces the same u128 as view_u128.
- //
- // SAFETY: view_u128 was a valid view, and the enclosing `len <= 12`
- // ensures it is inline
- unsafe { self.append_inline_view(view_u128) };
- payload
+ make_payload_fn(Some(value))
} else {
let value: &[u8] = values.value(i).as_ref();
- let payload = make_payload_fn(Some(value));
- self.append_value(value);
- payload
+ make_payload_fn(Some(value))
};
- let new_header = Entry { hash, payload };
+ let view_buffer_index = view_buffers.len() - 1;
+ view_buffers[view_buffer_index] = PendingViewBufferEntry::NonNull {
+ payload,
+ view: view_u128,
+ row: i,
+ };
+ debug_assert_eq!(
+ payload.group_index(),
+ persisted_num_group_values + view_buffer_index
+ );
+ let new_header = Entry { hash, payload };
self.map
.insert_accounted(new_header, |h| h.hash, &mut self.map_size);
payload
};
observe_payload_fn(payload);
}
+
+ for pending_entry in view_buffers {
+ match pending_entry {
+ PendingViewBufferEntry::Null { payload } => {
+ let null_index = self.views.len();
+ debug_assert_eq!(payload.group_index(), null_index);
+ self.views.push(0);
+ self.nulls.append_null();
+ self.null = Some((payload, null_index));
+ }
+ PendingViewBufferEntry::NonNull { payload, view, row } => {
+ debug_assert_eq!(payload.group_index(), self.views.len());
+ if (view as u32) <= 12 {
+ unsafe { self.append_inline_view(view) };
+ } else {
+ let value: &[u8] = values.value(row).as_ref();
+ self.append_value(value);
+ }
+ }
+ }
+ }
}
/// Converts this set into a `StringViewArray`, or `BinaryViewArray`,
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org