perf(in_list): add BranchlessFilter for small IN lists

Adds BranchlessFilter<T, N> - a const-generic filter that unrolls
membership checks into a fixed-size OR-chain comparison. For small
lists (≤16 elements), this outperforms hash lookups due to:

- No branching (uses bitwise OR to combine comparisons)
- Better CPU pipelining
- No hash computation overhead

Strategy selection thresholds (tuned via benchmarks):
- 4-byte types (Int32, Float32): branchless up to 16 elements
- 8-byte types (Int64, Float64): branchless up to 16 elements
- 16-byte types (Decimal128): branchless up to 4 elements

Author: geoffreyclaude

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

@@ -248,3 +248,92 @@ where
         ))
     }
 }
+
+// =============================================================================
+// BRANCHLESS FILTER (Const Generic for Small Lists)
+// =============================================================================
+
+/// A branchless filter for very small IN lists (0-16 elements).
+///
+/// Uses const generics to unroll the membership check into a fixed-size
+/// comparison chain, outperforming hash lookups for small lists due to:
+/// - No branching (uses bitwise OR to combine comparisons)
+/// - Better CPU pipelining
+/// - No hash computation overhead
+pub(crate) struct BranchlessFilter<T: ArrowPrimitiveType, const N: usize> {
+    null_count: usize,
+    values: [T::Native; N],
+}
+
+impl<T: ArrowPrimitiveType, const N: usize> BranchlessFilter<T, N>
+where
+    T::Native: Copy + PartialEq,
+{
+    /// Try to create a branchless filter if the array has exactly N non-null values.
+    pub(crate) fn try_new(in_array: &ArrayRef) -> Option<Result<Self>> {
+        let in_array = in_array.as_primitive_opt::<T>()?;
+        let non_null_count = in_array.len() - in_array.null_count();
+        if non_null_count != N {
+            return None;
+        }
+        let values: Vec<_> = in_array.iter().flatten().collect();
+        // Use default_value() from ArrowPrimitiveType trait instead of Default::default()
+        let mut arr = [T::default_value(); N];
+        arr.copy_from_slice(&values);
+        Some(Ok(Self {
+            null_count: in_array.null_count(),
+            values: arr,
+        }))
+    }
+
+    /// Branchless membership check using OR-chain.
+    #[inline(always)]
+    fn check(&self, needle: T::Native) -> bool {
+        self.values
+            .iter()
+            .fold(false, |acc, &v| acc | (v == needle))
+    }
+
+    /// Check membership using a raw values slice (zero-copy path for type reinterpretation).
+    #[inline]
+    pub(crate) fn contains_slice(
+        &self,
+        values: &[T::Native],
+        nulls: Option<&arrow::buffer::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<T: ArrowPrimitiveType, const N: usize> StaticFilter for BranchlessFilter<T, N>
+where
+    T::Native: Copy + PartialEq + Send + Sync,
+{
+    fn null_count(&self) -> usize {
+        self.null_count
+    }
+
+    fn contains(&self, v: &dyn Array, negated: bool) -> Result<BooleanArray> {
+        handle_dictionary!(self, v, negated);
+        let v = v.as_primitive_opt::<T>().ok_or_else(|| {
+            exec_datafusion_err!("Failed to downcast array to primitive type")
+        })?;
+        let input_values = v.values();
+        Ok(build_in_list_result(
+            v.len(),
+            v.nulls(),
+            self.null_count > 0,
+            negated,
+            // SAFETY: i is in bounds since we iterate 0..v.len()
+            #[inline(always)]
+            |i| self.check(unsafe { *input_values.get_unchecked(i) }),
+        ))
+    }
+}

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

@@ -15,39 +15,217 @@
 // specific language governing permissions and limitations
 // under the License.
 
-//! Filter strategy selection for InList expressions
+//! 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 (≤16) or hash (>16)
+//! - 8-byte types (Int64/Float64): branchless (≤16) or hash (>16)
+//! - 16-byte types (Decimal128): branchless (≤4) or hash (>4)
+//! - Other types: generic ArrayStaticFilter
 
+use std::hash::Hash;
 use std::sync::Arc;
 
 use arrow::array::ArrayRef;
 use arrow::datatypes::*;
 use datafusion_common::Result;
 
-use super::array_filter::{ArrayStaticFilter, StaticFilter};
+use super::array_filter::ArrayStaticFilter;
+use super::array_filter::StaticFilter;
 use super::primitive::{PrimitiveFilter, U8Config, U16Config};
-use super::transform::{make_bitmap_filter, make_primitive_filter};
+use super::transform::{make_bitmap_filter, make_branchless_filter, make_primitive_filter};
 
+// =============================================================================
+// LOOKUP STRATEGY THRESHOLDS (tuned via microbenchmarks)
+// =============================================================================
+//
+// Based on minimum batch time (8192 lookups per batch):
+// - Int8 (1 byte): BITMAP (32 bytes, always fastest)
+// - Int16 (2 bytes): BITMAP (8 KB, always fastest)
+// - Int32 (4 bytes): branchless up to 16, then hashset
+// - Int64 (8 bytes): branchless up to 16, then hashset
+// - Int128 (16 bytes): branchless up to 4, then hashset
+// - Other types: hashset (via ArrayStaticFilter)
+//
+// NOTE: Binary search and linear scan were benchmarked but consistently
+// lost to the strategies above at all tested list sizes.
+
+/// Maximum list size for branchless lookup on 4-byte primitives (Int32, UInt32, Float32).
+const BRANCHLESS_MAX_4B: usize = 16;
+
+/// Maximum list size for branchless lookup on 8-byte primitives (Int64, UInt64, Float64).
+const BRANCHLESS_MAX_8B: usize = 16;
+
+/// Maximum list size for branchless lookup on 16-byte types (Decimal128).
+const BRANCHLESS_MAX_16B: usize = 4;
+
+// =============================================================================
+// FILTER STRATEGY SELECTION
+// =============================================================================
+
+/// The lookup strategy to use for a given data type and list size.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum FilterStrategy {
+    /// Bitmap filter for u8/u16 - O(1) bit test, always fastest for these types.
+    Bitmap1B,
+    Bitmap2B,
+    /// Branchless OR-chain for small lists.
+    Branchless,
+    /// HashSet for larger lists.
+    Hashed,
+    /// Generic ArrayStaticFilter fallback.
+    Generic,
+}
+
+/// Determines the optimal lookup strategy based on data type and list size.
+///
+/// For 1-byte and 2-byte types, bitmap is always used (benchmarks show it's
+/// faster than both branchless and hashed at all list sizes).
+/// For larger types, cutoffs are tuned per byte-width.
+fn select_strategy(dt: &DataType, len: usize) -> FilterStrategy {
+    match dt.primitive_width() {
+        Some(1) => FilterStrategy::Bitmap1B,
+        Some(2) => FilterStrategy::Bitmap2B,
+        Some(4) => {
+            if len <= BRANCHLESS_MAX_4B {
+                FilterStrategy::Branchless
+            } else {
+                FilterStrategy::Hashed
+            }
+        }
+        Some(8) => {
+            if len <= BRANCHLESS_MAX_8B {
+                FilterStrategy::Branchless
+            } else {
+                FilterStrategy::Hashed
+            }
+        }
+        Some(16) => {
+            if len <= BRANCHLESS_MAX_16B {
+                FilterStrategy::Branchless
+            } else {
+                FilterStrategy::Hashed
+            }
+        }
+        _ => FilterStrategy::Generic,
+    }
+}
+
+// =============================================================================
+// FILTER INSTANTIATION
+// =============================================================================
+
+/// Creates the optimal static filter for the given array.
+///
+/// This is the main entry point for filter creation. It analyzes the array's
+/// data type and size to select the best lookup strategy.
 pub(crate) fn instantiate_static_filter(
     in_array: ArrayRef,
 ) -> Result<Arc<dyn StaticFilter + Send + Sync>> {
-    match in_array.data_type() {
-        // 1-byte types: use bitmap (256 bits = 32 bytes)
-        DataType::Int8 | DataType::UInt8 => make_bitmap_filter::<U8Config>(&in_array),
-        // 2-byte types: use bitmap (65536 bits = 8 KB)
-        DataType::Int16 | DataType::UInt16 => make_bitmap_filter::<U16Config>(&in_array),
-        // 4-byte integer types
-        DataType::Int32 => Ok(Arc::new(PrimitiveFilter::<Int32Type>::try_new(&in_array)?)),
-        DataType::UInt32 => Ok(Arc::new(PrimitiveFilter::<UInt32Type>::try_new(&in_array)?)),
-        // 8-byte integer types
-        DataType::Int64 => Ok(Arc::new(PrimitiveFilter::<Int64Type>::try_new(&in_array)?)),
-        DataType::UInt64 => Ok(Arc::new(PrimitiveFilter::<UInt64Type>::try_new(&in_array)?)),
-        // Float types: reinterpret as unsigned integers (same bit pattern = equal)
-        DataType::Float32 => make_primitive_filter::<UInt32Type>(&in_array),
-        DataType::Float64 => make_primitive_filter::<UInt64Type>(&in_array),
-        _ => {
-            /* fall through to generic implementation for unsupported types (Struct, etc.) */
-            Ok(Arc::new(ArrayStaticFilter::try_new(in_array)?))
+    let len = in_array.len();
+    let dt = in_array.data_type();
+
+    match select_strategy(dt, len) {
+        FilterStrategy::Bitmap1B => dispatch_bitmap_u8(&in_array),
+        FilterStrategy::Bitmap2B => dispatch_bitmap_u16(&in_array),
+        FilterStrategy::Branchless => dispatch_filter(&in_array, dispatch_branchless),
+        FilterStrategy::Hashed => dispatch_filter(&in_array, dispatch_hashed),
+        FilterStrategy::Generic => Ok(Arc::new(ArrayStaticFilter::try_new(in_array)?)),
+    }
+}
+
+/// Generic filter dispatcher with fallback to ArrayStaticFilter.
+fn dispatch_filter<F>(
+    in_array: &ArrayRef,
+    dispatch: F,
+) -> Result<Arc<dyn StaticFilter + Send + Sync>>
+where
+    F: Fn(&ArrayRef) -> Option<Result<Arc<dyn StaticFilter + Send + Sync>>>,
+{
+    dispatch(in_array).unwrap_or_else(|| {
+        Ok(Arc::new(ArrayStaticFilter::try_new(Arc::clone(in_array))?))
+    })
+}
+
+fn dispatch_bitmap_u8(
+    in_array: &ArrayRef,
+) -> Result<Arc<dyn StaticFilter + Send + Sync>> {
+    make_bitmap_filter::<U8Config>(in_array)
+}
+
+fn dispatch_bitmap_u16(
+    in_array: &ArrayRef,
+) -> Result<Arc<dyn StaticFilter + Send + Sync>> {
+    make_bitmap_filter::<U16Config>(in_array)
+}
+
+// =============================================================================
+// TYPE DISPATCH
+// =============================================================================
+
+/// Dispatch macro that routes primitive types by width to the appropriate UInt type.
+///
+/// All primitive types (Int*, UInt*, Float*, Timestamp*, Date*, Duration*, etc.) are
+/// automatically dispatched based on their width. The reinterpret function handles
+/// the fast path when source type already matches the destination UInt type.
+macro_rules! dispatch_primitive {
+    ($arr:expr, $reinterpret:ident) => {
+        match $arr.data_type().primitive_width() {
+            Some(1) => Some($reinterpret::<UInt8Type>($arr)),
+            Some(2) => Some($reinterpret::<UInt16Type>($arr)),
+            Some(4) => Some($reinterpret::<UInt32Type>($arr)),
+            Some(8) => Some($reinterpret::<UInt64Type>($arr)),
+            Some(16) => Some($reinterpret::<Decimal128Type>($arr)),
+            _ => None,
         }
+    };
+}
+
+fn dispatch_branchless(
+    arr: &ArrayRef,
+) -> Option<Result<Arc<dyn StaticFilter + Send + Sync>>> {
+    fn make<D: ArrowPrimitiveType + 'static>(
+        arr: &ArrayRef,
+    ) -> Result<Arc<dyn StaticFilter + Send + Sync>>
+    where
+        D::Native: Copy + PartialEq + Send + Sync + 'static,
+    {
+        make_branchless_filter::<D>(arr)
     }
+    dispatch_primitive!(arr, make)
 }
 
+fn dispatch_hashed(
+    arr: &ArrayRef,
+) -> Option<Result<Arc<dyn StaticFilter + Send + Sync>>> {
+    // Fast path: create PrimitiveFilter directly for common hashable types
+    macro_rules! direct_filter {
+        ($T:ty) => {
+            return Some(
+                PrimitiveFilter::<$T>::try_new(arr)
+                    .map(|f| Arc::new(f) as Arc<dyn StaticFilter + Send + Sync>),
+            )
+        };
+    }
+    match arr.data_type() {
+        DataType::Int32 => direct_filter!(Int32Type),
+        DataType::Int64 => direct_filter!(Int64Type),
+        DataType::UInt32 => direct_filter!(UInt32Type),
+        DataType::UInt64 => direct_filter!(UInt64Type),
+        _ => {}
+    }
+
+    // For other types (Float32, Float64, Timestamp, etc.), reinterpret to UInt
+    fn make<D: ArrowPrimitiveType + 'static>(
+        arr: &ArrayRef,
+    ) -> Result<Arc<dyn StaticFilter + Send + Sync>>
+    where
+        D::Native: Hash + Eq + Send + Sync + 'static,
+    {
+        make_primitive_filter::<D>(arr)
+    }
+    dispatch_primitive!(arr, make)
+}