Implement FixedSizeBinary zero-copy reinterpretation optimization

FixedSizeBinary(N) arrays share the same contiguous buffer layout as primitive arrays, so for power-of-2 widths (1, 2, 4, 8, 16) we can zero-copy reinterpret them and use the optimized primitive filters (bitmap, branchless, hash) instead of falling through to the NestedTypeFilter fallback.

Author: geoffreyclaude

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

@@ -853,6 +853,135 @@ mod tests {
         ])
     }
 
+    #[test]
+    fn test_in_list_fixed_size_binary_canonical_path() -> Result<()> {
+        let width = 16;
+        let matching = vec![0x10; width as usize];
+        let also_in_list = vec![0x20; width as usize];
+        let not_in_list = vec![0x30; width as usize];
+
+        let schema = Schema::new(vec![Field::new(
+            "a",
+            DataType::FixedSizeBinary(width),
+            true,
+        )]);
+        let col_a = col("a", &schema)?;
+        let list_array = Arc::new(FixedSizeBinaryArray::from(vec![
+            matching.as_slice(),
+            also_in_list.as_slice(),
+        ])) as ArrayRef;
+        let expr = Arc::new(InListExpr::try_new_from_array(
+            Arc::clone(&col_a),
+            list_array,
+            false,
+        )?) as Arc<dyn PhysicalExpr>;
+
+        let batch_array = Arc::new(FixedSizeBinaryArray::try_from_sparse_iter_with_size(
+            vec![
+                Some(matching.as_slice()),
+                Some(not_in_list.as_slice()),
+                None,
+            ]
+            .into_iter(),
+            width,
+        )?) as ArrayRef;
+        let batch = RecordBatch::try_new(Arc::new(schema), vec![batch_array])?;
+
+        let result = expr.evaluate(&batch)?.into_array(batch.num_rows())?;
+        let result = as_boolean_array(&result);
+        assert_eq!(
+            result,
+            &BooleanArray::from(vec![Some(true), Some(false), None])
+        );
+
+        Ok(())
+    }
+
+    #[test]
+    fn test_in_list_fixed_size_binary_offset_path() -> Result<()> {
+        let width = 16;
+        let before_slice = vec![0x01; width as usize];
+        let in_slice = vec![0x02; width as usize];
+        let not_in_slice = vec![0x03; width as usize];
+
+        let schema = Schema::new(vec![Field::new(
+            "a",
+            DataType::FixedSizeBinary(width),
+            false,
+        )]);
+        let col_a = col("a", &schema)?;
+        let parent = Arc::new(FixedSizeBinaryArray::from(vec![
+            before_slice.as_slice(),
+            in_slice.as_slice(),
+            not_in_slice.as_slice(),
+        ])) as ArrayRef;
+        let sliced_list = parent.slice(1, 1);
+        let expr = Arc::new(InListExpr::try_new_from_array(
+            Arc::clone(&col_a),
+            sliced_list,
+            false,
+        )?) as Arc<dyn PhysicalExpr>;
+
+        let batch_array = Arc::new(FixedSizeBinaryArray::from(vec![
+            in_slice.as_slice(),
+            before_slice.as_slice(),
+        ])) as ArrayRef;
+        let batch = RecordBatch::try_new(Arc::new(schema), vec![batch_array])?;
+
+        let result = expr.evaluate(&batch)?.into_array(batch.num_rows())?;
+        let result = as_boolean_array(&result);
+        assert_eq!(result, &BooleanArray::from(vec![Some(true), Some(false)]));
+
+        Ok(())
+    }
+
+    #[test]
+    fn test_in_list_fixed_size_binary_offset_path_with_nulls() -> Result<()> {
+        let width = 16;
+        let before_slice = vec![0xAA; width as usize];
+        let in_slice = vec![0xBB; width as usize];
+        let not_in_slice = vec![0xCC; width as usize];
+
+        let schema = Schema::new(vec![Field::new(
+            "a",
+            DataType::FixedSizeBinary(width),
+            true,
+        )]);
+        let col_a = col("a", &schema)?;
+        let parent = Arc::new(FixedSizeBinaryArray::try_from_sparse_iter_with_size(
+            vec![
+                Some(before_slice.as_slice()),
+                None,
+                Some(in_slice.as_slice()),
+            ]
+            .into_iter(),
+            width,
+        )?) as ArrayRef;
+        let sliced_list = parent.slice(1, 2);
+        let expr = Arc::new(InListExpr::try_new_from_array(
+            Arc::clone(&col_a),
+            sliced_list,
+            false,
+        )?) as Arc<dyn PhysicalExpr>;
+
+        let batch_array = Arc::new(FixedSizeBinaryArray::try_from_sparse_iter_with_size(
+            vec![
+                Some(in_slice.as_slice()),
+                Some(not_in_slice.as_slice()),
+                None,
+            ]
+            .into_iter(),
+            width,
+        )?) as ArrayRef;
+        let batch = RecordBatch::try_new(Arc::new(schema), vec![batch_array])?;
+
+        let result = expr.evaluate(&batch)?.into_array(batch.num_rows())?;
+        let result = as_boolean_array(&result);
+        assert_eq!(result, &BooleanArray::from(vec![Some(true), None, None]));
+
+        Ok(())
+    }
+
     /// Test IN LIST for date types (Date32, Date64).
     ///
     /// Test data: 0 (in list), 2 (not in list), [1, 3] (other list values)

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

