RowConverter encodes every dictionary value for each batch, not just the values its rows reference #11901

Is your feature request related to a problem or challenge?

RowConverter::convert_columns on a dictionary array row-encodes the dictionary's whole values array, then copies each row's bytes out of it (Codec::encoder, arrow-row/src/lib.rs:777 on main):

Codec::Dictionary(converter, nulls) => {
    let values = array.as_any_dictionary().values().clone();
    let rows = converter.convert_columns(&[values])?;

So converting a short slice of a large dictionary costs time proportional to the dictionary, not to its rows. #7996 fixed the same problem for lists by encoding only the referenced list values.

The repro below (arrow 60.0.0) converts 8192 rows taken from the middle of a Dictionary(Int32, Utf8) whose values are all distinct:

use arrow_array::types::Int32Type;
use arrow_array::{Array, ArrayRef, DictionaryArray, Int32Array, StringArray};
use arrow_cast::cast;
use arrow_row::{RowConverter, SortField};
use arrow_schema::DataType;
use std::sync::Arc;
use std::time::Instant;

fn dictionary(values: usize) -> DictionaryArray<Int32Type> {
    let v = StringArray::from_iter_values((0..values).map(|i| format!("value_{i}")));
    let keys = Int32Array::from_iter_values(0..values as i32);
    DictionaryArray::try_new(keys, Arc::new(v)).unwrap()
}

fn per_convert(column: &ArrayRef) -> std::time::Duration {
    let converter = RowConverter::new(vec![SortField::new(column.data_type().clone())]).unwrap();
    converter.convert_columns(&[Arc::clone(column)]).unwrap(); // warm up
    let start = Instant::now();
    for _ in 0..20 {
        converter.convert_columns(&[Arc::clone(column)]).unwrap();
    }
    start.elapsed() / 20
}

fn main() {
    let rows = 8192;
    for values in [8_192, 100_000, 1_000_000] {
        // 8192 rows from the middle of a dictionary with `values` distinct strings
        let slice: ArrayRef = Arc::new(dictionary(values).slice((values - rows) / 2, rows));
        let utf8 = cast(&slice, &DataType::Utf8).unwrap();
        println!(
            "{values} values: Dictionary {:?}, same rows as Utf8 {:?}",
            per_convert(&slice),
            per_convert(&utf8)
        );
    }
}

The input always has 8192 rows, but the time grows with the size of the dictionary. The table shows the range over 5 runs of the snippet (release build, Apple M4 Pro).

In DataFusion this makes a spilling GROUP BY on a high-cardinality dictionary column effectively never finish. ClickBench Q33 (GROUP BY "URL" with Parquet dictionary reads enabled) takes about 15 s without a memory limit. With an 8 GB limit, a three-iteration run was still going after 5 hours, when I stopped it. All 12 worker threads were in this encoder, called from the sort-preserving merge of the aggregation's spilled runs.

Describe the solution you'd like

When a dictionary has more values than the array has rows, encode only the values its keys reference, as #7996 does for lists. For example, collect the referenced keys, encode just those values, and map each key to its encoded row. Arrays whose dictionary isn't larger than their row count keep the current path.

Describe alternatives you've considered

• Compacting in each caller (take(values, keys) with keys 0..len) before converting. This works, but every caller that converts slices of dictionaries would have to do it.
• garbage_collect_dictionary first. It still allocates a values.len() remap for each batch (arrow-select/src/dictionary.rs:59).

Additional context

• concat of dictionary slices does work proportional to each input's whole dictionary, not its rows #11885 is the same kind of problem in concat_dictionaries.
• Converting a dictionary is slower than converting the equivalent Utf8 even when the dictionary has no extra values (about 2x at 8,192 values above). This issue is only about the part that grows with the dictionary size.

I'm not planning to fix this myself, so anyone is welcome to pick it up. Feel free to ping me with questions about the cause or when you'd like a review.