IN LIST: optimize Utf8View and BinaryView filters with standard hash sets

datafusion/physical-expr/benches/in_list_strategy.rs

@@ -313,19 +313,29 @@ fn bench_string_mixed_lengths<A>(
     name: &str,
     list_size: usize,
     match_rate: f64,
+    inline_rate: f64,
     to_scalar: fn(String) -> ScalarValue,
 ) where
     A: Array + FromIterator<Option<String>> + 'static,
 {
     let seed = 0xABCD_EF01_u64.wrapping_add(list_size as u64 * 0x5555);
     let mut rng = StdRng::seed_from_u64(seed);
 
-    // Mixed lengths: some short (<= 12), some long (> 12)
-    let lengths = [4, 8, 12, 16, 20, 24];
+    let inline_lengths = [4, 8, 12];
+    let long_lengths = [16, 20, 24];
+    let inline_count = ((list_size as f64 * inline_rate).round() as usize)
+        .max(1)
+        .min(list_size - 1);
 
     // Generate IN list with mixed lengths
     let haystack: Vec<String> = (0..list_size)
-        .map(|_| {
+        .map(|idx| {
+            let inline = idx < inline_count;
+            let lengths = if inline {
+                &inline_lengths
+            } else {
+                &long_lengths
+            };
             let len = *lengths.choose(&mut rng).unwrap();
             random_string(&mut rng, len)
         })
@@ -337,6 +347,11 @@ fn bench_string_mixed_lengths<A>(
             Some(if !haystack.is_empty() && rng.random_bool(match_rate) {
                 haystack.choose(&mut rng).unwrap().clone()
             } else {
+                let lengths = if rng.random_bool(inline_rate) {
+                    &inline_lengths
+                } else {
+                    &long_lengths
+                };
                 let len = *lengths.choose(&mut rng).unwrap();
                 random_string(&mut rng, len)
             })
@@ -602,6 +617,7 @@ fn bench_utf8(c: &mut Criterion) {
                 &format!("mixed_len/list={list_size}/match={match_pct}%"),
                 list_size,
                 match_pct as f64 / 100.0,
+                0.5,
                 to_scalar,
             );
         }
@@ -694,6 +710,23 @@ fn bench_utf8view(c: &mut Criterion) {
                 &format!("mixed_len/list={list_size}/match={match_pct}%"),
                 list_size,
                 match_pct as f64 / 100.0,
+                0.5,
+                to_scalar,
+            );
+        }
+    }
+
+    // Strongly skewed mixed lists exercise routing near the all-inline and
+    // all-long boundaries while retaining both representations.
+    for inline_pct in [2, 98] {
+        for match_pct in MATCH_RATES {
+            bench_string_mixed_lengths::<StringViewArray>(
+                c,
+                "utf8view",
+                &format!("mixed_len/inline={inline_pct}%/list=64/match={match_pct}%"),
+                64,
+                match_pct as f64 / 100.0,
+                inline_pct as f64 / 100.0,
                 to_scalar,
             );
         }

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

@@ -38,6 +38,7 @@ use datafusion_expr::{ColumnarValue, expr_vec_fmt};
 
 mod array_static_filter;
 mod branchless_filter;
+mod byte_view_filter;
 mod primitive_filter;
 mod result;
 mod static_filter;
@@ -215,7 +216,7 @@ impl InListExpr {
             expr,
             list,
             negated,
-            Some(instantiate_static_filter(array)?),
+            Some(instantiate_static_filter(array, &expr_data_type)?),
         ))
     }
 
@@ -242,7 +243,7 @@ impl InListExpr {
 
         // Try to create a static filter if all list expressions are constants
         let static_filter = match try_evaluate_constant_list(&list, schema)? {
-            Some(in_array) => Some(instantiate_static_filter(in_array)?),
+            Some(in_array) => Some(instantiate_static_filter(in_array, &expr_data_type)?),
             None => None, // Non-constant expressions, fall back to dynamic evaluation
         };
 
@@ -3576,6 +3577,23 @@ mod tests {
             )?
         );
 
+        // Utf8View in_array, Utf8View and Dict(Utf8View) needles
+        let utf8view_in =
+            Arc::new(StringViewArray::from(vec!["a", "b", "c"])) as ArrayRef;
+        let utf8view_needle =
+            Arc::new(StringViewArray::from(vec!["a", "d", "b"])) as ArrayRef;
+        assert_eq!(
+            expected,
+            eval_in_list_from_array(
+                Arc::clone(&utf8view_needle),
+                Arc::clone(&utf8view_in),
+            )?
+        );
+        assert_eq!(
+            expected,
+            eval_in_list_from_array(wrap_in_dict(utf8view_needle), utf8view_in)?
+        );
+
         // Struct in_array, Struct needle: multi-column join
         let struct_fields = Fields::from(vec![
             Field::new("c0", DataType::Utf8, true),

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

@@ -71,7 +71,7 @@
 use std::mem::size_of;
 
 use arrow::array::{Array, ArrayRef, AsArray, BooleanArray, PrimitiveArray};
-use arrow::buffer::{BooleanBuffer, ScalarBuffer};
+use arrow::buffer::{BooleanBuffer, NullBuffer, ScalarBuffer};
 use arrow::datatypes::*;
 use arrow::util::bit_iterator::BitIndexIterator;
 use datafusion_common::{Result, exec_datafusion_err, internal_datafusion_err};
@@ -244,6 +244,17 @@ where
             check_values,
         })
     }
+
+    #[inline]
+    pub(super) fn contains_slice(
+        &self,
+        input_values: &[BranchlessNative<T>],
+        nulls: Option<&NullBuffer>,
+        negated: bool,
+    ) -> BooleanArray {
+        let matches = (self.check_values)(self.in_list_values.as_ref(), input_values);
+        build_result_from_contains(nulls, self.null_count > 0, negated, matches)
+    }
 }
 
 impl<T> StaticFilter for BranchlessFilter<T>
@@ -272,14 +283,7 @@ where
             exec_datafusion_err!("BranchlessFilter: expected {} array", T::DATA_TYPE)
         })?;
         let input_values = branchless_values::<T>(v);
-        let matches =
-            (self.check_values)(self.in_list_values.as_ref(), input_values.as_ref());
-        Ok(build_result_from_contains(
-            v.nulls(),
-            self.null_count > 0,
-            negated,
-            matches,
-        ))
+        Ok(self.contains_slice(input_values.as_ref(), v.nulls(), negated))
     }
 }