IN LIST: centralize primitive filters and optimize Decimal128

datafusion/physical-expr/benches/in_list_strategy.rs

@@ -38,6 +38,7 @@
 //! | 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 |
 //! | 128-bit interval cases | IntervalMonthDayNano | small lists | 4 |
+//! | Decimal128 cases | Decimal128 | larger lists | 5, 64 |
 //! | Utf8 short-string cases | Utf8 | 8-byte strings | 4, 64, 256 |
 //! | Utf8 long-string cases | Utf8 | 24-byte strings | 4, 64, 256 |
 //! | Utf8View short-string cases | Utf8View | 8-byte strings | 4, 16, 64, 256 |
@@ -463,6 +464,23 @@ fn bench_primitive(c: &mut Criterion) {
         }
     }
 
+    // Decimal128: benchmark the first hash-set list size (5) and a larger list (64).
+    for list_size in [5, 64] {
+        for match_pct in MATCH_RATES {
+            bench_numeric::<i128, Decimal128Array>(
+                c,
+                "primitive",
+                &format!("decimal128/large_list/list={list_size}/match={match_pct}%"),
+                &NumericBenchConfig::new(
+                    list_size,
+                    match_pct as f64 / 100.0,
+                    |rng| i128::from(rng.random::<i64>()),
+                    |v| ScalarValue::Decimal128(Some(v), 38, 10),
+                ),
+            );
+        }
+    }
+
     // NOT IN benchmark: test negated path
     bench_numeric::<i32, Int32Array>(
         c,

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

@@ -43,15 +43,15 @@ mod result;
 mod static_filter;
 mod strategy;
 
-use static_filter::StaticFilter;
+use static_filter::StaticFilterRef;
 use strategy::instantiate_static_filter;
 
 /// InList
 pub struct InListExpr {
     expr: Arc<dyn PhysicalExpr>,
     list: Vec<Arc<dyn PhysicalExpr>>,
     negated: bool,
-    static_filter: Option<Arc<dyn StaticFilter + Send + Sync>>,
+    static_filter: Option<StaticFilterRef>,
 }
 
 impl Debug for InListExpr {
@@ -148,7 +148,7 @@ impl InListExpr {
         expr: Arc<dyn PhysicalExpr>,
         list: Vec<Arc<dyn PhysicalExpr>>,
         negated: bool,
-        static_filter: Option<Arc<dyn StaticFilter + Send + Sync>>,
+        static_filter: Option<StaticFilterRef>,
     ) -> Self {
         Self {
             expr,
@@ -222,8 +222,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 branchless, bitmap, or hash-set filters
+    /// when the list's physical representation supports them.
     ///
     /// 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 +2592,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(), which selects a primitive static 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

@@ -42,7 +42,7 @@
 //! different NaN values. [`BranchlessFilterType`] defines these safe,
 //! same-sized mappings and checks their sizes at compile time.
 //!
-//! The fast path is intentionally limited to short lists:
+//! The fast path is limited to short lists:
 //!
 //! - 16 values for 1-byte types
 //! - 8 values for 2-byte types