IN LIST: reuse primitive filters for FixedSizeBinary

datafusion/physical-expr/benches/in_list_strategy.rs

@@ -15,10 +15,10 @@
 // specific language governing permissions and limitations
 // under the License.
 
-//! Focused benchmarks for `InList` cases.
+//! Benchmarks for static `IN LIST` filters.
 //!
-//! This benchmark file adds targeted coverage for representative `IN LIST`
-//! workloads with controlled parameters:
+//! The cases control match rate and list size across several value types and
+//! string layouts:
 //!
 //! - **Controlled match rates**: Exercises both hit-heavy and miss-heavy paths
 //! - **List size scaling**: Measures behavior across small and large `IN` lists
@@ -27,7 +27,7 @@
 //! - **Shared-prefix strings**: Adds collision-heavy string cases where values
 //!   only differ late in the string
 //! - **Mixed-length strings**: Covers inputs that combine short and long values
-//! - **Null handling**: Includes representative `NULL` and `NOT IN` cases
+//! - **Null handling**: Covers `NULL` and `NOT IN` cases
 //!
 //! # Case Coverage
 //!
@@ -38,24 +38,27 @@
 //! | 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 |
 //! | 128-bit interval cases | IntervalMonthDayNano | small lists | 4 |
+//! | Decimal128 cases | Decimal128 | larger lists | 5, 64 |
 //! | Utf8 short-string cases | Utf8 | 8-byte strings | 4, 64, 256 |
 //! | Utf8 long-string cases | Utf8 | 24-byte strings | 4, 64, 256 |
 //! | Utf8View short-string cases | Utf8View | 8-byte strings | 4, 16, 64, 256 |
 //! | Utf8View length-12 cases | Utf8View | 12-byte strings | 16, 64 |
 //! | Utf8View long-string cases | Utf8View | 24-byte strings | 4, 16, 64, 256 |
 //! | Shared-prefix string cases | Utf8, Utf8View | same prefix, different suffix | 16, 32, 64 |
-//! | Fixed-size binary cases | FixedSizeBinary(16) | fixed-width binary values | 4, 64, 256, 10000 |
+//! | Fixed-size binary cases | FixedSizeBinary(1), FixedSizeBinary(2), FixedSizeBinary(16) | aligned values and unaligned 16-byte inputs | 16 (1 byte), 64 (2 bytes), 4/64/256/10000 (16 bytes) |
 
 use arrow::array::types::IntervalMonthDayNano;
 use arrow::array::*;
+use arrow::buffer::Buffer;
 use arrow::datatypes::{Field, Int32Type, IntervalMonthDayNanoType, Schema};
 use arrow::record_batch::RecordBatch;
 use criterion::{BenchmarkId, Criterion, criterion_group, criterion_main};
-use datafusion_common::ScalarValue;
+use datafusion_common::{HashSet, ScalarValue};
 use datafusion_physical_expr::expressions::{col, in_list, lit};
 use half::f16;
 use rand::distr::Alphanumeric;
 use rand::prelude::*;
+use std::mem::align_of;
 use std::sync::Arc;
 
 const ARRAY_SIZE: usize = 8192;
@@ -463,6 +466,23 @@ fn bench_primitive(c: &mut Criterion) {
         }
     }
 
