How CockroachDB Works Internally: Distributed SQL, Range-Based Replication, and the Raft-Backed Storage Engine Behind Horizontal Scaling

CockroachDB makes a specific claim: a relational database that survives node failures, scales horizontally across regions, and still gives you serializable transactions. Understanding whether that claim holds in your workload requires understanding exactly how the system achieves it, because every layer of the architecture has tradeoffs that affect latency, throughput, and operational complexity in ways that are not obvious from the documentation.

This article traces a write from SQL down to disk and a read from disk back up through the distributed query planner, touching every layer that matters.

The SQL-to-KV Mapping Layer

CockroachDB is, at its core, a distributed key-value store that interprets an MVCC-ordered byte range as a relational table. Everything at the SQL layer translates to operations on this KV store.

Each table row maps to a single KV entry. The key encodes the table ID, index ID, and primary key column values in a way that preserves sort order. The value encodes all non-primary-key columns using a columnar encoding format.

For a table like this:

// Schema we're working with conceptually
// CREATE TABLE orders (
//   id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
//   customer_id INT NOT NULL,
//   status TEXT NOT NULL,
//   total_cents INT NOT NULL
// );

The encoded key for a row with id = 'abc-123' looks roughly like:

/Table/52/1/"abc-123"/0

Where 52 is the table descriptor ID, 1 is the primary index ID, "abc-123" is the PK value encoded to preserve lexicographic sort order, and 0 is the column family ID.

Secondary indexes get their own KV entries. An index on (customer_id, status) produces entries keyed by:

/Table/52/2/<customer_id>/<status>/<primary_key>

This means a secondary index lookup is two KV reads: one to find the index entry, one to fetch the full row using the primary key found in that entry, unless the index covers all needed columns.

The ordered nature of this encoding is not incidental. It is the mechanism that makes range-based partitioning possible, and it is why CockroachDB can scatter adjacent primary keys across nodes without losing the ability to do range scans.

Range-Based Data Partitioning

The entire KV keyspace is divided into contiguous chunks called ranges. Each range covers a span of the ordered keyspace, from some start key up to but not including some end key. Ranges are the unit of replication, routing, and load balancing.

By default, a range grows until it reaches 512 MB, at which point the range splits at the midpoint of its key distribution. The split point is chosen to be the median key by data size, not by count, so skewed data distributions produce splits that roughly balance storage. Splits are initiated by the leaseholder node and coordinated through Raft before taking effect.

Merges happen when two adjacent ranges both fall below one quarter of the split threshold (128 MB). The system periodically scans for mergeable pairs and collapses them to reduce overhead, since each range carries Raft heartbeat traffic, lease renewal, and gossip metadata.

Every range has a leaseholder: the single node responsible for serving reads and coordinating writes for that range. The leaseholder holds an epoch-based lease with a configurable TTL (default 9 seconds). If a leaseholder goes silent, the lease expires and a new leaseholder is elected through Raft. Clients talk to a gateway node, which uses gossip to discover which node holds the lease for the target range and routes requests accordingly.

This means a write to a single row touches exactly one leaseholder, which then fans out replication to the other replicas in that range’s Raft group. There is no global write coordinator. This is the architectural choice that makes CockroachDB’s write scalability different from, say, a single-primary PostgreSQL cluster behind a proxy.

Raft Consensus Per Range

Each range is replicated across a configurable number of nodes (default 3). Replication is managed by a Raft group scoped to that range. This is a critical design point: there is no single global Raft group. A cluster with 10,000 ranges running 3x replication has 10,000 independent Raft groups, each making forward progress independently.

The leaseholder for a range is almost always the Raft leader for that range’s group, though these are separate roles. The leaseholder handles client traffic; the Raft leader drives log replication. CockroachDB collocates them to avoid an extra round trip for reads.

Writes go through Raft before being acknowledged:

1. Client sends write to the gateway node.
2. Gateway routes to the leaseholder for the target range.
3. Leaseholder proposes a Raft entry.
4. Raft replicates the entry to followers (a quorum, so 2 of 3 for a 3x replication factor).
5. Once a majority acknowledges, the entry is committed.
6. The leaseholder applies the entry to Pebble (the underlying storage engine, a LevelDB-style LSM tree) and responds to the client.

