Databases (6): Replication and Partitioning — Scaling Beyond One Machine

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

                    Writes
Client ────────────► Leader (Primary)
                       │
              ┌────────┼────────┐
              ▼        ▼        ▼
          Follower  Follower  Follower
          (Replica) (Replica) (Replica)
              ▲        ▲        ▲
              └────────┼────────┘
                    Reads

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

-- PostgreSQL: configure synchronous replication
-- postgresql.conf on primary
synchronous_standby_names = 'FIRST 1 (replica1, replica2)'
-- FIRST 1 = wait for at least 1 of the listed replicas
-- This means writes block until replica1 OR replica2 confirms

-- Check replication status
SELECT
    client_addr,
    state,
    sync_state,
    sent_lsn,
    write_lsn,
    flush_lsn,
    replay_lsn,
    (sent_lsn - replay_lsn) AS replication_lag_bytes
FROM pg_stat_replication;

1
2
3
4
5
6

Timeline:
1. User posts a comment (goes to leader)
2. Leader writes: OK
3. User reloads page (goes to follower)
4. Follower hasn't received step 2 yet
5. User sees: "No comments" — their comment disappeared!

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

# Solution 1: Read from leader for recently-written data
def get_user_profile(user_id, requesting_user_id):
    if user_id == requesting_user_id:
        # User is viewing their own profile — read from leader
        return db_leader.query("SELECT * FROM users WHERE id = %s", user_id)
    else:
        # Viewing someone else's profile — replica is fine
        return db_replica.query("SELECT * FROM users WHERE id = %s", user_id)

# Solution 2: Track the write timestamp
def get_comments(post_id, last_write_ts=None):
    if last_write_ts and (time.time() - last_write_ts) < 5:
        # Written within last 5 seconds — read from leader
        return db_leader.query("SELECT * FROM comments WHERE post_id = %s", post_id)
    return db_replica.query("SELECT * FROM comments WHERE post_id = %s", post_id)

1
2
3
4
5
6
7
8

Data Center A              Data Center B
┌──────────────┐          ┌──────────────┐
│   Leader A   │◄────────►│   Leader B   │
│  (read/write)│          │  (read/write)│
│      │       │          │      │       │
│  Follower    │          │  Follower    │
│  Follower    │          │  Follower    │
└──────────────┘          └──────────────┘

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

-- Example: last-writer-wins with a version timestamp
-- Both leaders accept updates independently
-- On sync, the row with the latest updated_at wins

-- Leader A: user updates their name
UPDATE users SET name = 'Alice Chen', updated_at = '2023-12-15T10:00:01Z'
WHERE user_id = 1;

-- Leader B: same user updates their name (at almost the same time)
UPDATE users SET name = 'Alice C.', updated_at = '2023-12-15T10:00:02Z'
WHERE user_id = 1;

-- After replication sync: Leader B's update wins (later timestamp)
-- But Leader A's change is silently lost

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

N = 3 (three replicas of each piece of data)
W = 2 (write must be acknowledged by 2 nodes)
R = 2 (read must be answered by 2 nodes)

Write "balance = 500" to key "account:1":
  Node 1: ✓ acknowledged (balance = 500)
  Node 2: ✓ acknowledged (balance = 500)
  Node 3: ✗ unreachable (still has balance = 1000)
  --> Write succeeds (W=2 met)

Read key "account:1":
  Node 1: balance = 500 (version 2)
  Node 2: balance = 500 (version 2)
  --> Returns 500 (latest version)
  OR
  Node 2: balance = 500 (version 2)
  Node 3: balance = 1000 (version 1)
  --> Returns 500 (client picks highest version)

1
2
3
4
5

Read key "account:1":
  Node 1: balance = 500 (version 2) ✓ latest
  Node 3: balance = 1000 (version 1) ✗ stale
  --> Return 500 to client
  --> Background: write balance=500 to Node 3 (read repair)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

