perf(in_list): add DirectProbeFilter for O(1) hash lookups

Replace HashMap-based lookups with a custom open-addressing hash table
optimized for the IN list use case:

- DirectProbeFilter: O(1) lookups using golden ratio hash + linear probing
- 25% load factor with power-of-2 table size for minimal collisions
- Simple hash function (2 ops) vs MurmurHash3 (~12 ops)
- Supports i32, i64, u32, u64, i128 for primitive and Decimal128/Utf8View

Also:
- Add SortedArrayFilter for binary search fallback
- Use DirectProbeFilter in ByteViewMaskedFilter Stage 1 rejection
- Simplify null handling in result.rs using bitmap operations

Author: geoffreyclaude

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

@@ -27,7 +27,7 @@ use datafusion_common::Result;
 use datafusion_common::hash_utils::with_hashes;
 use hashbrown::HashTable;
 
-use super::result::{build_in_list_result, handle_dictionary};
+use super::result::{build_in_list_result_with_null_shortcircuit, handle_dictionary};
 use super::static_filter::StaticFilter;
 
 // =============================================================================
@@ -136,13 +136,17 @@ impl<A: ByteAccess> StaticFilter for ByteFilter<A> {
         handle_dictionary!(self, v, negated);
 
         let needle_nulls = A::nulls(v);
+        let needle_null_count = v.null_count();
         let haystack_has_nulls = self.in_array.null_count() > 0;
 
         // Batch hash all needle values using SIMD-optimized hashing
         with_hashes([v], &self.state, |needle_hashes| {
-            Ok(build_in_list_result(
+            // Use null shortcircuit version: string comparison is expensive,
+            // so skipping lookups for null positions is worth the branch overhead
+            Ok(build_in_list_result_with_null_shortcircuit(
                 v.len(),
                 needle_nulls,
+                needle_null_count,
                 haystack_has_nulls,
                 negated,
                 #[inline(always)]

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

@@ -19,8 +19,6 @@
 //!
 //! This module provides high-performance membership testing for Arrow primitive types.
 
-use std::hash::Hash;
-
 use arrow::array::{Array, ArrayRef, AsArray, BooleanArray};
 use arrow::datatypes::ArrowPrimitiveType;
 use datafusion_common::{HashSet, Result, exec_datafusion_err};
@@ -256,76 +254,218 @@ where
 }
 
 // =============================================================================
-// PRIMITIVE FILTER (Hash-based)
+// DIRECT PROBE HASH FILTER (O(1) lookup with open addressing)
 // =============================================================================
 
-/// Hash-based filter for primitive types with larger IN lists.
-pub(crate) struct PrimitiveFilter<T: ArrowPrimitiveType> {
+/// Load factor inverse for DirectProbeFilter hash table.
+/// A value of 4 means 25% load factor (table is 4x the number of elements).
+const LOAD_FACTOR_INVERSE: usize = 4;
+
+/// Minimum table size for DirectProbeFilter.
+/// Ensures reasonable performance even for very small IN lists.
+const MIN_TABLE_SIZE: usize = 16;
+
+/// Golden ratio constant for 32-bit hash mixing.
+/// Derived from (2^32 / phi) where phi = (1 + sqrt(5)) / 2.
+const GOLDEN_RATIO_32: u32 = 0x9e3779b9;
+
+/// Golden ratio constant for 64-bit hash mixing.
+/// Derived from (2^64 / phi) where phi = (1 + sqrt(5)) / 2.
+const GOLDEN_RATIO_64: u64 = 0x9e3779b97f4a7c15;
+
+/// Secondary mixing constant for 128-bit hashing (from SplitMix64).
+/// Using a different constant for hi/lo avoids collisions when lo = hi * C.
+const SPLITMIX_CONSTANT: u64 = 0xbf58476d1ce4e5b9;
+
+/// Fast hash filter using open addressing with linear probing.
+///
+/// Uses a power-of-2 sized hash table for O(1) average-case lookups.
+/// Optimized for the IN list use case with:
+/// - Simple/fast hash function (golden ratio multiply + xor-shift)
+/// - 25% load factor for minimal collisions
+/// - Direct array storage for cache-friendly access
+pub(crate) struct DirectProbeFilter<T: ArrowPrimitiveType>
+where
+    T::Native: DirectProbeHashable,
+{
     null_count: usize,
-    set: HashSet<T::Native>,
+    /// Hash table with open addressing. None = empty slot, Some(v) = value present
+    table: Box<[Option<T::Native>]>,
+    /// Mask for slot index (table.len() - 1, always power of 2 minus 1)
+    mask: usize,
+}
+
+/// Trait for types that can be hashed for the direct probe filter.
+///
+/// Requires `Hash + Eq` for deduplication via `HashSet`, even though we use
+/// a custom `probe_hash()` for the actual hash table lookups.
+pub(crate) trait DirectProbeHashable:
+    Copy + PartialEq + std::hash::Hash + Eq
+{
+    fn probe_hash(self) -> usize;
+}
+
+// Simple but fast hash - golden ratio multiply + xor-shift
+impl DirectProbeHashable for i32 {
+    #[inline(always)]
+    fn probe_hash(self) -> usize {
+        let x = self as u32;
+        let x = x.wrapping_mul(GOLDEN_RATIO_32);
+        (x ^ (x >> 16)) as usize
+    }
 }
 
-impl<T: ArrowPrimitiveType> PrimitiveFilter<T>
+impl DirectProbeHashable for i64 {
+    #[inline(always)]
+    fn probe_hash(self) -> usize {
+        let x = self as u64;
+        let x = x.wrapping_mul(GOLDEN_RATIO_64);
+        (x ^ (x >> 32)) as usize
+    }
+}
+
+impl DirectProbeHashable for u32 {
+    #[inline(always)]
+    fn probe_hash(self) -> usize {
+        (self as i32).probe_hash()
+    }
+}
+
+impl DirectProbeHashable for u64 {
+    #[inline(always)]
+    fn probe_hash(self) -> usize {
+        (self as i64).probe_hash()
+    }
+}
+
+impl DirectProbeHashable for i128 {
+    #[inline(always)]
+    fn probe_hash(self) -> usize {
+        // Mix both halves with different constants to avoid collisions when lo = hi * C
+        let lo = self as u64;
+        let hi = (self >> 64) as u64;
+        let x = lo.wrapping_mul(GOLDEN_RATIO_64) ^ hi.wrapping_mul(SPLITMIX_CONSTANT);
+        (x ^ (x >> 32)) as usize
+    }
+}
+
+impl<T: ArrowPrimitiveType> DirectProbeFilter<T>
 where
-    T::Native: Hash + Eq,
+    T::Native: DirectProbeHashable,
 {
     pub(crate) fn try_new(in_array: &ArrayRef) -> Result<Self> {
         let arr = in_array.as_primitive_opt::<T>().ok_or_else(|| {
             exec_datafusion_err!(
-                "PrimitiveFilter: expected {} array",
+                "DirectProbeFilter: expected {} array",
                 std::any::type_name::<T>()
             )
         })?;
-        Ok(Self {
-            null_count: arr.null_count(),
-            set: arr.iter().flatten().collect(),
-        })
+
+        // Collect unique values using HashSet for deduplication
+        let unique_values: HashSet<_> = arr.iter().flatten().collect();
+
+        Ok(Self::from_values_inner(
+            unique_values.into_iter(),
+            arr.null_count(),
+        ))
     }
 
-    /// Check membership using a raw values slice (zero-copy path for type reinterpretation).
+    /// Creates a DirectProbeFilter from an iterator of values.
+    ///
+    /// This is useful when building the filter from pre-processed values
+    /// (e.g., masked views for Utf8View).
+    pub(crate) fn from_values(values: impl Iterator<Item = T::Native>) -> Self {
+        // Collect into HashSet for deduplication
+        let unique_values: HashSet<_> = values.collect();
+        Self::from_values_inner(unique_values.into_iter(), 0)
+    }
+
+    /// Internal constructor from deduplicated values
+    fn from_values_inner(
+        unique_values: impl Iterator<Item = T::Native>,
+        null_count: usize,
+    ) -> Self {
+        let unique_values: Vec<_> = unique_values.collect();
+
+        // Size table to ~25% load factor for fewer collisions
+        let n = unique_values.len().max(1);
+        let table_size = (n * LOAD_FACTOR_INVERSE)
+            .next_power_of_two()
+            .max(MIN_TABLE_SIZE);
+        let mask = table_size - 1;
+
+        let mut table: Box<[Option<T::Native>]> =
+            vec![None; table_size].into_boxed_slice();
+
+        // Insert all values using linear probing
+        for v in unique_values {
+            let mut slot = v.probe_hash() & mask;
+            loop {
+                if table[slot].is_none() {
+                    table[slot] = Some(v);
+                    break;
+                }
+                slot = (slot + 1) & mask;
+            }
+        }
+
+        Self {
+            null_count,
+            table,
+            mask,
+        }
+    }
+
+    /// O(1) single-value lookup with linear probing.
+    ///
+    /// Returns true if the value is in the set.
+    #[inline(always)]
+    pub(crate) fn contains_single(&self, needle: T::Native) -> bool {
+        let mut slot = needle.probe_hash() & self.mask;
+        loop {
+            // SAFETY: `slot` is always < table.len() because:
+            // - `slot = hash & mask` where `mask = table.len() - 1`
+            // - table size is always a power of 2
+            // - `(slot + 1) & mask` wraps around within bounds
+            match unsafe { self.table.get_unchecked(slot) } {
+                None => return false,
+                Some(v) if *v == needle => return true,
+                _ => slot = (slot + 1) & self.mask,
+            }
+        }
+    }
+
+    /// Check membership using a raw values slice
     #[inline]
     pub(crate) fn contains_slice(
         &self,
-        values: &[T::Native],
+        input: &[T::Native],
         nulls: Option<&arrow::buffer::NullBuffer>,
         negated: bool,
     ) -> BooleanArray {
-        build_in_list_result(
-            values.len(),
-            nulls,
-            self.null_count > 0,
-            negated,
-            // SAFETY: i is in bounds since we iterate 0..values.len()
-            |i| self.set.contains(unsafe { values.get_unchecked(i) }),
-        )
+        build_in_list_result(input.len(), nulls, self.null_count > 0, negated, |i| {
+            // SAFETY: i is in bounds since we iterate 0..input.len()
+            self.contains_single(unsafe { *input.get_unchecked(i) })
+        })
     }
 }
 
-impl<T> StaticFilter for PrimitiveFilter<T>
+impl<T> StaticFilter for DirectProbeFilter<T>
 where
     T: ArrowPrimitiveType + 'static,
-    T::Native: Hash + Eq + Send + Sync + 'static,
+    T::Native: DirectProbeHashable + Send + Sync + 'static,
 {
+    #[inline]
     fn null_count(&self) -> usize {
         self.null_count
     }
 
+    #[inline]
     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!(
-                "PrimitiveFilter: expected {} array",
-                std::any::type_name::<T>()
-            )
-        })?;
-        let 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()
-            |i| self.set.contains(unsafe { values.get_unchecked(i) }),
-        ))
+        // Use raw buffer access for better optimization
+        let data = v.to_data();
+        let values: &[T::Native] = data.buffer::<T::Native>(0);
+        Ok(self.contains_slice(values, v.nulls(), negated))
     }
 }