Fan-Out Is the Floor: Two Targeted Optimizations for Query Shapes Nobody Controls
Also in Chinese: 中文版
Where this sits
The analytics platform was rebuilt in three layers, and they solve different kinds of problem.
Migration replaced the substrate: off ClickHouse onto Apache Doris with compute-storage separation. Not because Doris was faster — ClickHouse still was — but because the precondition for ClickHouse’s speed had expired. Its performance depends on sorting keys and modelling the query shapes in advance, and once a growing share of queries arrives from AI agents, nobody knows the shapes in advance.
The elastic compute tier built a control plane on top: evaluate every query’s cost before execution, route it heavy or light at plan time, give the heavy tail a physically isolated compute group, and provision that group on demand from a floor of zero. That layer makes no query faster. It decides where each query lands and makes that place exist when needed.
This article is the third layer, and it is the only one that makes a query genuinely cheaper.
Why the optimization has to be targeted rather than general
Query text, workload mix, and query shape are all inputs we do not control, and a growing share of them is generated by product features and AI agents rather than written by hand and reviewed. “Tell the callers to write better SQL” is not a weak remedy here — it is architecturally unavailable.
So the only remaining lever is to make the shapes a caller might plausibly write structurally cheaper. But that cannot mean optimizing everything, so the selection rule matters:
A query shape is worth attacking when its cost is mostly avoidable fan-out. Attack waste, never necessary cost.
Two shapes qualified, and they are worth stating side by side, because they fail differently:
| Shape | What is waste | What is not |
|---|---|---|
| Point lookup on a non-bucketed column | opening 566 tablets when one holds the row | the one tablet that holds it |
| Wide read of hundreds of columns | reading its column files serially | the column files themselves |
This is why the second one is not “optimizing SELECT *”. A wide read does not read fewer columns afterwards. It stops reading them one at a time.
The measurement that reset the cost model
The first instinct was to reduce rows scanned, so bloom filters went in. Rows scanned fell 73% and wall-clock time did not move — because one user’s rows lived in 2,044 files. That result did not optimize anything; it invalidated the cost model. The currency is not rows scanned, it is how many independent IO units must be opened and whether they can overlap.
The mirror image arrived from the other direction: a 3,727-column SELECT * measured 253 s cold, while the planner’s byte-volume estimate (rows × average row bytes) classified it light. One overestimate and one underestimate, one root cause — the estimator was structurally blind to fan-out.
Fan-out also has two independent axes, which is why there are two optimizations and not one: the tablet axis can be pruned, the column axis cannot, but it can be overlapped.
Axis one: prune the tablet fan-out
Every index Doris ships — zonemap, bloom filter, inverted, short-key, primary-key — is local to a tablet. Each can only answer “is this value inside the tablet you already opened”. When a table is not bucketed by the column being filtered, a point query has no choice but to open every tablet. The missing structure is a global one that answers a different question, earlier: which tablets could contain X, before any tablet is opened.
The shape it took, on the fork: one bloom filter per rowset per indexed column, written as a self-validating side file next to the rowset’s segments in object storage. At plan time the frontend extracts =/IN predicates, encodes probe values byte-identically to the write path, and asks each involved backend — one RPC per backend, not one per tablet — to test its tablets’ blooms. Tablets whose blooms all miss are dropped, and the candidate set is always intersected with the planner’s own tablet set. A 64-tablet demo table prunes to 1 tablet (probes=1, degraded=0); a non-existent value prunes to 0.
What it costs, stated honestly
| Cost | Size |
|---|---|
| Write path | one extra hash and insert per value, consuming the column-value stream the writer already produces — no extra IO |
| Storage | planning target 10–20 GB for the table, not the 6–8 GB first estimated, because the bloom’s bit count rounds up to a power of two; measured 3–4 GiB per backend at four backends |
| Plan-time latency | one extra network round trip inside a point-query budget, capped at 500 ms; a cold backend can pay a real object-storage stall before degrading |
| Cache budget | the index blocks live in the file cache’s INDEX queue, which is carved out of the queue that holds column data — see the next section |
| Compaction coupling | normal compaction rebuilds blooms for free through the shared writer path, but the ordered-rowset fast path links segments without invoking the column writer, so its output silently has no index at all until that is handled explicitly |
| Lifecycle | the new side files are not in the garbage collector’s enumeration path, so without a fix they leak in object storage forever |
Why a false positive is acceptable and a miss never is
A false positive costs one extra scanned tablet: bounded, and budgeted by a knob. A false negative drops a tablet that actually holds data: unbounded, and it returns wrong results. Those two costs are not the same shape, so the design only ever leans one way — over-scan, never skip. Every failure path degrades to “this tablet must be scanned”: descriptor missing, bloom unreadable, not yet backfilled, RPC timeout, or a backend whose local view lags the query’s pinned snapshot version. The response field is deliberately named candidate rather than pruned, so any omission on the backend side naturally shows up as “returned too many candidates” (safe) instead of “pruned too many” (wrong).
Exactly one semantic path can break that guarantee: byte-encoding mismatch. If a predicate literal, after the planner’s type coercion, encodes differently from the stored value, the probe misses a value that is really there. So any encoding uncertainty means do not prune: implicit casts, CHAR(n) padding, datetime scale, narrowing conversions, and DECIMAL outright, whose scale differences would break byte-identity. The cost is real — those columns never benefit, and affected queries lose pruning silently, without an error. It is the correct trade: losing a benefit is recoverable, encoding wrong once is not.
The structural cost nobody notices at demo time
The false-positive budget must be defined per tablet, not per bloom, because a tablet accumulates one bloom per rowset: with B blooms of per-bloom rate p, the tablet’s union false-positive rate is B·p. And B is not a configuration knob — it is a function of how well compaction is keeping up.
| B (blooms per tablet) | tablet-level false positive | expected mis-scanned tablets |
|---|---|---|
| 1 | 1% | ~6 / 566 |
| 5 | 5% | ~28 / 566 |
| 50 | ~40% (union bound saturated) | ~226 / 566 — pruning has effectively stopped working |
So this index couples query performance to compaction health. That is the long-term liability it introduces, and it is worth saying out loud, because it is invisible in a demo and obvious in a quarter.
The tradeoff that actually cost us: one cache, two working sets
The index blocks and the column data compete for the same local file cache, and the competition is zero-sum by construction: the INDEX queue is sized as a percentage of capacity, and the queue holding column data is the remainder. Every point given to one is taken from the other — and the other is exactly what the 43-second wide read needs.
| index side | column-data side | |
|---|---|---|
| Working set | 3–4 GiB per backend; worst case ~14 GiB when one backend owns everything | 441 one-mebibyte blocks per wide read |
| Value when resident | probe costs 0.03 s, and pruning a tablet avoids all of that tablet’s downstream IO | each block saves one ~97 ms round trip |
| Cost when missing | ~100 ms per file, spent inside a hard 500 ms plan-time budget; blow the budget and the entire prune is abandoned, back to full fan-out | linearly slower |
| Failure curve | a cliff | a slope |
The conclusion follows from the shapes, not from preference: size the index queue for the worst-case single-backend working set, not the average — leave headroom on the cliff and let the slope absorb the squeeze. Both sides have to be computed before that number is set. Undersize it and a one-time cold-read cost becomes a permanent, non-converging object-storage bill.
The incident: “warmed once” is not “still resident”
A daemon warms each backend’s index files into cache on startup, so cold probes hit local disk instead of object storage. It kept a map of backend id to last start time and skipped any backend whose start time had not changed. That invariant is the wrong one.
What happened, over three days on preprod: four backends were ingesting; the cluster was scaled down to a single backend for a single-node performance test, and the daemon warmed it once — the only warm-up it ever received; two days of ingest and compaction produced new index files that no past warm-up covered; the cluster was scaled back to four backends, and the three re-added ones registered with fresh backend ids. Then a manual check:
| Backend | cached | missing | wrong queue |
|---|---|---|---|
| the three re-added | 386–405 each | — | — |
| the one that never restarted | 0 | 313 | 73 |
The three new backends looked new to the daemon, so they were warmed normally. The original backend kept its id and never restarted, so it was skipped on every round from that single warm-up onward, and every probe against it became a cold object-storage read.
The amplifier was capacity. The cache-sizing step had been skipped, leaving the INDEX queue at its 5% default; during the single-backend window that one backend owned every tablet’s blooms — roughly four times its normal footprint, which on its own exceeds any plausible 5% budget. It evicted its own warm-up.
Three things have to be said together about this:
- Correctness was never affected. Every failure path degraded to over-scan. This was a performance and cost problem, exactly as designed.
- It was silent. No alert, no error, no degraded flag — found by running a check script by hand. Residency was never an observable quantity.
- The fix order was itself the judgement. Size the cache first, before shipping any repair mechanism. Otherwise a mechanism that re-reads whatever is missing will re-read it on every round for as long as the budget is smaller than the working set — turning a one-time cost into a permanent one.
The repair replaced “process warmed once” with a residency-driven loop, governed by one hard constraint: a sweep over a fully resident backend must issue zero object-storage GETs. That single requirement eliminates most of the obvious designs.
| Rejected repair | Why |
|---|---|
| Re-warm every backend every round | violates the zero-GET rule; permanent bandwidth waste |
| Let each backend sweep itself, no coordinator | a backend does not know its own consistent-hash ownership, so it would miss tablets moved onto it by a membership change — the incident’s own scenario |
| Check residency over HTTP from the coordinator | hundreds of calls per sweep, and it duplicates cache-internal logic |
| Pin the index blocks so they can never be evicted | it removes them from budget accounting, which makes sizing mistakes invisible |
| Ship the queue resize only, treat it as a config incident | correct as phase one, insufficient alone |
And the counters — checked, resident, repaired, failed — shipped as metrics, because the first lesson of the incident is that residency has to be observable without a human remembering to look.
The bug that made the whole feature a no-op
Before any of that, plan-time pruning silently never fired for ordinary queries. Two independent causes, both in the planner:
The predicate map it read was always empty at that point, because the only code path that populates it is a short-circuit point-query path that normal queries do not take. And “just populate it during node initialization” would not have worked either: the translator recurses into the child — creating and initializing the scan node — before attaching predicates to it. That is not an accident, it is the inherent order of a top-down visit that constructs fragments bottom-up. At initialization time the predicate list is empty by definition.
Worse, the naive fix would have been unsafe. Pruning during initialization pins its own snapshot version, taken earlier than the version the query actually reads, which is pinned later and uniformly. Versions are monotonic, so the pruned rowset set could be a strict subset of what the query reads — violating the superset invariant and silently returning fewer rows. A performance bug repaired into a correctness bug.
The fix moved pruning to the point where the query’s real read version is computed, after the whole plan is translated. It incidentally collapsed per-partition metadata calls to zero extra and backend probe RPCs from one-per-partition-per-backend to one-per-backend.
The class matters more than the fix: a silent, ordering-dependent no-op. It compiled, ran, never threw, never logged, and never triggered. Its unit test passed because it bypassed the entire call chain and tested the encoding logic in isolation. Verification has to cover call order, not only function correctness.
Axis two: overlap the column fan-out
For a wide read, the column files are not waste — they are the work. The waste is doing them one at a time.
The measured baseline for one such read: 43.45 s wall clock, 98.6% of it remote IO, 441 object-storage fetches. The obvious story is “one serial read per column”. Reading the code showed it is up to three, in a strict order:
| # | Read | Offset comes from |
|---|---|---|
| 1 | ordinal index page | footer-resident — free |
| 2 | data page | the ordinal index — depends on read 1 |
| 3 | dictionary page (dictionary-encoded columns, 56% of this table) | footer-resident — free |
Only read 2 has a data dependency. Reads 1 and 3 have byte offsets that are already in memory once the segment footer is parsed, so both can be issued immediately, in parallel, with no prior IO. The entire design is a consequence of that one fact.
So: before the serial loop runs, warm every column’s meta pages concurrently, and open each column’s cache window ahead of the read that would have opened it later — deduplicated into cache-block-aligned ranges and submitted to the prefetch thread pool that already exists. The serial loop itself stays byte-for-byte unchanged; by the time it reaches column N, column N’s block is local.
Correctness is argued structurally rather than by testing: the prefetch tasks are dry-run reads into a null destination, they decode nothing and mutate no reader state; they run on a pool that only ever performs that one operation; a full queue or any other failure degrades to “the loop reads it synchronously, as today”; and a block already being fetched is in a downloading state that a concurrent reader waits on, so no duplicate object-storage GET is ever issued. Remove the two warming steps and results must be bit-identical.
What it costs, and where it stops working
The prefetch pool is backend-wide and its width is the throttle: with the default worker count and ~441 blocks per query, concurrent wide lookups queue rather than multiply object-storage concurrency. That is deliberate. The feature is off by default and gated on column count and row count so that scan-shaped queries are untouched. Two costs were accepted: a single master switch means the two warming steps cannot be rolled back independently, and the change touches code that compaction also runs through.
One clarification that matters more than it looks: the warming step is not free of IO. The index loads that follow it still block — they just block on a warm cache. What it removes is the serialization, not the reads.
And it stops working on array columns (2.5% of this table, deliberately skipped): an array’s item pages cannot be prefetched at all, because which item pages are needed depends on offsets that have not been read yet. Coverage is 97.5% of columns, measured from the table’s own schema, and the array pages are largely warmed incidentally anyway because they sit between scalar columns that are.
Verification status, stated plainly
This layer’s benefit is projected, not measured. The design and implementation are complete; the numbers are arithmetic over the measured baseline. Two things stand between it and a claim:
The baseline itself has to be rebuilt. It was measured on a build that predates the prefetch subsystem this design extends, so it must be re-measured on the target release with the switches off — and that run may well be slower, because it will expose thousands of serial index loads at iterator-init time, which is itself one of the things this change removes.
And the acceptance metric I originally wanted cannot be computed under this design. Prefetch reads are dry-run reads with the query id and cache statistics nulled out, so the profile’s remote-IO counters fall toward zero while wall clock falls: the IO did not disappear, it stopped being attributed. I cannot prove “we parallelized rather than moved the work” by showing per-fetch latency unchanged. The substitute checks are counting submitted prefetch tasks, and comparing total bytes written into the cache before and after, which should stay flat.
Killing my own first implementation
The first version of the prefetch was a self-contained path: its own stage enum, its own iterator method, its own block-folding arithmetic, its own configuration flags, sitting in front of the read loop. It was committed. Two days later I folded the whole thing into the prefetch subsystem that already existed and deleted it.
The reasons were mechanical, not aesthetic. The two overlapped on data pages, and each had derived the same block arithmetic separately — two submitters against one cache, with two caps that did not know about each other and no shared deduplication. And the genuinely new idea, warming meta pages, belonged somewhere the standalone version could not reach: ahead of the existing subsystem’s own per-column index loads, which is also where it fixes that subsystem’s worst behaviour on a wide table.
The part worth keeping is the condition that flips the answer, recorded at the time: had this needed to ship on the currently deployed release, where the subsystem does not exist and reuse would mean backporting more than a thousand lines across dozens of files, keeping both paths would have been correct. The answer depends entirely on one fact about the target release — which is the kind of thing that should be written down rather than remembered.
Alternatives rejected
| Alternative | Why not |
|---|---|
| An exact inverted map from value to tablet, in the shared metadata store | hundreds of gigabytes and billions of writes against a store other systems depend on. Zero false positives is genuinely better; kept as a swappable backend if precision ever becomes a hard requirement |
| A routing table plus a tablet hint | the strongest pragmatic alternative, and worth measuring as a baseline. Not chosen because tablet ids are not stable across split, merge, schema change, or restore, and it needs dual writes with a freshness window |
| One mutable per-tablet aggregate bloom instead of one per rowset | reintroduces the only question in this design that can produce wrong answers: has the aggregate already absorbed the just-committed rowset? Per-rowset decomposition eliminates it by construction |
| A transposed or bit-sliced global structure | same freshness problem; flagged as the preferred evolution if tablet counts reach tens of thousands |
| An in-memory prune directory in the frontend | the exact structure does not fit the heap, so it forces a new stateful service to save one cache read |
| Re-bucketing the table by the filtered column | destroys the locality every existing per-user query depends on, and requires rewriting billions of rows. Not changing the physical layout was a stated constraint |
| Writing the bloom into the segment footer | reading it would require opening the segment file, which defeats the entire point of pruning before data files are opened |
Takeaways
- Rows scanned is a proxy. Fan-out is the floor. A 73% reduction in rows with zero wall-clock change is the cheapest possible proof that you are optimizing the wrong quantity.
- When you cannot constrain the input, the only lever left is making the shapes cheap — but then the selection rule has to be explicit, or “optimization” becomes an infinite backlog. Attack avoidable fan-out; leave necessary cost alone.
- Compare the shape of costs, not their size. Bounded costs can be managed with a knob; unbounded ones have to be eliminated architecturally. That is why over-scan is acceptable and skipping is not, and why the cache budget goes to the failure curve that is a cliff rather than the one that is a slope.
- An index that depends on background maintenance inherits that maintenance’s health. Pruning power decays as blooms accumulate per tablet, so this feature’s value is a function of compaction keeping up.
- The characteristic failure of a cache-residency design is silence. “Warmed once” says nothing about “still resident”, and nothing will page you. Make residency a counter before you make it a mechanism.