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

@@ -923,6 +923,138 @@ 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::try_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,
+            &schema,
+        )?) 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::try_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,
+            &schema,
+        )?) as Arc<dyn PhysicalExpr>;
+
+        let batch_array = Arc::new(FixedSizeBinaryArray::try_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,
+            &schema,
+        )?) 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

@@ -15,6 +15,19 @@
 // specific language governing permissions and limitations
 // under the License.
 
+//! Filter selection strategy for InList expressions
+//!
+//! Selects the optimal lookup strategy based on data type and list size:
+//!
+//! - 1-byte types (Int8/UInt8): bitmap (32 bytes, O(1) bit test)
+//! - 2-byte types (Int16/UInt16): bitmap (8 KB, O(1) bit test)
+//! - 4-byte types (Int32/Float32): branchless (up to 32) or hash
+//! - 8-byte types (Int64/Float64): branchless (up to 16) or hash
+//! - 16-byte types (Decimal128): branchless (up to 4) or hash
+//! - Utf8View/BinaryView: short-string or masked-view filters
+//! - FixedSizeBinary(1/2/4/8/16): reinterpreted as fixed-width values
+//! - Other types: ArrayStaticFilter fallback
+
 use std::sync::Arc;
 
 use arrow::array::ArrayRef;
@@ -28,7 +41,8 @@ use super::static_filter::{StaticFilter, handle_dictionary};
 use super::transform::{
     make_bitmap_filter, make_branchless_filter, make_byte_view_masked_filter,
     make_utf8view_branchless_filter, make_utf8view_hash_filter,
-    reinterpret_any_primitive_to, utf8view_all_short_strings,
+    matches_reinterpreted_width, reinterpret_any_primitive_to,
+    utf8view_all_short_strings,
 };
 
 /// Maximum list size for branchless lookup on 1-byte primitives (Int8, UInt8).
@@ -127,6 +141,11 @@ pub(super) 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) {
@@ -160,6 +179,44 @@ pub(super) fn instantiate_static_filter(
     }
 }
 