@@ -25,6 +25,7 @@
 //! - 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)
+//! - FixedSizeBinary(N) for N ∈ {1,2,4,8,16}: reinterpreted as N-byte primitive
 //! - Byte arrays (Utf8, Binary, etc.): ByteArrayFilter / ByteViewFilter
 //! - Other types: NestedTypeFilter (fallback for List, Struct, Map, etc.)
 
@@ -146,6 +147,11 @@ pub(crate) fn instantiate_static_filter(
         };
     }
 
+    // FixedSizeBinary with power-of-2 width: reinterpret as primitive
+    if let &DataType::FixedSizeBinary(byte_width) = dt {
+        return instantiate_fixed_size_binary_filter(in_array, byte_width);
+    }
+
     let strategy = select_strategy(dt, len);
 
     match (dt, strategy) {
@@ -195,6 +201,49 @@ pub(crate) fn instantiate_static_filter(
     }
 }
 
+// =============================================================================
+// FIXED-SIZE BINARY DISPATCH
+// =============================================================================
+
+/// Creates the optimal filter for FixedSizeBinary(N) arrays.
+///
+/// For power-of-2 widths (1, 2, 4, 8, 16), FixedSizeBinary has the same
+/// contiguous buffer layout as primitive arrays, so we zero-copy reinterpret
+/// and use the optimized primitive filters (bitmap, branchless, hash).
+/// Non-power-of-2 widths fall back to NestedTypeFilter.
+fn instantiate_fixed_size_binary_filter(
+    in_array: ArrayRef,
+    byte_width: i32,
+) -> Result<Arc<dyn StaticFilter + Send + Sync>> {
+    let len = in_array.len() - in_array.null_count();
+    match byte_width {
+        1 => make_bitmap_filter::<U8Config>(&in_array),
+        2 => make_bitmap_filter::<U16Config>(&in_array),
+        4 => {
+            if len <= BRANCHLESS_MAX_4B {
+                make_branchless_filter::<UInt32Type>(&in_array, 4)
+            } else {
+                make_direct_probe_filter_reinterpreted::<UInt32Type>(&in_array)
+            }
+        }
+        8 => {
+            if len <= BRANCHLESS_MAX_8B {
+                make_branchless_filter::<UInt64Type>(&in_array, 8)
+            } else {
+                make_direct_probe_filter_reinterpreted::<UInt64Type>(&in_array)
+            }
+        }
+        16 => {
+            if len <= BRANCHLESS_MAX_16B {
+                make_branchless_filter::<Decimal128Type>(&in_array, 16)
+            } else {
+                make_direct_probe_filter_reinterpreted::<Decimal128Type>(&in_array)
+            }
+        }
+        _ => Ok(Arc::new(NestedTypeFilter::try_new(in_array)?)),
+    }
+}
+
 // =============================================================================
 // TYPE DISPATCH
 // =============================================================================

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

