Implement String View (Utf8View/BinaryView) Optimizations

Introduces a two-stage filter for ByteView types. Stage 1 uses a fast DirectProbeFilter on masked views (len + prefix) for quick rejection; Stage 2 performs full verification only for potential long-string matches. Triggers for Utf8View and BinaryView.

Author: geoffreyclaude

datafusion/physical-expr/src/expressions/in_list/primitive_filter.rs

@@ -376,6 +376,12 @@ where
         Ok(Self::from_unique_values(unique_values, arr.null_count()))
     }
 
+    /// Creates a DirectProbeFilter from pre-processed values.
+    pub(super) fn from_values(values: impl Iterator<Item = T::Native>) -> Self {
+        let unique_values: HashSet<_> = values.collect();
+        Self::from_unique_values(unique_values, 0)
+    }
+
     fn from_unique_values(unique_values: HashSet<T::Native>, null_count: usize) -> Self {
         // Size table to ~25% load factor for fewer collisions
         let n = unique_values.len().max(1);
@@ -410,7 +416,7 @@ where
     ///
     /// Returns true if the value is in the set.
     #[inline(always)]
-    fn contains_single(&self, needle: T::Native) -> bool {
+    pub(super) fn contains_single(&self, needle: T::Native) -> bool {
         let mut slot = needle.probe_hash() & self.mask;
         loop {
             // SAFETY: `slot` is always < table.len() because:

datafusion/physical-expr/src/expressions/in_list/result.rs

@@ -58,6 +58,69 @@ where
     build_result_from_contains(needle_nulls, haystack_has_nulls, negated, contains_buf)
 }
 
+/// Builds a BooleanArray result while skipping contains checks for null needles.
+///
+/// This is useful when `contains` is expensive, for example when a string path
+/// may need hashing and full byte comparison. If there are no actual nulls, it
+/// falls back to the same branch-free contains collection as
+/// `build_in_list_result`.
+#[inline]
+pub(crate) fn build_in_list_result_with_null_shortcircuit<C>(
+    len: usize,
+    needle_nulls: Option<&NullBuffer>,
+    needle_null_count: usize,
+    haystack_has_nulls: bool,
+    negated: bool,
+    mut contains: C,
+) -> BooleanArray
+where
+    C: FnMut(usize) -> bool,
+{
+    let effective_nulls = needle_nulls.filter(|_| needle_null_count > 0);
+
+    let contains_buf = match effective_nulls {
+        Some(nulls) => {
+            BooleanBuffer::collect_bool(len, |i| nulls.is_valid(i) && contains(i))
+        }
+        None => BooleanBuffer::collect_bool(len, contains),
+    };
+
+    build_result_from_premasked_contains(
+        effective_nulls,
+        haystack_has_nulls,
+        negated,
+        contains_buf,
+    )
+}
+
+/// Builds a result from a contains buffer that is already false at null needles.
+#[inline]
+fn build_result_from_premasked_contains(
+    needle_nulls: Option<&NullBuffer>,
+    haystack_has_nulls: bool,
+    negated: bool,
+    contains_buf: BooleanBuffer,
+) -> BooleanArray {
+    match (needle_nulls, haystack_has_nulls, negated) {
+        (_, true, false) => {
+            BooleanArray::new(contains_buf.clone(), Some(NullBuffer::new(contains_buf)))
+        }
+        (Some(v), true, true) => BooleanArray::new(
+            v.inner() ^ &contains_buf,
+            Some(NullBuffer::new(contains_buf)),
+        ),
+        (None, true, true) => {
+            BooleanArray::new(!&contains_buf, Some(NullBuffer::new(contains_buf)))
+        }
+        (Some(v), false, false) => BooleanArray::new(contains_buf, Some(v.clone())),
+        (Some(v), false, true) => {
+            BooleanArray::new(v.inner() & &(!&contains_buf), Some(v.clone()))
+        }
+        (None, false, false) => BooleanArray::new(contains_buf, None),
+        (None, false, true) => BooleanArray::new(!&contains_buf, None),
+    }
+}
+
 /// Builds a BooleanArray result from a pre-computed contains buffer.
 ///
 /// This version does not assume contains_buf is pre-masked at null positions.

datafusion/physical-expr/src/expressions/in_list/strategy.rs

@@ -26,7 +26,9 @@ use super::array_static_filter::ArrayStaticFilter;
 use super::primitive_filter::*;
 use super::static_filter::{StaticFilter, handle_dictionary};
 use super::transform::{
-    make_bitmap_filter, make_branchless_filter, reinterpret_any_primitive_to,
+    make_bitmap_filter, make_branchless_filter, make_byte_view_masked_filter,
+    make_utf8view_branchless_filter, make_utf8view_hash_filter,
+    reinterpret_any_primitive_to, utf8view_all_short_strings,
 };
 
 /// Maximum list size for branchless lookup on 4-byte primitives (Int32, UInt32, Float32).
@@ -97,6 +99,16 @@ pub(super) fn instantiate_static_filter(
 
     let len = in_array.len();
     let dt = in_array.data_type();
+
+    // Special case: Utf8View with short strings can be reinterpreted as i128
+    if matches!(dt, DataType::Utf8View) && utf8view_all_short_strings(in_array.as_ref()) {
+        return if len <= BRANCHLESS_MAX_16B {
+            make_utf8view_branchless_filter(&in_array)
+        } else {
+            make_utf8view_hash_filter(&in_array)
+        };
+    }
+
     let strategy = select_strategy(dt, len);
 
     match (dt, strategy) {
@@ -117,6 +129,14 @@ pub(super) fn instantiate_static_filter(
             exec_datafusion_err!("Hashed strategy selected but no filter for {:?}", dt)
         })?,
 
+        // Byte view filters (Utf8View, BinaryView)
+        (DataType::Utf8View, Generic) => {
+            make_byte_view_masked_filter::<StringViewType>(in_array)
+        }
+        (DataType::BinaryView, Generic) => {
+            make_byte_view_masked_filter::<BinaryViewType>(in_array)
+        }
+
         // Fallback for nested/complex types and strings.
         (_, Generic) => Ok(Arc::new(ArrayStaticFilter::try_new(in_array)?)),
     }