-- Each row has a version timestamp
CREATE TABLE products (
    product_id INT PRIMARY KEY,
    name VARCHAR(200),
    price DECIMAL(10,2),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

-- On conflict: keep the newer write
INSERT INTO products (product_id, name, price, updated_at)
VALUES (42, 'Widget', 19.99, '2024-03-15 10:30:00+00')
ON CONFLICT (product_id)
DO UPDATE SET
    name = EXCLUDED.name,
    price = EXCLUDED.price,
    updated_at = EXCLUDED.updated_at
WHERE products.updated_at < EXCLUDED.updated_at;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

# G-Counter: each node maintains its own counter
class GCounter:
    def __init__(self, node_id: str):
        self.node_id = node_id
        self.counts: dict[str, int] = {}

    def increment(self):
        self.counts[self.node_id] = self.counts.get(self.node_id, 0) + 1

    def value(self) -> int:
        return sum(self.counts.values())

    def merge(self, other: "GCounter"):
        """Merge is commutative, associative, idempotent."""
        for node, count in other.counts.items():
            self.counts[node] = max(self.counts.get(node, 0), count)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

-- Store conflicts explicitly
CREATE TABLE document_versions (
    doc_id INT,
    version_id UUID DEFAULT gen_random_uuid(),
    content TEXT,
    author VARCHAR(50),
    written_at TIMESTAMPTZ,
    is_conflict BOOLEAN DEFAULT FALSE,
    resolved_by UUID REFERENCES document_versions(version_id),
    PRIMARY KEY (doc_id, version_id)
);

-- Application presents conflicting versions to user for manual merge

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

import random

class DatabaseRouter:
    def __init__(self, primary: str, replicas: list[str]):
        self.primary = primary
        self.replicas = replicas

    def get_connection(self, operation: str, consistency: str = "eventual"):
        """Route to primary or replica based on operation type."""
        if operation in ("INSERT", "UPDATE", "DELETE"):
            return self.primary

        if consistency == "strong":
            return self.primary  # reads-after-writes need primary

        return random.choice(self.replicas)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

import time

class SessionConsistency:
    """Guarantee read-your-own-writes per session."""

    def __init__(self, router: DatabaseRouter):
        self.router = router
        self.last_write_time: float = 0
        self.lag_threshold: float = 2.0  # seconds

    def get_read_connection(self):
        if time.time() - self.last_write_time < self.lag_threshold:
            return self.router.primary  # too recent, use primary
        return self.router.get_connection("SELECT")

    def record_write(self):
        self.last_write_time = time.time()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

-- PostgreSQL: check replication lag
SELECT
    client_addr,
    state,
    sent_lsn,
    write_lsn,
    flush_lsn,
    replay_lsn,
    pg_wal_lsn_diff(sent_lsn, replay_lsn) AS replay_lag_bytes,
    reply_time
FROM pg_stat_replication;

-- MySQL: check seconds behind master
SHOW SLAVE STATUS\G
-- Look for: Seconds_Behind_Master

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

# patroni.yml
scope: my-cluster
name: node1

etcd3:
  hosts: etcd1:2379,etcd2:2379,etcd3:2379

bootstrap:
  dcs:
    ttl: 30
    loop_wait: 10
    maximum_lag_on_failover: 1048576  # 1MB max lag for promotion

postgresql:
  listen: 0.0.0.0:5432
  data_dir: /var/lib/postgresql/data
  parameters:
    max_connections: 200
    synchronous_commit: "on"

1
2
3
4
5
6
7

# orchestrator detects primary failure and promotes replica
$ orchestrator-client -c recover -i dead-primary:3306

# MySQL InnoDB Cluster (Group Replication)
# Auto-elects new primary from group members
mysqlsh> cluster = dba.getCluster()
mysqlsh> cluster.status()

1
2
3

Shard 1: user_id    1 - 1,000,000
Shard 2: user_id    1,000,001 - 2,000,000
Shard 3: user_id    2,000,001 - 3,000,000

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

-- PostgreSQL declarative partitioning (range)
CREATE TABLE orders (
    order_id    BIGSERIAL,
    user_id     INT NOT NULL,
    created_at  TIMESTAMP NOT NULL,
    total       DECIMAL(10,2)
) PARTITION BY RANGE (created_at);

CREATE TABLE orders_2023_q4 PARTITION OF orders
    FOR VALUES FROM ('2023-10-01') TO ('2024-01-01');

CREATE TABLE orders_2024_q1 PARTITION OF orders
    FOR VALUES FROM ('2024-01-01') TO ('2024-04-01');

CREATE TABLE orders_2024_q2 PARTITION OF orders
    FOR VALUES FROM ('2024-04-01') TO ('2024-07-01');

-- Queries automatically route to the correct partition
SELECT * FROM orders WHERE created_at = '2023-11-15';
-- Only scans orders_2023_q4, skips other partitions

1
2
3
4
5

partition = hash(user_id) % num_shards

hash("user:1")  = 0x3A2B... → shard 2
hash("user:2")  = 0x8F1C... → shard 0
hash("user:3")  = 0x12D4... → shard 1

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

-- PostgreSQL hash partitioning
CREATE TABLE sessions (
    session_id  UUID PRIMARY KEY,
    user_id     INT NOT NULL,
    data        JSONB,
    expires_at  TIMESTAMP
) PARTITION BY HASH (session_id);

CREATE TABLE sessions_p0 PARTITION OF sessions
    FOR VALUES WITH (MODULUS 4, REMAINDER 0);
CREATE TABLE sessions_p1 PARTITION OF sessions
    FOR VALUES WITH (MODULUS 4, REMAINDER 1);
CREATE TABLE sessions_p2 PARTITION OF sessions
    FOR VALUES WITH (MODULUS 4, REMAINDER 2);
CREATE TABLE sessions_p3 PARTITION OF sessions
    FOR VALUES WITH (MODULUS 4, REMAINDER 3);

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

                    0 / 2^32
                      │
              Node C  ●
                     ╱  ╲
                    ╱    ╲
           ●──────╱──────●
         Node A  ╱      Node B
                ╱
               ╱
Key "user:42" hashes to position X
→ Walk clockwise → first node encountered = owner

Adding Node D:
  Only keys between Node C and Node D need to move
  (not all keys in the cluster)

1
2
3
4
5

Physical Node A → Virtual nodes: A1, A2, A3, A4, A5 (5 positions on ring)
Physical Node B → Virtual nodes: B1, B2, B3, B4, B5

More virtual nodes = more even distribution
Cassandra default: 256 vnodes per physical node

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

Before (10 nodes, 1000 partitions):
  Node 1: partitions 0-99
  Node 2: partitions 100-199
  ...

After adding Node 11:
  Node 1:  partitions 0-89     (gave away 10)
  Node 2:  partitions 100-189  (gave away 10)
  ...
  Node 11: partitions 90-99, 190-199, ...  (received ~91 partitions)

1
2
3
4

-- Table partitioned by user_id
-- But we also query by email
SELECT * FROM users WHERE email = 'alice@example.com';
-- Which shard has this user? We don't know without checking all of them.

1
2
3
4
5

Shard 1: local index on email → {alice@...: row 1, bob@...: row 2}
Shard 2: local index on email → {carol@...: row 3, dave@...: row 4}

Query by email → scatter to ALL shards, gather results
(called "scatter-gather" — expensive for fan-out)

1
2
3
4
5

Email index shard A (emails a-m): alice@... → shard 1, carol@... → shard 2
Email index shard B (emails n-z): zara@... → shard 3

Query by email → go to the right index shard → then to the data shard
(2 hops but no scatter-gather)

1
2
3
4
5
6
7
8

# /etc/mysql/mysql.conf.d/mysqld.cnf on the leader
[mysqld]
server-id           = 1
log_bin              = /var/log/mysql/mysql-bin
binlog_format        = ROW          # safest format
binlog_expire_logs_seconds = 604800 # 7 days retention
sync_binlog          = 1            # sync binlog on every commit
innodb_flush_log_at_trx_commit = 1  # full durability

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

-- Create a replication user on the leader
CREATE USER 'repl_user'@'%' IDENTIFIED BY 'strong_password_here';
GRANT REPLICATION SLAVE ON *.* TO 'repl_user'@'%';
FLUSH PRIVILEGES;

-- Get the current binlog position
SHOW MASTER STATUS;
-- +------------------+----------+
-- | File             | Position |
-- +------------------+----------+
-- | mysql-bin.000003 |      785 |
-- +------------------+----------+

1
2
3
4
5
6
7
8

# Option 1: mysqldump with consistent snapshot
mysqldump --all-databases --single-transaction \
  --source-data=2 --routines --triggers \
  -u root -p > leader_backup.sql

# Option 2: For large databases, use xtrabackup
xtrabackup --backup --target-dir=/backup/full \
  --user=root --password=xxx

1
2
3
4
5
6

# /etc/mysql/mysql.conf.d/mysqld.cnf on the follower
[mysqld]
server-id            = 2       # must be unique
relay_log            = /var/log/mysql/mysql-relay
read_only            = ON      # prevent accidental writes
super_read_only      = ON      # even root can't write

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

-- Restore the backup on the follower
-- Then configure replication
CHANGE REPLICATION SOURCE TO
    SOURCE_HOST='leader-hostname',
    SOURCE_USER='repl_user',
    SOURCE_PASSWORD='strong_password_here',
    SOURCE_LOG_FILE='mysql-bin.000003',
    SOURCE_LOG_POS=785;

-- Start replication
START REPLICA;

-- Check replication status
SHOW REPLICA STATUS\G
-- Key fields to check:
--   Replica_IO_Running: Yes
--   Replica_SQL_Running: Yes
--   Seconds_Behind_Source: 0
--   Last_Error: (should be empty)

1
2
3
4
5
6
7
8
9

-- On the follower: check lag
SHOW REPLICA STATUS\G
-- Seconds_Behind_Source: 0  <-- healthy
-- Seconds_Behind_Source: 45 <-- concerning
-- Seconds_Behind_Source: NULL <-- replication broken!

-- On the leader: check connected replicas
SHOW REPLICAS;
-- or older syntax: SHOW SLAVE HOSTS;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

# Quick health check script
#!/bin/bash
LAG=$(mysql -e "SHOW REPLICA STATUS\G" | grep "Seconds_Behind_Source" | awk '{print $2}')
IO_RUNNING=$(mysql -e "SHOW REPLICA STATUS\G" | grep "Replica_IO_Running" | awk '{print $2}')
SQL_RUNNING=$(mysql -e "SHOW REPLICA STATUS\G" | grep "Replica_SQL_Running" | awk '{print $2}')

echo "Replication Lag: ${LAG}s"
echo "IO Thread: $IO_RUNNING"
echo "SQL Thread: $SQL_RUNNING"

if [ "$LAG" -gt 60 ] || [ "$IO_RUNNING" != "Yes" ] || [ "$SQL_RUNNING" != "Yes" ]; then
    echo "ALERT: Replication unhealthy!"
    exit 1
fi

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

-- On the follower to be promoted:
STOP REPLICA;
RESET REPLICA ALL;

-- The follower is now a standalone server
-- Reconfigure other followers to point to the new leader

-- On remaining followers:
STOP REPLICA;
CHANGE REPLICATION SOURCE TO
    SOURCE_HOST='new-leader-hostname',
    SOURCE_LOG_FILE='...',
    SOURCE_LOG_POS=...;
START REPLICA;

-- Application configuration must also be updated
-- (or use a proxy like ProxySQL / HAProxy)