Follower reads are not served by default because followers might be behind the leader. CockroachDB does support follower reads for queries that can tolerate a bounded staleness (controlled via the AS OF SYSTEM TIME clause with a negative offset, typically follower_read_timestamp()).

Learner replicas participate in replication during range splits and membership changes but do not vote in Raft elections until they catch up. Joint consensus (the two-phase membership change protocol from the Raft paper) handles the transition from one set of voting replicas to another, preventing split-brain during reconfigurations.

The Distributed SQL Query Planner

CockroachDB processes SQL in two phases: logical planning and physical planning.

The logical planner operates on a gateway node and produces an operator tree, which gets optimized by the cost-based optimizer using table statistics (histograms, row counts, null fractions). Statistics are collected automatically by a background job and stored in the system tables.

The physical planner takes the logical plan and decides whether to execute it locally on the gateway or to distribute execution across multiple nodes using DistSQL.

DistSQL works by sending partial execution plans to nodes that hold the relevant data ranges. Each such node is a leaf executor. Leaf executors scan their local ranges and stream rows back to the gateway, where the gateway performs any final aggregations or joins that require a full view of the data.

A query like:

// Conceptually: SELECT customer_id, SUM(total_cents)
// FROM orders
// WHERE status = 'completed'
// GROUP BY customer_id
// HAVING SUM(total_cents) > 100000;

Gets distributed roughly as:

• Each node scans its local orders ranges, applying the status = 'completed' filter locally.
• Each node performs a partial GROUP BY customer_id, computing partial sums.
• Partial aggregates stream to the gateway.
• Gateway merges partial sums, applies the HAVING filter, and returns results to the client.

This pushes the heavy lifting (full table scans, filter evaluation) to the nodes holding the data, which means network transfer is bounded by the output size rather than the full table size. For large analytic queries, this makes CockroachDB meaningfully more efficient than a proxy-based approach where all rows would flow through a single node.

The gateway coordinates through a flow infrastructure. Each physical plan is a directed acyclic graph of processors connected by channels, and the planner assigns processors to nodes based on where data lives. The physical plan serialization format is sent via gRPC to each participating node.

MVCC and the Transaction Model

CockroachDB uses MVCC (Multi-Version Concurrency Control) with timestamps as version identifiers. Every write produces a new version of a key, tagged with the write’s timestamp. Reads see a consistent snapshot at a chosen timestamp.

Timestamps are generated using a Hybrid Logical Clock (HLC). The HLC combines a physical component (wall clock time in nanoseconds) with a logical counter. The physical component ensures timestamps are meaningful for causality across the cluster. The logical counter ensures monotonicity within a node even when physical clocks momentarily go backward. Every message between nodes carries the sender’s HLC, and the receiver advances its HLC to the max of its own and the received value. This gives a partial causal ordering without requiring synchronized clocks (no GPS receivers required, unlike Spanner’s TrueTime).

Write Intents

When a transaction writes a key, it does not immediately commit. Instead, it lays down a write intent: a provisional version of the key, tagged with a special intent marker and the transaction’s unique ID. The transaction record (stored at a specific key in the same range as one of the transaction’s writes) tracks whether the transaction is pending, committed, or aborted.

Other transactions encountering a write intent must decide what to do:

• If the intent’s transaction is already committed or aborted (determinable by reading the transaction record), the reader can resolve the intent and proceed.
• If the intent’s transaction is still pending and the conflict is a write-write conflict, the encountering transaction pushes the intent’s transaction to a higher timestamp or waits.

TxnWaitQueue

When a transaction needs to wait for a conflicting write intent to resolve, it enters the TxnWaitQueue on the leaseholder node that holds the conflicting key. The queue tracks which transactions are blocking which others. When the blocking transaction commits or aborts, the queue notifies all waiters so they can retry their conflicting operations.

