Introduction to Sharding: Shard Keys, Application-Level Sharding, Citus and Distributed SQL

Version note: The examples in this post were run with PostgreSQL 18.7 (two independent containers as shards), Citus 14.2 (on PostgreSQL 18.4; one coordinator and one worker), .NET 10 and Npgsql 10.0.3 in a single-machine lab. The CockroachDB, YugabyteDB and Spanner section wasn't tried in the lab; it's based on the products' documentation.

Guess First

Bookstore, the online bookstore used as the shared example across this site's series, split its orders across two PostgreSQL servers: every customer's record, orders and order items live on a shard computed from customer_id. The marketing team wants a "top 10 best sellers of the last 30 days" list for the home page. The first idea is to ask each shard for its own top 10 and merge them in the application: two shards, 20 rows, fast. Guess: is this list the same as the correct one? Both methods side by side on the same data:

exact: 37644 rows transferred from 2 shards, 347 ms
top-10 per shard: 20 rows transferred, 209 ms
  #1  exact: book    90 sold  18 | shortcut: book   278 sold  16 (really 17)
  #2  exact: book   178 sold  18 | shortcut: book   954 sold  16 (really 17)
  #3  exact: book   266 sold  18 | shortcut: book  1510 sold  16 (really 17)
  #4  exact: book   462 sold  18 | shortcut: book  1746 sold  16 (really 17)
  #5  exact: book   550 sold  18 | shortcut: book  2166 sold  16 (really 17)
  #6  exact: book   638 sold  18 | shortcut: book  2422 sold  16 (really 17)
  #7  exact: book   834 sold  18 | shortcut: book  2578 sold  16 (really 17)
  #8  exact: book   922 sold  18 | shortcut: book  3078 sold  16 (really 17)
  #9  exact: book  1010 sold  18 | shortcut: book  4310 sold  16 (really 17)
  #10 exact: book  1206 sold  18 | shortcut: book  4390 sold  16 (really 17)

Not one of the ten books is shared. The books in the shortcut list sold well on one shard but only reached 17 in total; the books that really sold 18 were split across both shards and so didn't make either shard's top 10. Finding the right result took pulling per-book totals (37,644 rows) from every shard and adding them up in the application. That's exactly the price of sharding: queries that are a single GROUP BY on one server turn into a distributed algorithm. In this post we look at what sharding solves, choosing a shard key, an application-level shard resolver in .NET, database-level sharding with Citus, and why Bookstore doesn't need sharding yet.

What Does Sharding Solve, and What Does It Bring?

Sharding means splitting the rows of data with the same schema across different servers. Partitioning in posts 07 and 08 splits data inside a single server; sharding multiplies servers. It solves two problems: a write load a single primary can't carry, and a data size that won't fit on a single server. The read replicas of post 06 spread the read load, but every write still goes to a single primary; when the write side hits its limit, sharding is the ladder's last rung.

The problems it brings are permanent too: queries and transactions that touch more than one shard, the global unique ID that's free on a single server, redistributing data across shards (resharding), and as many backups, replicas, failovers and schema changes as there are shards. The theory and the comparison of hash, range and directory approaches are in the System Design series' posts on scaling the data layer and on consistent hashing. Here we look at the practice.

For Bookstore the numbers are clear. According to post 01's capacity estimate, 10x growth means about 12,900 orders a day and 3.1 GB of new data a year. That is a write load a single PostgreSQL or SQL Server comfortably carries (0.15 orders per second on average; even a hundred times that at a campaign peak is small for one server). In post 07's pgbench run a single server handled 9 thousand transactions per second, each one read and one write. Bookstore doesn't need sharding; this post is about showing what you'll run into when the need arises.

Choosing a Shard Key

The shard key is the column that decides which shard a row goes to, and it's the most expensive decision to change later. A good shard key provides three things: the load is spread evenly across shards, the most frequent queries go to a single shard, and data that changes together lives on the same shard. Bookstore has two candidates: customer_id and the order date. The distribution on the two shards (split by hash on customer_id):

== shard0
 year | orders | last_30_days
------+--------+--------------
 2022 |    556 |            0
 2023 |  39575 |            0
 2024 | 102091 |            0
 2025 | 164572 |            0
 2026 | 193197 |        19938
== shard1
 year | orders | last_30_days
