Calvin: Fast Distributed Transactions for Partitioned Database Systems

In one breath — the whole paper, compressed

Distributed transactions were expensive not mainly because of CPU or bandwidth, but because two-phase commit forces every participant to hold its locks across several network round trips, so a handful of hot records can throttle an entire cluster. Calvin removes that cost by doing all agreement before any lock is taken: a distributed sequencing layer batches incoming transaction requests into 10 ms epochs, replicates those inputs (asynchronously or through Paxos), and stamps out one global serial order that every replica follows. A partitioned deterministic lock manager then grants locks strictly in that order, and worker threads execute in five phases with one-way forwarding of remote read results, so nondeterministic events such as a node crash can never force an abort and no commit protocol is needed. Because only inputs are logged and replayed, physical REDO logging disappears and checkpoints can be taken asynchronously against a virtual point of consistency in the global order. On 100 EC2 nodes Calvin ran close to half a million TPC-C New Order transactions per second, competitive with the then-current Oracle world record of 504,161 obtained on far higher-end hardware.

Before this paper — the world it landed in

By 2012 the dominant answer to scaling was to give up transactions: Dynamo, MongoDB, CouchDB and Cassandra offered none, Bigtable offered single-row updates, and Azure, Megastore and the Oracle NoSQL Database restricted transactions to small predeclared subsets of the database. Systems that kept full ACID, such as VoltDB, did so by ceasing or limiting concurrent execution whenever a transaction spanned partitions. Anyone who really needed multi-partition atomicity fell back on the System R* design from the 1980s, where every participant runs two-phase commit at commit time while holding all of its locks. A parallel trend had weakened replication consistency in the name of CAP, though that was already reversing as Megastore and IBM's Spinnaker adopted synchronous Paxos replication. Calvin's authors had also built earlier deterministic prototypes, but those depended on a single-node sequencer and assumed the whole database fit in main memory.

The problem — what was actually breaking

System R*-style distributed transactions require an agreement protocol among all participants at commit time, and isolation demands that every lock be held for the full duration of that protocol.
Two-phase commit costs multiple network round trips between all participating machines, so the protocol frequently takes far longer than executing the transaction logic itself.
The paper names the resulting quantity the contention footprint, and when a few popular records are repeatedly touched by distributed transactions the extra lock-hold time on those records devastates overall throughput.
Allowing distributed transactions under pessimistic concurrency control also introduces distributed deadlock, whose detection forces aborts and restarts that add latency and shave throughput.
Because these costs are so severe, most scalable systems simply removed transactional support, leaving application programmers to hand-roll atomicity and isolation with complex, slow client-side scheduling.
Previous deterministic database prototypes used a single-node sequencer implemented as an echo server, which was both a single point of failure and a fixed throughput ceiling, and they only worked for databases resident entirely in main memory.

Core ideas — the contributions, and why they work

Agree before locking

Calvin's guiding move is to perform all inter-machine agreement outside transactional boundaries, before locks are acquired and execution begins. Once the machines have agreed on which transactions to run and in what order, that plan is binding: node failure and related nondeterministic problems cannot cause an abort, because a failed node can be replaced by a replica running the identical plan or can replay the plan history. Agreement therefore stops being a tax paid while locks are held and becomes a preprocessing step whose latency never enters the contention footprint. This is why Calvin can pay for Paxos across continents without losing any transactional throughput.

Determinism eliminates two-phase commit

Two-phase commit exists mainly to detect that some participant could not commit locally, and the causes split into nondeterministic events such as hardware failure and deterministic ones such as transaction logic aborting on zero inventory. If an active replica is executing exactly the same transaction sequence in parallel, other nodes never have to wait for a crashed node to recover, so nondeterministic failure no longer justifies aborting. Deterministic aborts do not need a full agreement protocol either: each node simply waits for a one-way message from every node whose logic could deterministically abort, then commits. Removing the round trips shrinks each distributed transaction's contention footprint to roughly the time its logic actually runs.

Deterministic locking with real concurrency

Earlier proposals for deterministic execution ran transactions serially in a single thread per node, which throws away throughput whenever a transaction stalls. Calvin instead keeps a lock manager that behaves like strict two-phase locking plus two extra invariants that pin conflict resolution to the sequencer's order. Because conflicting transactions can only be granted locks in the agreed order, and nonconflicting ones proceed freely, the execution is logically equivalent to the global serial order while many transactions run concurrently on a pool of worker threads. Concurrency is preserved; only the resolution of conflicts is made deterministic.

Replicate inputs, not effects

