IN LIST: add direct-probe hash filter for large primitive lists

datafusion/physical-expr/benches/in_list_strategy.rs

@@ -37,6 +37,7 @@
 //! | 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 |
+//! | 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 |
 //! | Utf8View short-string cases | Utf8View | 8-byte strings | 4, 16, 64, 256 |
@@ -45,8 +46,9 @@
 //! | Shared-prefix string cases | Utf8, Utf8View | same prefix, different suffix | 16, 32, 64 |
 //! | Fixed-size binary cases | FixedSizeBinary(16) | fixed-width binary values | 4, 64, 256, 10000 |
 
+use arrow::array::types::IntervalMonthDayNano;
 use arrow::array::*;
-use arrow::datatypes::{Field, Int32Type, Schema};
+use arrow::datatypes::{Field, Int32Type, IntervalMonthDayNanoType, Schema};
 use arrow::record_batch::RecordBatch;
 use criterion::{BenchmarkId, Criterion, criterion_group, criterion_main};
 use datafusion_common::ScalarValue;
@@ -528,6 +530,28 @@ fn bench_timestamp_ns(c: &mut Criterion) {
     }
 }
 
+fn bench_interval_month_day_nano(c: &mut Criterion) {
+    for match_pct in MATCH_RATES {
+        bench_numeric::<IntervalMonthDayNano, IntervalMonthDayNanoArray>(
+            c,
+            "interval_month_day_nano",
+            &format!("small_list/list=4/match={match_pct}%"),
+            &NumericBenchConfig::new(
+                4,
+                match_pct as f64 / 100.0,
+                |rng| {
+                    IntervalMonthDayNanoType::make_value(
+                        rng.random_range(-120..=120),
+                        rng.random_range(-31..=31),
+                        rng.random_range(-1_000_000_000..=1_000_000_000),
+                    )
+                },
+                |v| ScalarValue::IntervalMonthDayNano(Some(v)),
+            ),
+        );
+    }
+}
+
 // =============================================================================
 // UTF8 STRING CASE BENCHMARKS
 // =============================================================================
@@ -1049,7 +1073,7 @@ fn bench_fixed_size_binary(c: &mut Criterion) {
 criterion_group! {
     name = benches;
     config = Criterion::default();
-    targets = bench_narrow_integer, bench_primitive, bench_f32, bench_timestamp_ns, bench_utf8, bench_utf8view, bench_dictionary, bench_nulls, bench_fixed_size_binary
+    targets = bench_narrow_integer, bench_primitive, bench_f32, bench_timestamp_ns, bench_interval_month_day_nano, bench_utf8, bench_utf8view, bench_dictionary, bench_nulls, bench_fixed_size_binary
 }
 
 criterion_main!(benches);

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

@@ -2900,10 +2900,9 @@ mod tests {
 
     #[test]
     fn test_in_list_esoteric_types() -> Result<()> {
-        // Test esoteric/less common types to validate the transform and mapping flow.
-        // These types are reinterpreted to base primitive types (e.g., Timestamp -> UInt64,
-        // Interval -> Decimal128, Float16 -> UInt16). We just need to verify basic
-        // functionality works - no need for comprehensive null handling tests.
+        // Test less common types covered by IN-list evaluation. Some of these
+        // use specialized filters, and others fall back to the generic path;
+        // this keeps the end-to-end behavior covered either way.
 
         // Helper: simple IN test that expects [Some(true), Some(false)]
         let test_type = |data_type: DataType,
@@ -2926,7 +2925,7 @@ mod tests {
             Ok(())
         };
 
-        // Timestamp types (all units map to Int64 -> UInt64)
+        // Timestamp types
         test_type(
             DataType::Timestamp(TimeUnit::Second, None),
             Arc::new(TimestampSecondArray::from(vec![Some(1000), Some(2000)])),
@@ -2960,7 +2959,7 @@ mod tests {
             ],
         )?;
 
-        // Time32 and Time64 (map to Int32 -> UInt32 and Int64 -> UInt64 respectively)
+        // Time32 and Time64
         test_type(
             DataType::Time32(TimeUnit::Second),
             Arc::new(Time32SecondArray::from(vec![Some(3600), Some(7200)])),
@@ -3006,7 +3005,7 @@ mod tests {
             ],
         )?;
 
-        // Duration types (map to Int64 -> UInt64)
+        // Duration types
         test_type(
             DataType::Duration(TimeUnit::Second),
             Arc::new(DurationSecondArray::from(vec![Some(86400), Some(172800)])),
@@ -3052,7 +3051,7 @@ mod tests {
             ],
         )?;
 
-        // Interval types (map to 16-byte Decimal128Type)
+        // Interval types
         test_type(
             DataType::Interval(IntervalUnit::YearMonth),
             Arc::new(IntervalYearMonthArray::from(vec![Some(12), Some(24)])),
@@ -3114,8 +3113,7 @@ mod tests {
             ],
         )?;
 
-        // Decimal256 (maps to Decimal128Type for 16-byte width)
-        // Need to use with_precision_and_scale() to set the metadata
+        // Decimal256. Need to use with_precision_and_scale() to set the metadata.
         let precision = 38;
         let scale = 10;
         test_type(