Implement Zero-Copy Reinterpretation and enable Int8/Int16 Bitmaps

Introduces zero-copy buffer reinterpretation to allow signed integers and other 1 or 2-byte primitive types (e.g. Float16) to use the high-performance bitmap filters. Triggers for all types with 1-byte or 2-byte width.

Author: geoffreyclaude

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

@@ -41,6 +41,7 @@ mod primitive_filter;
 mod result;
 mod static_filter;
 mod strategy;
+mod transform;
 
 use static_filter::StaticFilter;
 use strategy::instantiate_static_filter;

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

@@ -131,6 +131,19 @@ impl<C: BitmapFilterConfig> BitmapFilter<C> {
     fn check(&self, needle: C::Native) -> bool {
         self.bits.get_bit(C::to_index(needle))
     }
+
+    /// Check membership using a raw values slice (zero-copy path for type reinterpretation).
+    #[inline]
+    pub(super) fn contains_slice(
+        &self,
+        values: &[C::Native],
+        nulls: Option<&NullBuffer>,
+        negated: bool,
+    ) -> BooleanArray {
+        build_in_list_result(values.len(), nulls, self.null_count > 0, negated, |i| {
+            self.check(unsafe { *values.get_unchecked(i) })
+        })
+    }
 }
 
 impl<C: BitmapFilterConfig> StaticFilter for BitmapFilter<C> {
@@ -345,9 +358,6 @@ macro_rules! primitive_static_filter {
     };
 }
 
-// Generate specialized filters for all integer primitive types
-primitive_static_filter!(Int8StaticFilter, Int8Type);
-primitive_static_filter!(Int16StaticFilter, Int16Type);
 primitive_static_filter!(Int32StaticFilter, Int32Type);
 primitive_static_filter!(Int64StaticFilter, Int64Type);
 primitive_static_filter!(UInt32StaticFilter, UInt32Type);

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

@@ -21,6 +21,8 @@
 //! from IN list membership tests, handling null propagation correctly
 //! according to SQL three-valued logic.
 
+#![expect(dead_code)]
+
 use arrow::array::BooleanArray;
 use arrow::buffer::{BooleanBuffer, NullBuffer};
 
@@ -43,6 +45,9 @@ use arrow::buffer::{BooleanBuffer, NullBuffer};
 ///
 /// This version computes contains for all positions, including nulls, then applies
 /// null masking via bitmap operations.
+///
+/// For expensive contains checks (like ByteViewMaskedFilter with string comparison),
+/// use `build_in_list_result_with_null_shortcircuit` instead.
 #[inline]
 pub(crate) fn build_in_list_result<C>(
     len: usize,
@@ -58,6 +63,106 @@ where
     build_result_from_contains(needle_nulls, haystack_has_nulls, negated, contains_buf)
 }
 