Because the schedule is deterministic, two replicas fed the same input sequence traverse the same sequence of database states, so Calvin replicates batches of transaction requests rather than the writes they produce. This makes strongly consistent replication dramatically cheaper: Megastore and Spinnaker must use Paxos on transactional effects, while Calvin only has to Paxos-agree on inputs. It also means the choice of replication mode is a pure latency decision — asynchronous master-slave, Paxos within a data center, or Paxos across continents — with no effect at all on throughput. Actively replicated nodes double as the failover mechanism that makes abort-free execution possible.

Unbundled sequencer, scheduler, storage

Calvin splits the system into a sequencing layer that fixes the global input order and handles replication and logging, a scheduling layer that orchestrates locking and execution, and a storage layer that owns physical layout behind a plain CRUD interface. All three layers are partitioned horizontally across shared-nothing nodes, and any storage engine with a CRUD interface can be plugged in. The price of this decoupling is that logging and concurrency control must be purely logical, referring to record keys and never to pages or index structures. Determinism is what makes that price affordable, since logical input logging fully determines database state.

Move the heavy lifting before the locks

Determinism normally hurts disk-based databases because a stalled transaction blocks everything ordered behind it, whereas a traditional system could reorder around the stall. Calvin applies its design principle instead: when the sequencer sees a request likely to incur a disk stall it inserts an artificial delay before forwarding the transaction to the scheduler and simultaneously asks the storage components to warm up the records. If the delay covers the fetch, the transaction touches only memory-resident data when it finally executes, so end-to-end latency is no worse than doing the I/O inline while none of the disk time lands inside the contention footprint. The same principle drives OLLP, which resolves unknown read/write sets before the transaction enters the sequence.

How it works — the mechanism, concretely

Sequencing layer: epochs and batches

The sequencing layer is distributed across all replicas and partitioned across every machine within each replica, removing the single-node echo server of earlier prototypes. Time is divided into 10-millisecond epochs; during an epoch each machine's sequencer accumulates client transaction requests, and at the epoch boundary compiles them into a batch that is then replicated. After replication succeeds, the sequencer sends each scheduler in its replica a message carrying the sequencer's node ID, the epoch number (incremented synchronously system-wide every 10 ms), and only those transaction inputs that recipient must participate in. Each scheduler reconstructs its own view of the global order by interleaving all sequencers' batches for that epoch in a deterministic round-robin manner.

Replicating transactional input

Nodes are organized into replication groups, each holding all replicas of one partition. In asynchronous mode one replica is master, requests go straight to its sequencers, and each master forwards its compiled batch to the slave sequencers in its group; latency is minimal but failover is complex, since survivors must agree on which batch was the last valid one and exactly what it contained, given that each scheduler only ever saw its own partial view. In synchronous mode all sequencers in a replication group use Paxos, implemented over ZooKeeper, to agree on a combined batch per epoch. ZooKeeper is not the fastest possible Paxos, but because this step happens before locking it never extends contention footprints, so transactional throughput is completely unaffected by the choice.

Deterministic lock manager

The lock manager is partitioned across the scheduling layer, and each node's scheduler locks only records stored in that node's own storage component, even for transactions touching remote data. It behaves like strict two-phase locking with two added invariants: if transactions A and B both want exclusive locks on a local record R and A precedes B in the sequencer's order, then A must request its lock first; and locks must be granted strictly in request order, so B waits until A has acquired, executed and released. Calvin enforces the first invariant by serializing all lock requests in a single thread that scans the serial order and requests every lock a transaction will ever need. This is precisely why all transactions must declare their full read and write sets in advance.

Five-phase transaction execution

Once a transaction holds all its locks it is handed to a worker thread which runs five phases. First, read/write set analysis identifies which elements are local and which nodes are active participants (they store part of the write set) versus passive participants (read set only). Second, the thread performs local reads; third, it forwards those results to counterpart threads on every active participant, after which a passive participant is finished and never runs transaction code. Fourth, active participants collect the remote read results; fifth, they execute the transaction logic and apply local writes, ignoring non-local writes because the counterpart thread will apply them as local writes at its own node. Assuming participants start at roughly the same time, all reads and all result deliveries happen in parallel and no worker ever requests data from another at execution time.

Committing without an agreement protocol

Nothing in this pipeline resembles a prepare phase. Nondeterministic failures do not abort transactions, because an actively replicated node is running the identical plan and other nodes can source their reads from it while the failed node recovers; the transaction commits on the strength of the replica's completion. Deterministic aborts, such as logic that rejects an order when inventory would go negative, are handled by having each node wait for a single one-way message from every node whose code could deterministically abort, and commit once those arrive. Recovery of a crashed machine means restoring its most recent checkpoint and replaying the more recent transaction inputs, with no physical REDO log involved.

