Implement Bitmap Filter for UInt8 (Stack-based)

Replaces HashSet<u8> with a 32-byte stack-allocated bitmap. Provides O(1) membership testing via bit-shifting, significantly reducing memory overhead and improving cache locality. Triggers for UInt8 arrays.

Author: geoffreyclaude

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

@@ -15,7 +15,9 @@
 // specific language governing permissions and limitations
 // under the License.
 
-//! Specialized primitive type filters for InList expressions
+//! Optimized primitive type filters for InList expressions
+//!
+//! This module provides high-performance membership testing for Arrow primitive types.
 
 use arrow::array::{
     Array, ArrayRef, AsArray, BooleanArray, downcast_array, downcast_dictionary_array,
@@ -26,8 +28,114 @@ use arrow::datatypes::*;
 use datafusion_common::{HashSet, Result, exec_datafusion_err};
 use std::hash::{Hash, Hasher};
 
+use super::result::{build_in_list_result, handle_dictionary};
 use super::static_filter::StaticFilter;
 
+// =============================================================================
+// BITMAP FILTERS (O(1) lookup for u8/u16 via bit test)
+// =============================================================================
+
+/// Trait for bitmap storage (stack-allocated for u8, heap-allocated for u16).
+pub(crate) trait BitmapStorage: Send + Sync {
+    fn new_zeroed() -> Self;
+    fn set_bit(&mut self, index: usize);
+    fn get_bit(&self, index: usize) -> bool;
+}
+
+impl BitmapStorage for [u64; 4] {
+    #[inline]
+    fn new_zeroed() -> Self {
+        [0u64; 4]
+    }
+    #[inline]
+    fn set_bit(&mut self, index: usize) {
+        self[index / 64] |= 1u64 << (index % 64);
+    }
+    #[inline(always)]
+    fn get_bit(&self, index: usize) -> bool {
+        (self[index / 64] >> (index % 64)) & 1 != 0
+    }
+}
+
+/// Configuration trait for bitmap filters.
+pub(crate) trait BitmapFilterConfig: Send + Sync + 'static {
+    type Native: ArrowNativeType + Copy + Send + Sync;
+    type ArrowType: ArrowPrimitiveType<Native = Self::Native>;
+    type Storage: BitmapStorage;
+
+    fn to_index(v: Self::Native) -> usize;
+}
+
+/// Config for u8 bitmap (256 bits = 32 bytes, fits in cache line).
+pub(crate) enum U8Config {}
+impl BitmapFilterConfig for U8Config {
+    type Native = u8;
+    type ArrowType = UInt8Type;
+    type Storage = [u64; 4];
+
+    #[inline(always)]
+    fn to_index(v: u8) -> usize {
+        v as usize
+    }
+}
+
+/// Bitmap filter for O(1) set membership via single bit test.
+///
+/// For small integer types (u8/u16), bitmap lookup outperforms both branchless
+/// and hashed approaches at all list sizes.
+pub(crate) struct BitmapFilter<C: BitmapFilterConfig> {
+    null_count: usize,
+    bits: C::Storage,
+}
+
+impl<C: BitmapFilterConfig> BitmapFilter<C> {
+    pub(crate) fn try_new(in_array: &ArrayRef) -> Result<Self> {
+        let prim_array =
+            in_array.as_primitive_opt::<C::ArrowType>().ok_or_else(|| {
+                exec_datafusion_err!("BitmapFilter: expected primitive array")
+            })?;
+        let mut bits = C::Storage::new_zeroed();
+        for v in prim_array.iter().flatten() {
+            bits.set_bit(C::to_index(v));
+        }
+        Ok(Self {
+            null_count: prim_array.null_count(),
+            bits,
+        })
+    }
+
+    #[inline(always)]
+    fn check(&self, needle: C::Native) -> bool {
+        self.bits.get_bit(C::to_index(needle))
+    }
+}
+
+impl<C: BitmapFilterConfig> StaticFilter for BitmapFilter<C> {
+    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::<C::ArrowType>().ok_or_else(|| {
+            exec_datafusion_err!("BitmapFilter: expected primitive array")
+        })?;
+        let input_values = v.values();
+        Ok(build_in_list_result(
+            v.len(),
+            v.nulls(),
+            self.null_count > 0,
+            negated,
+            #[inline(always)]
+            |i| self.check(unsafe { *input_values.get_unchecked(i) }),
+        ))
+    }
+}
+
+// =============================================================================
+// LEGACY FILTERS (to be replaced by optimized ones in subsequent commits)
+// =============================================================================
+
 /// Wrapper for f32 that implements Hash and Eq using bit comparison.
 #[derive(Clone, Copy)]
 pub(crate) struct OrderedFloat32(pub(crate) f32);
@@ -177,7 +285,6 @@ primitive_static_filter!(Int8StaticFilter, Int8Type);
 primitive_static_filter!(Int16StaticFilter, Int16Type);
 primitive_static_filter!(Int32StaticFilter, Int32Type);
 primitive_static_filter!(Int64StaticFilter, Int64Type);
-primitive_static_filter!(UInt8StaticFilter, UInt8Type);
 primitive_static_filter!(UInt16StaticFilter, UInt16Type);
 primitive_static_filter!(UInt32StaticFilter, UInt32Type);
 primitive_static_filter!(UInt64StaticFilter, UInt64Type);

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

@@ -114,3 +114,27 @@ pub(crate) fn build_result_from_contains(
         (None, false, true) => BooleanArray::new(!&contains_buf, None),
     }
 }
+
+// =============================================================================
+// DICTIONARY ARRAY HANDLING
+// =============================================================================
+
+/// Macro to handle dictionary arrays in StaticFilter::contains implementations.
+///
+/// This macro extracts the dictionary values, performs the contains check on
+/// the values array, and then uses `take` to map the results back to the
+/// dictionary keys.
+macro_rules! handle_dictionary {
+    ($self:ident, $v:ident, $negated:ident) => {
+        arrow::array::downcast_dictionary_array! {
+            $v => {
+                let values_contains = $self.contains($v.values().as_ref(), $negated)?;
+                let result = arrow::compute::take(&values_contains, $v.keys(), None)?;
+                return Ok(arrow::array::downcast_array(result.as_ref()))
+            }
+            _ => {}
+        }
+    };
+}
+
+pub(crate) use handle_dictionary;

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

@@ -39,7 +39,7 @@ pub(crate) fn instantiate_static_filter(
         DataType::Int16 => Ok(Arc::new(Int16StaticFilter::try_new(&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(UInt8StaticFilter::try_new(&in_array)?)),
+        DataType::UInt8 => Ok(Arc::new(BitmapFilter::<U8Config>::try_new(&in_array)?)),
         DataType::UInt16 => Ok(Arc::new(UInt16StaticFilter::try_new(&in_array)?)),
         DataType::UInt32 => Ok(Arc::new(UInt32StaticFilter::try_new(&in_array)?)),
         DataType::UInt64 => Ok(Arc::new(UInt64StaticFilter::try_new(&in_array)?)),