Scaling the Data Layer: Replication, Sharding and SQL vs NoSQL

Version note: The code in this post was built and run with C# 14 and .NET 10; the sharding example ran with Microsoft.Data.Sqlite 10.0.14, with each shard as a separate SQLite file.

Guess First

Bookstore has grown: a million orders, twenty thousand customers. The orders table has been split across four databases, and customer_id decides which database each order goes to; each database also has a replica for reads. The "My orders" page is faster than ever.

Two questions. The finance team wants "the 20 biggest orders of the last 30 days"; how many databases does that query go to, and how many rows travel to the application? Second: Ada places an order and is immediately redirected to the "My orders" page, which is read from a replica. Does Ada see her new order?

Access Patterns First, Database Second

Scaling the data layer does not start with picking a database product, it starts with listing how the data is read and written. That is also the core stance of Kleppmann and Riccomini's Designing Data-Intensive Applications (2nd edition, 2026): the data model is chosen according to the questions the application will ask. Bookstore's list:

• Orders: a customer reads their own orders and updates order status; money and stock are involved, so transactions and constraints are required.
• Catalog: 200 searches per second, full text and filters; read-heavy, a few seconds of staleness is acceptable.
• Cart: a small document per user, written often, losing it is not a disaster.
• View and like counters: written very often, an approximate value is enough.
• Reporting: aggregations over large ranges, delays of minutes are acceptable.

This list does not give a single answer to "SQL (Structured Query Language) or NoSQL", it gives several. Orders stay in a relational database, because what they need is transactions and constraints. A search engine for catalog search, a key-value store like Redis for the cart, a write-heavy store for counters and a separate analytical copy for reporting can all make sense. NoSQL is not a product category but an umbrella name for "everything outside the relational model"; document, key-value, wide-column and graph databases each fit a different access pattern well. The cost is that every new store is an operational burden and a problem of keeping data in sync. For a small team like Bookstore's, the right start is to use the single relational database as far as it goes; the rest of this post looks at what happens when you need to go beyond that point.

Replication: Leader-Follower and Read Replicas

Replication means keeping a copy of the same data on more than one machine. Its most common form is leader-follower (primary-replica): all writes go to the leader, the leader sends its change log to the followers, and the followers can serve reads. Readable secondaries of Always On availability groups in SQL Server and streaming replication in PostgreSQL are examples.

Replication solves two separate problems. Availability: if the leader goes down a follower takes its place (failover). Read capacity: reads are spread across several replicas. It does not increase write capacity; every write still goes through a single leader and has to be applied on every follower too.

The critical choice is synchronous or asynchronous replication. In synchronous mode the leader does not say "done" to the client until the follower confirms it has the write; the follower is always current, but every write is as slow as the follower, and if the follower is unreachable writes stop. In asynchronous mode the leader confirms immediately and ships the change afterwards; writes are fast, but the follower lags behind, and if the leader crashes at that moment the last acknowledged writes can be lost. In practice most systems use one synchronous follower in the same datacenter and asynchronous followers in a remote region. The theory of replication, quorums and leaderless models is in the upcoming Distributed Systems Fundamentals series post "Replication: Leader-Based, Multi-Leader, Leaderless and Quorums".

Replication Lag and Read-Your-Writes

The everyday face of asynchronous replication is replication lag: how many milliseconds or seconds the follower is behind the leader. Let's answer the second opening question with a small simulation. Instead of setting up real database replication, we write a fake replica that applies the change log in order with a 200 ms delay:

public long Write(long orderId, string status)
{
    _leader[orderId] = status;
    long position = Interlocked.Increment(ref _leaderPosition);
    _log.Writer.TryWrite((position, orderId, status, DateTime.UtcNow));
    return position;   // the client keeps this position in its session
}

// read from the replica if it has reached the client's last written position, otherwise from the leader
public (string? Status, string ServedBy) ReadYourWrites(long orderId, long lastWrittenPosition) =>
    ReplicaPosition >= lastWrittenPosition
        ? (_replica.GetValueOrDefault(orderId), "replica")
        : (_leader.GetValueOrDefault(orderId), "leader");

replica, right after write: not found
read-your-writes:           Placed (served by leader)
after 300 ms:               Placed (served by replica), replica applied position 1

The first line is the answer to the opening question: Ada placed her order, the page was read from the replica, and the order is not there. To the user this looks like data loss; she places the order again and two orders exist. This is called a violation of read-your-writes (read-after-write) consistency.

