← Back
Home › Systems › Databases and storage engines

distributed-sql

Presents one relational database with transactions across many machines, built on a sorted key-value map, and keeps working when some machines fail.

Spanner, CockroachDB

Open the distributed-sql pagedistributed-sql Bloom filterdata-structures: Bit array answering "definitely not present" or "probably present" in about 10 bits per key for 1 % false positives; inserts are cheap, deletes impossible.›Hdr histogramdata-structures: Log-linear bucketed histograms with bounded relative error for latency percentilesClock synchronizationdistributed: Estimate a clock's offset from four timestamps, filter out queueing noise, and steer the clock smoothly; NTP gets about a millisecond, PTP with hardware timestamps under a microsecond.›Data shuffledistributed: Hash/range repartitioning of intermediate data between workers (exchange operator)Deterministic simulationtestingdistributed: Run the whole system in one thread under a seeded scheduler with simulated network, disk and clock, so every failure replays from its seed.›Hybrid logical clocksdistributed: A 64-bit timestamp that tracks wall-clock time but never goes backwards and respects causality; the default timestamp for MVCC databases without atomic clocks.›Leasesdistributed: A lock that expires on its own: bounded-time ownership that survives crashes and partitions, given bounded clock drift and fencing tokens on writes.›Multi raftdistributed: Thousands of Raft groups per node: batched heartbeats, shared log engine, range split/mergePrimary-backup replicationdistributed: One node orders all writes and ships them to standbys; the acknowledgement rule (none, one, all) sets what a crash loses, and an external arbiter decides who takes over.›Raftdistributed: Leader-based consensus on a replicated log; one leader per term, majority commit; the default for a strongly consistent replicated state machine.›Range partitioningdistributed: Cut the sorted key space into contiguous half-open ranges kept in a boundary map; split hot or big ranges, merge small ones, move ranges between nodes.›SWIM gossip membershipdistributed: Each node probes one random peer per period, asks k others on timeout, and piggybacks membership updates; constant load per node at any cluster size.›Truetime commit waitdistributed: Commit timestamps from bounded-uncertainty clocks plus commit wait for external consistencyTwo-phase commitdistributed: Atomic commit across shards: participants durably vote yes, then the coordinator logs and broadcasts one decision; blocks if the coordinator is lost.›Process supervisionhardware-os: A parent process starts services, reaps their exits and restarts them under a delay and burst-limit policy, then stops them with SIGTERM before SIGKILL.›Coverage-guided fuzzingsecurity: Mutate inputs, run an instrumented target, and keep any input that reaches a new code edge; feedback turns blind guessing into incremental search.›Mutual TLSsecurity: TLS where the client also proves a certificate key, so every connection carries a verified, rotatable machine identity.›TLS 1.3security: The standard encrypted, authenticated transport; one round trip, only forward-secret AEAD suites, and little left to misconfigure.›Cardinality estimationstorage: Row-count estimates from histograms, MCVs, sketches and sampling for the optimizerExpand contract migrationstorage: Multi-step schema changes (expand, migrate, contract) that keep old and new code workingLSM compactionstorage: Compaction policies (leveled, tiered, time-window, FIFO) decide which sorted files to merge and when, trading write, read and space amplification and purging tombstones.›LSM-treestorage: Buffer writes in memory, flush them as immutable sorted files and merge in the background; turns random writes into sequential I/O.›MVCCstorage: Keep timestamped versions of each row so readers see a consistent snapshot without blocking writers; the default concurrency control in modern databases.›Point-in-time recoverystorage: Periodic base backups plus a continuously archived change log let you rebuild the state as of any instant, then branch from there.›Query optimizerstorage: Cost-based plan search: join ordering (DP/greedy), rewrites, access path selectionSerializable snapshot isolationstorage: SSI: detect rw-antidependency cycles on top of snapshot isolation to get serializabilityGraceful drainingsystems: On a termination signal, stop taking new work, finish in-flight work until a deadline, tell clients to reconnect elsewhere, then exit; clients resume from a token.›PID controllerrobotics: Drive a measured value to a setpoint with three terms on the error: proportional, integral, derivative. The default feedback loop for motors, heaters, drones and gimbals.›Phase-locked loop (digital)signal-processing: Feedback loop that locks a local oscillator to the phase and frequency of an input tone; the core of carrier recovery, clock recovery and GNSS tracking.›Robust statisticsstatistics: Location and scale estimates (median, MAD, trimmed mean, Huber M-estimator) that stay correct when a minority of the data is garbage.›External merge sortsearch-sort: Sort data larger than RAM: write memory-sized sorted runs, then k-way merge them with large sequential buffers; terabytes in one merge pass.›Columnar memory formatstorage: In-memory column buffers with validity bitmaps and offsets (Arrow style) for zero-copy exchangeCooperative schedulingconcurrency: Green threads/coroutines yielding at await points on a user-space schedulerFault injectiondistributed: Inject latency, errors, kills and partitions via proxies, netem or hooksRecord replaydistributed: Record every nondeterministic input (time, randomness, I/O, model and tool outputs, per-tick player input) and re-execute deterministically for debugging, replays and state rebuildFencing tokensdistributed: A number that grows with every lock grant and is checked by the protected resource, so a paused ex-holder's late write is rejected.›Group commitstorage: Let concurrent committers share one log write and fsync; throughput grows with concurrency while each commit still waits at most about two fsyncs.›Lamport clocksdistributed: One integer per process, max-plus-one on receive; gives a total order consistent with causality using 8 bytes per timestamp.›Health checkingsystems: Decide per backend whether to route to it using probes and real-traffic errors, with consecutive-count hysteresis, capped ejection and a panic fallback.›+ 34 moreBuffer pools forzero-allocation I/Ohardware-os: Preallocate a fixed set of aligned I/O buffers and recycle them through a free list, so the hot path never mallocs, faults or unmaps. The bounded count doubles as back pressure.›Pattern-defeating quicksort(pdqsort)search-sort: Introsort successor: quicksort with adaptive pivots, linear on sorted and equal-heavy input, heapsort fallback; the unstable sort of Go and (until 1.81) Rust.›Transparent interceptionnetworking: Redirect traffic to a local proxy (iptables TPROXY/REDIRECT, eBPF) without app changesSyscall interceptionsecurity: Intercept syscalls via ptrace, seccomp user notify or a user-space kernelFutex-based mutexconcurrency: A mutex that is one atomic word in user space and enters the kernel (futex wait/wake) only under contention; how std and glibc locks work.›+ 15 moreMemory orderingconcurrency: What Relaxed, Acquire, Release and SeqCst atomics actually guarantee, what they cost on x86 and ARM, and which one each pattern needs.›Ebpfhardware-os: Verified in-kernel programs on hooks (kprobes, XDP, tc) with mapsPtrace process controlhardware-os: Ptrace: attach, read and write registers and memory, single-step, intercept signalsSeccomp bpfsecurity: Syscall filtering with seccomp BPF programs+ 6 more+ 1 more

