Tag: partitioning

  • Scaling databases: read replicas, sharding, and partitioning

    Scaling databases: read replicas, sharding, and partitioning

    Almost every scaling story I have lived through ends at the database. You can cache, you can queue, you can add a hundred stateless app nodes, and it all helps right up until a single primary database is the thing every one of those nodes is waiting on. At that point you have to scale the data layer itself, and that is a different kind of problem because data has weight. Moving compute is easy. Moving and splitting data without losing or corrupting it is where the real engineering lives.

    I think about this as a ladder. Each rung is more powerful and more painful than the last, and you should climb only as high as you actually need.

    First, exhaust the cheap wins

    Before any architectural change I make sure the database is not just doing unnecessary work. The most common cause of a database that feels too small is a database that is missing indexes, running sequential scans on large tables, or being hammered by queries that should have been cached. Connection pool exhaustion masquerades as a scaling problem constantly. A well indexed schema on a properly sized instance handles far more than people expect, and I cover the groundwork in database schema best practices. Do not shard a database that just needs an index.

    Read replicas, the first real lever

    Most applications read far more than they write. Timelines, product pages, dashboards, search results, all reads. So the first structural move is almost always read replicas. You keep one primary that accepts all writes, and you stream its changes to one or more replica copies that serve reads. Now your read capacity scales with the number of replicas while writes stay on the primary.

    The catch is replication lag. A replica is a copy that is always slightly behind, usually by milliseconds but sometimes by seconds under load. This creates a subtle bug class: a user writes something, gets redirected, the read goes to a replica that has not caught up yet, and their own change appears to have vanished. The fix is read your writes routing. After a write, send that user reads to the primary for a short window, or for anything where the user expects to see their own action immediately.

    def get_connection(query_type, just_wrote=False):
        if query_type == "write" or just_wrote:
            return primary_pool.get()
        return replica_pool.get()  # round robin across replicas
    
    # After updating a profile, read from primary briefly
    update_profile(user_id, data)
    profile = read_profile(user_id, just_wrote=True)

    Replicas also give you something beyond capacity. They are a warm standby. If the primary dies, you can promote a replica, which makes replicas part of your availability story and not only your performance story.

    Partitioning, splitting one big table sensibly

    Replicas multiply your read capacity but every replica still holds the entire dataset, and writes still all go to one primary. When a single table grows to hundreds of millions of rows, the table itself becomes the problem. Indexes get deep, vacuum and maintenance get slow, and queries that touch the whole table get sluggish. Partitioning splits one logical table into many physical pieces while keeping a single primary.

    The most useful kind is range partitioning by time. Most large tables are append heavy and time ordered: events, logs, orders, messages. If you partition by month, a query for last week only touches one partition, and dropping old data becomes an instant detach of a whole partition instead of a massive delete.

    CREATE TABLE events (
        id        BIGSERIAL,
        user_id   BIGINT NOT NULL,
        created_at TIMESTAMPTZ NOT NULL,
        payload   JSONB
    ) PARTITION BY RANGE (created_at);
    
    CREATE TABLE events_2026_06 PARTITION OF events
        FOR VALUES FROM ('2026-06-01') TO ('2026-07-01');
    
    CREATE TABLE events_2026_07 PARTITION OF events
        FOR VALUES FROM ('2026-07-01') TO ('2026-08-01');

    The thing to understand is that partitioning is still one database server. It helps with table size, maintenance, and queries that can prune to a single partition. It does nothing for write throughput on the machine as a whole, because every partition lives on the same box.

    Sharding, the last and heaviest rung

    When a single primary cannot keep up with writes no matter how big the machine is, you have to split the data across multiple independent database servers. That is sharding. Each shard is its own database holding a subset of the data, and there is no single machine that has all of it. This is the move that finally scales writes horizontally, and it is also the move that costs you the most.

    Everything depends on the shard key, the column you use to decide which shard a row lives on. Pick well and most queries hit a single shard. Pick badly and you create hot shards that take all the traffic while others sit idle, or you force queries to fan out across every shard and gather results, which is slow and fragile.

    • Hash based sharding spreads rows evenly by hashing the key. Great for uniform distribution, bad for range queries because related rows scatter everywhere.
    • Range based sharding keeps related keys together, which is good for range scans but tends to create hotspots on the newest range.
    • Directory based sharding keeps an explicit lookup table mapping keys to shards. Most flexible, lets you rebalance, but the directory itself becomes something you have to scale and protect.

    The shard key has to match how you actually query. If you run a multi tenant application, sharding by tenant id is usually ideal because almost every query is already scoped to one tenant, so it naturally lands on one shard. If you ever need a query that does not include the shard key, you are looking at a scatter gather across all shards, and you should design hard to avoid those on hot paths.

    What you give up when you shard

    I want to be blunt about the costs, because sharding gets romanticized. Cross shard joins effectively stop existing. You denormalize or you join in the application layer. Transactions that span shards require distributed transaction machinery that is slow and complex, so you redesign to keep each transaction inside one shard. Globally unique ids need a scheme that does not depend on a single sequence, so you move to UUIDs or a snowflake style generator. Rebalancing when one shard fills up is a genuine project, not a config change. And every one of these costs is permanent. Once you shard, you live with it.

    This is exactly why sharding is the last rung. You climb to it only after replicas, partitioning, caching, and queues have all been pushed as far as they go.

    How I sequence the whole thing

    The path I follow almost every time looks like this. Index and tune until the cheap wins are gone. Add read replicas and route reads off the primary. Partition the giant tables so maintenance and pruning stay sane. Use caching and queues, which I cover in backend scalability patterns, to take pressure off both reads and writes. Only when the write primary itself is the hard ceiling do I shard, and I treat the shard key choice as the most important decision in the entire effort.

    The overarching lesson is that database scaling is a sequence of trade offs, not a single upgrade. Each rung buys capacity and charges complexity. The engineers who get into trouble are the ones who jump straight to the most powerful pattern because it sounds impressive, and then spend two years paying for distributed complexity they could have deferred for a long time with a couple of replicas and a good index.