The fix is to guarantee that users always see their own writes. The method in the example is the most robust: the leader gives every write an increasing log position (an LSN (Log Sequence Number) in SQL Server, a WAL (Write-Ahead Log) position in PostgreSQL), the client keeps that position in its session, and if the replica has not reached it the read goes to the leader. On the second line the read came from the leader; on the third the replica had caught up and the read went back to the replica. There are simpler alternatives too: always reading a user's own data from the leader, or reading from the leader for a few seconds after the last write. Both put load on the leader; but pages showing the user's own data, like "My orders", are a small fraction, and data everyone reads, like the catalog, can stay on replicas.

Partitioning: Range, Hash and Directory

Replication scales reads, not writes. When write capacity or data size outgrows a single leader you have to split the data: partitioning, or sharding. Each shard is an independent database holding a subset of the data, and the shard key decides which record lives on which shard.

There are three basic methods. In range partitioning, ranges of the key are split across shards: customers A–F on one shard, January orders on another. Range queries are easy, but the load depends on the key's distribution. In hash partitioning, the key's hash is split by the number of shards; distribution is even, but range queries have to go to every shard. In directory-based partitioning, a lookup table holds which shard each key lives on; it is the most flexible method, moving a big customer to a shard of its own is a one-row change, but the lookup table itself becomes a component on every request's path that needs high availability. The shard map of Azure SQL's Elastic Database client library is an example of this approach.

Experiment: Sharding Orders by customer_id

Four shards, each a separate SQLite file; a million orders, twenty thousand customers. That the engine is SQLite does not matter, what we are showing is how a query spreads across shards. The shard key is customer_id:

// shard key = customer_id; all orders of a customer live on the same shard
public static int ShardOf(int customerId, int shardCount) => (int)((uint)customerId * 2654435761u % (uint)shardCount);

The multiplication is a simple multiplicative hash (Knuth's golden ratio constant for 2^32) so that consecutive ids spread evenly across shards. The "20 biggest orders of the last 30 days" query has no shard key, so it goes to all four shards at once and each shard returns its own top 20:

// each shard returns its own top 20, the application merges and sorts again
public async Task<ShardResult> TopOrdersLast30DaysAsync(int take)
{
    string sql = $"SELECT id, created_on, total FROM orders WHERE created_on >= $p ORDER BY total DESC LIMIT {take}";
    string since = Today.AddDays(-30).ToString("yyyy-MM-dd");
    // Microsoft.Data.Sqlite's async methods run synchronously; Task.Run for parallelism
    var perShard = await Task.WhenAll(Enumerable.Range(0, shardCount).Select(s => Task.Run(() => QueryShardAsync(s, sql, since))));
    List<(long Id, double Total)> merged = [.. perShard.SelectMany(r => r.Items).OrderByDescending(o => o.Total).Take(take)];
    return new ShardResult(shardCount, merged.Count, perShard.Sum(r => r.Rows), perShard.Max(r => r.Ms), perShard.Min(r => r.Ms));
}

orders of customer 1234   shards touched: 1, rows: 47, 0.1 ms
top 20 of last 30 days    shards touched: 4, rows: 20, rows moved to app: 80, 6.4 ms
  slowest shard 6.3 ms, fastest shard 6.1 ms
plan, by customer: SEARCH orders USING COVERING INDEX ix_orders_customer (customer_id=?)
plan, last 30 days: SCAN orders | USE TEMP B-TREE FOR ORDER BY

The answer to the first opening question is here. The customer query went to a single shard, read 47 rows from the index and took 0.1 ms. The last-30-days query went to four shards, each shard scanned its 250,000 rows from start to end (SCAN orders) and sorted them; 80 rows travelled to the application and 20 were kept. The total time equals the slowest shard: the latency of a scatter-gather query is set by the shards' p99, and the fan-out math from "Back-of-the-Envelope Estimation: QPS (queries per second), Storage, Bandwidth and Availability Math" applies here too. When the shard count goes from 4 to 40, the same query goes to 40 shards, moves 800 rows, and the chance that one of the 40 shards is slow grows 40 times.

The fix is not making the query faster but not asking this data model that question. The rule of sharding is: the most frequent and most critical queries must include the shard key. A reporting query like "orders of the last 30 days" should be asked of a separate reporting copy that orders flow into, not of the operational shards. Pagination is another trap: OFFSET 1000 LIMIT 20 asks each of the four shards for 1,020 rows; that is why cursor-based pagination is close to mandatory with sharding.

Hot Partitions and Resharding

A good shard key balances not only queries but also load. Let's look at how today's 600 orders spread under hash and month-based range sharding:

today's 600 writes  hash(customer): [158, 146, 148, 148]  range(month): [0, 0, 0, 600]
4 -> 5 shards with hash % N: 16,001 of 20,000 customers move (80%)

With month-based range sharding all of today's writes go to a single shard; the write capacity of the other three sits idle. Every key that grows with time (a date, an auto-increment id) produces this hot partition. With a hash the spread is even. But a hash cannot split a single hot key either: a corporate customer that places all its orders under the same customer_id heats up wherever it lands in the hash. The remedy is either moving that customer to its own shard with a directory, or adding something to the key (a sub-customer, a branch).

The second line is the long-term cost of sharding. When four shards become five, with hash % N 80% of customers have to move to another shard: eight hundred thousand of a million orders copied in a live system. That is why resharding is one of the most expensive operations, with two common remedies: defining far more logical partitions than shards (for example 1,024) and moving only partitions between physical servers, and consistent hashing. The second is the topic of the series' next post.

Denormalization

The silent cost of sharding is joins. If orders are split by customer_id and books by some other method, "orders with book titles" is no longer a JOIN running in a single database but two separate queries merged in the application. The common solution is denormalization: copying the book title and the price at that moment into the order line. This is right from a business point of view anyway; an order should show the price at the time it was placed, not today's price. When copied data changes (a book's title gets corrected), it is updated with an event and can stay stale for a while.

