Add physical plan optimization to decide dictionary encoding

[Rich-T-kid]

Is your feature request related to a problem or challenge?

Even with good stats available, there's no optimizer rule that uses them to decide whether a string column should be dictionary-encoded.

Describe the solution you'd like

Add a physical plan optimization pass that looks at column statistics (NDV, aggregation heaviness) as well as user-defined flags/thresholds, and decides whether to encode a given string column as a dictionary array.

Will need extensive testing and benchmarking to decide whether this should be enabled by default.

Describe alternatives you've considered

Final piece of : #22891

part of epic : #24111

Additional context

No response

[Rich-T-kid]

Something that popped into mind is looking at any filters in the physical plan to determine if its worth converting to a dictionary a well. for example a filter that retains a 'small' set of values that is later aggregated may benefit from dict<k,v>

[Rich-T-kid]

constant filters to be exact ^

[Rich-T-kid]

closing #22891 in favor of this issue

[asolimando]

Something that popped into mind is looking at any filters in the physical plan to determine if its worth converting to a dictionary a well. for example a filter that retains a 'small' set of values that is later aggregated may benefit from dict<k,v>

As discussed offline, I agree it's important but I think relying on existing statistics propagation around distinct_count/NDV will be enough to that effect, you'd just need to "consume" the statistics propagated bottom-up in the subplan.

[yinli-systems]

I'd like to pick this up as a staged implementation, provided nobody already has parallel code for #24117.

