Distributed SQL: Consensus Meets Query Processing

Distributed SQL tries to preserve a demanding abstraction: applications issue ordinary relational transactions while data is split, replicated, moved, and queried across machines. Consensus is necessary to make replicated storage agree, but it is only the lowest coordination layer. It does not choose a join plan, make secondary indexes atomic with base rows, assign a serial order across unrelated replica groups, or decide whether a distant follower may serve a read.

The useful mental model is a stack of contracts. Ordered key ranges provide placement units. Each range has a replicated log that decides its local mutations. A transaction layer combines decisions across ranges at a logical timestamp. A SQL layer maps tables and indexes into keys, routes operations, and builds distributed execution plans. Distributed SQL works when every layer exposes enough ordering and locality information for the layer above, without leaking placement churn into application semantics.

This article starts after consensus fundamentals. Elections, term rules, and log repair belong to the consensus protocol itself. Here the question is how a database uses an already-correct replicated state machine to implement SQL, and where the resulting latency and operational tradeoffs appear.

Make Ranges the Unit of Placement #

A distributed SQL database commonly encodes table rows and secondary-index entries into one ordered key space. Contiguous intervals of that space are called ranges, tablets, or shards depending on the product. To avoid confusing them with application-managed sharding, this article uses range: a database-owned placement unit whose location can change without changing SQL table names.

An illustrative key encoding might look like this:

/table/customers/primary/<customer_id>
/table/orders/primary/<order_id>
/table/orders/by_customer/<customer_id>/<created_at>/<order_id>
/table/inventory/primary/<sku>

The database partitions these keys at boundaries. Small adjacent keys share a range; a range that grows beyond policy can split; underfilled neighbors can merge. Replicas of each range live on several nodes, usually across distinct failure domains. Placement policy can require replicas or voting members in specified regions, zones, or hardware classes.

Ranges solve two problems at once. They bound consensus membership, so every write need not involve every database node, and they provide movable units for balancing bytes, traffic, and replicas. The range descriptor records its start key, end key, replica set, and an epoch or generation. Clients or SQL gateways cache descriptors, refresh them after a routing error, and retry against the current owner.

The abstraction needs fencing. During a split or relocation, a stale gateway may send a mutation to the former owner. The range layer must reject requests carrying obsolete placement knowledge rather than allowing two owners to accept writes. Routing errors such as range moved or not leaseholder are expected control flow; application-visible failures should occur only after bounded internal refresh and retry cannot complete.

Range boundaries are physical, not relational. One table can span many ranges, and one small range can contain keys from adjacent indexes or tables. A row in a base table and its secondary-index entry may therefore belong to different consensus groups. That observation is why local replication alone cannot provide complete SQL transactions.

Use Consensus as a Local Ordering Primitive #

For one range, consensus exposes a compact contract: submit a command, establish one durable order accepted by a quorum, and apply committed commands deterministically on replicas. The SQL storage layer uses that contract for writes such as placing a transaction intent, updating a timestamp record, or resolving a committed value.

Consensus provides per-range linearizable mutation order when used correctly. It does not automatically provide:

• Atomicity for a transaction touching several ranges.
• A global sequence shared by all ranges.
• Snapshot visibility across independent replica groups.
• Exactly-once client behavior after an unknown response.
• A query plan or a relational constraint spanning keys.

Those properties require additional protocols. This separation matters during design reviews. Saying “the database uses consensus” does not answer whether a multi-row transaction is serializable or whether a follower read is current.

Most implementations designate a leader, leaseholder, or primary replica to coordinate a range’s strongly consistent operations. The exact authority differs: a consensus leader orders log entries, while a read lease may authorize serving reads without a quorum round on every request. Keeping the lease near active clients reduces latency, but moving it too eagerly can create churn. A replica placement can be durable and healthy while its request authority is in the wrong region for the workload.

Commit acknowledgment also has a physical meaning. A quorum write can survive the failure policy represented by the replica set, but its latency includes the required durable network path. Adding distant replicas for recovery does not always require every write to wait for all of them, yet quorum composition and placement determine which failures can be tolerated without losing progress. Operators must reason about voters, replicas, and serving locality separately.

Turn Local Versions Into SQL Transactions #

Distributed SQL engines typically combine consensus with multiversion concurrency control. A committed value is stored with a logical timestamp, and a transaction reads from a snapshot while buffering or writing provisional changes. The transaction has an identity, a read timestamp, a status, and enough metadata to resolve uncertainty after failures.