The TxnWaitQueue is also the mechanism CockroachDB uses to detect and break deadlocks. Each node tracks transaction dependencies. Cycles in the wait graph indicate a deadlock. CockroachDB uses a wound-wait scheme: an older transaction (lower timestamp) can push a younger transaction (higher timestamp), but a younger transaction cannot push an older one. It must wait. This biases abort decisions toward aborting younger transactions, which have done less work.

Serializable Isolation

CockroachDB’s default and only isolation level is serializable. It achieves this through timestamp ordering. When a transaction reads a key, it records a read timestamp. When it goes to commit, the system checks whether any writes have occurred to those keys at timestamps between the transaction’s read timestamp and its commit timestamp. If yes, the transaction must restart with a higher timestamp (or abort if the keys have been overwritten by a committed transaction that the restarted read would also need to see differently).

This is serializable snapshot isolation (SSI), similar in spirit to PostgreSQL’s SSI but implemented differently because reads and writes are distributed across many nodes rather than coordinated through a single lock manager.

Here is a minimal TypeScript example showing how to work with serializable transactions and handle retries:

import { Pool } from "pg";

const pool = new Pool({ connectionString: process.env.DATABASE_URL });

async function transferFunds(
  fromAccountId: number,
  toAccountId: number,
  amountCents: number,
  maxRetries = 5
): Promise<void> {
  for (let attempt = 0; attempt < maxRetries; attempt++) {
    const client = await pool.connect();
    try {
      await client.query("BEGIN");

      const { rows: fromRows } = await client.query<{ balance_cents: number }>(
        "SELECT balance_cents FROM accounts WHERE id = $1 FOR UPDATE",
        [fromAccountId]
      );

      if (fromRows.length === 0) throw new Error("Source account not found");
      if (fromRows[0].balance_cents < amountCents) {
        throw new Error("Insufficient funds");
      }

      await client.query(
        "UPDATE accounts SET balance_cents = balance_cents - $1 WHERE id = $2",
        [amountCents, fromAccountId]
      );

      await client.query(
        "UPDATE accounts SET balance_cents = balance_cents + $1 WHERE id = $2",
        [amountCents, toAccountId]
      );

      await client.query("COMMIT");
      return;
    } catch (err: unknown) {
      await client.query("ROLLBACK");

      // CockroachDB signals retriable errors with SQLSTATE 40001
      const pgErr = err as { code?: string; message?: string };
      if (pgErr.code === "40001") {
        // Retriable serialization failure: back off and retry
        const backoffMs = Math.min(100 * 2 ** attempt, 2000);
        await new Promise((resolve) => setTimeout(resolve, backoffMs));
        continue;
      }

      throw err;
    } finally {
      client.release();
    }
  }

  throw new Error(`Transaction failed after ${maxRetries} attempts`);
}

The 40001 SQLSTATE is the standard serialization failure code. CockroachDB returns this for both write-write conflicts and SSI-detected anomalies. Your application layer must handle retries. CockroachDB also provides a SAVEPOINT cockroach_restart mechanism that allows in-transaction retries without reconnecting, which reduces overhead for high-contention workloads.

Production Considerations

Hotspot ranges. Sequential primary keys (auto-increment integers, time-based UUIDs v1) cause all writes to funnel into a single range at the high end of the keyspace. This produces a leaseholder hotspot. Use UUIDs (v4) or gen_random_uuid() as primary keys to distribute writes across the keyspace. Alternatively, use hash-sharded indexes for high-write tables with range scan requirements.

Cross-range transactions. A transaction touching keys in multiple ranges is a distributed transaction. The coordinator node manages a two-phase commit across all involved leaseholders, using write intents as the first phase and transaction record commit as the second. Cross-region transactions are expensive because Raft round trips span geographic latency. Structure your schema so that commonly co-transacted keys land in the same range by co-locating related data using locality-aware placement policies.

Follower reads for read-heavy workloads. If you can tolerate reading data that is 4.8 seconds stale (the default closed_timestamp interval), use follower reads to serve reads from a nearby replica rather than the leaseholder. This is particularly useful in multi-region clusters where clients in region B would otherwise have to route reads to the leaseholder in region A.