How it is built

Architecture overview

A distributed SQL database presents one relational database with transactions over many machines, and keeps working when some of them fail. Google Spanner is the original; CockroachDB, TiDB and YugabyteDB are open designs in the same family. The idea that makes it work is a layering: SQL tables are encoded as rows in one huge sorted key-value map; the map is cut into contiguous ranges (TiKV calls them Regions, Spanner and YugabyteDB use tablets and directories); each range is a small replicated state machine run by a consensus group (Raft or Paxos), so its writes survive the loss of a minority of replicas; and a transaction layer with timestamped versions makes operations that touch several ranges atomic and consistently ordered. SQL planning sits on top and storage engines sit underneath. Everything interesting is in the seams: how a range is found, who may serve a read, and where the commit timestamp comes from.

Parts:

 client ──SQL──► any node = gateway ───────────────────────────────────────┐
                 parse → optimise → plan (rows/indexes → KV ops)            │
                          │                                                 │
                          ▼  KV batches, routed by key                      │
              range lookup (cached 2-level index)                           │
                          │                                                 │
        ┌─────────────────┼─────────────────┐                               │
        ▼                 ▼                 ▼                               │
   range A leader    range B leader    range C leader   ◄── 2PC / intents ──┘
   (consensus group) (consensus group) (consensus group)   transaction record + timestamps
    │ log │ log │      │ log │ log │
    ▼     ▼     ▼      ▼     ▼     ▼
   replicas: each node holds many ranges, one LSM store per node/disk (WAL + SST files)

How the parts fit together

SQL layer

