Rebuilding a 5-billion-row event store, down to the engine
Also in Chinese: 中文版
The one-paragraph version
Our fraud-analytics event layer serves two increasingly conflicting workloads: millisecond point lookups on the serving path, and a growing stream of unpredictable ad-hoc analytics — increasingly generated by AI-agent-facing product features rather than human analysts. The old ClickHouse shared-nothing layer could neither isolate the second workload nor scale for it: one heavy query starves the shared pool, and we measured a SELECT * LIMIT 10 that waited 61 s of wall-clock on 60 ms of CPU, queued behind a bulk insert. What I did was not a migration. I rebuilt this layer on compute-storage separation — Apache Doris shared-data mode with Iceberg on S3, on Kubernetes — and built query routing into the Doris frontend itself, so the system classifies every query heavy/light in 6–8 ms before execution and drives a spot compute pool from zero to N and back. The move was certified 99.945% row-complete over ~5.2B rows / ~4 TiB / up to ~3,700 columns; point-lookup latency went from ~8 s to ~20 ms, proven flat across 19× data growth; and along the way I root-caused a chain of engine-level failures, from a Parquet footer explosion that OOMed every backend on a bare COUNT(*) to a compaction death spiral driven by a 46× compression collapse.
Why we left ClickHouse (it was still faster)
ClickHouse was winning flat-table scans by 1.2–2.5× on our own primary-source benchmarks when we decided to leave it. The migration case was structural, and it took discipline to keep it that way, because “the new engine is faster” is the story everyone wants to tell and the one that falls apart under cross-examination.
Four compromises drove the decision:
-
No workload isolation. One heavy query starves a shared node — the 61 s / 60 ms starvation case above was routine, not exceptional.
-
Rigid, always-on compute. Shared-nothing couples compute to local disk. Idle periods and heavy-query bursts pay for the same resident hardware. Genuinely stateless elastic compute exists in the ClickHouse ecosystem only as a closed-source managed service; that is an architecture gap, not a configuration gap.
-
Fragile wide-schema evolution. Per-tenant schemas evolve dynamically and land as thousands of sparse columns, maintained by brittle
ALTER ADD COLUMN. -
Sampling as a correctness tax. Analytical queries defaulted to a 1M-row sample because full scans were too expensive — trading correct answers for latency.
The load-bearing argument, and the one that survives even the newest ClickHouse releases, is hard memory isolation: Doris workload groups enforce cgroup-backed hard limits per query pool, where ClickHouse’s equivalents remain best-effort. When an AI-agent-facing product layer started generating unpredictable query shapes at unpredictable rates, isolation and elasticity stopped being nice-to-haves.
Target architecture
Doris 4.0.5 in shared-data mode: tablet data lives in an S3 storage vault, metadata in MetaService backed by FoundationDB, and backends are stateless compute with local EBS acting only as file cache. Compute is physically partitioned into compute groups — a resident serving pool on on-demand nodes, and an elastic heavy pool on spot instances that rests at zero replicas. Within each pool, workload groups govern CPU, memory, and concurrency. Iceberg on the same S3 serves as the open lake tier.
The design insight worth stating explicitly: compute-storage separation does not remove state. It concentrates state into one layer (MetaService, FoundationDB, the S3 recycler) — and that layer is precisely where you can no longer cut corners. Three new stateful failure domains replaced “disks attached to every node,” and the operational work below reflects that trade.
The migration was a memory-engineering problem
A ~3,700-column table breaks the naive “export to Parquet, bulk load” path in four places, none of them obvious, and adding hardware fixes none of them:
-
The footer bomb. A Parquet footer scales with
row_groups × columns. Our early objects carried a 2.12 GB footer; the reader OOMed all eight backends before touching a single row of data. The tell: even a bareCOUNT(*)died — proof that metadata parsing, not data scanning, was exploding. Fix: restructure the objects, footer down ~9×. -
The untracked scanner. Doris’s vectorized Parquet column reader allocates outside
exec_mem_limit. Memory scales withscanner_concurrency × columns; at defaults that was ~78 GB per backend. No query-level memory limit helps a buffer the engine does not track. -
The export buffer seesaw. Each export stream buffers
row_group_size × columns. Small slices do nothing for export memory; small row groups blow up the footer. Only cutting both — small objects and small row groups — decouples the seesaw. -
Load memory grows with the table, not the batch. At ~1.7 B rows loaded, identical load jobs began breaching the memory ceiling and livelocking: merge-on-write delete-bitmap maintenance and wide-table compaction scale with accumulated table size. A watchdog (soft-stop / hard-kill thresholds, atomic aborts, idempotent retries) held the line; dropping to single-stream concurrency resolved it with zero data loss — throughput was export-paced anyway.
The pipeline that emerged is byte-budget-scheduled, checkpointed per object, safe to kill at any point, and moved ~19,500 rows/s on four backends — about 50× the small-batch alternative on a 3,700-column schema. One result I have not seen published anywhere: on schemas this wide, column count alone swings write throughput ~5×, while the full correctness stack (merge-on-write dedup, three inverted indexes, ZSTD) costs under 4–12%. The expensive-looking features are nearly free; the width is the cost.
The workload was not what we thought
The original serving-table plan was built on a curated 50-query benchmark. Before committing to an irreversible layout, I mined 14 days of production system.query_log — 74.7 M log rows — and the picture inverted: the overwhelming majority of traffic (well over 90%) was point lookups keyed on one high-cardinality user identifier, and most of those were unbounded “latest value” reconstructions that no time filter could be retrofitted onto. A second serving table went from “maybe” to mandatory. Designing from the benchmark would have optimized the wrong workload.
Then the physics lesson. Point lookups on the Iceberg layout ran 7–13 s against ClickHouse’s 21 ms. The reflex fix — bloom filters — cut rows scanned by 73% and moved wall-clock time not at all: one active user’s rows were scattered across 2,044 partition files, and the floor was file-open count, not rows scanned. (We also proved, byte-level, that two common Iceberg writer paths silently ignore declared bloom-filter table properties.) Indexes cannot fix a layout problem. The fix was physical re-clustering: a Doris-native table distributed by hash on the user key. Because a distribution key is the one decision you cannot ALTER later, we ran a two-table A/B — same data, mirrored keys, four measurements — before committing. Result: 16–24 ms warm, re-measured at three checkpoints during the live migration (15M → 107M → 286M rows) to prove the latency stays flat as the table grows 19×.
Routing is admission control, not performance tuning
With serving and analytics on one platform, the <1% of heavy queries had to be caught before execution — autoscaling is a minutes-scale mechanism and OOM is a seconds-scale event, so elasticity alone can never be the safety guarantee.
Stock Doris EXPLAIN routed our production-shaped SQL suite at 45% accuracy with a 27% heavy-miss rate. The decisive observation: identical query templates with different parameters produced byte-identical EXPLAIN text while real memory differed 23× — a signal problem, not a rule problem. The optimizer computes per-operator cardinality, selectivity, and state sizes internally; it just never prints them.
So I fixed it inside the engine, on an Apache Doris fork:
-
EXPLAIN ESTIMATE PLAN— a read-only visitor over the finalized physical plan that emits the CBO’s existing per-operator estimates as JSON. Purely additive (zero deletions); the mechanism was prepared as an upstream PR. -
EXPLAIN ROUTE PLAN— the estimate plus a 15-rule classifier returning a routing verdict in 6–8 ms, independent of data volume. The classifier was ported from a Python reference implementation treated as an executable oracle: 189/189 golden-corpus parity before any cluster was involved. All ~21 thresholds are hot-reloadable, so tuning never requires a rebuild.
One judgment call mattered as much as the code: leadership initially wanted the classifier upstreamed too. I argued the opposite, using the first PR’s own principle — the engine should emit data; callers decide policy. The classifier carries our calibrated thresholds and recall-first bias; that is policy, and it stays internal. Knowing what not to contribute changed the plan.
Heavy recall went 7% → 100% on the canonical suite, and the residual is covered by design: misrouted heavy queries hit a fixed-slot memory hard limit (~404 MB/query, no overcommit) in the light pool, die within ~3 s, and auto-escalate to the heavy pool. The classifier is allowed to be imperfect because the safety guarantee never depended on it.
Two compaction incidents most operators never see
The death spiral. Post-migration, the wide table froze at ~2,500 segments per tablet; manual compaction failed silently; cold point queries took 40–68 s. Root cause was three factors multiplied: auto-compaction had been disabled during bulk load, letting debt accumulate until a single task had to merge ~2,506 segments × 3,727 columns (aborted by the memory guard at 43 GB — a task-level abort, not a crash); cloud mode has no peer cache, so every segment was a ~7 s cold S3 read; and the segments were tiny in the first place because a sparse wide schema suffers a 46× compression collapse between the in-memory write buffer and the on-disk segment. We also proved that adding backends was useless before anyone spent money on it: a single tablet’s compaction cannot be split across nodes — the logs showed 632 no-op wakeups against 9 real merges. Concentration-bound, not resource-bound. The verified recipe (auto-compaction always on, 4× larger write buffer, per-month bucket counts to hold tablets at ~10 GB, wider column groups to cut S3 re-scans ~40×) took cold point queries from 40 s to <100 ms server-side.
The stuck gauge. A second incident: compaction score pinned at ~4,500, unmoved by scaling or restarts. Three independent read-only investigation tracks — log enumeration, tablet-meta HTTP endpoints, catalog recycle-bin — converged on the same orphaned tablet from an already-dropped table, structurally excluded from scheduling. One chain of reasoning can be led astray; three independent evidence chains pointing at the same tablet is a verdict. That investigation was executed as parallel AI agents fanned out over the diagnostic surfaces, which is how I run forensics generally now.
Elasticity that earns the word
The heavy pool rests at zero replicas. The scaler (an internal controller I rebuilt) does exactly one thing: JSON-patch the replica count on the Doris operator’s custom resource — an HPA role, nothing more. Its first version deployed backends itself and registered them via SQL; I killed that design before implementation because in cloud mode, anything not declared in the operator’s CR gets erased on the next reconcile. When the platform primitive matures, bespoke tooling should shrink.
The full loop runs live: verdict says heavy → controller patches 0→N → spot node up in ~66 s, backend registered in ~2 min, layered readiness checks (a SELECT 1 probe is constant-folded by the frontend and never touches a backend — it will happily report an empty pool as healthy) → query executes → pool reaps to zero. Scaling the serving pool 2→4 backends live moved zero data — tablets live in S3, ownership just rebalances. That is the operational dividend of shared-data, captured on a real cluster.
What makes the numbers trustworthy
Every claim above was measured, and several were retracted on the way. An early benchmark showed Doris beating ClickHouse 3–5× on aggregations; I red-teamed my own result, found the confound (a loaded 8-core production node versus an isolated 14-core benchmark node), and withdrew the multiplier. Public “Doris beats ClickHouse 6–40×” claims all traced back to the vendor; primary-source ClickBench data says ClickHouse still leads flat scans. Migration completeness was certified at 99.945% — 5,170,688,484 of 5,173,508,562 rows — with the 0.055% gap attributed line-by-line to deduplication semantics, and the verification method itself had to be engineered for safety after a naive COUNT(*) audit took down all eight backends.
The habit underneath all of it: when a number gets less flattering every time you re-measure it, that is usually the sign you are converging on the truth.
Engine-level work referenced here lives on an Apache Doris fork; the mechanism (EXPLAIN ESTIMATE PLAN) was prepared for upstream contribution, the policy layer deliberately kept internal. Names, tenants, and exact identifiers are anonymized; scale figures are rounded.