IN LIST: optimize Utf8View and BinaryView filters

datafusion/physical-expr/benches/in_list_strategy.rs

@@ -35,8 +35,8 @@
 //! |------|-------|-----------------|-------------------|
 //! | Narrow integer cases | UInt8 | small value domain | 4, 16 |
 //! | Narrow integer cases | Int16, Float16 | larger value domain | 4, 64, 256 |
-//! | 32-bit primitive cases | Int32, Float32 | small and large lists | 4, 32, 64, 256 |
-//! | 64-bit primitive cases | Int64, TimestampNs | small and large lists | 4, 16, 32, 128 |
+//! | 32-bit primitive cases | Int32, Float32 | small and large lists | 4, 32, 33, 64, 256, 1024, 10000 |
+//! | 64-bit primitive cases | Int64, TimestampNs | small and large lists | 4, 16, 17, 32, 128, 1024, 10000 |
 //! | 128-bit interval cases | IntervalMonthDayNano | small lists | 4 |
 //! | Utf8 short-string cases | Utf8 | 8-byte strings | 4, 64, 256 |
 //! | Utf8 long-string cases | Utf8 | 24-byte strings | 4, 64, 256 |
@@ -166,11 +166,11 @@ fn random_string(rng: &mut StdRng, len: usize) -> String {
 fn strings_with_shared_prefix(
     rng: &mut StdRng,
     count: usize,
-    prefix_len: usize,
+    prefix: &str,
+    discriminator: char,
 ) -> Vec<String> {
-    let prefix = random_string(rng, prefix_len);
     (0..count)
-        .map(|_| format!("{}{}", prefix, random_string(rng, 8))) // prefix + random 8-char suffix
+        .map(|_| format!("{prefix}{}{discriminator}", random_string(rng, 7)))
         .collect()
 }
 
@@ -275,12 +275,14 @@ fn bench_string_shared_prefix<A>(
         .wrapping_add(prefix_len as u64 * 0x4444);
     let mut rng = StdRng::seed_from_u64(seed);
 
-    // Generate IN list with a shared prefix.
-    let haystack = strings_with_shared_prefix(&mut rng, list_size, prefix_len);
+    // Use the same prefix and equal-length, disjoint suffixes for both pools so
+    // misses exercise length/prefix collisions rather than immediate rejection.
+    let prefix = random_string(&mut rng, prefix_len);
+    let haystack = strings_with_shared_prefix(&mut rng, list_size, &prefix, 'h');
 
     // Generate non-matching strings with the same prefix to keep misses close
     // to the matching set.
-    let non_match_pool = strings_with_shared_prefix(&mut rng, 100, prefix_len);
+    let non_match_pool = strings_with_shared_prefix(&mut rng, 100, &prefix, 'm');
 
     // Generate array with controlled match rate
     let values: A = (0..ARRAY_SIZE)
@@ -313,19 +315,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 +349,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)
             })
@@ -420,7 +437,7 @@ fn bench_narrow_integer(c: &mut Criterion) {
 
 fn bench_primitive(c: &mut Criterion) {
     // Int32: small and larger list sizes
-    for list_size in [4, 32, 64, 256] {
+    for list_size in [4, 32, 33, 64, 256, 1024, 10_000] {
         let list_case = if list_size <= 32 {
             "small_list"
         } else {
@@ -442,7 +459,7 @@ fn bench_primitive(c: &mut Criterion) {
     }
 
     // Int64: small and larger list sizes
-    for list_size in [4, 16, 32, 128] {
+    for list_size in [4, 16, 17, 32, 128, 1024, 10_000] {
         let list_case = if list_size <= 16 {
             "small_list"
         } else {
@@ -602,6 +619,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 +712,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,8 @@ use datafusion_expr::{ColumnarValue, expr_vec_fmt};
 
 mod array_static_filter;
 mod branchless_filter;
+mod byte_view_filter;
+mod frozen_set;
 mod primitive_filter;
 mod result;
 mod static_filter;
@@ -215,15 +217,15 @@ impl InListExpr {
             expr,
             list,
             negated,
-            Some(instantiate_static_filter(array)?),
+            Some(instantiate_static_filter(array, &expr_data_type)?),
         ))
     }
 
     /// Create a new InList expression, using a static filter when possible.
     ///
     /// This validates data types and attempts to create a static filter for constant
-    /// list expressions. Uses specialized StaticFilter implementations for better
-    /// performance (e.g., Int32StaticFilter for Int32).
+    /// list expressions. Uses specialized `StaticFilter` implementations for better
+    /// performance (for example, primitive filters for fixed-width values).
     ///
     /// Returns an error if data types don't match. If the list contains non-constant
     /// expressions, falls back to dynamic evaluation at runtime.
@@ -242,7 +244,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
         };
 
@@ -2592,7 +2594,7 @@ mod tests {
         // Create IN list with Int32 literals: (100, 200, 300)
         let list = vec![lit(100i32), lit(200i32), lit(300i32)];
 
-        // Create InListExpr via in_list() - this uses Int32StaticFilter for Int32 lists
+        // Create InListExpr via in_list() - this uses a specialized primitive filter
         let expr = in_list(col_a, list, &false, &schema)?;
 
         // Create dictionary-encoded batch with values [100, 200, 500]
@@ -3576,6 +3578,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

@@ -58,8 +58,9 @@
 //! would be useful. Wider types have too many possible values for such a
 //! bitmap, so their limits are tuned separately.
 //!
-//! Larger lists use the standard filter strategy, including bitmap filters for
-//! one- and two-byte types.
+//! Larger lists use another filter strategy: bitmap filters for one- and
+//! two-byte types, frozen sets for supported four- and eight-byte types, and
+//! the standard fallback for the remaining types.
 //!
 //! # What about nulls?
 //!
@@ -71,7 +72,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 +245,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 +284,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))
     }
 }