I traced the current plan boundary and the related work (#24227, #24113, #24822, #25185). Two constraints seem important:

1. enable_rle_to_dictionary in introduce optional rle reads from parquet #24227 affects inferred table / scan schema before physical optimization, while the aggregate shape and bottom-up NDV statistics are only fully visible in the physical plan. A physical rule therefore cannot safely just flip that global flag; selective native reads need an internal per-scan/per-column decision channel (or an earlier planning hook) and downstream schema handling.
2. NDV alone is not a safe cost model. Dictionary-encoded group keys pay 27-37% aggregation tax vs Utf8View in TPC-H (Q2/Q19/Q20), DF 54 & 55 both #24822 shows that dictionary grouping can still lose for both low-cardinality single keys and high-cardinality long strings; perf: reuse cached dictionary value hashes in vectorized_append #25185 improves the repeated shared-values case, but does not make arbitrary runtime encoding free.

My proposed first reviewable slice is deliberately narrow:

• establish an end-to-end benchmark matrix for plain strings vs native/pre-encoded dictionaries vs runtime casts;
• cover low/high NDV, short/long values, batch size/count, reused vs changing dictionary values, and partial/final aggregation;
• verify identical results and separately measure conversion, aggregate, total time, and memory;
• use those results to define a conservative, opt-in policy for direct Utf8/Utf8View GROUP BY columns;
• missing or unsuitable statistics => no rewrite; uncertain benefit => no rewrite; no correctness or visible output-schema change; idempotent decision;
• stack the scan mechanism on introduce optional rle reads from parquet #24227 and avoid duplicating the user threshold work already owned in Allow users to define the cardinality threshold for dictionary reads #24113.

For the implementation path, I plan to prototype a private column-selection mechanism on the Parquet scan rather than expose a new public allowlist immediately. The optimizer/planning integration would only select columns whose physical encoding and statistics make native dictionary materialization viable; it would not insert an unproven generic per-batch cast.

@Rich-T-kid @kumarUjjawal does that boundary match the direction you had in mind, and is anyone already implementing #24117? If not, please assign it to me and I will post the benchmark evidence and a concrete plan diff before expanding the surface area.

[yinli-systems]

Benchmark evidence update from the first local prototype (not a proposed universal threshold yet).

Method

• DataFusion c149764, release/LTO build, Apple M5 Max, Rust 1.97
• GROUP BY key, COUNT(*), 10M identical logical rows, 8,192-row input batches
• median of iterations 2-6 after two warmups
• timer includes planning + execution; input construction is excluded
• dictionary cases cast the final group key back to Utf8, so the comparison includes preserving the user-visible output type
• compared plain Utf8, pre-encoded dictionaries with a shared values array, pre-encoded dictionaries with independent per-batch values arrays, and a runtime arrow_cast from Utf8

Single target partition results (milliseconds; lower is better):

value bytes	NDV	plain	dict/shared	dict/per-batch	runtime cast
8	10	48.87	35.43 (-27.5%)	35.49 (-27.4%)	94.82 (+94.0%)
8	100	47.85	38.87 (-18.8%)	37.95 (-20.7%)	98.37 (+105.6%)
8	1,000	42.66	44.88 (+5.2%)	45.33 (+6.3%)	111.74 (+161.9%)
64	10	59.09	35.64 (-39.7%)	36.44 (-38.3%)	114.25 (+93.4%)
64	100	57.09	37.63 (-34.1%)	38.41 (-32.7%)	119.16 (+108.7%)
64	1,000	59.36	49.04 (-17.4%)	48.17 (-18.9%)	132.70 (+123.6%)
64	10,000	64.36	115.09 (+78.8%)	118.20 (+83.6%)	401.75 (+524.2%)

A 4-target-partition check at NDV=100 kept the result: dictionary was ~19.5% faster for 8-byte values and ~36.8% faster for 64-byte values. At 1M rows / NDV=100 / 8-byte values, it remained ~21.6% faster. Shared vs per-batch values did not materially change these low-NDV cases; I still expect reuse to matter as dictionary cardinality grows, consistent with #25185.

What this rules out

• A generic physical-plan Utf8 -> Dictionary runtime cast is not viable here, even at NDV=10. It loses by roughly 2x and gets much worse as NDV grows.
• NDV ratio alone is not enough. At the same 1,000 / 10M ratio, short strings lose while 64-byte strings win.
• Absolute NDV alone is not enough either; row count and value width affect the crossover.

This supports a scan-native-only first policy: use dictionary materialization only when the source can supply it without a generic runtime build; fail closed when NDV/row count/byte-width evidence is missing; preserve the original query output type; and keep the rule opt-in until it has cross-machine / Parquet results.

I now have a single-file benchmark change that adds explicit input encoding, absolute NDV, value length, input batch size, shared/per-batch values, and output normalization while preserving the existing default dictionary benchmark behavior. Its four focused tests and default-feature clippy pass locally. If this benchmark slice is useful as a prerequisite PR, I can submit it independently while we settle the selective scan/planning hook for #24117.

[yinli-systems]

Implementation update: I now have a working, fail-closed end-to-end prototype stacked on #24227:

yinli-systems@2bdf91d9f

Key points:

• Adds an opt-in physical optimizer rule for direct Utf8View GROUP BY columns.
• Carries per-file physical facts from the Parquet footer through listing metadata and the metadata cache.
• Selectively asks the Parquet source to emit Dictionary<Int32, Utf8> only for profitable group keys; it does not insert a runtime dictionary cast.
• Requires exact row count, null count, and NDV, every data page in every input file to be dictionary encoded, average decoded value width >= 16 bytes, and at least 10,000 non-null rows per distinct value.
• Missing or ambiguous evidence, filters, limits, file ranges, complex scan projections, grouping expressions, or aggregate references to a retyped key all leave the plan unchanged.
• Restores the original aggregate output schema and field metadata above the aggregate, and the rewrite is idempotent.

The conservative gates come from the expanded benchmark in #25303, now including Utf8View. With 10M rows, dictionaries still win at NDV 1,000 for 16-byte and 64-byte values, but lose clearly by NDV 10,000; short inline Utf8View values are also a poor target. I updated the PR description with those results.

Validation completed locally:

• Real Parquet end-to-end test proves the footer facts, verifies the scan actually changes from Utf8View to native Dictionary<Int32, Utf8>, compares results byte-for-byte with the disabled baseline, and repeats planning through the metadata-cache path.
• datafusion-physical-optimizer: 41 passed.
• datafusion-datasource-parquet: 265 passed.
• datafusion-common: 610 passed.
• datafusion-datasource: 176 passed.
• datafusion-execution: 113 passed.
• datafusion-catalog-listing: 19 passed.
• targeted information_schema sqllogictests: 4 files passed.
• affected packages pass all-target no-deps clippy with -D warnings; Rust formatting, diff checks, and Prettier 2.7.1 checks pass.

I have pushed the implementation branch but deliberately have not opened it against main yet: #24227 is still unmerged, and opening now would mix that contributor's dependency commits into this PR. Once #24227 lands, I will rebase this single feature commit onto current main, regenerate the config docs there, and open the clean implementation PR.

[Rich-T-kid]

Taking a glance, and I agree with your analysis for the most part, @yinli-systems. I think #25185 should be a pretty big factor in the performance boost we expect to gain from dictionary encoding. The narrowed goal of this epic is to support CBDE for parquet files, which gives us the privilege of working with the arrow-rs parquet reader. From my understanding, one row group maps to a dictionary values array, in aggregations, we can move work away from O(row_count) to O(row_group_set), which would be huge. with the linked PR we would run computation once and then cache the results even if we recieve thousands of .intern() calls.

I took a brief look some time ago, but it may be worth looking at other operators/kernels to see if there are any dictionary-specific optimizations that can be applied to give us more leeway to decide between utf8 and dict<_,utf8>.

once #24227 is shipped it will also be easier to benchmark what changes are moving the needle in terms of performance.