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 81bab74b3b Instrument byte view group comparisons
81bab74b3b is described below
commit 81bab74b3b233de36f905a60eb29c7ab78faf4a5
Author: kamille <[email protected]>
AuthorDate: Thu Jul 9 05:51:43 2026 +0800
Instrument byte view group comparisons
---
.../physical-expr-common/src/binary_view_map.rs | 61 ++++++++++++++++++++++
.../src/aggregates/group_values/mod.rs | 8 ++-
.../aggregates/group_values/single_group_by/mod.rs | 1 +
3 files changed, 69 insertions(+), 1 deletion(-)
diff --git a/datafusion/physical-expr-common/src/binary_view_map.rs
b/datafusion/physical-expr-common/src/binary_view_map.rs
index 96bc8a49e8..4cbf827e38 100644
--- a/datafusion/physical-expr-common/src/binary_view_map.rs
+++ b/datafusion/physical-expr-common/src/binary_view_map.rs
@@ -119,6 +119,49 @@ 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,
@@ -273,6 +316,7 @@ 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();
for i in 0..values.len() {
let view_u128 = input_views[i];
let hash = self.hashes_buffer[i];
@@ -310,18 +354,22 @@ 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,
)
}
})
@@ -366,15 +414,26 @@ where
};
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 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) };
@@ -401,6 +460,8 @@ 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;
diff --git a/datafusion/physical-plan/src/aggregates/group_values/mod.rs
b/datafusion/physical-plan/src/aggregates/group_values/mod.rs
index a3a34d3fd3..ee253e5d7a 100644
--- a/datafusion/physical-plan/src/aggregates/group_values/mod.rs
+++ b/datafusion/physical-plan/src/aggregates/group_values/mod.rs
@@ -41,7 +41,7 @@ pub(crate) use single_group_by::primitive::HashValue;
use crate::aggregates::{
group_values::single_group_by::{
boolean::GroupValuesBoolean, bytes::GroupValuesBytes,
- primitive::GroupValuesPrimitive,
+ bytes_view::GroupValuesBytesView, primitive::GroupValuesPrimitive,
},
order::GroupOrdering,
};
@@ -181,12 +181,18 @@ pub fn new_group_values(
DataType::LargeUtf8 => {
return
Ok(Box::new(GroupValuesBytes::<i64>::new(OutputType::Utf8)));
}
+ DataType::Utf8View => {
+ return
Ok(Box::new(GroupValuesBytesView::new(OutputType::Utf8View)));
+ }
DataType::Binary => {
return
Ok(Box::new(GroupValuesBytes::<i32>::new(OutputType::Binary)));
}
DataType::LargeBinary => {
return
Ok(Box::new(GroupValuesBytes::<i64>::new(OutputType::Binary)));
}
+ DataType::BinaryView => {
+ return
Ok(Box::new(GroupValuesBytesView::new(OutputType::BinaryView)));
+ }
DataType::Boolean => {
return Ok(Box::new(GroupValuesBoolean::new()));
}
diff --git
a/datafusion/physical-plan/src/aggregates/group_values/single_group_by/mod.rs
b/datafusion/physical-plan/src/aggregates/group_values/single_group_by/mod.rs
index e257596f8a..89c6b624e8 100644
---
a/datafusion/physical-plan/src/aggregates/group_values/single_group_by/mod.rs
+++
b/datafusion/physical-plan/src/aggregates/group_values/single_group_by/mod.rs
@@ -19,4 +19,5 @@
pub(crate) mod boolean;
pub(crate) mod bytes;
+pub(crate) mod bytes_view;
pub(crate) mod primitive;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