Every node can accept SQL: in CockroachDB the node a client connects to is the gateway, it parses the statement, builds a logical plan, runs a cost-based optimiser, then plans physically across nodes and runs the plan by issuing key-value requests (source: CockroachDB docs, SQL layer). TiDB makes the split explicit: its SQL servers are stateless, "only for computing and SQL analyzing", and the data lives in TiKV (source: TiDB docs, Architecture). The planning blocks are the same as in any database (query-optimizer (no page yet) with relational-algebra (no page yet), cardinality-estimation (no page yet), join-order-dynamic-programming (no page yet) against the alternative of a Cascades-style search, subquery-decorrelation (no page yet)) with one addition that only distribution needs: predicate-pushdown (no page yet) and partition-pruning (no page yet) decide how much work can be done next to the data on the node that owns it, because moving rows across the network is the expensive step. Equi-joins that cannot be pushed down are run as a Hash join after a data shuffle (data-shuffle (no page yet)): rows are repartitioned between workers by Fast hash functions of the join key, or by range when a sorted result is needed (External merge sort). The contract between the SQL layer and everything below is a key-value interface: SQL never sees a replica, only keys, timestamps and batches.

For TiDB, table data and index data get key prefixes derived from the table and index ids, so all rows of a table sit together in the key space (source: TiDB docs, Computing). That is what makes a range scan of a table a range scan in the map, and what lets splits follow table boundaries.

Distribution: ranges and how to find them

The key space is one sorted map, cut into contiguous half-open ranges (Range partitioning). This is the alternative to hash partitioning, chosen because ordered data keeps range scans and index lookups local; the cost is hot spots on sequential keys, which splitting and load-based rebalancing have to handle. CockroachDB describes it as a "monolithic sorted map" where every key lives in exactly one range, and a range splits when it grows past its maximum (source: CockroachDB docs, Distribution layer). TiKV's Regions are left-closed, right-open key intervals with a default size limit of 256 MiB (source: TiDB docs, Storage). A directory of boundaries is needed to route a request: CockroachDB keeps a two-level index in the key space itself (meta ranges), caches it on every node, and distributes the location of the first level by gossip; the docs' arithmetic for the address space assumes 2^26-byte ranges (source: the distribution-layer page). TiDB instead has a central service, PD (Placement Driver), which stores the region map and issues scheduling commands (source: TiDB docs, Architecture). The trade is a lookup that is cached and decentralised against a simple, central brain that can become a dependency.

Which nodes are alive is a separate question. CockroachDB's docs name gossip for distributing some cluster metadata; SWIM gossip membership is the failure-detecting variant (probe, indirect probe, piggybacked updates), with a phi-accrual detector as the alternative. Health checking and Graceful draining let a node being restarted hand off its leases and connections first.

Replication: one consensus group per range

Each range is replicated by its own consensus group, so a node runs thousands of groups at once. That arrangement is multi-raft (no page yet): heartbeats are batched across groups so that idle ranges cost almost nothing, the groups share a log engine, and ranges split and merge without a reconfiguration of the whole cluster. The protocol is Raft in CockroachDB, TiKV and YugabyteDB, and Paxos in Spanner; Multi-Paxos is the closest alternative. CockroachDB's docs give the quorum arithmetic: three is the smallest count that can lose one replica and still reach quorum, so high availability needs three nodes (source: CockroachDB docs, Replication layer). A write is a command proposed to the leader, appended to the log, replicated to a majority and applied; the log is a Write-ahead log in all but name, and Group commit lets many proposals share one log write and fsync. Spanner's paper notes that its Paxos leaders hold timed leases of 10 seconds by default and that every Paxos write is logged twice, once in the tablet's log and once in the Paxos log (source: Spanner, OSDI 2012).

A lease (Leases) is how reads avoid consensus. In CockroachDB one replica, the leaseholder, serves reads and proposes writes; the docs say it is also kept as the Raft leader, and that reads at the leaseholder bypass Raft because the data was already committed by consensus (source: replication-layer page). A lease only works with bounded clock drift, and a deposed leaseholder that does not know it yet is the classic hazard that Fencing tokens address. Lagging or new replicas catch up from the log, or, if the log has been truncated, from a snapshot of the range (State-machine snapshots). Retried RPCs use Exponential backoff with jitter. Non-voting replicas, which CockroachDB documents, serve reads in far regions without adding to write latency, because they are outside the quorum.

Storage engine