A transaction that touches one range can often use a fast path because its writes, conflict checks, and final status share one replicated order. A transaction that touches several ranges needs one atomic outcome across several independent logs. The engine may place intents at participant ranges and maintain a transaction record at a chosen home range. Finalization makes every participant interpret those intents as committed or aborted. This is an atomic-commit problem; the implementation may optimize messages and cleanup, but it cannot wish away the need for a durable decision.

Intent cleanup may be asynchronous after the outcome is durable. A reader encountering an unresolved intent consults transaction status and helps resolve it. This lets the client receive a committed result without waiting for every physical cleanup, but it means transaction records and intent-resolution queues are part of the critical health surface.

Serializable isolation requires handling conflicts that arise after the initial timestamp choice. Engines use combinations of latches, lock tables, timestamp caches, dependency tracking, and transaction refresh or restart. A transaction may be logically valid yet receive a retry because a concurrent write established an ordering incompatible with its earlier reads. Client libraries should retry the complete transaction callback only when the database classifies the error as retryable, and side effects outside the transaction must not run twice.

Atomic secondary indexes illustrate the value of the transaction layer. Inserting an order writes the primary row and one or more index entries, potentially on different ranges. Readers must not observe an index entry without its row or a row missing its committed index entry. Treating every index as an eventually updated projection would weaken ordinary SQL semantics; distributed transactions keep those representations atomic.

Follow One Order Through the Stack #

Consider a transaction that creates order O-701 for customer C-42 and reserves two units of SKU P-9. The schema has a customer lookup index and inventory keyed by SKU:

BEGIN;

INSERT INTO orders (order_id, customer_id, created_at, status)
VALUES ('O-701', 'C-42', :created_at, 'accepted');

UPDATE inventory
SET available = available - 2
WHERE sku = 'P-9' AND available >= 2;

-- The application verifies exactly one inventory row changed.
COMMIT;

The SQL gateway parses and plans the statements, then encodes keys. The order primary row may route to range R17; the (customer_id, created_at, order_id) secondary-index entry to R8; inventory P-9 to R31; and the transaction record to one selected home range. Each participant’s replicas agree on local commands through its own consensus group.

The inventory update is conditional. If available starts at 5, its committed value becomes 3. If it starts at 1, zero rows change and the application aborts the transaction. The condition must execute where the authoritative inventory version is serialized; reading 5 at a gateway and later issuing an unconditional decrement would allow concurrent overselling.

A client timeout creates an unknown outcome. The transaction may have committed while the response was lost. Reissuing the same business operation with a new order_id could reserve inventory twice. The application needs a stable request key or order identity protected by a unique constraint, then a status lookup or idempotent retry. Consensus prevents replicas from disagreeing about one accepted command; it does not infer that two client commands represent the same business intent.

Now query the customer’s latest orders:

SELECT order_id, created_at, status
FROM orders
WHERE customer_id = 'C-42'
  AND created_at >= :from_time
ORDER BY created_at DESC
LIMIT 20;

The secondary index makes the relevant key interval contiguous. The gateway looks up the range descriptors covering that interval and sends bounded scans to them. If recent entries fit one range, execution remains local to one replica group. If the interval crosses split boundaries, scans run on several ranges and their ordered streams are merged. SQL hides the placements, but locality still determines latency and resource use.

Build Distributed Query Plans Around Locality #

Point lookups and narrow index scans mostly exercise routing. Analytical joins, aggregations, and sorts require a distributed execution graph. A gateway obtains table statistics, selects access paths and join order, then places operators near data when possible. Exchange operators move rows between stages when data must be repartitioned, broadcast, or gathered.

For a join between large orders and small status_codes, broadcasting the small input to nodes scanning orders can avoid moving the large input. For two large relations joined by customer_id, hash-repartitioning both sides by that key may co-locate matching rows. If both tables are already keyed and placed compatibly, a locality-aware join can avoid that shuffle. These are cost choices, not semantic differences.

range scans -> partial filters -> partial aggregates
             -> hash exchange by customer_id
             -> distributed join
             -> final aggregate -> gateway

The planner needs network and locality costs in addition to CPU, row count, and index selectivity. Stale statistics can choose a broadcast for an input that is no longer small. Correlated tenants can create severe skew even when global histograms look balanced. One destination then receives a disproportionate hash bucket and becomes a straggler.

Push predicates, projections, limits, and partial aggregation toward range scans. Sending ten qualifying columns is cheaper than sending complete rows, and combining local counts before a network exchange reduces traffic. However, pushing an operator is valid only if it preserves SQL semantics for nulls, collation, ordering, numeric behavior, and the chosen snapshot.

Admission control matters because one broad query can contact many ranges and consume workers across the cluster. Bound fan-out concurrency, memory for hash tables and sorts, temporary disk, and result bytes. Cancellation must propagate to every remote stage. A client that disconnects should not leave a distributed query running until its original deadline.

