Distributed Database Design: Consistency, Partitioning, and Quorum

In the age of global applications, databases must scale beyond a single machine. Distributed databases spread data across multiple nodes to handle massive traffic and ensure availability. However, this distribution introduces challenges: how do you keep data consistent across nodes? How do you split data efficiently? How do you ensure reads and writes are reliable even when some nodes fail?

This tutorial dives into three core concepts: consistency models, partitioning strategies, and quorum mechanisms. You'll learn how they interact, how to configure them in popular NoSQL databases like Cassandra and MongoDB, and how to avoid common pitfalls that lead to production outages.

We'll start with a real-world incident where a misconfigured quorum caused a major e-commerce platform to sell the same item twice. Then we'll explore each concept with practical examples, SQL-like queries (adapted for NoSQL), and debugging guides. By the end, you'll be equipped to design distributed databases that are both scalable and reliable.

Understanding Consistency Models

Consistency models define how and when updates become visible to all nodes. The strongest model is linearizability (strong consistency), where every read returns the most recent write. Weaker models like eventual consistency allow temporary staleness but improve availability.

In distributed databases, you often choose a consistency level per operation. For example, in Cassandra, you can set CONSISTENCY ONE (fast but weak) or CONSISTENCY QUORUM (balanced). MongoDB offers 'majority' read concern for strong consistency.

Consider an e-commerce inventory system. If you use eventual consistency, two users might see different stock levels for the same item, leading to overselling. Strong consistency prevents this but may increase latency.

Example: In Cassandra, to ensure strong consistency:

CREATE TABLE inventory (item_id text PRIMARY KEY, quantity int); -- Write with QUORUM INSERT INTO inventory (item_id, quantity) VALUES ('SKU123', 10) USING CONSISTENCY QUORUM; -- Read with QUORUM SELECT quantity FROM inventory WHERE item_id = 'SKU123' USING CONSISTENCY QUORUM;

-- Create table
CREATE TABLE inventory (
  item_id text PRIMARY KEY,
  quantity int
);

-- Write with QUORUM
INSERT INTO inventory (item_id, quantity) VALUES ('SKU123', 10) USING CONSISTENCY QUORUM;

-- Read with QUORUM
SELECT quantity FROM inventory WHERE item_id = 'SKU123' USING CONSISTENCY QUORUM;

Partitioning Strategies: Range, Hash, and List

Partitioning splits data across nodes. Three common strategies: - Range partitioning: Data is divided by key ranges (e.g., A-M on node1, N-Z on node2). Simple but can cause hot spots if keys are not uniformly distributed. - Hash partitioning: A hash function on the partition key distributes data uniformly. Avoids hot spots but makes range queries inefficient. - List partitioning: Data is assigned to partitions based on a list of values (e.g., region: US, EU, ASIA). Good for geographic sharding.

In Cassandra, partitioning is hash-based on the partition key. In MongoDB, you can choose range or hash sharding.

Example: Creating a partitioned table in Cassandra with a composite partition key:

CREATE TABLE orders ( customer_id text, order_id timeuuid, amount decimal, PRIMARY KEY ((customer_id), order_id) ) WITH CLUSTERING ORDER BY (order_id DESC);

Here, customer_id is the partition key, and order_id is the clustering key. All orders for a customer are stored together, enabling efficient queries by customer.

-- Create partitioned table
CREATE TABLE orders (
  customer_id text,
  order_id timeuuid,
  amount decimal,
  PRIMARY KEY ((customer_id), order_id)
) WITH CLUSTERING ORDER BY (order_id DESC);

-- Insert sample data
INSERT INTO orders (customer_id, order_id, amount) VALUES ('cust1', now(), 100.00);
INSERT INTO orders (customer_id, order_id, amount) VALUES ('cust1', now(), 200.00);
INSERT INTO orders (customer_id, order_id, amount) VALUES ('cust2', now(), 150.00);

-- Query all orders for a customer
SELECT * FROM orders WHERE customer_id = 'cust1';

Quorum: The Math of Consistency

