feat(parquet): DictionaryFallback::Adaptive — measured, checkpointed dictionary fallback decision

[adriangb]

Which issue does this PR close?

• Part of Parquet dictionary fallback heuristics #9699.
• Related to [Parquet] Allow sampling some values to decide the encoding pattern #8378: implements measured sampling for the dictionary-fallback decision specifically.

Important

Stacked PR (draft): this builds on #10775 (DictionaryFallback policy) and #10777 (re-encode on fallback) and includes both as its base. Please review only the top commits (the merge of the two branches plus one feature commit); this PR will be rebased as those merge.

Rationale for this change

#10775 added an opt-in DictionaryFallback::WhenProfitable { worth_ratio, .. } policy that keeps a dictionary past dictionary_page_size_limit while the dictionary page stays below worth_ratio × the PLAIN-encoded size of the appended values. That PR disclosed two structural weaknesses of the ratio heuristic, both reproduced in its benchmarks:

1. The ratio has to be tuned timidly. PLAIN size is a pessimistic bound for the fallback encoding, so a "profitable" dictionary can still lose to the real fallback — most visibly on sorted/high-cardinality keys where DELTA_BINARY_PACKED (or PLAIN + compression) crushes a dictionary of distinct keys. At worth_ratio: 0.5 ClickBench gained 8.5% overall but paid +13.2 MB on UserID and +8.3 MB on FUniqID, and TPC-H regressed +2.1% on l_orderkey/ps_partkey; the suggested default had to be a conservative 0.1, which captures only a fraction of the available wins.
2. The decision is blind to shuffled data. The decision is made when the dictionary crosses the grace floor, after only ~floor / value_size values — e.g. 64 samples of 16 KiB values — before repeats are even visible. On the disclosed uniformly-shuffled reproducer every policy behaved exactly like stock.

This PR adds the measured variant that #10775's API documentation anticipated: no ratio to tune, decisions based on measuring the actual encodings (including compression), and deferral with a trend signal instead of a one-shot decision.

#[non_exhaustive]
pub enum DictionaryFallback {
    OnPageSizeLimit,                                                  // default, unchanged
    WhenProfitable { worth_ratio: f64, max_dictionary_page_size: usize },
    /// Keep the dictionary past the grace floor while a *measured*
    /// comparison shows it beating the fallback encoding, re-evaluated at
    /// geometric checkpoints; fall back unconditionally at the hard cap.
    Adaptive { max_dictionary_page_size: usize },
}

How the decision works

Whenever the dictionary page size crosses a checkpoint — the grace floor (dictionary_page_size_limit), then 2×, 4×, … the floor, up to the hard cap — the writer compares:

• dictionary cost: the dictionary page, scaled by the measured compressibility of a sampled prefix, plus an upper-bound estimate of the RLE/bit-packed indices for every value appended so far;
• fallback cost: a bounded sample of the values buffered for the in-progress page, resolved through the in-memory dictionary (the machinery from fix(parquet): re-encode buffered values on dictionary fallback when the dictionary would be unreferenced #10777) and re-encoded through a scratch fallback encoder, compressed with the chunk's codec, and scaled to the PLAIN bytes appended so far.

Compressing both sides with the chunk's codec is what fixes the sorted-key class: a dictionary of distinct keys barely compresses, while the PLAIN/delta encoding of sorted-with-repeats keys compresses extremely well, so the dictionary correctly loses even when it is nominally 2–4× smaller before compression. The two compression samples use matched input sizes, since codec efficiency varies strongly with input size.

The decision is not final at the first crossing:

• A column whose repeat fraction (share of appended PLAIN bytes that were already in the dictionary — measured exactly, at zero cost, from two counters feat(parquet): configurable DictionaryFallback policy for dictionary encoding fallback #10775 already maintains) is below 1% always falls back: the repeat fraction bounds what a dictionary can possibly win, so below it a measured win is sampling noise. In particular, a column with no repeats at all produces output identical to stock.
• A measured near-tie at the first checkpoint defers to the next checkpoint: the floor is typically crossed after too few values to judge a shuffled distribution.
• At later checkpoints, a losing-but-close dictionary keeps deferring while the windowed hit rate is rising (repeats are arriving) or at break-even; otherwise it falls back. Thanks to fix(parquet): re-encode buffered values on dictionary fallback when the dictionary would be unreferenced #10777 the buffered values re-encode cleanly, and the dictionary page is only written if already-flushed pages reference it.

Bounded cost: each checkpoint re-encodes and compresses at most 1 MiB (PLAIN size) of sampled values plus a ≤1 MiB dictionary prefix — about one extra data-page encode — and the geometric spacing bounds the number of checkpoints per column chunk by log2(cap / floor) + 1 (7 for the 1 MiB floor / 64 MiB cap defaults). Between checkpoints the policy does no work. There is no continuously-running parallel encoder.

Benchmarks

Methodology as in #10775: every file rewritten with ArrowWriter, ZSTD level 1, defaults otherwise, only the fallback policy differing; totals are summed output bytes. "Stock" is the default policy on this branch (i.e. including #10777's fallback improvements, which is why totals differ slightly from the numbers in #10775). WhenProfitable arms use the cap 64 MiB; Adaptive { max_dictionary_page_size: 64 MiB }.

Dataset	Stock	WhenProfitable 0.1	WhenProfitable 0.5	Adaptive
Repetitive 16 KiB blobs, short runs (seed 42)	51,015,911	−38.9%	−38.9%	−38.9%
Same pool, uniformly shuffled	794,807,940	±0	±0	−46.9%
Same pool, sorted	61,447,608	−19.9%	−19.9%	−19.9%
ClickBench (hits_0..29, ~4.1 GB)	2,666,195,053	−0.35%	−8.69%	−12.09%
TPC-H SF1	242,833,808	—	+2.12%	+0.01%
TPC-DS SF1	301,005,084	—	—	+0.01%
large_values (unique 16 KiB values)	803,839,649	—	—	byte-identical

• Shuffled reproducer (the case disclosed as a miss in feat(parquet): configurable DictionaryFallback policy for dictionary encoding fallback #10775): 2 of the 4 seeded files flip from 198.7 MB to 12.3 MB each — the dictionary is deferred at the floor, wins once repeats accumulate, and the 1000-value pool exhausts at a 16.4 MB dictionary, far below the cap. The other 2 files happen to cross the grace floor with zero repeats among the first ~64 samples and fall back exactly like stock — that is the deliberate price of keeping unique-value columns byte-identical to stock, and it is a missed win, never a regression.
• ClickBench: total −322.4 MB (−12.09%), beating both ratio arms while also removing their regression class. Top wins Title −183.9 MB, URL −92.2 MB, Referer −54.4 MB; the WhenProfitable{0.5} regressions UserID +13.0 MB → +1.1 MB and FUniqID +8.2 MB → +2.4 MB. Aggregate: −334.0 MB of wins vs +11.7 MB of regressions (largest single: RefererHash +2.9 MB).
• TPC-H / TPC-DS: +0.01% each (no per-column mover above 0.15 MB) — the sorted-key columns (l_orderkey, ps_partkey) that cost the ratio heuristic +2.1% now measure as losing post-compression and fall back exactly like stock.
• Wall time: ClickBench 53.6 s (stock) → 58.3 s (+8.8%, the measurement bound in action); TPC-H 4.0 s → 3.8 s; the shuffled reproducer is faster than stock (1.7 s → 1.6 s) since far fewer bytes get compressed.

Generator for the shuffled reproducer (pyarrow, seed 42; the runs/sorted variants from #10775 differ only in the idx line):

import numpy as np, pyarrow as pa, pyarrow.parquet as pq, os, string
OUT = "repetitive_large_values"
os.makedirs(OUT, exist_ok=True)
rng = np.random.default_rng(42)
alphabet = np.frombuffer(bytes(string.ascii_letters + string.digits, "ascii"), dtype=np.uint8)
pool = ["".join(map(chr, rng.choice(alphabet, size=16384))) for _ in range(1000)]
for f in range(4):
    idx = rng.integers(0, 1000, size=16384)          # runs: np.repeat(rng.integers(0, 1000, 1024), 16); sorted: np.sort(...)
    t = pa.table({
        "id": pa.array(np.arange(f * 16384, (f + 1) * 16384), pa.int64()),
        "key": pa.array(idx, pa.int64()),
        "val": pa.array([pool[i] for i in idx], pa.string()),
    })
    pq.write_table(t, f"{OUT}/part-{f}.parquet", compression="zstd")

What changes are included in this PR?

• DictionaryFallback::Adaptive { max_dictionary_page_size } in file/properties.rs (the enum is #[non_exhaustive]; plumbing was added by feat(parquet): configurable DictionaryFallback policy for dictionary encoding fallback #10775).
• A defaulted crate-private ColumnValueEncoder::sample_dictionary_cost, implemented by both dictionary encoders (generic typed and arrow byte-array): resolves a bounded sample of the buffered ids through the dictionary, measures it through a scratch fallback encoder, and returns the dictionary/indices/PLAIN byte counts plus a dictionary-prefix compressibility sample. Encoders without measurement support keep the absolute-limit behavior.
• The checkpoint state machine and decision procedure in GenericColumnWriter (adaptive_should_fallback), with the thresholds documented as implementation details.
• Tests: sorted keys with repeats fall back byte-identically to stock while WhenProfitable{0.5} keeps the dictionary and produces a 2×+ larger file; repetitive large values keep the dictionary (no fallback pages, less than half the stock size); a uniformly-shuffled bounded pool keeps the dictionary where stock falls back (deterministic LCG sampling, less than half the stock size); unique values are byte-identical to stock; a trickle-of-repeats column defers exactly one checkpoint and then falls back with a chunk ≈ stock; the hard cap forces fallback on a clearly-winning dictionary; the non-byte-array encoder path; property plumbing incl. into_builder() roundtrip; roundtrip equality in every scenario.

Are these changes tested?

Yes, as above; cargo test -p parquet --all-features passes (1791 tests), fmt and clippy clean.

Are there any user-facing changes?

New opt-in DictionaryFallback::Adaptive variant. Default behavior is unchanged (OnPageSizeLimit remains the default; with the policy enabled, columns with no repeated values still produce byte-identical output to the default).

🤖 Generated with Claude Code

[adriangb]

Some read-side evidence for this policy, measured on the branch as pushed (all public datasets; same methodology as the PR body, ZSTD-1, identical writer settings, only the fallback policy differing between file sets).

ClickBench query A/B (all 100 hits_partitioned files)

Both arms are rewrites of the standard dataset — one with this branch's default policy, one with Adaptive { max_dictionary_page_size: 64 MiB } — queried by the same DataFusion binary (built from datafusion main @ c429919c7c), 43 ClickBench queries, 15 interleaved iterations per query per arm (5 rounds × 3, stock/adaptive/original alternating within each round to cancel load drift), medians.

	stock rewrite	adaptive rewrite	Δ
Total bytes	9,982,078,788	9,055,616,486	−9.28%
Query geomean (43 queries)	153.4 ms	145.5 ms	−5.19%

As a harness sanity check, the untouched original dataset's geomean lands within ~2% of the stock rewrite.

Consistent movers (≥5% and same sign in ≥4 of 5 rounds), which line up exactly with the columns whose dictionary retention changed (Title −40% bytes, URL −17%, Referer −12%):

• Faster: Q22 −29.9% (Title LIKE), Q27 −17.5% (URL full scan), Q20 −17.3% (URL LIKE '%google%'), Q21 −17.3%, Q5 −16.6% (COUNT(DISTINCT SearchPhrase)), Q31 −12.0%, Q23 −10.3% (SELECT * with URL filter), Q30 −8.7%.
• Slower: Q38 +32.7% and Q39 +6.9% — highly selective filters that read URL/Referer for a few thousand matching rows, so the reader decompresses and materializes a much larger dictionary page to serve very few values. This is the policy's read-side trade stated plainly: full scans of retained-dictionary columns get faster, highly selective point-ish reads of them get slower. (Q32 also moved −13.9% but touches only byte-identical columns; we attribute it to environment residue, not the policy.)

Row counts were identical across arms for all 43 queries in every iteration; two apparent value diffs (Q32/Q38) reproduce between runs on the same files and are LIMIT-over-tied-counts order nondeterminism — order-insensitive aggregation checksums are exactly equal.

Decode microbenchmark (the repetitive-blob class)

Full-column scans (single leaf projected, single-threaded, release, medians of warmed runs) of the PR's runs-reproducer val column (16 KiB blobs, 1000-value pool):

Variant	Input bytes stock → adaptive	Scan speedup (adaptive)
ZSTD-1	50.7 MB → 31.0 MB	1.7–2.0×
Uncompressed	1,074 MB → 41.4 MB	2.4–2.8×

The uncompressed variant isolates the mechanism: the speedup grows with no decompression in the picture, so the win is dominated by having ~26× fewer input bytes to move and parse — the fallen-back file re-reads every 16 KiB repeat in full, the dictionary file expands from a hot ~10 MB dictionary. On general string columns (ClickBench Title/URL) the same microbenchmark shows roughly read-neutral results for a one-shot local full-column scan (the retained dictionary must be materialized per chunk, offsetting the 16–18% byte savings), consistent with the query-level picture above where parallel query execution does cash the bytes in.

Caveats: warm-cache medians on a busy 12-core laptop (interleaving cancels drift, but treat individual sub-10% per-query deltas as noise); local NVMe, so the −9.3% bytes understates the benefit for cold/object-store reads, while the Q38-style regression would likely shrink there (I/O-dominated).

🤖 Generated with Claude Code