------+--------+--------------
 2022 |    555 |            0
 2023 |  39591 |            0
 2024 | 102076 |            0
 2025 | 164595 |            0
 2026 | 193194 |        19937

With hash, the last 30 days' orders split into 19,938 and 19,937 across the two shards. Had range sharding by date been chosen (one shard per year), all 39,875 orders of the last 30 days would have been written to the 2026 shard, while the 2022 shard sat idle with 1,111 orders in total. Range sharding on a monotonically increasing key (a date, an identity, UUIDv7) always produces a hot shard: all new writes go to the last shard. Hash, on the other hand, keeps queries that need all of a customer's data, such as "My orders", on a single shard, but spreads date-range queries across all shards. On the Institutional Orders side the natural key is institution_id: a school's units, users and orders stay on the same shard. That's the generalised form of the database-per-tenant model in the Solution Patterns series' post on data isolation in multi-tenancy.

Application-Level Sharding: IShardResolver

The simplest sharding is the application choosing the connection. In Bookstore the customer number is first mapped to one of 64 fixed buckets, and the bucket is mapped to a physical shard through a table:

// customer -> one of 64 buckets (fixed, never changes) -> physical shard (a directory that can change)
public sealed class BucketShardResolver(IReadOnlyList<NpgsqlDataSource> shards, int[] bucketToShard) : IShardResolver
{
    public const int BucketCount = 64;

    // same formula as bucket_of() in SQL: Knuth's multiplicative hash, top 6 bits of 32
    public static int BucketOf(int customerId) => (int)((((ulong)(uint)customerId * 2654435761UL) % 4294967296UL) >> 26);

    public int ShardFor(int customerId) => bucketToShard[BucketOf(customerId)];

    public NpgsqlDataSource DataSourceFor(int customerId) => shards[ShardFor(customerId)];

    public IReadOnlyList<NpgsqlDataSource> AllShards => shards;
}

The two layers are deliberate. A direct formula such as customer_id % 2 would require two thirds of the customers to move when the shard count goes from 2 to 3. The bucket count, on the other hand, is chosen large up front and never changes; when a new shard is added, only some buckets' data moves and their rows in the directory are updated. Consistent hashing's virtual nodes apply the same idea. The formula also has a SQL counterpart (bucket_of); when spreading data to shards and moving it, the application and the database must reach the same answer. string.GetHashCode() or HashCode.Combine can't be used for this: .NET computes them with a random seed per process, so a restarted application looks for the customer on another shard.

Queries for a customer's own data go to a single shard:

customer 4242: bucket 44, shard 1, 10 orders (80.4 ms)
customer 4243: bucket 20, shard 0, 10 orders (9.2 ms)
customer 77777: bucket 53, shard 1, 10 orders (0.6 ms)

The timings on the first two lines are the first connection being opened to that shard; after that the pool takes over. Each shard is a separate NpgsqlDataSource, that is a separate connection pool; post 02's connection budget is now multiplied by the number of shards. Moving a bucket (resharding) is a three-step job: copy the rows of the bucket's customers to the new shard, carry over the writes that arrive while copying (logical replication or a short write freeze), then change the row in the directory and delete the old rows. The logical replication from post 04 is used for this job too.

Global IDs: Identity Is No Longer Unique

Every shard keeps its own identity sequence. When two customers on two different shards order at the same time:

customer 4242 on shard 1: order_id 1000001
customer 4243 on shard 0: order_id 1000001

Two different orders got the same number. Within a shard that's fine, but the order number is used in e-mails, on shipping labels, in the reporting database and in the accounting system without any shard information. The fix is generating the ID independently of the shard: UUIDv7 (Guid.CreateVersion7(), which we used for the idempotency key in post 05), 64-bit IDs made of time, machine and counter such as Snowflake, or sequences that give each shard its own range. The comparison of the options is in the System Design series' post on unique ID generation; separating the number shown to customers from the internal ID is in the Solution Patterns series' post on exposing IDs.

Cross-Shard Queries

The best-sellers query at the top of the post shows the right way to write cross-shard queries: partial totals on each shard, final totals in the application:

List<(int BookId, long Sold)>[] perShard = await Task.WhenAll(resolver.AllShards.Select(shard => QueryAsync(shard, perBook)));
Dictionary<int, long> totals = perShard.SelectMany(rows => rows)
    .GroupBy(r => r.BookId)
    .ToDictionary(g => g.Key, g => g.Sum(r => r.Sold));
