How CockroachDB Works Internally: Distributed SQL, Raft Consensus Per Range, and the Architecture Behind Serializable Transactions at Global Scale

CockroachDB makes a promise that most databases avoid: serializable isolation, horizontal scalability, and automatic survivability across node and zone failures, all under a familiar PostgreSQL wire protocol. The interesting part is not the promise but the mechanism. Each of these properties is expensive to implement correctly in a distributed setting, and CockroachDB achieves all three by composing several independently complex subsystems: a range-based key-value store, per-range Raft consensus, MVCC with hybrid logical clocks, and a distributed SQL execution engine layered on top.

This article traces what actually happens when you run a transaction against a CockroachDB cluster, from SQL parsing through KV operations through Raft log replication through commit.

The Range-Based Data Model

The fundamental storage unit in CockroachDB is the range: a contiguous interval of the sorted key space, stored in a local RocksDB instance on each node. Every range is approximately 512 MB by default. All SQL data, system metadata, and internal structures live in this single key space, identified by structured key prefixes.

A row in a SQL table maps to a KV entry. The key encodes the table ID, index ID, and primary key column values in a byte-comparable encoding that preserves sort order across types. The value encodes the non-primary-key columns. Secondary indexes get their own KV entries, with keys that encode the indexed column values followed by the primary key, enabling index-only scans without a primary key lookup when the projected columns are all in the index.

When a range grows beyond the size threshold, CockroachDB splits it at the midpoint of the key space. The split is atomic: the Raft group for the original range replicates a split command, which causes the range to divide into two ranges, each with its own independent Raft group going forward. Merges happen in reverse: two adjacent ranges with combined size below a merge threshold are consolidated if they have been small for long enough, reducing the number of Raft groups the cluster must maintain.

The split and merge thresholds interact with a class of production problem called range hotspots. A monotonically increasing primary key (serial, timestamp, ULIDv1) concentrates all new writes on the rightmost range in the key space, because every insert lands at the current maximum key. That single range becomes a bottleneck regardless of how many nodes the cluster has. The fix is key scattering: use UUID or gen_random_uuid() as the primary key, prepend a hash bucket, or use CockroachDB’s HASH SHARDED INDEX syntax which automates the hash prefix.

Raft Consensus Per Range

Each range is replicated to a configurable number of nodes (default: 3). The replicas form a Raft group for that range specifically. There is no global consensus mechanism; consensus is per-range. This is the core architectural decision that enables horizontal scalability: adding a node adds capacity for more Raft groups without increasing the coordination cost of existing ones.

Within a Raft group, one replica is the leaseholder (often co-located with the Raft leader). The leaseholder serves all reads and coordinates all writes for that range without requiring a round-trip to other replicas for reads, because it holds a range lease that guarantees it has seen all committed writes. Leases are time-bounded and must be renewed before expiry.

A write operation to a range goes through this sequence:

1. The leaseholder receives the write request.
2. It proposes a log entry to the Raft group.
3. Followers write the entry to their RocksDB WAL and acknowledge.
4. Once a quorum acknowledges, the leaseholder applies the entry to its RocksDB state machine and sends the response to the client.
5. Followers apply the entry asynchronously.

// Simplified representation of a write going through Raft
interface RaftLogEntry {
  index: number;          // monotonically increasing
  term: number;           // current leader term
  command: KVCommand;     // the actual write
}

interface KVCommand {
  type: "put" | "delete" | "conditionalPut";
  key: Uint8Array;
  value?: Uint8Array;
  expectedValue?: Uint8Array;  // for conditional writes
  timestamp: HybridLogicalClock;
}

// The leaseholder only responds after quorum ack
async function proposeAndWait(
  entry: RaftLogEntry,
  group: RaftGroup
): Promise<void> {
  const proposal = group.propose(entry);
  await proposal.quorumAcknowledged(); // n/2 + 1 nodes wrote to WAL
  group.applyToStateMachine(entry);    // local RocksDB commit
}

The RocksDB layer underneath each replica uses an LSM tree for all writes, which means Raft log entries translate to RocksDB batch writes. CockroachDB’s Pebble (its own Go-based RocksDB-compatible engine, now the default) provides the same LSM mechanics with better write amplification control.

MVCC and Hybrid Logical Clocks

CockroachDB’s storage engine is fully multi-version: every write stamps the KV entry with a timestamp rather than overwriting the previous value. Older versions are garbage collected by a background process. This MVCC model enables snapshot reads at a point in time without holding locks.

