IN LIST: add keyed frozen set for large primitive lists

datafusion/physical-expr/benches/in_list_strategy.rs

@@ -35,8 +35,8 @@
 //! |------|-------|-----------------|-------------------|
 //! | Narrow integer cases | UInt8 | small value domain | 4, 16 |
 //! | Narrow integer cases | Int16, Float16 | larger value domain | 4, 64, 256 |
-//! | 32-bit primitive cases | Int32, Float32 | small and large lists | 4, 32, 64, 256 |
-//! | 64-bit primitive cases | Int64, TimestampNs | small and large lists | 4, 16, 32, 128 |
+//! | 32-bit primitive cases | Int32, Float32 | small and large lists | 4, 32, 33, 64, 256, 1024, 10000 |
+//! | 64-bit primitive cases | Int64, TimestampNs | small and large lists | 4, 16, 17, 32, 128, 1024, 10000 |
 //! | 128-bit interval cases | IntervalMonthDayNano | small lists | 4 |
 //! | Utf8 short-string cases | Utf8 | 8-byte strings | 4, 64, 256 |
 //! | Utf8 long-string cases | Utf8 | 24-byte strings | 4, 64, 256 |
@@ -420,7 +420,7 @@ fn bench_narrow_integer(c: &mut Criterion) {
 
 fn bench_primitive(c: &mut Criterion) {
     // Int32: small and larger list sizes
-    for list_size in [4, 32, 64, 256] {
+    for list_size in [4, 32, 33, 64, 256, 1024, 10_000] {
         let list_case = if list_size <= 32 {
             "small_list"
         } else {
@@ -442,7 +442,7 @@ fn bench_primitive(c: &mut Criterion) {
     }
 
     // Int64: small and larger list sizes
-    for list_size in [4, 16, 32, 128] {
+    for list_size in [4, 16, 17, 32, 128, 1024, 10_000] {
         let list_case = if list_size <= 16 {
             "small_list"
         } else {

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

@@ -38,6 +38,7 @@ use datafusion_expr::{ColumnarValue, expr_vec_fmt};
 
 mod array_static_filter;
 mod branchless_filter;
+mod frozen_set;
 mod primitive_filter;
 mod result;
 mod static_filter;
@@ -222,8 +223,8 @@ impl InListExpr {
     /// Create a new InList expression, using a static filter when possible.
     ///
     /// This validates data types and attempts to create a static filter for constant
-    /// list expressions. Uses specialized StaticFilter implementations for better
-    /// performance (e.g., Int32StaticFilter for Int32).
+    /// list expressions. Uses specialized `StaticFilter` implementations for better
+    /// performance (for example, primitive filters for fixed-width values).
     ///
     /// Returns an error if data types don't match. If the list contains non-constant
     /// expressions, falls back to dynamic evaluation at runtime.
@@ -2592,7 +2593,7 @@ mod tests {
         // Create IN list with Int32 literals: (100, 200, 300)
         let list = vec![lit(100i32), lit(200i32), lit(300i32)];
 
-        // Create InListExpr via in_list() - this uses Int32StaticFilter for Int32 lists
+        // Create InListExpr via in_list() - this uses a specialized primitive filter
         let expr = in_list(col_a, list, &false, &schema)?;
 
         // Create dictionary-encoded batch with values [100, 200, 500]

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

@@ -58,8 +58,9 @@
 //! would be useful. Wider types have too many possible values for such a
 //! bitmap, so their limits are tuned separately.
 //!
-//! Larger lists use the standard filter strategy, including bitmap filters for
-//! one- and two-byte types.
+//! Larger lists use another filter strategy: bitmap filters for one- and
+//! two-byte types, frozen sets for supported four- and eight-byte types, and
+//! the standard fallback for the remaining types.
 //!
 //! # What about nulls?
 //!

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

@@ -0,0 +1,213 @@
+// 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.
+
+//! Immutable set for fixed-width values.
+
+use std::hash::{BuildHasher, Hash};
+
+use datafusion_common::{Result, exec_datafusion_err};
+use hashbrown::{DefaultHashBuilder, HashTable};
+
+/// Immutable set optimized for repeated membership tests.
+///
+/// Each hash bucket holds two values directly. Values that collide beyond those
+/// two slots use a `HashTable`, which keeps the common lookup path to one hash
+/// and two bucket comparisons without sacrificing collision handling.
+///
+/// The first member is used as the empty-slot sentinel and handled before
+/// lookup, so primary slots store `V` directly without an `Option<V>` wrapper.
+pub(super) struct FrozenSet<V> {
+    hash_builder: DefaultHashBuilder,
+    sentinel: Option<V>,
+    buckets: Box<[[V; 2]]>,
+    overflowed: Box<[bool]>,
+    overflow: HashTable<V>,
+}
+
+impl<V> FrozenSet<V>
+where
+    V: Copy + Eq + Hash,
+{
+    pub(super) fn try_new(values: &[V]) -> Result<Self> {
+        let Some((&sentinel, values)) = values.split_first() else {
+            return Ok(Self {
+                hash_builder: DefaultHashBuilder::default(),
+                sentinel: None,
+                buckets: Box::default(),
+                overflowed: Box::default(),
+                overflow: HashTable::new(),
+            });
+        };
+
+        if values.is_empty() {
+            return Ok(Self {
+                hash_builder: DefaultHashBuilder::default(),
+                sentinel: Some(sentinel),
+                buckets: Box::default(),
+                overflowed: Box::default(),
+                overflow: HashTable::new(),
+            });
+        }
+
+        let bucket_count = values
+            .len()
+            .checked_add(1)
+            .ok_or_else(|| exec_datafusion_err!("FrozenSet capacity overflow"))?;
+        bucket_count
+            .checked_mul(2)
+            .ok_or_else(|| exec_datafusion_err!("FrozenSet capacity overflow"))?;
+        let mut set = Self {
+            hash_builder: DefaultHashBuilder::default(),
+            sentinel: Some(sentinel),
+            buckets: vec![[sentinel; 2]; bucket_count].into_boxed_slice(),
+            overflowed: vec![false; bucket_count].into_boxed_slice(),
+            overflow: HashTable::with_capacity(bucket_count / 8),
+        };
+        values.iter().copied().for_each(|value| set.insert(value));
+        Ok(set)
+    }
+
+    fn insert(&mut self, value: V) {
+        let sentinel = self.sentinel.expect("non-empty frozen set");
+        if value == sentinel {
+            return;
+        }
+
+        let hash = self.hash_builder.hash_one(value);
+        let bucket = reduce_hash(hash, self.buckets.len());
+        let slots = &mut self.buckets[bucket];
+        if slots[0] == sentinel {
+            slots[0] = value;
+            return;
+        }
+        if slots[0] == value {
+            return;
+        }
+        if slots[1] == sentinel {
+            slots[1] = value;
+            return;
+        }
+        if slots[1] == value {
+            return;
+        }
+
+        let hash_builder = &self.hash_builder;
+        self.overflow
+            .entry(
+                hash,
+                |stored| *stored == value,
+                |stored| hash_builder.hash_one(stored),
+            )
+            .or_insert(value);
+        self.overflowed[bucket] = true;
+    }
+
+    #[inline(always)]
+    pub(super) fn contains(&self, value: V) -> bool {
+        let Some(sentinel) = self.sentinel else {
+            return false;
+        };
+        if value == sentinel {
+            return true;
+        }
+        if self.buckets.is_empty() {
+            return false;
+        }
+
+        let hash = self.hash_builder.hash_one(value);
+        let bucket = reduce_hash(hash, self.buckets.len());
+        // SAFETY: `reduce_hash` returns an index in `0..buckets.len()`.
+        let slots = unsafe { self.buckets.get_unchecked(bucket) };
+        if (slots[0] == value) | (slots[1] == value) {
+            return true;
+        }
+
+        // SAFETY: `overflowed` has exactly one entry per bucket.
+        (unsafe { *self.overflowed.get_unchecked(bucket) })
+            && self
+                .overflow
+                .find(hash, |stored| *stored == value)
+                .is_some()
+    }
+}
+
+#[inline(always)]
+fn reduce_hash(hash: u64, len: usize) -> usize {
+    #[cfg(target_pointer_width = "64")]
+    return ((hash as u128 * len as u128) >> 64) as usize;
+
+    #[cfg(target_pointer_width = "32")]
+    return (((hash as u32) as u64 * len as u64) >> 32) as usize;
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::hash::{Hash, Hasher};
+
+    #[derive(Clone, Copy, Debug, Eq, PartialEq)]
+    struct Colliding(u64);
+
+    impl Hash for Colliding {
+        fn hash<H: Hasher>(&self, state: &mut H) {
+            0_u8.hash(state);
+        }
+    }
+
+    #[test]
+    fn handles_empty_duplicates_and_sentinel_member() {
+        let empty = FrozenSet::<u64>::try_new(&[]).unwrap();
+        assert!(!empty.contains(0));
+
+        let singleton = FrozenSet::try_new(&[42]).unwrap();
+        assert!(singleton.contains(42));
+        assert!(!singleton.contains(7));
+
+        let values = [42, 7, 42, 9, 11, 7];
+        let set = FrozenSet::try_new(&values).unwrap();
+        for value in [42, 7, 9, 11] {
+            assert!(set.contains(value));
+        }
+        for value in [0, 8, 10, 12] {
+            assert!(!set.contains(value));
+        }
+    }
+
+    #[test]
+    fn handles_many_values() {
+        let values = (0_u64..10_000).collect::<Vec<_>>();
+        let set = FrozenSet::try_new(&values).unwrap();
+        assert!(values.iter().all(|&value| set.contains(value)));
+        assert!((10_000..20_000).all(|value| !set.contains(value)));
+    }
+
+    #[test]
+    fn handles_u128_values() {
+        let values = [1_u128, (1_u128 << 64) | 1, (2_u128 << 96) | 7, u128::MAX];
+        let set = FrozenSet::try_new(&values).unwrap();
+        assert!(values.iter().all(|&value| set.contains(value)));
+        assert!(!set.contains(1_u128 << 96));
+    }
+
+    #[test]
+    fn handles_collisions() {
+        let values = (0..128).map(Colliding).collect::<Vec<_>>();
+        let set = FrozenSet::try_new(&values).unwrap();
+        assert!(values.iter().all(|&value| set.contains(value)));
+        assert!((128..256).all(|value| !set.contains(Colliding(value))));
+    }
+}