// Follower read: reads data as of ~4.8s ago from a nearby replica
const { rows } = await client.query(
  "SELECT * FROM orders AS OF SYSTEM TIME follower_read_timestamp() WHERE customer_id = $1",
  [customerId]
);

Range count overhead. Each range has background overhead: Raft heartbeats (every 200ms by default), lease renewal, and gossip updates. A cluster with millions of small rows across many tables can accumulate thousands of tiny ranges. Watch the ranges metric per store and tune kv.range_merge.queue_enabled if you have a workload that fragments ranges but rarely fills them.

Transaction retries are real. Under contention, retry rates can climb. Use EXPLAIN ANALYZE to identify hot tables and check crdb_internal.transaction_contention_events to see which transactions are conflicting. For known-contention scenarios (counters, queue-like patterns), consider batching writes or using SELECT FOR UPDATE to acquire locks eagerly rather than letting conflicts surface at commit time.

Schema changes are online. CockroachDB performs online schema changes (adding columns, creating indexes) without locking the table. The process uses a multi-version schema state machine borrowed from the F1 paper. Migrations run in the background and the table remains fully available. The tradeoff: schema changes are slower than PostgreSQL’s instant column additions for small tables.

Tradeoffs Comparison

Dimension	CockroachDB	PostgreSQL	Spanner	YugabyteDB	TiDB
Isolation level	Serializable only	Read committed to serializable	Serializable (external consistency)	Serializable	Snapshot + serializable
Horizontal write scaling	Yes, range-based	No (single primary)	Yes, split-key	Yes, tablet-based	Yes, region-based
Clock model	HLC (no GPS needed)	Local wall clock	TrueTime (GPS + atomic)	HLC	HLC
Geo-partitioning	Yes (partition by region)	Manual sharding only	Yes	Yes	Limited
Follower reads	Yes (bounded staleness)	No (hot standby only)	Yes (stale reads)	Yes	Yes
Wire protocol	PostgreSQL (pgwire)	PostgreSQL (native)	Proprietary gRPC	PostgreSQL (pgwire)	MySQL + TiDB
Schema changes	Online (F1 model)	Mixed (instant for small)	Online	Online	Online
Managed offering	CockroachDB Serverless / Dedicated	RDS, Cloud SQL, Aurora	Cloud Spanner	Yugabyte Cloud	TiDB Cloud
Operational complexity	Medium (self-hosted)	Low	Low (managed only)	Medium	Medium-high
Sweet spot	Multi-region OLTP, survivability	Single-region OLTP, familiarity	Global OLTP at Google scale	PostgreSQL API, multi-region	Hybrid OLTP/OLAP (HTAP)

Spanner is the reference implementation for what CockroachDB is trying to approximate without proprietary hardware. The TrueTime API gives Spanner external consistency (a stronger guarantee than serializable: the commit order matches real time, not just logical order). HLC gets close but cannot guarantee this without atomic clocks. In practice, the difference surfaces when you have cross-region transactions and need to reason about absolute time ordering of commits.

YugabyteDB uses a similar Raft-per-tablet architecture and also exposes a PostgreSQL wire protocol, making it CockroachDB’s most architecturally similar competitor. The primary differentiator is storage engine (YugabyteDB uses DocDB, a RocksDB fork with a custom value encoding) and the degree of PostgreSQL compatibility each team has achieved.

TiDB separates OLTP storage (TiKV, a Raft-based KV store) from OLAP execution (TiFlash, a columnar replica) within the same cluster, making it a genuine HTAP system. If your workload mixes heavy analytical queries with transactional writes, TiDB’s architecture is a better fit than CockroachDB’s, which requires exporting to an OLAP system for serious analytical work.

Closing

CockroachDB’s architecture is coherent: every layer (ordered KV encoding, range-based partitioning, Raft per range, DistSQL, HLC-based MVCC) fits together to produce a system that can horizontally scale OLTP workloads while maintaining serializable isolation. The cost is latency for cross-range and cross-region transactions, retry handling in the application layer, and operational complexity that PostgreSQL does not require. Whether that cost is worth it depends entirely on whether your workload genuinely needs horizontal write scaling and multi-region survivability. For a single-region workload that fits on one primary, it does not.