The challenge in a distributed system is clock synchronization. Two nodes with skewed clocks can produce timestamps that violate causality: a write on node A at time T1, followed by a read on node B that sees a timestamp T2 < T1, would miss the write. CockroachDB solves this with Hybrid Logical Clocks (HLC), a combination of wall clock time and a logical counter.

Every HLC timestamp is a pair (wall, logical). The rules are:

• When a node receives a message, it advances its wall component to max(local_wall, message_wall).
• If the wall components are equal, it increments the logical counter.
• When sending a message, the node includes its current HLC.

The result is a clock that tracks causality (if A happens before B, A’s HLC is less than B’s HLC) while staying anchored to real time within the configured max-offset (default: 500ms). If a node’s clock drifts beyond this bound, CockroachDB will reject operations from that node rather than risk violating consistency guarantees.

interface HybridLogicalClock {
  wallNanos: bigint;   // nanoseconds since Unix epoch
  logical: number;     // tie-breaker within the same wall time
}

function advance(local: HybridLogicalClock, received: HybridLogicalClock): HybridLogicalClock {
  const wall = local.wallNanos > received.wallNanos
    ? local.wallNanos
    : received.wallNanos;

  const logical =
    wall === local.wallNanos && wall === received.wallNanos
      ? Math.max(local.logical, received.logical) + 1
      : wall === local.wallNanos
      ? local.logical + 1
      : wall === received.wallNanos
      ? received.logical + 1
      : 0;

  return { wallNanos: wall, logical };
}

The Transaction Protocol: Write Intents and Timestamp Ordering

CockroachDB implements serializable isolation using a protocol based on write intents and timestamp ordering, without a two-phase locking approach for reads.

When a transaction writes a key, it does not write the final MVCC value immediately. Instead, it writes a write intent: an MVCC entry at the transaction’s provisional timestamp, tagged with the transaction ID. The actual transaction record, stored in a special system key, tracks the transaction status (pending, committed, aborted) and the commit timestamp.

When another transaction reads a key and finds a write intent:

• If the intent’s transaction is committed, the reader resolves the intent (finalizes the value) and continues.
• If the intent’s transaction is pending, the reader must either wait or push the intent’s timestamp forward.
• If the intent’s transaction is aborted, the reader cleans up the intent.

The push mechanism is key to liveness. When reader R1 encounters a write intent from writer W1, it contacts the transaction coordinator for W1 and requests a timestamp push. If W1 has lower priority (determined by transaction age, by default), W1’s timestamp gets bumped forward. W1 then has to refresh its reads: it must verify that no committed write exists between its original read timestamp and its new timestamp for each key it has read. If the refresh succeeds, the transaction continues with the new timestamp. If not, the transaction aborts and retries.

// Write intent structure stored in Pebble
interface WriteIntent {
  key: Uint8Array;
  timestamp: HybridLogicalClock;  // provisional commit timestamp
  txnID: string;                  // UUID of owning transaction
  txnKey: Uint8Array;             // key to the transaction record
}

// Transaction record stored at a system key
interface TransactionRecord {
  id: string;
  status: "pending" | "committed" | "aborted" | "staging";
  writeTimestamp: HybridLogicalClock;
  readTimestamp: HybridLogicalClock;
  intents: Array<{ key: Uint8Array; timestamp: HybridLogicalClock }>;
  epoch: number;  // incremented on restart
}

For multi-range transactions, CockroachDB uses a parallel commit optimization. Rather than running a traditional two-phase commit where phase 1 locks and phase 2 releases, it writes all intents in parallel and then marks the transaction as “staging” in a single atomic write. A staging transaction is considered committed if all its write intents exist at the declared timestamps. This allows the transaction to be considered committed in a single round-trip to the Raft groups holding the relevant ranges, with cleanup happening asynchronously.

DistSQL: Distributed Query Execution

When a SQL query touches multiple ranges, CockroachDB’s DistSQL engine builds a physical execution plan that pushes computation close to the data.

The gateway node (the node receiving the SQL query) parses and plans the query, producing a logical plan. The optimizer, based on Cascades-style rule application with a cost model, chooses join orders, index selections, and scan directions. The output is a physical plan: a DAG of processors connected by streams.

For a query like a filtered join across a large table:

1. The gateway identifies which ranges hold the relevant key spans.
2. It schedules TableReader processors on the leaseholder nodes for those ranges. Each TableReader scans its local Pebble store and applies the filter.
3. Results flow through HashJoiner or MergeJoiner processors, placed on intermediate nodes.
4. Final aggregation and sorting run on the gateway.

// Logical shape of a DistSQL flow (simplified)
interface DistSQLFlow {
  processors: Array<{
    id: number;
    type: "tableReader" | "hashJoiner" | "sorter" | "aggregator";
    nodeID: number;       // which CockroachDB node runs this
    inputs: number[];     // processor IDs feeding this one
    outputs: number[];    // processor IDs this feeds
    spec: ProcessorSpec;  // filter, join condition, sort key, etc.
  }>;
  resultRouterID: number; // final processor that sends rows to the client
}

The DistSQL engine uses a columnar vectorized execution path (the “vectorized engine”) for CPU-intensive operations like aggregations and joins, processing batches of 1024 rows at a time rather than one row at a time. This dramatically reduces per-row overhead for analytical queries, though OLTP point lookups bypass this path entirely.

Closed Timestamps and Follower Reads

A significant latency optimization for read-heavy workloads is follower reads: the ability to serve a read from any replica, not just the leaseholder. This avoids the latency of routing reads to the leaseholder, which may be in a different region.

Follower reads require that the serving replica’s state is consistent at the read timestamp. CockroachDB achieves this through closed timestamps: a mechanism where the leaseholder periodically broadcasts a closed timestamp to followers, indicating that no new writes will be accepted at or before that timestamp. A follower that has applied the Raft log up to a point covering the closed timestamp can serve reads at or before that timestamp without contacting the leaseholder.

The closed timestamp advances at a configured interval (default: 3 seconds). This means follower reads are available for timestamps at least 3 seconds in the past. Queries using AS OF SYSTEM TIME follower_read_timestamp() or the equivalent AS OF SYSTEM TIME '-4.8s' (a safe default) automatically route to the nearest replica.

-- Historical follower read (routes to nearest replica)
SELECT account_id, balance
FROM accounts
AS OF SYSTEM TIME follower_read_timestamp()
WHERE account_id = $1;

-- Useful in a global deployment where the leaseholder is in us-east
-- and the reader is in eu-west: serves locally with ~4s staleness

For global tables, CockroachDB supports pinning leaseholders per region using zone configurations, with follower reads providing low-latency historical reads in secondary regions while writes still round-trip to the primary leaseholder region.

Online Schema Changes

CockroachDB implements schema changes without blocking reads or writes on the affected table. This uses a multi-version schema approach inspired by the F1 paper from Google.

Schema changes move through a sequence of states: absent, delete-only, write-only, delete-and-write-only, public. At each transition, nodes across the cluster are given time to adopt the new schema version before the change advances. The key invariant is that at any moment, no two schema versions in active use across the cluster differ by more than one step.

For an ADD COLUMN operation:

1. Column enters delete-only: existing code ignores it; new writes see it but delete any values for it.
2. Column enters write-only: new writes include the column; reads do not yet return it.
3. A backfill job writes the default value for the new column across all existing rows.
4. Column enters public: all reads and writes use the new column.

The backfill job runs as a background scan over the table’s key range, writing batches of updates. It respects the cluster’s admission control to avoid saturating the range leaseholders. If the job fails partway through, it restarts from a checkpoint.

// Simplified schema change descriptor state machine
type SchemaElementState =
  | "absent"
  | "deleteOnly"
  | "writeOnly"
  | "deleteAndWriteOnly"
  | "public";

interface ColumnDescriptor {
  id: number;
  name: string;
  type: ColumnType;
  nullable: boolean;
  defaultExpr?: string;
  state: SchemaElementState;  // governs visibility across the cluster
  version: number;            // descriptor version at last state transition
}

DROP COLUMN runs the same state machine in reverse. The column moves from public to write-only (so new writes stop populating it) to delete-only (so old values get cleaned up during compaction) before being removed entirely. This means schema changes are always reversible until the final step completes.

Production Considerations

Write amplification from Raft and Pebble. Every write goes through Raft log replication (at least 3 writes across nodes), then into Pebble’s WAL, then into Pebble’s memtable, and eventually through compaction. For write-heavy workloads, monitor rocksdb.block.cache.hits, raftlog.size, and the Pebble compaction metrics in the DB console. Compaction stalls are a common source of write latency spikes under sustained write load.