+/// Builds a BooleanArray result with null short-circuit (optimized for expensive contains).
+///
+/// Unlike `build_in_list_result`, this version checks nulls INSIDE the loop and
+/// skips the contains check for null positions. This is optimal for expensive
+/// contains checks (like ByteViewMaskedFilter with hash lookup + string comparison) where
+/// skipping lookups outweighs the branch overhead.
+///
+/// The shortcircuit is only applied when `needle_null_count > 0` - if there are
+/// no actual nulls, we avoid the branch overhead entirely.
+///
+/// Use this for: ByteViewMaskedFilter, Utf8TwoStageFilter (string/binary types)
+/// Use `build_in_list_result` for: DirectProbeFilter, BranchlessFilter (primitive types)
+#[inline]
+pub(crate) fn build_in_list_result_with_null_shortcircuit<C>(
+    len: usize,
+    needle_nulls: Option<&NullBuffer>,
+    needle_null_count: usize,
+    haystack_has_nulls: bool,
+    negated: bool,
+    mut contains: C,
+) -> BooleanArray
+where
+    C: FnMut(usize) -> bool,
+{
+    // When null_count=0, treat as no validity buffer to avoid extra work.
+    // The validity buffer might exist but have all bits set to true.
+    let effective_nulls = needle_nulls.filter(|_| needle_null_count > 0);
+
+    match effective_nulls {
+        Some(nulls) => {
+            // Has nulls: check validity inside loop to skip expensive contains()
+            let contains_buf =
+                BooleanBuffer::collect_bool(len, |i| nulls.is_valid(i) && contains(i));
+            build_result_from_contains_premasked(
+                Some(nulls),
+                haystack_has_nulls,
+                negated,
+                contains_buf,
+            )
+        }
+        None => {
+            // No nulls: compute contains for all positions without branch overhead
+            let contains_buf = BooleanBuffer::collect_bool(len, contains);
+            // Use premasked path since contains_buf is "trivially premasked" (no nulls to mask)
+            build_result_from_contains_premasked(
+                None,
+                haystack_has_nulls,
+                negated,
+                contains_buf,
+            )
+        }
+    }
+}
+
+/// Builds result from a contains buffer that was pre-masked at null positions.
+///
+/// This is used by `build_in_list_result_with_null_shortcircuit` where the
+/// contains buffer already has `false` at null positions due to the short-circuit.
+///
+/// Since contains_buf is pre-masked (false at null positions), we can simplify:
+/// - `valid & contains_buf` = `contains_buf` (already 0 where valid is 0)
+/// - XOR can replace AND+NOT for the negated case: `valid ^ contains = valid & !contains`
+#[inline]
+fn build_result_from_contains_premasked(
+    needle_nulls: Option<&NullBuffer>,
+    haystack_has_nulls: bool,
+    negated: bool,
+    contains_buf: BooleanBuffer,
+) -> BooleanArray {
+    match (needle_nulls, haystack_has_nulls, negated) {
+        // Haystack has nulls: result is null unless value is found
+        (_, true, false) => {
+            // contains_buf is already masked (false at null positions)
+            BooleanArray::new(contains_buf.clone(), Some(NullBuffer::new(contains_buf)))
+        }
+        (Some(v), true, true) => {
+            // NOT IN with nulls: true if valid and not found, null if found or needle null
+            // XOR: valid ^ contains = 1 iff valid=1 and contains=0 (not found)
+            BooleanArray::new(
+                v.inner() ^ &contains_buf,
+                Some(NullBuffer::new(contains_buf)),
+            )
+        }
+        (None, true, true) => {
+            BooleanArray::new(!&contains_buf, Some(NullBuffer::new(contains_buf)))
+        }
+        // Haystack has no nulls: result validity follows needle validity
+        (Some(v), false, false) => {
+            // contains_buf is already masked, just use needle validity for nulls
+            BooleanArray::new(contains_buf, Some(v.clone()))
+        }
+        (Some(v), false, true) => {
+            // Need AND because !contains_buf is 1 at null positions
+            BooleanArray::new(v.inner() & &(!&contains_buf), Some(v.clone()))
+        }
+        (None, false, false) => BooleanArray::new(contains_buf, None),
+        (None, false, true) => BooleanArray::new(!&contains_buf, None),
+    }
+}
+
 /// Builds a BooleanArray result from a pre-computed contains buffer.
 ///
 /// This version does not assume contains_buf is pre-masked at null positions.

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

@@ -25,6 +25,7 @@ use datafusion_common::Result;
 use super::array_static_filter::ArrayStaticFilter;
 use super::primitive_filter::*;
 use super::static_filter::StaticFilter;
