# Inside the StarRocks and Doris Query Engine: CBO, Colocate Joins, Merge-on-Write, and Materialized View Rewrite 🧭 **Deciding whether to adopt?** I scored StarRocks and Apache Doris side by side — ten axes, an interactive radar, workload fit, and a 4-week POC plan — in the [Architecture Radar assessment](/architecture-radar-doris-vs-starrocks). A few months back I spent a week chasing a query that should have been colocated and wasn't. Two fact tables, both bucketed on the same key, both in what I was sure was the same colocation group — and the profile kept showing a shuffle exchange eating half the runtime. The bug turned out to be a bucket count mismatch introduced by a table that got recreated with a different default during a migration script, six weeks earlier, by someone who has since left the team. Nothing in the query itself was wrong. The optimizer was doing exactly what it should with the inputs it had. That week is basically why this article exists — the FE/BE architecture and the four data models get you to "it works," but the difference between "it works" and "it's fast, and I know why" lives one layer down, in the optimizer's cost model, the write path's index structures, and the compaction scheduler nobody looks at until it falls behind. I already covered the shared FE/BE architecture and the basic data models — Duplicate, Aggregate, Unique, Primary Key — in [StarRocks & Apache Doris Internals](starrocks-doris-architecture). Go read that first if the terms FE, BE, and tablet are new to you; I'm not re-explaining them here. This article picks up where that one stops: how the cost-based optimizer actually decides among join strategies, what the vectorized pipeline engine does once a plan is chosen, how Doris's merge-on-write primary-key model resolves an upsert at the byte level, why compaction is the mechanism that makes merge-on-write sustainable rather than a slow leak, and how materialized view rewrite finds and substitutes a matching MV without the query ever naming it. If you want the business case for any of this — why a team would adopt these engines in the first place — that's the companion piece, [StarRocks and Apache Doris in Production](starrocks-doris-production-use-cases). ## What does the cost-based optimizer actually decide? StarRocks' CBO takes a parsed, validated query plan and a set of gathered table statistics — row counts, NDVs, histograms, null fractions — and searches for the lowest-estimated-cost physical execution plan among many logically equivalent ones. It's built on Cascades/ORCA-style principles: rather than applying a fixed sequence of rewrite rules, it explores a search space of alternative plans (different join orders, different join strategies, different aggregation placements) and scores each candidate by an estimated cost function covering CPU, memory, network, and I/O per operator, then keeps the cheapest. This is the mechanism that separates these engines from a plain columnar scanner — a scanner doesn't have to decide anything about the shape of a six-table join; these engines do, on every query, in milliseconds. The optimizer's single most consequential decision, for most analytical workloads, is **which physical join strategy to use for each join in the plan**. StarRocks' CBO chooses per-join among five strategies, and the choice is driven almost entirely by table size estimates and the physical distribution (bucketing) of the tables involved: | Strategy | Data movement | When the CBO picks it | | --- | --- | --- | | **Broadcast** | Copy the entire small side to every node holding a fragment of the large side | One side is small enough to replicate cheaply (below a cost threshold derived from row count/size stats) | | **Shuffle (hash)** | Repartition *both* sides by the join key across all nodes | Both sides are large and neither is pre-bucketed on the join key | | **Bucket-shuffle** | Repartition *only one* side to match the other side's existing bucket distribution | One side is already bucketed on the join key; only the other side needs to move | | **Colocate** | None — matching tablets already sit on the same node | Both tables are in the same colocation group, bucketed identically on the join key | | **Replicated** | Every node holds a full copy of one table ahead of time | A small dimension table that's joined constantly, worth fully replicating rather than broadcasting per query | ```mermaid graph TD subgraph Shuffle["Shuffle join — both sides move"] direction LR S1["Node Afact rows 1-N"] -->|"repartition by key"| SX["exchange"] S2["Node Bfact rows N-2N"] -->|"repartition by key"| SX S3["Node Adim rows 1-N"] -->|"repartition by key"| SX S4["Node Bdim rows N-2N"] -->|"repartition by key"| SX end subgraph BucketShuffle["Bucket-shuffle join — one side moves"] direction LR B1["Node Afact, bucketed on key"] -->|"stays put"| BJ["local join"] B2["Node Adim, re-bucketed to match"] -->|"shuffled once"| BJ end subgraph Colocate["Colocate join — no shuffle"] direction LR C1["Node Afact tablet, bucket 3"] -->|"local join"| CJ["local join"] C2["Node Adim tablet, bucket 3"] -->|"local join"| CJ end ``` *Three of the five join strategies, ordered by how much data crosses the network. Shuffle moves everything; bucket-shuffle moves half; colocate moves nothing because both tables were bucketed on the same key ahead of time and matching rows already live on the same node. The CBO picks among these (plus broadcast and replicated) per join, per query, from cost estimates — but colocate only exists as an option if you designed your bucketing to make it exist.* Notice that colocate and bucket-shuffle aren't things the optimizer conjures from nothing — they're only available because a schema decision made them available. If your fact and dimension tables are bucketed on different keys, or with different bucket counts, the CBO can't pick colocate no matter how much it would help; it falls back to shuffle. This is the thing that bit me in the story above: a bucket count mismatch silently removed an option from the optimizer's search space, and everything downstream — the plan, the profile, the runtime — was a correct response to a broken input. **Colocate joins are a schema commitment, not a query hint.** To get one, both tables need the same bucketing key, the same bucket count, and the same replication factor, and they need to be assigned to the same colocation group at DDL time. Change any of those later — resize buckets, add a table with a mismatched count, rebalance replicas unevenly — and the optimizer silently falls back to a shuffle join with no error, just a slower profile. If a join you're counting on suddenly gets slower after a schema change, check colocation group membership before you look anywhere else. ## Where do the CBO's cost estimates actually come from? A cost-based optimizer is only as good as the statistics it's costing plans against, and this is the part evaluations tend to skip past. StarRocks collects table statistics through `ANALYZE TABLE` — either run manually or, more commonly in production, scheduled automatically on a cadence or triggered by a change-volume threshold. Collection comes in two granularities: a cheap full-table statistics pass (row counts, column NDVs, min/max, null fraction) that's fast enough to run frequently, and a more expensive histogram collection on specific columns you name explicitly, which buckets a column's value distribution finely enough for the optimizer to estimate selectivity on skewed predicates accurately. Doris's Nereids optimizer follows the same shape: automatic statistics collection jobs plus manual histogram collection for columns the optimizer needs finer-grained selectivity on. The failure mode worth knowing before it bites you: **stale statistics don't cause query errors, they cause silently bad plans.** If a table's row count triples after a large backfill and statistics haven't been refreshed, the optimizer is costing joins against numbers that no longer describe reality — it might broadcast a table it thinks is small but which is now genuinely large, or pick a join order that made sense for the old row-count ratio. Nothing errors. The query just runs slower than it should, and unless someone happens to check `SHOW ANALYZE STATUS` or compare estimated versus actual row counts in a query profile, the cause isn't obvious. After any bulk load, backfill, or partition swap materially changing a table's size or distribution, triggering a fresh `ANALYZE` before trusting the query profile is worth the habit. ## How does the vectorized pipeline engine make a chosen plan fast? The CBO decides *what* to run; the execution engine decides *how fast* it runs. These are genuinely separate layers, and it's worth being precise about the split, because a lot of tuning conversations conflate them. Once the FE has picked a physical plan, it's compiled into a DAG of **pipeline drivers** — a StarRocks/Doris-specific execution model where each operator (scan, join, aggregate, exchange) processes data as a stream of column-oriented batches rather than pulling one row at a time through the classic Volcano iterator model. Each BE runs many pipeline drivers concurrently across its CPU cores, and drivers yield cooperatively when they're waiting on I/O or an exchange, instead of blocking a thread — which is what lets a BE keep every core busy under a mixed workload instead of stalling on the slowest fragment. "Vectorized" is doing real work in that sentence, not just marketing language: operators like filters, hash probes, and aggregations process a batch of, typically, a few thousand rows at once through tight C++ loops that the compiler can auto-vectorize with SIMD instructions, rather than branching per row. A hash join probe against a batch of 4,096 rows is one tight loop with predictable branching, not 4,096 individual function calls. That's the same principle behind ClickHouse's scan speed, applied here to every operator in a distributed, join-capable plan rather than just to a single-table scan. The practical upshot: a bad plan executed by a fast engine is still a bad plan — vectorization doesn't fix a shuffle join that should have been a colocate join, it just makes the shuffle join less catastrophically slow than it would be on a row-at-a-time engine. Tune the CBO's inputs (statistics, bucketing) first; the pipeline engine is what makes a good decision pay off, not what compensates for a bad one. The other detail worth being precise about: pipeline parallelism is elastic within a single query in a way a fixed-thread execution model isn't. A classic MPP engine that assigns a fixed number of execution threads per fragment at plan time can leave cores idle when one fragment finishes early and another is still grinding through a big scan. The pipeline-driver model lets the BE's scheduler move available cores to whichever drivers have runnable work, across fragments, rather than a query being bottlenecked by whichever fragment got under-provisioned at plan time. This is also why these engines tend to degrade more gracefully under mixed concurrent workloads — a burst of small dashboard queries sharing a BE with one large batch query isn't fighting over statically pinned resources, it's competing for scheduler time the same way any cooperative scheduler resolves contention. ## What does merge-on-write actually do on every upsert? This is the mechanism behind the real-time CDC use case I described in the companion article's VBill Payment example — billions of records, sub-second reads, continuous upserts from a change stream. Doris's **merge-on-write (MoW)** mode, the default for the Unique Key model since Doris 2.1, is the concrete answer to "how does an upsert into a 10-billion-row table not become a query-time merge tax." StarRocks' Primary Key model works on the same underlying principle. Here's the mechanism, end to end, for a single incoming row during a write: - Every data segment, when it's flushed to disk, gets a **per-segment primary-key index** built alongside it — a sorted, paginated structure mapping each key in that segment to its row position. This index exists so a key lookup doesn't require scanning the segment's data. - When a write batch arrives, the BE looks up each incoming key against the primary-key indexes of existing segments (using an in-memory cache of recently active indexes to avoid disk reads for hot keys) to find whether that key already exists, and if so, in which rowset, segment, and row position. - If the key already exists, its old row isn't touched. Instead, the BE marks that (rowset_id, segment_id, row position) as logically deleted in a **delete bitmap** scoped to that rowset/segment/version. The old bytes stay physically on disk. - The new row's values — full row or, for a partial update, just the changed columns merged with the unchanged ones — are written into a fresh rowset at the next version number. - If the table has `function_column.sequence_col` configured, the incoming row's sequence value is compared against the existing row's stored sequence value before any of this happens. If the incoming sequence value is smaller, the write is a no-op for that key — the existing row wins. This is what resolves out-of-order delivery from a CDC source correctly: a Debezium connector replaying a burst of events, or a Kafka consumer restarting after a rebalance, won't silently let an older event clobber a newer one. ```sql -- Doris table using merge-on-write Unique Key with a sequence column -- for correct out-of-order CDC resolution CREATE TABLE orders ( order_id BIGINT, status VARCHAR(32), updated_at DATETIME, amount DECIMAL(18,2) ) UNIQUE KEY(order_id) DISTRIBUTED BY HASH(order_id) BUCKETS 32 PROPERTIES ( "enable_unique_key_merge_on_write" = "true", "function_column.sequence_col" = "updated_at", "replication_num" = "3" ); -- A late-arriving event with an older updated_at than what's already -- stored for this order_id is a no-op: the existing row's sequence -- value is compared and the larger one wins, at write time. ``` The reason this matters at query time is what it *doesn't* do: a read never has to merge candidate rows for a key on the fly. Merge-on-read models (the plain Unique Key model, or ClickHouse's `ReplacingMergeTree` without `FINAL`) push that reconciliation to query time — every scan has to check whether it's holding the latest version of a row, which is exactly the cost that scales badly under high concurrency. MoW pays that cost once, on write, using an index built for the purpose, and leaves every subsequent read looking at rows that are already known-current except for whatever the delete bitmap has marked dead — which the scan simply skips. Partial updates deserve a specific mention because they're where the primary-key index earns its keep twice over. A CDC event from a source system frequently touches only a handful of columns — an order-status change updates `status` and `updated_at`, not the customer address or line-item details on the same row. Without a fast way to find the existing row's other column values, a partial update would force a read of the full row before it could write a complete replacement. The primary-key index is what makes that lookup cheap enough to do on every write: it resolves the existing row's location, the BE reads just the columns needed to fill in the ones the incoming batch didn't touch, and the merged row is written as a whole into the new rowset. This is also why partial-update support is tied specifically to the merge-on-write model and not available on merge-on-read Unique Key tables — merge-on-read doesn't maintain a structure that makes finding "the rest of this row" cheap. **The trade-off, stated plainly.** Merge-on-write moves cost from the read path to the write path. Every write now pays for a primary-key index lookup and a delete-bitmap update, on top of the write itself. For a table with heavy write concurrency and rare reads, that's a bad trade. For the actual target workload — CDC mirrors and dimension tables read constantly by dashboards and joins, written continuously but each write touching one row — it's close to the ideal trade, because you're paying the lookup cost once per write instead of a merge cost on every one of thousands of reads. ## Why does compaction matter, and how does it connect to merge-on-write? Both engines use an LSM-tree-style storage layer: writes append new rowsets rather than modifying existing files in place, which is fast to write but means a tablet's data accumulates across an increasing number of small files over time. Left alone, this hurts scans (each scan has to read from more files and merge more candidate rows) and eventually hurts writes too (too many small rowsets slows the primary-key index lookups described above). **Compaction** is the background process that merges smaller rowsets into larger ones to keep this under control. The scheduler tracks a per-tablet **compaction score** — roughly, a measure of how many separate rowset "layers" a query against that tablet would need to touch if nothing were merged. A tablet with a high compaction score is a tablet whose queries are about to get slower. There are two tiers of compaction: - **Cumulative Compaction** runs frequently and merges recently written, smaller rowsets together — cheap, fast, keeps the score from climbing under steady write load. - **Base Compaction** runs less often and merges accumulated rowsets into the tablet's large base rowset — more expensive per run, but this is what actually reclaims disk space and keeps long-term scan cost flat. Here's the connection to merge-on-write that's easy to miss: **the delete bitmap only marks old rows as logically dead — it doesn't remove them.** A row superseded by ten subsequent upserts is still sitting on disk ten times over until compaction physically merges the rowsets and drops the bitmapped-dead rows. A table under heavy upsert churn with compaction falling behind will show growing disk usage and growing scan latency that look like unrelated problems but have the same root cause: the write path is generating dead rows faster than the background compaction is reclaiming them. This is the "oh, that's why" I mentioned at the top — merge-on-write's speed depends on compaction keeping pace, and if you only monitor query latency you'll see the symptom (scans getting slower) long after the actual cause (compaction score climbing) was visible in the metrics. In practice, compaction competes with foreground query and write traffic for the same CPU and I/O, which is why both engines expose throttles and priority controls rather than just running compaction flat-out. Push compaction too aggressively and it starves the query traffic it's supposed to be protecting the latency of; throttle it too conservatively and the compaction score climbs faster than it's reclaimed, especially during a sustained high-write period like a backfill or a CDC replay after downtime. The operational habit that actually works: watch compaction score as a leading indicator, not scan latency as a lagging one — by the time scan latency visibly degrades from a compaction backlog, you're already behind and catching up means running compaction harder while it's competing with the traffic you're trying to protect. ## How does materialized view rewrite find a match without being told? An **asynchronous materialized view** in StarRocks or Doris is a precomputed, incrementally refreshed result of a query — typically a join-and-aggregate — stored as its own physical table. The interesting part isn't the storage; it's that the optimizer can substitute an MV into a query plan for a query that never mentions the MV's name. Doris does this with an **SPJG-based** (SELECT-PROJECT-JOIN-GROUP-BY) rewrite algorithm: it normalizes both the incoming query and each candidate MV's defining query into that canonical shape — the tables and join predicates involved, the grouping keys, the aggregate functions — and checks whether the MV's shape is a structural subset or equivalent of what the incoming query needs. If it matches, the optimizer substitutes a scan of the MV for the underlying joins and aggregations, transparently, as one candidate plan among the others the CBO is costing. ```mermaid graph LR Q["Incoming query:SELECT region, SUM(amount)FROM orders JOIN customersGROUP BY region"] N["Normalize to SPJG shape:tables + join keys +group-by keys + aggregates"] M["Compare againstcandidate MV definitions"] R{"Structural matchfound?"} MV["Rewrite plan:scan the MV directly"] ORIG["Original plan:join + aggregatefrom base tables"] Q --> N --> M --> R R -->|"yes"| MV R -->|"no"| ORIG ``` *SPJG-based transparent rewrite. The optimizer never needs the query to name the materialized view — it normalizes both sides to a comparable shape and matches structurally. When several MVs could satisfy the same query, cost-based selection picks among them the same way the CBO picks among join strategies.* ```sql -- An async materialized view Doris can transparently substitute -- into any query whose SPJG shape matches this one CREATE MATERIALIZED VIEW mv_orders_by_region_daily BUILD DEFERRED REFRESH ASYNC EVERY(INTERVAL 1 HOUR) DISTRIBUTED BY HASH(region) BUCKETS 8 AS SELECT o.order_date, c.region, SUM(o.amount) AS total_amount, COUNT(*) AS order_count FROM orders o JOIN customers c ON o.customer_id = c.customer_id GROUP BY o.order_date, c.region; -- A later query never referencing the MV by name can still -- be transparently rewritten to scan it, if the optimizer -- matches the shape and the cost favors it: SELECT region, SUM(total_amount) FROM orders o JOIN customers c ON o.customer_id = c.customer_id WHERE order_date >= '2025-10-01' GROUP BY region; ``` Incremental refresh is where the partition-level bookkeeping matters. Doris tracks a **partition version correspondence** between an MV and its base tables — for example, MV partition `p202003` might be built from base-table partitions `p20200301` and `p20200302`. At each refresh, Doris records which base-table partition versions were used to build each MV partition; on the next refresh cycle, it checks whether those specific base-table partition versions have changed. If `p20200301` got new data since the last refresh but `p20200302` didn't, only the MV partition depending on the changed one gets rebuilt — not the whole MV. For an MV spanning months of history refreshed hourly against a table where only the current day is actively written, this is the difference between an hourly refresh that touches megabytes and one that rescans everything. **Transparent rewrite is only as good as the optimizer's ability to recognize a match.** The SPJG matching is structural, not semantic — a query written in a logically equivalent but differently shaped way (a different join order expressed explicitly, an extra redundant predicate, a CTE that obscures the join structure from the matcher) can fail to match an MV you built specifically for it, and you'll get correct results at the original, un-accelerated cost with no error or warning telling you the rewrite didn't fire. If a dashboard isn't getting the speedup you expected from an MV, check the query profile for whether the MV was actually scanned before assuming the MV itself is the problem — the more common cause is a query shape the matcher doesn't recognize. ## What actually breaks when transparent rewrite doesn't fire? I want to spend one more section on this because it's the mechanism most likely to quietly cost a team money without anyone noticing. An MV is a table — it has storage cost, and if it's being refreshed hourly or more frequently, a real ongoing compute cost to keep current. If the queries it was built to accelerate stop matching because someone rewrote a dashboard's SQL, added an extra join for a new filter, or introduced a CTE the SPJG matcher can't see through, you're paying full price for the MV's storage and refresh cost while getting zero acceleration benefit from it. Nothing alerts on this. The dashboard just quietly runs at un-accelerated speed, and the MV sits there looking productive in a table listing. The check that actually catches this: pull the query profile for a dashboard you expect to be MV-accelerated and look for whether the plan shows a scan against the MV's table or a scan against the base tables with a live join and aggregate. If it's the latter, the rewrite didn't fire, and the fix is almost always restructuring the query to match the MV's SPJG shape more directly — moving predicates, simplifying an unnecessary CTE, matching the exact aggregate functions the MV computes — rather than assuming the MV mechanism itself is broken. I've seen teams debug this backwards more than once: rebuilding or "refreshing" an MV that was working fine, when the actual problem was a query shape drift that happened weeks earlier and nobody connected to the dashboard getting slower. ## How these four mechanisms fit together None of these pieces are independent trivia — they chain. The CBO's join strategy selection only has colocate and bucket-shuffle available because of upfront bucketing decisions; the vectorized pipeline engine is what makes whichever strategy gets picked actually fast to execute; merge-on-write is what makes the Primary Key/Unique Key model's real-time upserts viable without a query-time merge tax; compaction is the unglamorous background process that keeps merge-on-write's write-path optimism from turning into unbounded disk growth and creeping scan latency; and materialized view rewrite is the mechanism that lets you precompute the expensive joins these engines are good at without forcing every query author to know the MV exists. Miss any one link and the others degrade quietly — a bucketing mismatch removes a join strategy with no error, a compaction backlog turns a fast upsert model into a slow one over weeks, an MV that never gets matched just silently costs you the same as if you'd never built it. ## What to carry away The CBO picks a physical plan — join order and strategy (broadcast, shuffle, bucket-shuffle, colocate, replicated) — from cost estimates over gathered statistics, but colocate and bucket-shuffle only exist as options if your bucketing design made them possible. The vectorized pipeline engine is a separate layer that executes whatever plan gets chosen, processing columnar batches through cooperatively scheduled drivers; it makes a good plan fast, it doesn't fix a bad one. Doris's merge-on-write model resolves upserts at write time via a per-segment primary-key index and a delete bitmap, with a sequence column to correctly resolve out-of-order CDC events — trading write-path cost for read-path simplicity. Compaction is what makes that trade sustainable, physically reclaiming what the delete bitmap marks logically dead. And materialized view rewrite uses SPJG structural matching plus partition-version tracking to transparently accelerate queries and refresh incrementally — but only for query shapes the matcher actually recognizes. For the FE/BE fundamentals and the four data models this article assumes you already know, see [StarRocks & Apache Doris Internals](starrocks-doris-architecture). For where these mechanisms show up in real production deployments — including the CDC/upsert use case merge-on-write exists for — see [StarRocks and Apache Doris in Production](starrocks-doris-production-use-cases). And for how these engines stack up against ClickHouse on the same dimensions, see [StarRocks vs ClickHouse vs Doris](starrocks-vs-clickhouse-vs-doris).