Transaction retries. The timestamp advancement and read refresh protocol means transactions can fail with a TransactionRetryWithProtoRefreshError. Application code must wrap transactions in a retry loop. CockroachDB provides BEGIN; SAVEPOINT cockroach_restart; ...; RELEASE SAVEPOINT cockroach_restart; for explicit retry control in client applications.

Leaseholder placement. In multi-region deployments, every write to a range must reach the leaseholder. If your application is in us-west but leaseholders for your hot tables are in us-east, every write pays a cross-region round-trip. Use ALTER TABLE ... CONFIGURE ZONE USING lease_preferences to pin leaseholders to the region where writes originate, accepting that reads in other regions will either use follower reads (with staleness) or pay the round-trip.

Range count growth. A busy OLTP cluster with many tables accumulates tens of thousands of ranges. Each range carries Raft overhead: leader elections, heartbeats, lease renewals. Monitor ranges.unavailable, ranges.underreplicated, and replicas.leaders per node. If ranges are concentrated on a few nodes, the rebalancer queue metrics will show pending rebalances. Tune kv.snapshot_rebalance.max_rate to control how aggressively rebalancing uses network bandwidth.

Clock synchronization. The HLC max-offset (default 500ms) is the maximum permitted clock skew. If NTP drift on any node exceeds this, that node removes itself from the cluster. Running timedatectl status or chronyc tracking on each node to confirm sub-100ms offset is standard operational hygiene in production CockroachDB deployments.

Tradeoffs Comparison

Dimension	CockroachDB	Cloud Spanner	TiDB	YugabyteDB	Vitess
Consistency model	Serializable (SSI)	External consistency (Linearizable + Serializable)	Serializable (Percolator)	Serializable (SSI)	Read committed per shard
Clock mechanism	HLC (software, NTP-dependent)	TrueTime (GPS + atomic clocks)	TSO service (PD node)	HLC (software, NTP-dependent)	N/A (MySQL clocks)
Replication unit	Range (512 MB, Raft per range)	Directory (8 GB, Paxos per directory)	Region (96 MB, Raft per region)	Tablet (Raft per tablet)	MySQL replication (per shard)
Horizontal writes	Auto-split ranges, no resharding	Auto-split directories	Auto-split regions	Auto-split tablets	Manual sharding config
HTAP	Limited (vectorized engine for analytics)	No native columnar (use BigQuery integration)	Yes (TiFlash columnar learner replica)	Limited	No
Wire protocol	PostgreSQL	Proprietary (client libraries required)	MySQL	PostgreSQL	MySQL
Self-hosted	Yes	No (GCP only)	Yes	Yes	Yes
Schema changes	Online, multi-version (F1 approach)	Online, phased with backfill	Online (F1 approach)	Online	DDL per shard, not atomic
Geographic distribution	Multi-region zone configs, survival goals	Global by default	Placement rules per region	Multi-region tablespaces	Manual cross-region sharding
Sweet spot	Global OLTP, PostgreSQL teams, multi-region survivability	GCP-native global OLTP, financial workloads	MySQL shops needing horizontal write scale with HTAP	PostgreSQL teams, latency-sensitive global OLTP	MySQL shops adding read scale without re-architecture

Where This Architecture Fits

CockroachDB’s architecture is a good match for workloads where global write scalability, multi-region survivability, and strong consistency are genuinely required at the same time. The Raft-per-range model means you are running thousands of independent consensus groups, which is efficient compared to a single global consensus mechanism but adds operational surface area that single-node or sharded systems do not have.

The HLC-based transaction protocol provides serializable guarantees without hardware clock support, which makes it deployable anywhere. The cost is that clock skew above 500ms causes node self-exclusion, so NTP discipline is a hard operational requirement rather than a nice-to-have.

For workloads that are primarily OLTP with occasional analytical queries on recent data, the closed timestamp follower reads and the vectorized execution path provide a useful escape valve. For workloads where heavy analytics on live data is central, a dedicated columnar store like TiFlash in TiDB or a separate OLAP system is worth the architecture complexity.

The distributed SQL model means query planning and physical execution are genuinely distributed, which eliminates single-node query bottlenecks but introduces distributed execution overhead for queries that are fundamentally point lookups. Benchmark both simple KV patterns and complex join patterns before committing to a deployment topology.