Design a Distributed OLTP Database

A distributed OLTP database needs to do what a single-node PostgreSQL does (ACID transactions, SQL queries, sub-10ms latency) but across dozens of nodes in multiple regions. The challenge is not SQL support. It is maintaining consistency guarantees when data is split across machines that fail independently and communicate over unreliable networks. Think of it this way: a single-node database commits a transaction by writing one WAL entry to one disk. A distributed database must coordinate that same guarantee across three continents.

Functional

ACID transactions across distributed nodes: Every transaction is atomic (all-or-nothing), consistent, isolated, and durable. This holds even when a single transaction touches data on multiple shards. A bank transfer from account A (shard 1) to account B (shard 3) must either complete fully or not at all, even if a node crashes mid-way.

OLTP workloads are defined by concurrency, not storage. 10,000 app servers with 50 connections each means 500,000 concurrent connections. At 100,000 TPS peak, each transaction touching 5 to 10 rows, the system executes up to 1 million row operations per second. Storage is modest (roughly 1TB per year) but throughput and latency drive every architectural decision.

Connections and Throughput

Concurrent connections: 500,000 total from application servers. Connection pooling at the gateway multiplexes them onto roughly 10,000 backend connections. Without pooling, each TCP connection consumes approximately 10KB of kernel buffer space plus thread context, 500K connections would require 5GB of memory just for connection state.

A distributed OLTP database exposes two interfaces: a SQL wire protocol for developers (PostgreSQL-compatible) and an admin API for operators. The wire protocol is the critical path; every microsecond of overhead multiplied by 100K TPS matters. The admin API handles rare operations where latency is irrelevant.

SQL Wire Protocol (PostgreSQL-Compatible)

1BEGIN;
2SELECT balance FROM accounts WHERE id = 'acc_123';
3UPDATE accounts SET balance = balance - 100 WHERE id = 'acc_123';
4UPDATE accounts SET balance = balance + 100 WHERE id = 'acc_456';
5COMMIT;

The storage engine is where distributed and single-node databases diverge most. A single-node database has one WAL, one B-tree, one buffer pool. A distributed OLTP database has a WAL per Raft group, an LSM-tree per shard, and a distributed transaction log coordinating across all of them. The design decisions here directly determine write throughput, read latency, and space amplification.

Core Schema

1CREATE TABLE system_catalog.shard_map (
2shard_id UUID PRIMARY KEY,
3key_range_start BYTEA NOT NULL,
4key_range_end BYTEA NOT NULL,
5raft_group_id UUID NOT NULL,
6leader_node_id UUID NOT NULL,
7replica_count INT DEFAULT 3
8);
9
10CREATE TABLE system_catalog.transactions (
11txn_id UUID PRIMARY KEY,
12status TEXT NOT NULL,
13involved_shards UUID[] NOT NULL,
14start_ts BIGINT NOT NULL,
15commit_ts BIGINT,
16coordinator_id UUID NOT NULL
17);

Three layers with strict separation of concerns. The Gateway layer parses SQL and routes queries. The Coordination layer manages distributed transactions via 2PC. The Storage layer persists data in Raft-replicated shards. This separation matters because each layer scales independently: you can add gateway nodes to handle more connections without adding storage nodes, and vice versa.

Gateway Layer (Stateless)

Two flows, radically different cost. A single-shard read returns in under 5ms. A cross-shard write requires 2PC, Raft consensus on each shard, and WAL persistence, adding up to at least 4 network round-trips. This difference is why data placement (keeping related data on the same shard) is the single biggest performance optimization.

Single-Shard Write

Two transactions on different continents both update the same bank account. Without coordination, one overwrites the other. The GATE is the interplay of three mechanisms: 2PC for atomicity across shards, Raft for durability within each shard, and MVCC for isolation between concurrent transactions. Each solves one dimension of the distributed ACID problem. Remove any one and the system breaks in a specific, predictable way.

Two-Phase Commit (2PC): Cross-Shard Atomicity

Every architectural decision in a distributed OLTP database trades latency for correctness, throughput for isolation, or simplicity for flexibility. For financial OLTP, correctness wins every trade-off because the cost of losing or corrupting a single transaction (regulatory fines, customer trust) exceeds the cost of any performance optimization.

Synchronous vs. Asynchronous Replication

The three hardest failures are coordinator failure during 2PC (locks held indefinitely), network partitions (split-brain risk), and clock skew (MVCC timestamps become unreliable). Node crashes are comparatively easy: Raft handles them automatically. The dangerous failures are the ones that are invisible or that interact with the consistency model.

Node Failure Mid-Transaction

The Raft group elects a new leader within 1 to 3 seconds. In-flight transactions on the failed leader are rolled back via WAL undo logs on the new leader. Committed data is safe because it was already replicated to a majority before acknowledgment. Clients retry against the new leader, which the gateway discovers by querying the shard map for the updated leader. The impact is a brief latency spike (1-3 seconds of unavailability for that shard) and aborted in-flight transactions that must be retried.

The system handles 100K TPS with strong consistency across 3 regions. These improvements target the next order of magnitude in scale, efficiency, and operational simplicity.

Active-Active Multi-Region Writes

Currently, cross-region writes route to a single leader per shard, adding inter-region latency to every write. The Calvin protocol offers an alternative: a sequencer assigns a global ordering to all transactions before execution. Each region executes the same deterministic sequence, reaching the same state without per- transaction consensus. The trade-off: a 10ms batching delay for the sequencer to collect transactions, but this eliminates per- transaction cross-region round-trips (200-300ms). Calvin suits workloads with many cross-shard transactions but adds complexity in handling non-deterministic operations (current time, random values).