perf(in_list): add Utf8View short string optimization

Adds specialized filters for Utf8View arrays where all strings are
short enough (≤12 bytes) to be stored inline in the 16-byte view struct.
These can be reinterpreted as i128 for fast equality comparison:

- Utf8ViewHashFilter: uses PrimitiveFilter<Decimal128Type> internally
- Utf8ViewBranchless<N>: branchless filter for ≤4 short strings
- utf8view_all_short_strings(): helper to check if optimization applies

The strategy module checks for Utf8View type and verifies all strings
are short before using the optimized path, falling back to generic
ArrayStaticFilter for longer strings.

Author: geoffreyclaude

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

@@ -62,7 +62,6 @@ impl Debug for InListExpr {
     }
 }
 
-
 /// Evaluates the list of expressions into an array, flattening any dictionaries
 fn evaluate_list(
     list: &[Arc<dyn PhysicalExpr>],

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

@@ -175,30 +175,48 @@ impl<C: BitmapFilterConfig> StaticFilter for BitmapFilter<C> {
 }
 
 // =============================================================================
-// PRIMITIVE FILTER (Hash-based)
+// BRANCHLESS FILTER (Const Generic for Small Lists)
 // =============================================================================
 
-/// Hash-based filter for primitive types with larger IN lists.
-pub(crate) struct PrimitiveFilter<T: ArrowPrimitiveType> {
+/// A branchless filter for very small IN lists (0-16 elements).
+///
+/// Uses const generics to unroll the membership check into a fixed-size
+/// comparison chain, outperforming hash lookups for small lists due to:
+/// - No branching (uses bitwise OR to combine comparisons)
+/// - Better CPU pipelining
+/// - No hash computation overhead
+pub(crate) struct BranchlessFilter<T: ArrowPrimitiveType, const N: usize> {
     null_count: usize,
-    set: HashSet<T::Native>,
+    values: [T::Native; N],
 }
 
-impl<T: ArrowPrimitiveType> PrimitiveFilter<T>
+impl<T: ArrowPrimitiveType, const N: usize> BranchlessFilter<T, N>
 where
-    T::Native: Hash + Eq,
+    T::Native: Copy + PartialEq,
 {
-    pub(crate) fn try_new(in_array: &ArrayRef) -> Result<Self> {
-        let arr = in_array.as_primitive_opt::<T>().ok_or_else(|| {
-            exec_datafusion_err!(
-                "PrimitiveFilter: expected {} array",
-                std::any::type_name::<T>()
-            )
-        })?;
-        Ok(Self {
-            null_count: arr.null_count(),
-            set: arr.iter().flatten().collect(),
-        })
+    /// Try to create a branchless filter if the array has exactly N non-null values.
+    pub(crate) fn try_new(in_array: &ArrayRef) -> Option<Result<Self>> {
+        let in_array = in_array.as_primitive_opt::<T>()?;
+        let non_null_count = in_array.len() - in_array.null_count();
+        if non_null_count != N {
+            return None;
+        }
+        let values: Vec<_> = in_array.iter().flatten().collect();
+        // Use default_value() from ArrowPrimitiveType trait instead of Default::default()
+        let mut arr = [T::default_value(); N];
+        arr.copy_from_slice(&values);
+        Some(Ok(Self {
+            null_count: in_array.null_count(),
+            values: arr,
+        }))
+    }
+
+    /// Branchless membership check using OR-chain.
+    #[inline(always)]
+    fn check(&self, needle: T::Native) -> bool {
+        self.values
+            .iter()
+            .fold(false, |acc, &v| acc | (v == needle))
     }
 
     /// Check membership using a raw values slice (zero-copy path for type reinterpretation).
@@ -214,16 +232,14 @@ where
             nulls,
             self.null_count > 0,
             negated,
-            // SAFETY: i is in bounds since we iterate 0..values.len()
-            |i| self.set.contains(unsafe { values.get_unchecked(i) }),
+            |i| self.check(unsafe { *values.get_unchecked(i) }),
         )
     }
 }
 
-impl<T> StaticFilter for PrimitiveFilter<T>
+impl<T: ArrowPrimitiveType, const N: usize> StaticFilter for BranchlessFilter<T, N>
 where
-    T: ArrowPrimitiveType + 'static,
-    T::Native: Hash + Eq + Send + Sync + 'static,
+    T::Native: Copy + PartialEq + Send + Sync,
 {
     fn null_count(&self) -> usize {
         self.null_count
@@ -232,66 +248,46 @@ where
     fn contains(&self, v: &dyn Array, negated: bool) -> Result<BooleanArray> {
         handle_dictionary!(self, v, negated);
         let v = v.as_primitive_opt::<T>().ok_or_else(|| {
-            exec_datafusion_err!(
-                "PrimitiveFilter: expected {} array",
-                std::any::type_name::<T>()
-            )
+            exec_datafusion_err!("Failed to downcast array to primitive type")
         })?;
-        let values = v.values();
+        let input_values = v.values();
         Ok(build_in_list_result(
             v.len(),
             v.nulls(),
             self.null_count > 0,
             negated,
             // SAFETY: i is in bounds since we iterate 0..v.len()
-            |i| self.set.contains(unsafe { values.get_unchecked(i) }),
+            #[inline(always)]
+            |i| self.check(unsafe { *input_values.get_unchecked(i) }),
         ))
     }
 }
 
 // =============================================================================
-// BRANCHLESS FILTER (Const Generic for Small Lists)
+// PRIMITIVE FILTER (Hash-based)
 // =============================================================================
 
-/// A branchless filter for very small IN lists (0-16 elements).
-///
-/// Uses const generics to unroll the membership check into a fixed-size
-/// comparison chain, outperforming hash lookups for small lists due to:
-/// - No branching (uses bitwise OR to combine comparisons)
-/// - Better CPU pipelining
-/// - No hash computation overhead
-pub(crate) struct BranchlessFilter<T: ArrowPrimitiveType, const N: usize> {
+/// Hash-based filter for primitive types with larger IN lists.
+pub(crate) struct PrimitiveFilter<T: ArrowPrimitiveType> {
     null_count: usize,
-    values: [T::Native; N],
+    set: HashSet<T::Native>,
 }
 
-impl<T: ArrowPrimitiveType, const N: usize> BranchlessFilter<T, N>
+impl<T: ArrowPrimitiveType> PrimitiveFilter<T>
 where
-    T::Native: Copy + PartialEq,
+    T::Native: Hash + Eq,
 {
-    /// Try to create a branchless filter if the array has exactly N non-null values.
-    pub(crate) fn try_new(in_array: &ArrayRef) -> Option<Result<Self>> {
-        let in_array = in_array.as_primitive_opt::<T>()?;
-        let non_null_count = in_array.len() - in_array.null_count();
-        if non_null_count != N {
-            return None;
-        }
-        let values: Vec<_> = in_array.iter().flatten().collect();
-        // Use default_value() from ArrowPrimitiveType trait instead of Default::default()
-        let mut arr = [T::default_value(); N];
-        arr.copy_from_slice(&values);
-        Some(Ok(Self {
-            null_count: in_array.null_count(),
-            values: arr,
-        }))
-    }
-
-    /// Branchless membership check using OR-chain.
-    #[inline(always)]
-    fn check(&self, needle: T::Native) -> bool {
-        self.values
-            .iter()
-            .fold(false, |acc, &v| acc | (v == needle))
+    pub(crate) fn try_new(in_array: &ArrayRef) -> Result<Self> {
+        let arr = in_array.as_primitive_opt::<T>().ok_or_else(|| {
+            exec_datafusion_err!(
+                "PrimitiveFilter: expected {} array",
+                std::any::type_name::<T>()
+            )
+        })?;
+        Ok(Self {
+            null_count: arr.null_count(),
+            set: arr.iter().flatten().collect(),
+        })
     }
 
     /// Check membership using a raw values slice (zero-copy path for type reinterpretation).
@@ -307,14 +303,16 @@ where
             nulls,
             self.null_count > 0,
             negated,
-            |i| self.check(unsafe { *values.get_unchecked(i) }),
+            // SAFETY: i is in bounds since we iterate 0..values.len()
+            |i| self.set.contains(unsafe { values.get_unchecked(i) }),
         )
     }
 }
 
-impl<T: ArrowPrimitiveType, const N: usize> StaticFilter for BranchlessFilter<T, N>
+impl<T> StaticFilter for PrimitiveFilter<T>
 where
-    T::Native: Copy + PartialEq + Send + Sync,
+    T: ArrowPrimitiveType + 'static,
+    T::Native: Hash + Eq + Send + Sync + 'static,
 {
     fn null_count(&self) -> usize {
         self.null_count
@@ -323,17 +321,19 @@ where
     fn contains(&self, v: &dyn Array, negated: bool) -> Result<BooleanArray> {
         handle_dictionary!(self, v, negated);
         let v = v.as_primitive_opt::<T>().ok_or_else(|| {
-            exec_datafusion_err!("Failed to downcast array to primitive type")
+            exec_datafusion_err!(
+                "PrimitiveFilter: expected {} array",
+                std::any::type_name::<T>()
+            )
         })?;
-        let input_values = v.values();
+        let values = v.values();
         Ok(build_in_list_result(
             v.len(),
             v.nulls(),
             self.null_count > 0,
             negated,
             // SAFETY: i is in bounds since we iterate 0..v.len()
-            #[inline(always)]
-            |i| self.check(unsafe { *input_values.get_unchecked(i) }),
+            |i| self.set.contains(unsafe { values.get_unchecked(i) }),
         ))
     }
 }

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

@@ -24,6 +24,7 @@
 //! - 4-byte types (Int32/Float32): branchless (≤16) or hash (>16)
 //! - 8-byte types (Int64/Float64): branchless (≤16) or hash (>16)
 //! - 16-byte types (Decimal128): branchless (≤4) or hash (>4)
+//! - Utf8View (short strings): branchless (≤4) or hash (>4)
 //! - Other types: generic ArrayStaticFilter
 
 use std::hash::Hash;
@@ -36,7 +37,10 @@ use datafusion_common::Result;
 use super::array_filter::ArrayStaticFilter;
 use super::array_filter::StaticFilter;
 use super::primitive::{PrimitiveFilter, U8Config, U16Config};
-use super::transform::{make_bitmap_filter, make_branchless_filter, make_primitive_filter};
+use super::transform::{
+    make_bitmap_filter, make_branchless_filter, make_primitive_filter,
+    make_utf8view_branchless_filter, make_utf8view_hash_filter, utf8view_all_short_strings,
+};
 
 // =============================================================================
 // LOOKUP STRATEGY THRESHOLDS (tuned via microbenchmarks)
@@ -128,6 +132,15 @@ pub(crate) 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)
+        };
+    }
+
     match select_strategy(dt, len) {
         FilterStrategy::Bitmap1B => dispatch_bitmap_u8(&in_array),
         FilterStrategy::Bitmap2B => dispatch_bitmap_u16(&in_array),