+    // Decimal128: benchmark the first hash-set list size (5) and a larger list (64).
+    for list_size in [5, 64] {
+        for match_pct in MATCH_RATES {
+            bench_numeric::<i128, Decimal128Array>(
+                c,
+                "primitive",
+                &format!("decimal128/large_list/list={list_size}/match={match_pct}%"),
+                &NumericBenchConfig::new(
+                    list_size,
+                    match_pct as f64 / 100.0,
+                    |rng| i128::from(rng.random::<i64>()),
+                    |v| ScalarValue::Decimal128(Some(v), 38, 10),
+                ),
+            );
+        }
+    }
+
     // NOT IN benchmark: test negated path
     bench_numeric::<i32, Int32Array>(
         c,
@@ -852,7 +872,7 @@ fn bench_dictionary(c: &mut Criterion) {
 // NULL HANDLING BENCHMARKS
 // =============================================================================
 //
-// Tests representative null-containing inputs across primitive and string cases.
+// Null-containing primitive and string cases.
 
 fn bench_nulls(c: &mut Criterion) {
     // =========================================================================
@@ -996,74 +1016,133 @@ fn bench_nulls(c: &mut Criterion) {
 }
 
 // =============================================================================
-// FIXED SIZE BINARY BENCHMARKS (FixedSizeBinary<16>, e.g. UUIDs)
+// FIXED SIZE BINARY BENCHMARKS
 // =============================================================================
 
-/// Generates a random 16-byte value (UUID-sized).
-fn random_fixed_binary_16(rng: &mut StdRng) -> Vec<u8> {
-    let mut buf = vec![0u8; 16];
+fn random_fixed_binary(rng: &mut StdRng, width: i32) -> Vec<u8> {
+    let mut buf = vec![0u8; width as usize];
     rng.fill(&mut buf[..]);
     buf
 }
 
-/// Benchmarks FixedSizeBinary(16) IN list evaluation.
+fn unaligned_fixed_size_binary_16(values: &[Vec<u8>]) -> FixedSizeBinaryArray {
+    const WIDTH: usize = 16;
+    let alignment = align_of::<i128>();
+    let payload_len = values.len() * WIDTH;
+    let mut bytes = vec![0_u8; payload_len + alignment];
+    let offset = usize::from(bytes.as_ptr().align_offset(alignment) == 0);
+    for (target, value) in bytes[offset..offset + payload_len]
+        .chunks_exact_mut(WIDTH)
+        .zip(values.iter())
+    {
+        assert_eq!(value.len(), WIDTH);
+        target.copy_from_slice(value);
+    }
+    let buffer = Buffer::from(bytes).slice_with_length(offset, payload_len);
+    assert_ne!(
+        buffer.as_ptr().align_offset(alignment),
+        0,
+        "benchmark input must be unaligned"
+    );
+    FixedSizeBinaryArray::new(WIDTH as i32, buffer, None)
+}
+
 /// FixedSizeBinary doesn't use the generic numeric helpers since its array
 /// construction differs from primitive types.
 fn bench_fixed_size_binary_inner(
     c: &mut Criterion,
-    name: &str,
+    width: i32,
     list_size: usize,
-    match_rate: f64,
+    match_pct: u32,
+    unaligned_input: bool,
 ) {
-    let seed = 0xF1ED_B1A7_u64.wrapping_add(list_size as u64 * 0x6666);
+    assert!(match_pct <= 100);
+    if let Some(domain_size) = match width {
+        1 => Some(1_usize << 8),
+        2 => Some(1_usize << 16),
+        _ => None,
+    } {
+        // The input generator needs at least one value outside the list.
+        assert!(list_size < domain_size);
+    }
+    let match_rate = f64::from(match_pct) / 100.0;
+
+    let seed = 0xF1ED_B1A7_u64
+        .wrapping_add(list_size as u64 * 0x6666)
+        .wrapping_add(width as u64 * 0x7777);
     let mut rng = StdRng::seed_from_u64(seed);
 
-    // Generate IN list values (16-byte each)
-    let haystack: Vec<Vec<u8>> = (0..list_size)
-        .map(|_| random_fixed_binary_16(&mut rng))
-        .collect();
+    // Keep the number of distinct values equal to the configured list size.
+    let mut haystack_set = HashSet::with_capacity(list_size);
+    let mut haystack = Vec::with_capacity(list_size);
+    while haystack.len() < list_size {
+        let value = random_fixed_binary(&mut rng, width);
+        if haystack_set.insert(value.clone()) {
+            haystack.push(value);
+        }
+    }
 
     // Generate array with controlled match rate
     let values: Vec<Vec<u8>> = (0..ARRAY_SIZE)
         .map(|_| {
             if !haystack.is_empty() && rng.random_bool(match_rate) {
                 haystack.choose(&mut rng).unwrap().clone()
             } else {
-                random_fixed_binary_16(&mut rng)
+                loop {
+                    let value = random_fixed_binary(&mut rng, width);
+                    if !haystack_set.contains(&value) {
+                        break value;
+                    }
+                }
             }
         })
         .collect();
+    drop(haystack_set);
 
-    let refs: Vec<&[u8]> = values.iter().map(|v| v.as_slice()).collect();
-    let array = FixedSizeBinaryArray::try_from_iter(refs.into_iter()).unwrap();
+    let array = if unaligned_input {
+        assert_eq!(width, 16);
+        unaligned_fixed_size_binary_16(&values)
+    } else {
+        FixedSizeBinaryArray::try_from_iter(values.iter().map(Vec::as_slice)).unwrap()
+    };
 
     let schema = Schema::new(vec![Field::new("a", array.data_type().clone(), true)]);
     let exprs: Vec<_> = haystack
         .iter()
-        .map(|v| lit(ScalarValue::FixedSizeBinary(16, Some(v.clone()))))
+        .map(|v| lit(ScalarValue::FixedSizeBinary(width, Some(v.clone()))))
         .collect();
     let expr = in_list(col("a", &schema).unwrap(), exprs, &false, &schema).unwrap();
     let batch = RecordBatch::try_new(Arc::new(schema), vec![Arc::new(array) as ArrayRef])
         .unwrap();
 
     c.bench_with_input(
-        BenchmarkId::new("fixed_size_binary", name),
+        BenchmarkId::new("fixed_size_binary", {
+            let name = format!("fsb{width}/list={list_size}/match={match_pct}%");
+            if unaligned_input {
+                format!("{name}/input=unaligned")
+            } else {
+                name
+            }
+        }),
         &batch,
         |b, batch| b.iter(|| expr.evaluate(batch).unwrap()),
     );
 }
 
 fn bench_fixed_size_binary(c: &mut Criterion) {
-    for list_size in [4, 64, 256, 10000] {
+    for (width, list_size) in
+        [(1, 16), (2, 64), (16, 4), (16, 64), (16, 256), (16, 10000)]
+    {
         for match_pct in MATCH_RATES {
-            bench_fixed_size_binary_inner(
-                c,
-                &format!("fsb16/list={list_size}/match={match_pct}%"),
-                list_size,
-                match_pct as f64 / 100.0,
-            );
+            bench_fixed_size_binary_inner(c, width, list_size, match_pct, false);
         }
     }
+
+    // At 16 bytes per value, an unaligned 8,192-row input copies 128 KiB per
+    // evaluation. List size 64 exercises the larger-list hash-set path.
+    for match_pct in MATCH_RATES {
+        bench_fixed_size_binary_inner(c, 16, 64, match_pct, true);
+    }
 }
 
 // =============================================================================

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

@@ -38,20 +38,21 @@ use datafusion_expr::{ColumnarValue, expr_vec_fmt};
 
 mod array_static_filter;
 mod branchless_filter;
+mod fixed_size_binary_filter;
 mod primitive_filter;
 mod result;
 mod static_filter;
 mod strategy;
 
-use static_filter::StaticFilter;
+use static_filter::StaticFilterRef;
 use strategy::instantiate_static_filter;
 
 /// InList
 pub struct InListExpr {
     expr: Arc<dyn PhysicalExpr>,
     list: Vec<Arc<dyn PhysicalExpr>>,
     negated: bool,
-    static_filter: Option<Arc<dyn StaticFilter + Send + Sync>>,
+    static_filter: Option<StaticFilterRef>,
 }
 
 impl Debug for InListExpr {
@@ -148,7 +149,7 @@ impl InListExpr {
         expr: Arc<dyn PhysicalExpr>,
         list: Vec<Arc<dyn PhysicalExpr>>,
         negated: bool,
-        static_filter: Option<Arc<dyn StaticFilter + Send + Sync>>,
+        static_filter: Option<StaticFilterRef>,
     ) -> Self {
         Self {
             expr,
@@ -222,8 +223,8 @@ impl InListExpr {
     /// 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 branchless, bitmap, or hash-set filters
+    /// when the list's physical representation supports them.
     ///
     /// Returns an error if data types don't match. If the list contains non-constant
     /// expressions, falls back to dynamic evaluation at runtime.
@@ -2592,7 +2593,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(), which selects a primitive static filter.
         let expr = in_list(col_a, list, &false, &schema)?;
 
         // Create dictionary-encoded batch with values [100, 200, 500]
@@ -3548,6 +3549,38 @@ mod tests {
             );
         }
 
+        // FixedSizeBinary in_array, FixedSizeBinary and Dictionary needles
+        let fsb_in = Arc::new(FixedSizeBinaryArray::try_from_iter(
+            [
+                [1, 2, 3, 4].as_slice(),
+                [5, 6, 7, 8].as_slice(),
+                [9, 10, 11, 12].as_slice(),
+            ]
+            .into_iter(),
+        )?) as ArrayRef;
+        let fsb_needle = Arc::new(FixedSizeBinaryArray::try_from_iter(
+            [
+                [1, 2, 3, 4].as_slice(),
+                [13, 14, 15, 16].as_slice(),
+                [5, 6, 7, 8].as_slice(),
+            ]
+            .into_iter(),
+        )?) as ArrayRef;
+        assert_eq!(
+            expected,
+            eval_in_list_from_array(Arc::clone(&fsb_needle), Arc::clone(&fsb_in))?
+        );
+        // The dictionary does not reference its second value, so that value
+        // must not become a member of the flattened list.
+        let dict_fsb_in = Arc::new(DictionaryArray::new(
+            Int32Array::from(vec![0, 2]),
+            Arc::clone(&fsb_in),
+        ));
+        assert_eq!(
+            BooleanArray::from(vec![Some(true), Some(false), Some(false)]),
+            eval_in_list_from_array(wrap_in_dict(fsb_needle), dict_fsb_in)?
+        );
+
         // Utf8 (falls through to ArrayStaticFilter)
         let utf8_in = Arc::new(StringArray::from(vec!["a", "b", "c"])) as ArrayRef;
         let utf8_needle = Arc::new(StringArray::from(vec!["a", "d", "b"])) as ArrayRef;

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

@@ -42,7 +42,7 @@
 //! different NaN values. [`BranchlessFilterType`] defines these safe,
 //! same-sized mappings and checks their sizes at compile time.
 //!
-//! The fast path is intentionally limited to short lists:
+//! The fast path is limited to short lists:
 //!
 //! - 16 values for 1-byte types
 //! - 8 values for 2-byte types