Make Time and Read Locality Honest #

Transactions need a comparable ordering domain even though ranges commit independently. Many systems use logical or hybrid logical timestamps: a physical-clock component preserves useful wall-time proximity, while a logical component orders events that would otherwise tie or arrive from a clock ahead of the receiver. The clock algorithm is infrastructure; SQL cares about the guarantees attached to a timestamp.

A read at timestamp $t$ must use replicas known to have applied all relevant committed writes through $t$, and it must handle provisional transactions that could affect that snapshot. The current leaseholder can usually serve a linearizable read after satisfying its authority checks. A nearby follower may serve a historical read if its closed or safe timestamp has advanced beyond $t$. This creates a valuable latency tradeoff: explicitly stale reads can stay local without claiming to be current.

Clock uncertainty affects strict real-time ordering. If the database cannot know whether another transaction with an overlapping physical-time interval happened first, it may wait out uncertainty, choose a higher timestamp, or restart work according to its protocol. Better clock synchronization reduces these costs but does not replace logical ordering and conflict checks.

Expose read semantics in the API:

Silently routing a supposedly current read to a lagging follower is not a performance optimization. It changes the database contract. Historical reads also depend on MVCC retention; a timestamp older than garbage-collection policy cannot be reconstructed merely because a replica exists.

Operate Placement, Schema, and Hotspots #

Automatic balancing watches range bytes, request rate, CPU, replica health, and placement constraints. Splitting a hot range can add parallelism only if traffic covers separable keys. A single hot key remains one serialization point after every surrounding split. Mitigations may require caching, write coalescing, bucketing a mergeable counter, or redesigning the invariant rather than asking the balancer to move it faster.

Lease placement should follow serving traffic while replica placement follows durability and failure policy. Moving leases too slowly leaves cross-region hops on the critical path; moving them on every short traffic shift causes instability. Track per-range lease transfers, routing retries, request origin, and the reason placement constraints prevent a better location.

Schema changes are distributed data transformations. Adding a nullable metadata-only column may be cheap, while adding an index requires backfilling entries across ranges and then making the index public at a consistent schema version. Nodes running mixed binaries and transactions started under older descriptors must remain compatible. Backfills need checkpoints, admission limits, and validation; otherwise a schema change can consume the same capacity required by foreground replication and intent cleanup.

Backups must capture a transactionally coherent timestamp across ranges, plus descriptors and schema metadata needed to restore key ownership. A set of unrelated node filesystem copies is not necessarily a database snapshot. Restore drills should rebuild ranges, validate SQL constraints and indexes, and test a point-in-time target under the documented retention policy.

Capacity planning must include quorum write paths, not only aggregate disk. A cluster can have free bytes yet be unable to place replicas across required zones, or have spare CPU in a remote region while a locality-bound leaseholder is saturated. Show headroom by failure domain and placement policy.

Test the Contracts Between Layers #

Storage tests should assert that committed per-range commands survive replica loss and that stale descriptors cannot mutate an old owner. Transaction tests should pause participants before and after outcome recording, lose client responses, leave intents unresolved, and verify that readers converge on one atomic result. Generate concurrent histories and check serializability rather than only final row counts.

For the order example, test two transactions racing for the last two inventory units. At most one may commit when each requests two. Then drop the successful response and retry with the same order identity; inventory must still decrease once. Move the inventory range during the transaction and ensure routing refresh does not weaken the condition.

Query tests should compare distributed results with a trusted single-node execution over the same snapshot. Exercise range splits mid-scan, node loss, duplicate RPC responses, skewed joins, spill to disk, cancellation, and limits with non-unique sort keys. Deterministic ordering requires a complete tie-breaker; otherwise different parallel arrival orders may legitimately change which tied rows appear under LIMIT.

Operational signals should include transaction retry reasons, intent age, unresolved transaction records, range unavailability, consensus commit latency, lease location, descriptor-cache misses, fan-out width, exchange bytes, spill bytes, skew by operator, clock uncertainty, follower safe timestamp, and schema-job progress. Correlate them by query and transaction identifiers without logging sensitive row values.

Distributed SQL offers a familiar interface by coordinating several distinct distributed mechanisms. Its principal tradeoff is that relational invariants spanning distant keys require communication: consensus within each range, atomic outcome across ranges, and network exchanges for nonlocal queries. Good schemas keep transactions and dominant access paths local where possible, while the database retains correctness when they are not. Consensus supplies durable local order; the transaction and query layers turn those orders into SQL. Operating the system well means observing every boundary between them rather than treating the SQL endpoint as one indivisible machine.