perf(in_list): Optimize primitive and ByteView filters with Bloom filter and single hashing

This commit introduces a unified high-performance path for IN list operations:
- Implements an optimal 3-bit Bloom filter (512-bit space) for both `PrimitiveFilter` and `ByteViewMaskedFilter`.
- Refactors filters to use `hashbrown::HashTable` directly, ensuring the hash is computed only once and reused for Bloom check, stage-1 probe, and stage-2 verification.
- Centralizes chunked `u64` result building in `build_no_nulls_result` to bypass bit-by-bit builder overhead in null-free paths.
- Adds a saturation threshold (256 elements) to automatically bypass the Bloom check for large lists, preventing overhead.
- Includes an exhaustive structured test suite verifying 3,000+ combinations of types, sizes, and null logic.
- Fixes several clippy warnings and ensures compliance with project standards.
- These changes significantly reduce probe latency and memory pressure for medium-to-large IN lists across all optimized types.

Author: geoffreyclaude

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

@@ -3033,4 +3033,302 @@ mod tests {
 
         Ok(())
     }
+
+    /// Exhaustive structured test that verifies all lookup strategies, boundaries, and DataFusion types.
+    #[test]
+    fn test_in_list_exhaustive() -> Result<()> {
+        let types = vec![
+            DataType::Int8,
+            DataType::Int16,
+            DataType::Int32,
+            DataType::Int64,
+            DataType::Float32,
+            DataType::Float64,
+            DataType::Utf8,
+            DataType::LargeUtf8,
+            DataType::Binary,
+            DataType::LargeBinary,
+            DataType::Utf8View,
+            DataType::BinaryView,
+            DataType::Date32,
+            DataType::Date64,
+            DataType::Timestamp(TimeUnit::Nanosecond, None),
+            DataType::Timestamp(TimeUnit::Microsecond, Some("UTC".into())),
+            DataType::Decimal128(10, 2),
+            DataType::Decimal256(38, 10),
+            DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::Utf8)),
+            DataType::Dictionary(Box::new(DataType::Int8), Box::new(DataType::Int32)),
+            DataType::Struct(Fields::from(vec![Field::new("f1", DataType::Int32, true)])),
+        ];
+
+        // Critical thresholds: 4 (Decimal boundary), 16 (8B boundary), 32 (4B boundary), 256 (Bloom boundary)
+        // Test all sizes 1..33 to cover ALL unrolled branchless filters, plus saturation boundaries.
+        let mut sizes: Vec<usize> = (1..=33).collect();
+        sizes.extend(vec![100, 256, 257, 512]);
+
+        for dt in &types {
+            for size in &sizes {
+                for input_has_nulls in [false, true] {
+                    for list_has_nulls in [false, true] {
+                        for negated in [false, true] {
+                            for input_is_scalar in [false, true] {
+                                run_exhaustive_test_case(
+                                    dt,
+                                    *size,
+                                    input_has_nulls,
+                                    list_has_nulls,
+                                    negated,
+                                    input_is_scalar,
+                                )?;
+                            }
+                        }
+                    }
+                }
+            }
+        }
+        Ok(())
+    }
+
+    fn run_exhaustive_test_case(
+        dt: &DataType,
+        list_size: usize,
+        input_has_nulls: bool,
+        list_has_nulls: bool,
+        negated: bool,
+        input_is_scalar: bool,
+    ) -> Result<()> {
+        let array_len = 100;
+        let (input_array, in_list_scalars) = generate_deterministic_data(
+            dt,
+            array_len,
+            list_size,
+            input_has_nulls,
+            list_has_nulls,
+        );
+
+        let schema = Schema::new(vec![Field::new("a", dt.clone(), true)]);
+        let batch = RecordBatch::try_new(
+            Arc::new(schema.clone()),
+            vec![Arc::clone(&input_array)],
+        )?;
+        let col_a = col("a", &schema)?;
+        let list_exprs: Vec<Arc<dyn PhysicalExpr>> =
+            in_list_scalars.iter().map(|s| lit(s.clone())).collect();
+
+        let opt_result_arr = if input_is_scalar {
+            let scalar = ScalarValue::try_from_array(&input_array, 0)?;
+            let expr =
+                InListExpr::try_new(lit(scalar), list_exprs.clone(), negated, &schema)?;
+            expr.evaluate(&batch)?.into_array(batch.num_rows())?
+        } else {
+            let expr = InListExpr::try_new(
+                Arc::clone(&col_a),
+                list_exprs.clone(),
+                negated,
+                &schema,
+            )?;
+            expr.evaluate(&batch)?.into_array(batch.num_rows())?
+        };
+
+        let opt_result_bool = opt_result_arr
+            .as_any()
+            .downcast_ref::<BooleanArray>()
+            .unwrap();
+        let ref_result = compute_reference_result(
+            &batch,
+            &col_a,
+            &in_list_scalars,
+            negated,
+            input_is_scalar,
+        )?;
+
+        assert_eq!(
+            opt_result_bool, &ref_result,
+            "FAIL: Type={dt:?}, Size={list_size}, InNull={input_has_nulls}, ListNull={list_has_nulls}, Neg={negated}, Scalar={input_is_scalar}"
+        );
+
+        Ok(())
+    }
+
+    fn generate_deterministic_data(
+        dt: &DataType,
+        len: usize,
+        list_size: usize,
+        input_has_nulls: bool,
+        list_has_nulls: bool,
+    ) -> (ArrayRef, Vec<ScalarValue>) {
+        let base_dt = if let DataType::Dictionary(_, v) = dt {
+            v.as_ref()
+        } else {
+            dt
+        };
+
+        let input_scalars: Vec<ScalarValue> = (0..len)
+            .map(|i| {
+                if input_has_nulls && i % 10 == 0 {
+                    ScalarValue::try_new_null(base_dt).unwrap()
+                } else {
+                    deterministic_scalar(base_dt, i)
+                }
+            })
+            .collect();
+        let mut input_array = ScalarValue::iter_to_array(input_scalars).unwrap();
+
+        if let DataType::Dictionary(k, _) = dt {
+            match k.as_ref() {
+                DataType::Int32 => {
+                    let keys = Int32Array::from(
+                        (0..len as i32).map(|i| i % 10).collect::<Vec<_>>(),
+                    );
+                    input_array =
+                        Arc::new(DictionaryArray::<Int32Type>::new(keys, input_array))
+                            as ArrayRef;
+                }
+                DataType::Int8 => {
+                    let keys = Int8Array::from(
+                        (0..len as i8).map(|i| i % 10).collect::<Vec<_>>(),
+                    );
+                    input_array =
+                        Arc::new(DictionaryArray::<Int8Type>::new(keys, input_array))
+                            as ArrayRef;
+                }
+                _ => unreachable!(),
+            };
+        }
+
+        let mut in_list: Vec<ScalarValue> = (0..list_size)
+            .map(|i| deterministic_scalar(base_dt, i * 2))
+            .collect();
+
+        if list_has_nulls && !in_list.is_empty() {
+            in_list[0] = ScalarValue::try_new_null(base_dt).unwrap();
+        }
+
+        (input_array, in_list)
+    }
+
+    fn deterministic_scalar(dt: &DataType, i: usize) -> ScalarValue {
+        match dt {
+            DataType::Int8 => ScalarValue::Int8(Some(i as i8)),
+            DataType::Int16 => ScalarValue::Int16(Some(i as i16)),
+            DataType::Int32 => ScalarValue::Int32(Some(i as i32)),
+            DataType::Int64 => ScalarValue::Int64(Some(i as i64)),
+            DataType::Float32 => ScalarValue::Float32(Some(i as f32)),
+            DataType::Float64 => ScalarValue::Float64(Some(i as f64)),
+            DataType::Utf8 | DataType::Utf8View | DataType::LargeUtf8 => {
+                let s = format!("val-{i}");
+                if dt == &DataType::Utf8View {
+                    ScalarValue::Utf8View(Some(s))
+                } else if dt == &DataType::LargeUtf8 {
+                    ScalarValue::LargeUtf8(Some(s))
+                } else {
+                    ScalarValue::Utf8(Some(s))
+                }
+            }
+            DataType::Binary | DataType::LargeBinary | DataType::BinaryView => {
+                let b = format!("bin-{i}").into_bytes();
+                if dt == &DataType::BinaryView {
+                    ScalarValue::BinaryView(Some(b))
+                } else if dt == &DataType::LargeBinary {
+                    ScalarValue::LargeBinary(Some(b))
+                } else {
+                    ScalarValue::Binary(Some(b))
+                }
+            }
+            DataType::Date32 => ScalarValue::Date32(Some(i as i32)),
+            DataType::Date64 => ScalarValue::Date64(Some(i as i64)),
+            DataType::Timestamp(unit, tz) => {
+                let val = (i * 1000) as i64;
+                match unit {
+                    TimeUnit::Nanosecond => {
+                        ScalarValue::TimestampNanosecond(Some(val), tz.clone())
+                    }
+                    TimeUnit::Microsecond => {
+                        ScalarValue::TimestampMicrosecond(Some(val), tz.clone())
+                    }
+                    TimeUnit::Millisecond => {
+                        ScalarValue::TimestampMillisecond(Some(val), tz.clone())
+                    }
+                    TimeUnit::Second => {
+                        ScalarValue::TimestampSecond(Some(val), tz.clone())
+                    }
+                }
+            }
+            DataType::Decimal128(p, s) => {
+                ScalarValue::Decimal128(Some(i as i128), *p, *s)
+            }
+            DataType::Decimal256(p, s) => {
+                ScalarValue::Decimal256(Some(i256::from_i128(i as i128)), *p, *s)
+            }
+            DataType::Struct(fields) => {
+                let values: Vec<ScalarValue> = fields
+                    .iter()
+                    .map(|f| deterministic_scalar(f.data_type(), i))
+                    .collect();
+                ScalarValue::Struct(Arc::new(
+                    StructArray::try_new(
+                        fields.clone(),
+                        values.iter().map(|s| s.to_array().unwrap()).collect(),
+                        None,
+                    )
+                    .unwrap(),
+                ))
+            }
+            _ => ScalarValue::Int32(Some(i as i32)),
+        }
+    }
+
+    fn compute_reference_result(
+        batch: &RecordBatch,
+        col_a: &Arc<dyn PhysicalExpr>,
+        list: &[ScalarValue],
+        negated: bool,
+        input_is_scalar: bool,
+    ) -> Result<BooleanArray> {
+        let input_array = if input_is_scalar {
+            let val = col_a.evaluate(batch)?;
+            let arr = val.into_array(batch.num_rows())?;
+            let s = ScalarValue::try_from_array(&arr, 0)?;
+            s.to_array_of_size(batch.num_rows())?
+        } else {
+            col_a.evaluate(batch)?.into_array(batch.num_rows())?
+        };
+
+        let num_rows = batch.num_rows();
+        let mut result = Vec::with_capacity(num_rows);
+        for i in 0..num_rows {
+            let mut val = ScalarValue::try_from_array(&input_array, i)?;
+            if let ScalarValue::Dictionary(_, v) = val {
+                val = *v;
+            }
+
+            if val.is_null() {
+                result.push(None);
+                continue;
+            }
+
+            let mut found = false;
+            let mut has_null = false;
+            for list_val in list {
+                if list_val.is_null() {
+                    has_null = true;
+                    continue;
+                }
+                if list_val == &val {
+                    found = true;
+                    break;
+                }
+            }
+
+            let res = if found {
+                Some(!negated)
+            } else if has_null {
+                None
+            } else {
+                Some(negated)
+            };
+            result.push(res);
+        }
+        Ok(BooleanArray::from(result))
+    }
 }