Each node stores its replicas in a local ordered key-value engine; CockroachDB uses Pebble, TiKV uses RocksDB, and Spanner stored tablet state in B-tree-like files plus a write-ahead log on its distributed file system (sources: CockroachDB docs, Storage layer; TiDB docs, Storage; Spanner paper). The usual pick is the LSM-tree with LSM compaction, sstable (no page yet) files and per-file Bloom filters, and the alternative is the B+ tree. The reasons are workload-driven: replication already makes every write sequential through the Raft log, so an engine that also turns random updates into sequential file writes fits, and MVCC versions (below) are naturally append-only. The lsm-engine system page shows that engine as a system of its own. Contract: the replicated state machine applies committed commands to the engine in log order, and the engine returns versions by timestamp. Block checksums (CRC32C checksums) catch silent corruption, which consensus alone does not.

Transactions: versions, intents and atomic commit across ranges

All of these systems store several versions per key, stamped with a timestamp (MVCC); CockroachDB states that its MVCC values are stored in Pebble and used throughout the transaction layer (source: storage-layer page). A reader at timestamp t sees the newest version at or below t and never blocks on writers, which is also what makes historical and follower reads possible. For writes that span ranges the shape is Two-phase commit, over replicated participants so a coordinator crash cannot stall the system for ever. Spanner's participant leaders coordinate two-phase commit, with state stored in the replicated groups, and read-write transactions take locks in a lock table at each leader (source: Spanner paper). CockroachDB writes write intents, provisional values that point to a transaction record on the range holding the transaction's first key; committing flips the record and the intents are resolved lazily, and "Parallel Commits" lets a transaction be treated as committed once its intents are replicated (source: transaction-layer page). TiKV exposes snapshot isolation by default at the key-value level (source: TiDB docs, Architecture). The alternative is serializable-snapshot-isolation (no page yet), which detects dangerous read-write dependency cycles on top of snapshot isolation, as the route to serializability (an Interval tree for overlapping key ranges and a Swiss table for fast lock and intent lookup support it); it sits against Two-phase locking and Optimistic concurrency control. Isolation levels differ per system and per version, and that should be checked in each product's documentation.

Time

Commit order across nodes needs a clock that is comparable across nodes. Spanner's answer is TrueTime: the API returns an interval guaranteed to contain absolute time, built from GPS and atomic clocks, with uncertainty generally below 10 ms and a typical bound of about 4 ms (the paper reports a sawtooth between 1 and 7 ms). A transaction picks its commit timestamp and then waits until that time has definitely passed, commit wait, so that any transaction that starts after it commits gets a larger timestamp: external consistency (source: Spanner paper; truetime-commit-wait (no page yet)). The price is the wait, equal to the clock uncertainty. Hardware time references are not available everywhere, so the open systems use Hybrid logical clocks, which combine a physical component with a logical counter; CockroachDB and YugabyteDB both do, and YugabyteDB's docs say the maximum clock skew influences conflict resolution and so the number of conflicts and the latency (sources: CockroachDB docs, Transaction layer; YugabyteDB docs, Transactions overview). TiDB's PD allocates transaction ids centrally, trading a shared service for no dependence on clock quality. All of these rest on Clock synchronization (offset estimation, outlier filtering) and on ordering rules like Lamport clocks. The choice between TrueTime-with-wait, HLC-with-uncertainty and a timestamp oracle is the central design decision of the family.

CockroachDB also uses closed timestamps: the leaseholder promises that no new writes will arrive at or below a timestamp that trails the present, and followers can serve reads at or below it once they have applied the log (source: transaction-layer page). This is how follower reads give stale-but-consistent data from the nearest replica.

Operations, change and recovery

Schema changes must not stop the world: add the new structure, backfill existing rows in bounded chunks (Chunked backfill), switch, then remove the old one (expand-contract-migration (no page yet)). Backups combine periodic base copies with the archived change log on cheap storage (Point-in-time recovery, Object storage); on one node, crash recovery follows ARIES recovery, while across nodes the Raft log and snapshots do that job. Traffic is encrypted and mutually authenticated (TLS 1.3, Mutual TLS). Consensus and clock bugs do not show up in ordinary tests, so the system is run in a deterministic simulator with seeded fault injection (Deterministic simulation testing, fault-injection (no page yet)) and fuzzed (Coverage-guided fuzzing).