List<(int BookId, long Sold)> exact = [.. totals
    .Select(t => (BookId: t.Key, Sold: t.Value))
    .OrderByDescending(r => r.Sold).ThenBy(r => r.BookId)
    .Take(10)];

The queries go out in parallel with Task.WhenAll, so the total time is that of the slowest shard. COUNT, SUM, MIN and MAX combine correctly from partial results; for an average, sum and count must be fetched separately; values such as medians and percentiles can't be computed from partial results at all. ORDER BY ... LIMIT, as we saw at the top, gives the wrong result if the sort key isn't the shard key. As the number of shards grows, so do the rows transferred and the time spent waiting for the slowest shard.

Cross-shard writes are harder. An operation that must write to two shards together (for example an institution's order affecting customers' stock on different shards) needs a distributed transaction; why that's avoided is in the Distributed Systems Fundamentals series' post on distributed transactions and two-phase commit. The practical rule is to choose the shard key so that data changing together lives on the same shard, and to leave the remaining cases to eventual consistency through events. Reporting is usually done not on the shards but on a separate analytical store fed from them. The OLTP/OLAP split and star schemas are in the Data Modeling and Normalization series' post on OLTP and OLAP modeling.

Database-Level Sharding: Citus

Citus makes PostgreSQL distributed as an extension: a coordinator receives queries, tables are split into shards spread over worker nodes, and the query plan is sent to every shard. Citus has been fully open source since version 11; on Azure its managed form is offered as Azure Database for PostgreSQL elastic clusters. In the lab we set up one coordinator and one worker and let the coordinator hold shards as well. The same rule as at the application level showed up here too:

CREATE TABLE orders_try (
    order_id    bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
    customer_id int NOT NULL,
    created_at  timestamptz NOT NULL
);
SELECT create_distributed_table('orders_try', 'customer_id');

ERROR:  cannot create constraint on "orders_try"
DETAIL:  Distributed relations cannot have UNIQUE, EXCLUDE, or PRIMARY KEY constraints that do not include the partition column (with an equality operator if EXCLUDE).

The same rule as in partitioning: each shard can only guarantee uniqueness within itself. Bookstore's schema was built on Citus with three kinds of tables:

SELECT create_reference_table('books');
SELECT create_distributed_table('customers', 'customer_id');
SELECT create_distributed_table('orders', 'customer_id', colocate_with => 'customers');
SELECT create_distributed_table('order_items', 'customer_id', colocate_with => 'orders');
ALTER TABLE order_items ADD FOREIGN KEY (customer_id, order_id) REFERENCES orders;
ALTER TABLE order_items ADD FOREIGN KEY (book_id) REFERENCES books;

books is a reference table: every node has a full copy, so every shard can join with books locally. Customers, orders and items were distributed by customer_id and co-located: the same customer's rows live on the same node, with the same shard number. For that the primary key of orders became (customer_id, order_id) and a customer_id column was added to order_items. The foreign keys were added after the tables were distributed; a foreign key from a local table to a distributed table can't be defined. The default 32 shards spread evenly over the two nodes:

 nodename | shards |  size
----------+--------+--------
 citus-c  |     16 | 136 MB
 citus-w1 |     16 | 136 MB

The "My orders" query was routed to a single shard (a router query; plan shortened):

 Custom Scan (Citus Adaptive)
   Task Count: 1
   Tasks Shown: All
   ->  Task
         Node: host=citus-c port=5432 dbname=postgres
         ->  Limit
               ->  Sort
                     Sort Key: created_at DESC
                     ->  Bitmap Heap Scan on orders_102053 orders
                           Recheck Cond: (customer_id = 4242)

The best-sellers query went to all 32 shards, and Citus did itself what we did by hand in the application: a partial total per book_id on each shard (HashAggregate), final totals and sorting on the coordinator (plan shortened):

 Limit
   ->  Sort
         Sort Key: ((pg_catalog.sum(remote_scan.sold))::bigint) DESC, remote_scan.book_id
         ->  HashAggregate
               Group Key: remote_scan.book_id
               ->  Custom Scan (Citus Adaptive)
                     Task Count: 32
                     Tasks Shown: One of 32
                     ->  Task
                           ->  HashAggregate
                                 Group Key: oi.book_id
                                 ->  Hash Join
                                       Hash Cond: ((oi.customer_id = o.customer_id) AND (oi.order_id = o.order_id))