OLLP for dependent transactions

Transactions that must read data in order to discover their own read/write sets — the paper calls them dependent transactions — cannot be handled natively, since locks are requested before execution. Optimistic Lock Location Prediction precedes such a transaction with an inexpensive, low-isolation, unreplicated, read-only reconnaissance query that performs the reads needed to determine the full read/write set. The real transaction is then submitted to the global sequence carrying that predicted set; at execution time the read results are rechecked, and if the reconnoitered set is no longer valid the transaction is deterministically restarted. Restarts are expected to be rare in practice because dependent transactions usually hinge on secondary indexes, which are costly to modify and therefore kept on stable fields; TPC-C's Payment transaction is exactly this shape and never has to restart, since the benchmark never modifies the index it depends on.

Checkpointing against a virtual point of consistency

Since only input is logged, Calvin needs periodic full-database checkpoints to bound replay, and it supports three modes. The naive synchronous mode freezes one entire replica and snapshots it, which is invisible to clients but leaves that replica lagging and slow to catch up. The second mode adapts Cao et al.'s Zig-Zag algorithm, which keeps two copies AS[K]0 and AS[K]1 of each record plus bits MR[K] and MW[K] selecting the version to read and the version to overwrite; Calvin's variant avoids Zig-Zag's requirement to quiesce the database by defining a virtual point of consistency, a prespecified position in the global serial order. Records then carry a before version writable only by earlier transactions and an after version written by later ones; once all earlier transactions finish, an asynchronous thread checkpoints the immutable before versions and duplicates are garbage collected. A third mode exists when the storage layer is fully multiversioned, where a checkpoint is just a SELECT * whose result is logged to disk instead of returned.

What the paper showed — measurements and proofs

On TPC-C limited to New Order transactions, with 10 warehouses per node and every multi-warehouse order touching a warehouse on a different machine, Calvin sustained roughly 5000 transactions per second per node beyond 10 nodes and scaled linearly to nearly half a million transactions per second on 100 nodes — against the then-standing Oracle world record of 504,161 New Order transactions per second on much higher-end hardware.
In the microbenchmark, a single machine reached about 27000 transactions per second, and adding distributed work cost a 5x to 7x drop to about 5000 (4 nodes) or 4000 (8 nodes) per node; per-node throughput then flattened by around 10 machines at contention index 0.0001 and declined more gradually at contention index 0.01, scaling to 100 nodes in all cases.
Replication mode changed latency but not throughput: with 4 EC2 High-CPU machines per replica running 40000 microbenchmark transactions per second of which 10 percent were multipartition, three-replica Paxos in one data center (about 1 ms ping) and across Virginia, Northern California and Ireland (100 to 170 ms ping) both left total transactional throughput unchanged.
With a simple filesystem-backed cold store, throughput was unaffected as long as no more than 0.9 percent of transactions (90 out of 10,000 per second per machine) went to disk, the limit being local random-access disk throughput on commodity hardware rather than contention; a 40 ms artificial delay was needed to get 99 percent of disk-accessing transactions scheduled after prefetch at contention index 0.01, while at contention index 0.001 or lower the 5 ms average batching delay sufficed.
When cold data was served by a separate machine after a configurable delay, each machine still sustained 10,000 transactions per second regardless of how many of them touched cold data, even at contention index 0.01.
Under 100 percent multipartition workloads, Calvin's slowdown stayed far below an analytical lower bound for a System R*-style system, which given contention index C can execute at most 1/(C * D_2PC) transactions per second with D_2PC estimated at about 8 ms from measured 2 ms one-way inter-thread latencies; the model ignores CPU costs, local commit decisions and execution progress skew, so a real 2PC system would fare worse still.

Limits and trade-offs — conceded and discovered

The paper concedes that the deterministic locking protocol requires every transaction to declare its complete read and write set before execution, which excludes dependent transactions natively and forces the OLLP workaround, whose reconnaissance results can go stale and trigger deterministic restarts.
The paper measures, and cannot remove, the cost of forbidding on-the-fly reordering: a machine cannot run far ahead of or behind the pack, so slow EC2 instances and ordinary execution progress skew from thread scheduling and network jitter drag down the whole cluster, and the effect worsens as contention rises. Later deterministic systems were largely motivated by relaxing exactly this coupling.
The paper concedes that disk support rests on two fragile requirements — predicting disk latency accurately, where overestimates waste latency and flood memory with cold records while underestimates stall transactions holding locks, and having every sequencer track which keys are memory-resident across the whole system, which it explicitly calls not a scalable solution.
The paper concedes that unbundling storage from transaction management makes ARIES-style physiological logging and next-key locking impossible; locking key ranges and handling phantoms would require logically lockable virtual resources, and implementing that remained future work.
The paper concedes that failover is not yet seamless: a crashed machine is recovered from its last complete snapshot plus replay, and because other nodes in the same replica depend on remote reads from it, throughput across the rest of the replica is apt to slow or halt until recovery finishes.