+use super::transform::make_bitmap_filter;
 
 pub(super) fn instantiate_static_filter(
     in_array: ArrayRef,
@@ -37,17 +38,14 @@ pub(super) fn instantiate_static_filter(
         _ => in_array,
     };
     match in_array.data_type() {
-        // Integer primitive types
-        DataType::Int8 => Ok(Arc::new(Int8StaticFilter::try_new(&in_array)?)),
-        DataType::Int16 => Ok(Arc::new(Int16StaticFilter::try_new(&in_array)?)),
+        DataType::Int8 | DataType::UInt8 => {
+            make_bitmap_filter::<UInt8BitmapConfig>(&in_array)
+        }
+        DataType::Int16 | DataType::UInt16 => {
+            make_bitmap_filter::<UInt16BitmapConfig>(&in_array)
+        }
         DataType::Int32 => Ok(Arc::new(Int32StaticFilter::try_new(&in_array)?)),
         DataType::Int64 => Ok(Arc::new(Int64StaticFilter::try_new(&in_array)?)),
-        DataType::UInt8 => Ok(Arc::new(BitmapFilter::<UInt8BitmapConfig>::try_new(
-            &in_array,
-        )?)),
-        DataType::UInt16 => Ok(Arc::new(BitmapFilter::<UInt16BitmapConfig>::try_new(
-            &in_array,
-        )?)),
         DataType::UInt32 => Ok(Arc::new(UInt32StaticFilter::try_new(&in_array)?)),
         DataType::UInt64 => Ok(Arc::new(UInt64StaticFilter::try_new(&in_array)?)),
         // Float primitive types (use ordered wrappers for Hash/Eq)

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

@@ -0,0 +1,92 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Type transformation utilities for InList filters
+//!
+//! This module provides type reinterpretation for optimizing filter dispatch.
+//! For equality comparison, only the bit pattern matters, so we can:
+//! - Reinterpret signed integers as unsigned (Int32 → UInt32)
+//! - Reinterpret floats as unsigned integers (Float64 → UInt64)
+//!
+//! This allows using a single filter implementation (e.g., for UInt64) to handle
+//! multiple types (Int64, Float64, Timestamp, Duration) that share the same
+//! byte width, reducing code duplication.
+
+use std::sync::Arc;
+
+use arrow::array::{Array, ArrayRef, BooleanArray, PrimitiveArray};
+use arrow::buffer::ScalarBuffer;
+use arrow::datatypes::ArrowPrimitiveType;
+use datafusion_common::Result;
+
+use super::primitive_filter::{BitmapFilter, BitmapFilterConfig};
+use super::static_filter::{StaticFilter, handle_dictionary};
+
+// =============================================================================
+// REINTERPRETING FILTERS (zero-copy type conversion)
+// =============================================================================
+
+/// Reinterpreting filter for bitmap lookups (u8/u16).
+struct ReinterpretedBitmap<C: BitmapFilterConfig> {
+    inner: BitmapFilter<C>,
+}
+
+impl<C: BitmapFilterConfig> StaticFilter for ReinterpretedBitmap<C> {
+    fn null_count(&self) -> usize {
+        self.inner.null_count()
+    }
+
+    fn contains(&self, v: &dyn Array, negated: bool) -> Result<BooleanArray> {
+        handle_dictionary!(self, v, negated);
+
+        let data = v.to_data();
+        let values: &[C::Native] = data.buffer::<C::Native>(0);
+
+        Ok(self.inner.contains_slice(values, data.nulls(), negated))
+    }
+}
+
+/// Reinterprets any primitive-like array as the target primitive type T by extracting
+/// the underlying buffer.
+///
+/// This is a zero-copy operation that works for all primitive types (Int*, UInt*, Float*,
+/// Timestamp*, Date*, Duration*, etc.) by directly accessing the underlying buffer,
+/// ignoring any metadata like timezones or precision/scale.
+#[inline]
+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();
+    Arc::new(PrimitiveArray::<T>::new(buffer, array.nulls().cloned()))
+}
+
+/// Creates a bitmap filter for u8/u16 types, reinterpreting if needed.
+pub(crate) fn make_bitmap_filter<C>(
+    in_array: &ArrayRef,
+) -> Result<Arc<dyn StaticFilter + Send + Sync>>
+where
+    C: BitmapFilterConfig,
+{
+    if in_array.data_type() == &C::ArrowType::DATA_TYPE {
+        return Ok(Arc::new(BitmapFilter::<C>::try_new(in_array)?));
+    }
+
+    let reinterpreted = reinterpret_any_primitive_to::<C::ArrowType>(in_array.as_ref());
+    let inner = BitmapFilter::<C>::try_new(&reinterpreted)?;
+    Ok(Arc::new(ReinterpretedBitmap { inner }))
+}