Quorum is the number of nodes that must agree on a read or write operation. For a replication factor N, write quorum W and read quorum R must satisfy W + R > N to ensure strong consistency (no stale reads). Typically, you set W = R = floor(N/2) + 1 (majority).

In Cassandra, you can set consistency per operation. For example, with N=3, setting W=2 and R=2 ensures that any read sees the latest write because at least one node overlaps.

Example: Setting quorum in Cassandra:

-- Write with quorum INSERT INTO users (id, name) VALUES (1, 'Alice') USING CONSISTENCY QUORUM; -- Read with quorum SELECT * FROM users WHERE id = 1 USING CONSISTENCY QUORUM;

If you set W=1 and R=1, you risk reading stale data. If you set W=3 and R=1, writes are slow but reads are fast (but still may be stale if write fails on some nodes).

-- Assume replication factor = 3
-- Write with QUORUM (W=2)
INSERT INTO users (id, name) VALUES (1, 'Alice') USING CONSISTENCY QUORUM;

-- Read with QUORUM (R=2)
SELECT * FROM users WHERE id = 1 USING CONSISTENCY QUORUM;

-- Check consistency level
CONSISTENCY;

CAP Theorem and Trade-offs

The CAP theorem states that a distributed system can only guarantee two of three properties: Consistency, Availability, and Partition Tolerance. In practice, partitions are inevitable, so you must choose between CP (consistency + partition tolerance) and AP (availability + partition tolerance).

• CP systems (e.g., HBase, MongoDB with majority read concern) sacrifice availability during partitions.
• AP systems (e.g., Cassandra, DynamoDB) sacrifice consistency, becoming eventually consistent.

Your choice depends on application requirements. For a banking system, consistency is critical (CP). For a social media feed, availability is more important (AP).

Example: In Cassandra, you can tune consistency per operation, effectively choosing between CP and AP dynamically. For critical writes, use QUORUM (CP); for non-critical, use ONE (AP).

-- CP: Strong consistency
INSERT INTO payments (id, amount) VALUES (1, 100) USING CONSISTENCY QUORUM;

-- AP: Eventual consistency
INSERT INTO logs (id, message) VALUES (1, 'user login') USING CONSISTENCY ONE;

Replication and Consistency in Practice

Replication copies data across nodes for fault tolerance. Two common replication strategies: synchronous (all replicas updated before acknowledging) and asynchronous (acknowledge after one replica).

Synchronous replication ensures strong consistency but increases latency. Asynchronous replication improves performance but risks data loss if the primary fails.

In Cassandra, replication is configurable per keyspace. Example:

CREATE KEYSPACE mykeyspace WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 3};

This replicates data to three nodes. Consistency levels then determine how many replicas must respond.

In MongoDB, replica sets use primary-secondary replication. Reads can be directed to secondaries for scalability but may return stale data unless using 'majority' read concern.

-- Create keyspace with replication factor 3
CREATE KEYSPACE mykeyspace WITH replication = {
  'class': 'SimpleStrategy',
  'replication_factor': 3
};

-- Use keyspace
USE mykeyspace;

-- Create table
CREATE TABLE users (id int PRIMARY KEY, name text);

-- Insert data
INSERT INTO users (id, name) VALUES (1, 'Bob') USING CONSISTENCY QUORUM;

Conflict Resolution and Last-Write-Wins

In eventually consistent systems, concurrent writes can cause conflicts. The most common resolution is Last-Write-Wins (LWW), where the write with the latest timestamp is accepted. However, clock skew can cause issues.

Cassandra uses LWW by default. To avoid data loss, you can use lightweight transactions (compare-and-set) for critical updates.

Example: Using lightweight transactions in Cassandra:

UPDATE users SET balance = 50 WHERE id = 1 IF balance = 100;

This ensures the update only happens if the current balance is 100, preventing lost updates.

In MongoDB, you can use findAndModify with conditions for atomic updates.

-- Lightweight transaction: update only if current balance is 100
UPDATE users SET balance = 50 WHERE id = 1 IF balance = 100;

-- Check result
APPLY BATCH;

-- If another concurrent update happened, this will fail