The result matched the correct list we computed in the application (books 90, 178, 266… each 18 copies). When we forgot to add customer_id to the join condition, Citus refused the query:

ERROR:  the query contains a join that requires repartitioning
HINT:  Set citus.enable_repartition_joins to on to enable repartitioning

Co-located tables are joined within a shard only when joined on the distribution column; otherwise rows have to be redistributed between nodes. Citus keeps that off by default, because its cost grows with the data size. Carrying the distribution column into every query in application code is the first rule of using Citus efficiently.

Distributed SQL: CockroachDB, YugabyteDB and Spanner

Another family is the distributed SQL databases, which build sharding, replication and distributed transactions into the engine itself. CockroachDB and YugabyteDB speak the PostgreSQL wire protocol, and Google Cloud Spanner has a PostgreSQL interface. They split data into ranges automatically, replicate the ranges with consensus (Raft or Paxos), rebalance by themselves when nodes are added, and hide cross-shard transactions from the application. The price: since every write waits for a consensus round, single-row write latency is higher than on a single PostgreSQL server, PostgreSQL compatibility is not complete (extensions, some types and functions), and the operating and licensing models are different. When to consider them: when writing in several regions at once, automatic regional failover and a load beyond a single server's write limit are needed together. None of these apply to Bookstore's current scale. We didn't try the products in this section in the lab; check their licensing and pricing on their official pages on the day you read this. The ideas behind them (Spanner's TrueTime, CockroachDB's hybrid logical clock) are in the Distributed Systems Fundamentals series' post on lessons from real systems. MongoDB's sharding is in the MongoDB series' post on replica sets and sharding.

In Production

Keep sharding as the last rung. The earlier rungs of post 01's ladder (query and index, pooling, cache, vertical scaling, read replicas, partitioning and archiving) carry most systems for years. Before moving to sharding, measure which resource is running out, as in post 01: disk writes, log volume, CPU or storage.

Choose a large bucket count up front. A bucket count such as 64 or 256 lets you grow later to 4, 8 or 16 shards by moving buckets, even with two shards on day one. Changing the bucket count means redistributing all the data.

Every shard is a database server. The replicas and failover of posts 04 and 05, the partitioning and archive jobs of posts 07 and 08, schema migrations and backup/restore tests now come as many times as there are shards. You need a tool that applies migrations to all shards and notices one that's half done.

Measure and limit cross-shard queries. Every query going to all shards is as slow as the slowest shard. Keep a log or metric that spots these queries; if their number grows, the shard key may have been chosen wrong.

Common Pitfalls

Range sharding on a monotonic shard key: with shards split by date, all new writes go to the last shard. Use hash or a directory.

Merging each shard's top N: the list at the top of the post was completely wrong. If the sort key isn't the shard key, fetch partial totals and add them up in the application, or leave the job to an analytical store.

Embedding cross-shard JOINs in application code: code that pulls data from two shards and joins it in memory gets slower and heavier on memory as data grows. Keep data read together on the same shard (co-location) or move the join to the reporting side.

Treating identity as a global ID: two shards produced the same order_id. Generate IDs independently of the shard.

Choosing the shard with GetHashCode(): .NET string hashes change per process; a restarted application looks for data on another shard.

A Small Experiment

★☆☆ Add a third shard to BucketShardResolver and write a directory that spreads the 64 buckets over three shards. How many buckets, and so what percentage of customers, have to move? Answer the same question for customer_id % shardCount.

★★☆ Add "average order amount per category" to the best-sellers query. Work out which partial values you need from each shard, and verify the result against the same query on a single-server PostgreSQL.

★★★ Add a third node to Citus with SELECT citus_add_node(...) and rebalance the shards with citus_rebalance_start(). While the rebalance runs, run the "My orders" query in a loop similar to post 06's rw-routing loop and measure whether there are errors or delays.

Over the series we have seen every rung of the ladder. In the last post we put them together: starting measurements for Bookstore's 10x growth, the target SLO, the steps taken in order, a decision record for each step, verification with the same metric, and a failover drill.