What it became — the systems that inherited it

Calvin turned deterministic execution from a curiosity into a credible architecture, and its order-first-then-execute pipeline became the reference design for a whole line of systems. FaunaDB built its distributed transaction engine directly on Calvin's model of agreeing on a global input order before execution, and CalvinFS carried the sequencer/scheduler split into distributed file-system metadata. The Yale group's follow-on work extended the idea in the directions Calvin left open: deterministic multiversion engines such as Bohm and PWV attacked the execution-order coupling, and Aria later removed the requirement that read/write sets be known in advance, which had been Calvin's sharpest constraint. Arriving the same year as Spanner, Calvin also framed the enduring architectural debate in distributed SQL — pay TrueTime plus Paxos plus two-phase commit at execution time, or pay consensus once on the input log and execute deterministically afterwards — an argument Abadi continued to press in print for years. More broadly, Calvin is the clearest database-side statement of state machine replication applied to transactions, the same pattern that underlies log-first designs built on ordered, replayable event logs.

In the paper’s words — verbatim

“when multiple machines need to agree on how to handle a particular transaction, they do it outside of transactional boundaries—that is, before they acquire locks and begin executing the transaction.”

§1.3

“Since all Calvin nodes reach an agreement regarding what transactions to attempt and in what order, it is able to completely eschew distributed commit protocols, reducing the contention footprints of distributed transactions, thereby allowing throughput to scale out nearly linearly despite the presence of multipartition transactions.”

§1.3

“move as much as possible of the heavy lifting to earlier in the transaction processing pipeline, before locks are acquired.”

§4

Vocabulary — as this paper uses it

Contention footprint: The total duration for which a transaction holds its locks, including any commit protocol it must run. Calvin's central claim is that two-phase commit is expensive mainly because it inflates this quantity, not because of its CPU or network overhead.
Deterministic locking: A locking protocol resembling strict two-phase locking but with the extra rules that conflicting transactions request locks in the sequencer's global order and that locks are granted strictly in request order. It makes every replica's execution logically equivalent to the same serial order while still running many transactions concurrently.
Sequencing layer: The layer that intercepts client transaction requests, batches them into 10 ms epochs, replicates the batches, and publishes the global transactional input sequence. It is distributed across all replicas and partitioned within each replica, so there is no single-node sequencer bottleneck.
Scheduling layer: The layer holding the partitioned deterministic lock manager and the pool of transaction execution threads. Each node's scheduler locks only the records stored locally, even for transactions that also read or write data on other nodes.
Epoch: A 10-millisecond window during which each sequencer collects incoming transaction requests before compiling them into one replicated batch. The epoch number is incremented synchronously system-wide and lets every scheduler interleave all sequencers' batches deterministically.
Active and passive participants: For a given transaction, active participants are nodes storing part of its write set and passive participants store only read-set elements. Passive participants forward their local read results and then stop, never executing the transaction code.
Optimistic Lock Location Prediction (OLLP): The scheme for dependent transactions, in which a cheap unreplicated read-only reconnaissance query discovers the transaction's read/write set before the real transaction enters the global sequence. The prediction is rechecked at execution time and the transaction is deterministically restarted if it has become invalid.
Contention index: The microbenchmark parameter giving the fraction of the hot record set that each transaction updates at a participating machine. An index of 0.01 permits at most 100 concurrent transactions, while an index of 1 forces fully serial execution.
Virtual point of consistency: A prespecified position in the global serial order used as the logical instant a checkpoint captures. It lets Calvin's Zig-Zag variant take a consistent snapshot without ever quiescing the database into a physical point of consistency.

On the timeline — where this sits in the story

Year2012

MilestoneCalvin deterministic transactions

Problem solvedDistributed transactions incurred high coordination cost, especially under contention across partitions.

How it solved itCalvin used deterministic transaction ordering to reduce runtime coordination complexity.

Why it matteredInfluenced distributed SQL and deterministic transaction-processing architectures.

DDIACh. 8: Transactions Ch. 10: Consistency & Consensus

View on the timeline