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 74c9b4f03b653d32af829430e5c6c93ca167dc90 Author: kamille <[email protected]> AuthorDate: Fri Jul 10 09:03:50 2026 +0800 add fast path. --- .../physical-expr-common/src/binary_view_map.rs | 146 +++++++++------------ 1 file changed, 62 insertions(+), 84 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..0e88656eec 100644 --- a/datafusion/physical-expr-common/src/binary_view_map.rs +++ b/datafusion/physical-expr-common/src/binary_view_map.rs @@ -273,29 +273,21 @@ 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 current_group_num = self.views.len(); + 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 @@ -311,14 +303,11 @@ where 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; + let input_buffers = values.data_buffers(); self.map .find(hash, |header| { @@ -326,27 +315,32 @@ where return false; } - let stored_view = self.views[header.payload.group_index()]; + let group_index = header.payload.group_index(); + let stored_view = self.views[group_index]; - // Fast path: inline strings can be compared directly if len <= 12 { return stored_view == view_u128; } - // For larger strings: first compare the 4-byte prefix + if group_index >= current_group_num && stored_view == view_u128 { + return true; + } + let stored_prefix = (stored_view >> 32) as u32; let input_prefix = (view_u128 >> 32) as u32; if stored_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() { + let stored_value = if group_index >= current_group_num { + &input_buffers[buffer_index].as_slice() + [offset..offset + stored_len] + } else if buffer_index < completed.len() { &completed[buffer_index].as_slice() [offset..offset + stored_len] } else { @@ -361,37 +355,68 @@ where let payload = if let Some(payload) = maybe_payload { payload } else { - // no existing value, make a new one 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 }; + debug_assert_eq!(payload.group_index(), self.views.len()); + self.views.push(view_u128); + self.nulls.append_non_null(); + let new_header = Entry { hash, payload }; self.map .insert_accounted(new_header, |h| h.hash, &mut self.map_size); payload }; observe_payload_fn(payload); } + + self.materialize_new_non_inline_views::<B>(current_group_num, values); + } + + fn materialize_new_non_inline_views<B>( + &mut self, + current_group_num: usize, + values: &arrow::array::GenericByteViewArray<B>, + ) where + B: ByteViewType, + { + for group_index in current_group_num..self.views.len() { + let view = self.views[group_index]; + if (view as u32) <= 12 { + continue; + } + + let byte_view = ByteView::from(view); + let buffer_index = byte_view.buffer_index as usize; + let offset = byte_view.offset as usize; + let length = byte_view.length as usize; + let value = + &values.data_buffers()[buffer_index].as_slice()[offset..offset + length]; + self.views[group_index] = self.append_value_to_buffers(value); + } + } + + fn append_value_to_buffers(&mut self, value: &[u8]) -> u128 { + debug_assert!(value.len() > 12); + + if self.in_progress.len() + value.len() > BYTE_VIEW_MAX_BLOCK_SIZE { + let flushed = std::mem::replace( + &mut self.in_progress, + Vec::with_capacity(BYTE_VIEW_MAX_BLOCK_SIZE), + ); + self.completed.push(Buffer::from_vec(flushed)); + } + + let buffer_index = self.completed.len() as u32; + let offset = self.in_progress.len() as u32; + self.in_progress.extend_from_slice(value); + make_view(value, buffer_index, offset) } /// Converts this set into a `StringViewArray`, or `BinaryViewArray`, @@ -425,53 +450,6 @@ where } } - /// Append an already-computed inline view (len <= 12) directly, bypassing - /// buffer allocation. - /// - /// Returns the view that was stored (identical to the argument). - /// - /// # Safety - /// - /// `view` must be a valid inline `ByteView`: the length field in the low - /// 32 bits must be <= 12, and the remaining 12 bytes must hold the - /// value's bytes (zero-padded if shorter). Calling with a non-inline view - /// would store a value that downstream `views` consumers interpret as - /// `[buffer_index, offset]` into the `completed`/`in_progress` buffers, - /// which is unsound for any view that didn't originate from a real - /// allocation in those buffers. - unsafe fn append_inline_view(&mut self, view: u128) -> u128 { - self.views.push(view); - self.nulls.append_non_null(); - view - } - - /// Append a value to our buffers and return the view pointing to it - fn append_value(&mut self, value: &[u8]) -> u128 { - let len = value.len(); - let view = if len <= 12 { - make_view(value, 0, 0) - } else { - // Ensure buffer is big enough - if self.in_progress.len() + len > BYTE_VIEW_MAX_BLOCK_SIZE { - let flushed = std::mem::replace( - &mut self.in_progress, - Vec::with_capacity(BYTE_VIEW_MAX_BLOCK_SIZE), - ); - self.completed.push(Buffer::from_vec(flushed)); - } - - let buffer_index = self.completed.len() as u32; - let offset = self.in_progress.len() as u32; - self.in_progress.extend_from_slice(value); - - make_view(value, buffer_index, offset) - }; - - self.views.push(view); - self.nulls.append_non_null(); - view - } - /// Total number of entries (including null, if present) pub fn len(&self) -> usize { self.non_null_len() + self.null.map(|_| 1).unwrap_or(0) --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
