adriangb opened a new pull request, #10780:
URL: https://github.com/apache/arrow-rs/pull/10780
# Which issue does this PR close?
- Part of #9699.
- Related to #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.
```rust
#[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
#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
#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 #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 #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):
```python
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
#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](https://claude.com/claude-code)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
