IN LIST: handle dictionary-encoded needles once (rather than in each filter type) to reduce code size

[alamb]

Is your feature request related to a problem or challenge?

• Part of Further improve performance of IN list evaluation #19241

Every StaticFilter in the in_list module handles dictionary-encoded needles itself, via the handle_dictionary! macro (or a hand-rolled copy in ArrayStaticFilter). The macro expands downcast_dictionary_array! — a match over all eight dictionary key types — into every monomorphized contains implementation.

This basically increases the binary size with a lot of repetition for almost no performance gain

You can measure the code size using

cargo llvm-lines -p datafusion-physical-expr --lib
...
<datafusion_physical_expr[3705280180f342a4]::expressions::in_list::branchless_filter::BranchlessFilter<arrow_array[30e2684e55267fac]::types::DurationMillisecondType> as datafusion_physical_expr[3705280180f342a4]::expressions::in_list::static_filter::StaticFilter>::contains
     775 (0.1%,  2.5%)      1 (0.0%,  0.0%)  <datafusion_physical_expr[3705280180f342a4]::expressions::in_list::branchless_filter::BranchlessFilter<arrow_array[30e2684e55267fac]::types::DurationNanosecondType> as datafusion_physical_expr[3705280180f342a4]::expressions::in_list::static_filter::StaticFilter>::contains
     775 (0.1%,  2.7%)      1 (0.0%,  0.0%)  <datafusion_physical_expr[3705280180f342a4]::expressions::in_list::branchless_filter::BranchlessFilter<arrow_array[30e2684e55267fac]::types::DurationSecondType> as datafusion_physical_expr[3705280180f342a4]::expressions::in_list::static_filter::StaticFilter>::contains
     775 (0.1%,  2.8%)      1 (0.0%,  0.1%)  <datafusion_physical_expr[3705280180f342a4]::expressions::in_list::branchless_filter::BranchlessFilter<arrow_array[30e2684e55267fac]::types::Float16Type> as datafusion_physical_expr[3705280180f342a4]::expressions::in_list::static_filter::StaticFilter>::contains
     775 (0.1%,  2.9%)      1 (0.0%,  0.1%)  <datafusion_physical_expr[3705280180f342a4]::expressions::in_list::branchless_filter::BranchlessFilter<arrow_array[30e2684e55267fac]::types::Float32Type> as datafusion_physical_expr[3705280180f342a4]::expressions::in_list::static_filter::StaticFilter>::contains
     775 (0.1%,  3.0%)      1 (0.0%,  0.1%)  <datafusion_physical_expr[3705280180f342a4]::expressions::in_list::branchless_filter::BranchlessFilter<arrow_array[30e2684e55267fac]::types::Float64Type> as datafusion_physical_expr[3705280180f342a4]::expressions::in_list::static_filter::StaticFilter>::contains
     775 (0.1%,  3.1%)      1 (0.0%,  0.1%)  <datafusion_physical_expr[3705280180f342a4]::expressions::in_list::branchless_filter::BranchlessFilter<arrow_array[30e2684e55267fac]::types::Int16Type> as datafusion_physical_expr[3705280180f342a4]::expressions::in_list::static_filter::StaticFilter>::contains
     775 (0.1%,  3.2%)      1 (0.0%,  0.1%)  <datafusion_physical_expr[3705280180f342a4]::expressions::in_list::branchless_filter::BranchlessFilter<arrow_array[30e2684e55267fac]::types::Int32Type> as datafusion_physical_expr[3705280180f342a4]::expressions::in_list::static_filter::StaticFilter>::contains
     775 (0.1%,  3.3%)      1 (0.0%,  0.1%)  <datafusion_physical_expr[3705280180f342a4]::expressions::in_list::branchless_filter::BranchlessFilter<arrow_array[30e2684e55267fac]::types::Int64Type> as 
...

According to Claude

1. most of the various InList contains copies ( ~85%) is the repeated dictionary match.
2. In total the duplicated dictionary handling accounts for ~27,000 IR lines, about 4% of the whole crate.

Each new specialized filter in the #19241 series is generating more copies of the dictionary handling

Describe the solution you'd like

Avoid generating the code over and over again for dictionary handling

Describe alternatives you've considered

Implement dictionary handling once as a wrapper filter that delegates to any inner StaticFilter for the values, and wrap it once in instantiate_static_filter:

/// Unwraps dictionary-encoded needles for the wrapped filter.
struct DictionaryFilter {
    /// The haystack's (non-dictionary) value type.
    values_type: DataType,
    inner: StaticFilterRef,
}

impl StaticFilter for DictionaryFilter {
    fn null_count(&self) -> usize {
        self.inner.null_count()
    }

    fn contains(&self, v: &dyn Array, negated: bool) -> Result<BooleanArray> {
        downcast_dictionary_array! {
            v => {
                if v.values().data_type() == &self.values_type {
                    let values_contains =
                        self.inner.contains(v.values().as_ref(), negated)?;
                    let result = take(&values_contains, v.keys(), None)?;
                    return Ok(downcast_array(result.as_ref()));
                }
            }
            _ => {}
        }
        self.inner.contains(v, negated)
    }
}

The handle_dictionary! macro and ArrayStaticFilter's inline dictionary block are then deleted.

A prototype of this change was promising and reduced the size of datafusion-physical-expr by ~4.1% with all in_list unit tests and sqllogictests passing.

The runtime cost is one additional virtual call plus a DataType match per batch; per-row hot loops are unchanged.

Another benefit of this design is that guture representation adapters (e.g. #24088) then get dictionary support for free instead of embedding the match per instantiation.

Additional context

To measure the code size impact, use cargo-llvm-lines (cargo install cargo-llvm-lines):

Details

```bash # Baseline (run on main or the PR base), then again with the change applied cargo llvm-lines -p datafusion-physical-expr --lib > /tmp/llvm_lines_before.txt cargo llvm-lines -p datafusion-physical-expr --lib > /tmp/llvm_lines_after.txt

Crate-wide totals (first data line of each report)

grep "(TOTAL)" /tmp/llvm_lines_before.txt /tmp/llvm_lines_after.txt

IR lines attributable to the in_list module

grep "in_list" /tmp/llvm_lines_before.txt | awk '{s+=$1} END {print s}'
grep "in_list" /tmp/llvm_lines_after.txt | awk '{s+=$1} END {print s}'

Size of the monomorphized contains() copies specifically

grep -E "in_list::(primitive_filter|branchless_filter|fixed_size_binary_filter)"
/tmp/llvm_lines_after.txt | grep "StaticFilter>::contains"

Note llvm-lines counts pre-optimization LLVM IR in a debug build, so it is a proxy for generated code and compile time rather than exact final binary bytes.
</details>
Discussed while reviewing #24102; the prototype applies on top of that PR.