Raft keeps the replicas of one range consistent inside a cluster, but a whole-cluster loss (a region outage, an operator error that deletes the cluster) needs a copy outside it. Primary-backup replication is that second layer: the live cluster is the primary and orders all writes, and it ships its committed change stream to a standby cluster in another region. The acknowledgement rule is chosen deliberately, because waiting for the standby on every commit adds a cross-region round trip while not waiting means a disaster loses the last few seconds of writes; most deployments accept the loss window. Promotion of the standby is decided by an external arbiter (an operator or a separate witness), never by the two clusters themselves, so a network split cannot leave two primaries. The standby's change stream is also the source for point-in-time restores.

Data and control flow

A single-range write (UPDATE accounts SET balance = balance - 10 WHERE id = 7).

  1. The gateway parses and plans the statement. The optimiser chooses the primary-key lookup. The row key is computed.
  2. The cached range index maps the key to a range and its leaseholder. On a miss the index is read through the meta ranges (or PD), which takes more round trips.
  3. The request goes to the leaseholder. It reads the latest version at the transaction's timestamp, checks for conflicting intents and writes a new version.
  4. The write is proposed to the Raft group, appended to the leader's log, sent to the followers, and applied once a majority has acknowledged. This costs one round trip to the nearest majority plus the log fsync on each member; across regions that round trip dominates, so replicas are placed to keep a majority close to the leader.
  5. The gateway receives the acknowledgement and answers the client.

A multi-range transaction (transfer between two accounts on different ranges).

  1. The transaction gets a timestamp from the HLC (or TSO, or TrueTime).
  2. It writes provisional versions on each range's leader. Each write is replicated by that range's group (step 4 above).
  3. At commit, the coordinator (gateway or participant leader) records the decision in the transaction record, which is itself replicated, so a coordinator crash does not lose the decision.
  4. Commit is acknowledged after the decision is durable. With commit wait (Spanner) the acknowledgement is further delayed by the clock uncertainty. Intents are then converted to committed versions in the background.
  5. A reader who meets an unresolved intent consults the transaction record and either waits, helps resolve, or aborts it.

Scaling

The first limit is single-range throughput, because one range's writes go through one leader and one log. Levers, in order: split hot ranges and move leaders, choose keys that do not append at the end of the key space, and use Group commit to batch log writes. The second limit is cross-range transactions: every extra range adds a consensus round and widens the failure window of two-phase commit, so schema design that keeps a transaction on one range (interleaving or co-locating, partitioning by tenant) is the strongest lever. The third limit is geography: write latency is the round trip to the nearest majority, plus commit wait where it applies, so place voters in a few nearby zones, add non-voting or follower replicas for far readers, and use stale reads where the application allows.

Failure modes

FailureSymptomCauseFix
Node crashBrief latency rise for its rangesLeaders lost; lease and election timeouts must passShort election/lease settings within clock drift; three or more replicas per range; drain nodes before planned restarts
Majority lossRanges unavailableTwo of three replicas downRestore the nodes; recovery from a lone surviving replica gives up safety and is a last resort
Clock drift beyond boundReads restart or conflicts rise; in the worst case violated lease or ordering assumptionsHLC and lease reasoning assume bounded skew (Yugabyte documents skew effects on conflicts)Run time sync with monitoring; keep the configured maximum offset above the real one
Transaction contentionHigh retry rate, restart errorsMany writers on the same rows, long transactions, closed timestamp pushing writesShorter transactions, retries with backoff, tune isolation, avoid hot rows

Real implementations

namewhat it isrepolicencenotes
SpannerGoogle's globally distributed relational database: Paxos groups, TrueTime, two-phase commit— (not open source; paper)—The reference design; OSDI 2012
CockroachDBDistributed SQL on a sorted KV map, Raft per range, HLC, Pebble storagecockroachdb/cockroachCockroachDB Software License (LICENSE read; source-available, not OSI open source)Its architecture docs are the best public walkthrough of the layers
TiDBMySQL-compatible SQL layer over TiKV (Raft + RocksDB), PD scheduler, optional TiFlash columnar replicaspingcap/tidb, tikv/tikvApache-2.0 (LICENSE read)Stateless SQL servers, central timestamp and placement service
YugabyteDBPostgreSQL-compatible YSQL on DocDB with Raft per tablet and hybrid timeyugabyte/yugabyte-dbApache-2.0 for the core (README)Leader preferences by region; follower reads are timeline-consistent

Sources

Similar systems

Share the most building blocks with this one.

In-house systems

Named systems of this kind at large companies, with their paper or write-up.