@@ -26,15 +26,16 @@
 //! multiple types (Int64, Float64, Timestamp, Duration) that share the same
 //! byte width, reducing code duplication.
 
+use std::hash::BuildHasher;
 use std::marker::PhantomData;
 use std::mem::size_of;
 use std::sync::Arc;
 
-use ahash::RandomState;
 use arrow::array::{Array, ArrayRef, AsArray, BooleanArray, PrimitiveArray};
 use arrow::buffer::ScalarBuffer;
-use arrow::datatypes::{ArrowPrimitiveType, ByteViewType, Decimal128Type};
+use arrow::datatypes::{ArrowPrimitiveType, ByteViewType, DataType, Decimal128Type};
 use arrow::util::bit_iterator::BitIndexIterator;
+use datafusion_common::hash_utils::RandomState;
 use datafusion_common::hash_utils::with_hashes;
 use datafusion_common::{Result, exec_datafusion_err};
 use hashbrown::HashTable;
@@ -88,11 +89,17 @@ where
     }
 
     fn contains(&self, v: &dyn Array, negated: bool) -> Result<BooleanArray> {
+        fn matches_reinterpreted_width(dt: &DataType, width: usize) -> bool {
+            dt.primitive_width() == Some(width)
+                || matches!(dt, DataType::FixedSizeBinary(byte_width) if *byte_width as usize == width)
+        }
+
         arrow::array::downcast_dictionary_array! {
             v => {
-                if v.values().data_type().primitive_width()
-                    != Some(size_of::<T::Native>())
-                {
+                if !matches_reinterpreted_width(
+                    v.values().data_type(),
+                    size_of::<T::Native>(),
+                ) {
                     return Err(exec_datafusion_err!(
                         "Failed to downcast array to primitive type"
                     ));
@@ -105,7 +112,7 @@ where
             _ => {}
         }
 
-        if v.data_type().primitive_width() != Some(size_of::<T::Native>()) {
+        if !matches_reinterpreted_width(v.data_type(), size_of::<T::Native>()) {
             return Err(exec_datafusion_err!(
                 "Failed to downcast array to primitive type"
             ));
@@ -151,8 +158,9 @@ impl StaticFilter for Utf8ViewHashFilter {
 pub(crate) fn reinterpret_any_primitive_to<T: ArrowPrimitiveType>(
     array: &dyn Array,
 ) -> ArrayRef {
-    let values = array.to_data().buffers()[0].clone();
-    let buffer: ScalarBuffer<T::Native> = values.into();
+    let data = array.to_data();
+    let values = data.buffers()[0].clone();
+    let buffer = ScalarBuffer::<T::Native>::new(values, data.offset(), data.len());
     Arc::new(PrimitiveArray::<T>::new(buffer, array.nulls().cloned()))
 }
 
@@ -417,7 +425,7 @@ where
         let views: &[i128] = bv.views().inner().typed_data();
 
         let mut masked_views = Vec::new();
-        let state = RandomState::new();
+        let state = RandomState::default();
         let mut long_value_table = HashTable::new();
 
         // Build hash table for long strings using batch hashing
@@ -617,7 +625,7 @@ impl<O: arrow::array::OffsetSizeTrait + 'static> Utf8TwoStageFilter<O> {
 
         let len = arr.len();
         let mut encoded_values = Vec::with_capacity(len);
-        let state = RandomState::new();
+        let state = RandomState::default();
         let mut long_string_table = HashTable::new();
         let mut all_short = true;