The same idea applies to reporting: streaming changes from the operational shards to a separate analytical store with CDC (Change Data Capture) or events, and asking questions like "last 30 days" there. This is the data-layer counterpart of the read model idea in the CQRS (Command Query Responsibility Segregation) and Event Sourcing series.

CAP and PACELC: A Decision Lens per Feature

In every replication and sharding decision the same question comes back: when a replica falls behind or a shard is unreachable, do we show stale data or return an error? The CAP (Consistency, Availability, Partition tolerance) theorem names this choice during a network partition, and Daniel Abadi's PACELC (if Partition, then Availability or Consistency; Else, Latency or Consistency) formulation names the choice between latency and consistency when there is no partition. The precise statement of the theory and the common misreadings are in the upcoming Distributed Systems Fundamentals series post "Reading CAP and PACELC Correctly"; here we use it only as a decision lens.

In practice you make the decision per feature, not per system. In Bookstore, payments and stock decrements are read and written on the leader; if the replica is behind, we wait, and if the leader is unreachable the operation is rejected, because a wrong balance or overselling is more expensive than a stale page. Order history is read from replicas with read-your-writes. Catalog and book details come from replicas and the cache and can be seconds stale. Like and view counters are eventual: if a counter lags a few seconds nobody notices, but if a page fails to open because of a counter everyone does. Same database, same replicas; a different read rule per feature.

Common Pitfalls

Taking read replicas for write scaling. Replicas scale reads; every write still goes through the leader and is applied again on every replica.

Reading users' own writes from a replica. Not seeing an order after placing it pushes the user to a second order; design read-your-writes explicitly.

Choosing the shard key without looking at the queries. If the most frequent query does not include the shard key, every request becomes a scatter-gather and the gain from sharding is lost.

Range sharding on a key that grows with time. All new writes go to the last shard; a hot partition is guaranteed.

Starting with hash % N and leaving resharding for later. Changing the shard count means moving most of the data; consider logical partitions or consistent hashing from the start.

Sharding too early. Moving to sharding before vertical scaling, indexes, caching and read replicas are exhausted means taking the most expensive and least reversible step first.

A Small Experiment

★☆☆ Add an index on created_on in the example (CREATE INDEX ix_orders_created ON orders (created_on)) and look at the last-30-days query's plan and time again. Did the scatter-gather go away?

★★☆ Add a second replica to ReplicatedStore; make one 50 ms and the other 2 seconds behind. If you spread reads across the two replicas in turn, can a user see an order and then not see it? What is this anomaly called, and how do you prevent it?

★★★ Turn ShardOf into a structure that maps customers to 1,024 logical partitions and partitions to four physical shards through a lookup table. When a fifth shard is added, how many partitions, and so how many orders, have to move? Compare with the 80% of hash % N.

The 80% figure from resharding is the starting point of the next post: a distribution method that moves only what is necessary when the number of nodes changes. We will write consistent hashing in C# ourselves, with a hash ring and virtual nodes, and measure the fraction of moved keys with an xUnit test.