+/// Creates the optimal filter for FixedSizeBinary(N) arrays.
+///
+/// Widths 1, 2, 4, 8, and 16 share the same contiguous values-buffer layout
+/// as the corresponding primitive width, so they can use the optimized
+/// bitmap, branchless, or direct-probe filters without copying values.
+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::<UInt8Type>(&in_array),
+        2 => make_bitmap_filter::<UInt16Type>(&in_array),
+        4 => {
+            if len <= BRANCHLESS_MAX_4B {
+                make_branchless_filter::<UInt32Type>(&in_array)
+            } else {
+                make_direct_probe_filter_reinterpreted::<UInt32Type>(&in_array)
+            }
+        }
+        8 => {
+            if len <= BRANCHLESS_MAX_8B {
+                make_branchless_filter::<UInt64Type>(&in_array)
+            } else {
+                make_direct_probe_filter_reinterpreted::<UInt64Type>(&in_array)
+            }
+        }
+        16 => {
+            if len <= BRANCHLESS_MAX_16B {
+                make_branchless_filter::<Decimal128Type>(&in_array)
+            } else {
+                make_direct_probe_filter_reinterpreted::<Decimal128Type>(&in_array)
+            }
+        }
+        _ => Ok(Arc::new(ArrayStaticFilter::try_new(in_array)?)),
+    }
+}
+
 fn dispatch_branchless(
     arr: &ArrayRef,
 ) -> Option<Result<Arc<dyn StaticFilter + Send + Sync>>> {
@@ -215,7 +272,7 @@ where
     if in_array.data_type() == &D::DATA_TYPE {
         return Ok(Arc::new(DirectProbeFilter::<D>::try_new(in_array)?));
     }
-    if in_array.data_type().primitive_width() != Some(size_of::<D::Native>()) {
+    if !matches_reinterpreted_width(in_array.data_type(), size_of::<D::Native>()) {
         return Err(exec_datafusion_err!(
             "DirectProbeFilter: expected {}-byte primitive array, got {}",
             size_of::<D::Native>(),
@@ -253,7 +310,7 @@ where
         negated: bool,
     ) -> Result<arrow::array::BooleanArray> {
         handle_dictionary!(self, v, negated);
-        if v.data_type().primitive_width() != Some(size_of::<D::Native>()) {
+        if !matches_reinterpreted_width(v.data_type(), size_of::<D::Native>()) {
             return Err(exec_datafusion_err!(
                 "DirectProbeFilter: expected {}-byte primitive array, got {}",
                 size_of::<D::Native>(),
@@ -271,7 +328,7 @@ where
 mod tests {
     use super::*;
 
-    use arrow::array::{ArrayRef, BooleanArray, Float64Array};
+    use arrow::array::{ArrayRef, BooleanArray, FixedSizeBinaryArray, Float64Array};
 
     #[test]
     fn direct_probe_strategy_handles_reinterpreted_slices() -> Result<()> {
@@ -312,4 +369,39 @@ mod tests {
 
         Ok(())
     }
+
+    #[test]
+    fn fixed_size_binary_uses_bitmap_and_direct_probe_paths() -> Result<()> {
+        let haystack: ArrayRef = Arc::new(FixedSizeBinaryArray::try_from(vec![
+            [0x12, 0x34].as_slice(),
+            [0xab, 0xcd].as_slice(),
+        ])?);
+        let filter = instantiate_static_filter(haystack)?;
+        let needles = FixedSizeBinaryArray::try_from(vec![
+            [0xab, 0xcd].as_slice(),
+            [0xff, 0xff].as_slice(),
+        ])?;
+        assert_eq!(
+            filter.contains(&needles, false)?,
+            BooleanArray::from(vec![Some(true), Some(false)])
+        );
+
+        let values: Vec<_> = (0_u32..33).map(u32::to_le_bytes).collect();
+        let haystack_values: Vec<_> = values.iter().map(|v| v.as_slice()).collect();
+        let haystack: ArrayRef =
+            Arc::new(FixedSizeBinaryArray::try_from(haystack_values)?);
+        let filter = instantiate_static_filter(haystack)?;
+        let direct_hit = 7_u32.to_le_bytes();
+        let direct_miss = 99_u32.to_le_bytes();
+        let needles = FixedSizeBinaryArray::try_from(vec![
+            direct_hit.as_slice(),
+            direct_miss.as_slice(),
+        ])?;
+        assert_eq!(
+            filter.contains(&needles, false)?,
+            BooleanArray::from(vec![Some(true), Some(false)])
+        );
+
+        Ok(())
+    }
 }

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

@@ -28,7 +28,7 @@ use std::sync::Arc;
 
 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::{Result, exec_datafusion_err};
@@ -49,6 +49,16 @@ fn views_as_i128(views: &ScalarBuffer<u128>) -> &[i128] {
     views.inner().typed_data()
 }
 
+#[inline]
+pub(crate) fn matches_reinterpreted_width(dt: &DataType, width: usize) -> bool {
+    dt.primitive_width() == Some(width)
+        || matches!(
+            dt,
+            DataType::FixedSizeBinary(byte_width)
+                if usize::try_from(*byte_width).ok() == Some(width)
+        )
+}
+
 /// Reinterpreting filter for bitmap lookups (u8/u16).
 struct ReinterpretedBitmap<T: BitmapFilterType> {
     inner: BitmapFilter<T>,
@@ -62,7 +72,7 @@ impl<T: BitmapFilterType> StaticFilter for ReinterpretedBitmap<T> {
     fn contains(&self, v: &dyn Array, negated: bool) -> Result<BooleanArray> {
         handle_dictionary!(self, v, negated);
 
-        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!(
                 "BitmapFilter: expected {}-byte primitive array, got {}",
                 size_of::<T::Native>(),
@@ -94,7 +104,7 @@ where
     fn contains(&self, v: &dyn Array, negated: bool) -> Result<BooleanArray> {
         handle_dictionary!(self, v, negated);
 
-        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!(
                 "BranchlessFilter: expected {}-byte primitive array, got {}",
                 size_of::<T::Native>(),
@@ -156,7 +166,7 @@ where
     if in_array.data_type() == &T::DATA_TYPE {
         return Ok(Arc::new(BitmapFilter::<T>::try_new(in_array)?));
     }
-    if in_array.data_type().primitive_width() != Some(size_of::<T::Native>()) {
+    if !matches_reinterpreted_width(in_array.data_type(), size_of::<T::Native>()) {
         return Err(exec_datafusion_err!(
             "BitmapFilter: expected {}-byte primitive array for {} bitmap, got {}",
             size_of::<T::Native>(),
@@ -190,7 +200,7 @@ where
     let arr = if is_native {
         Arc::clone(in_array)
     } else {
-        if in_array.data_type().primitive_width() != Some(width) {
+        if !matches_reinterpreted_width(in_array.data_type(), width) {
             return Err(exec_datafusion_err!(
                 "BranchlessFilter: expected {width}-byte primitive array, got {}",
                 in_array.data_type()