Database Sharding and Partitioning

Advanced System Design & Software Architecture · Lesson 30 · High-Level Design and Distributed Systems

Lesson Overview

The orders table from Database Design and Indexing for Scale is indexed, and Database Replication and Consistency has taken history reads off the primary. Checkout writes still land on one PostgreSQL primary. Enrollment week pushes that primary's WAL, CPU, and connection count together. A replica cannot take the insert. Another B-tree cannot take the insert. The write path needs more than one writer, or it needs the table broken into pieces a single primary can still maintain.

Those are different moves, and teams mix the names. Partitioning in PostgreSQL keeps one database and one catalog. Declarative partitions are child tables. The planner can skip partitions that cannot match the query. A transaction can still touch two partitions with ordinary commit. You gain maintenance boundaries and pruning. You do not gain a second primary. Sharding places rows in different databases, each with its own primary. Writes scale because tenant A and tenant B do not share a WAL. You lose cross-shard foreign keys, cross-shard unique constraints, and a single commit. Citus is a PostgreSQL extension that shards by turning a distribution column into shards on worker nodes and routing SQL through a coordinator. It is still sharding. PARTITION BY HASH on one instance is not.

This lesson chooses a shard key for TechLearningPro's multi-tenant academy, shows why a time-ordered key creates a hotspot, and treats resharding as a data movement project. Native partitioning concepts and Citus functions that actually exist (create_distributed_table, create_reference_table, rebalance_table_shards) are the concrete tools. Consistency across shards is CAP Theorem and Consistency Models and Distributed Transactions and the Saga Pattern. Companion lessons: Partitioning, Sharding Concepts, Row-Level Security.

Learning Objectives

By the end of this lesson you will be able to:

1. Separate single-node declarative partitioning from sharding across databases, including what each does to transactions and unique constraints.
2. Choose a shard key from the queries that must stay on one node, and reject a key that piles writes onto one shard.
3. Explain a hotspot caused by a monotonic key, a low-cardinality key, or one tenant that dominates a hash bucket.
4. State the PostgreSQL rule that a primary key or unique constraint on a partitioned table includes the partition key.
5. Use range partitions for time-based detach and hash partitions for tenant spread on one cluster, and say which one adds write throughput.
6. Describe Citus distribution, reference tables, and colocation well enough to keep a tenant's orders and lines on the same worker.
7. List what breaks when the shard count or the distribution column must change: movement, dual writes, identifiers, and cross-shard queries.
8. Route a Spring request to a shard from a tenant id stored in a lookup table rather than only from hash % N.
9. Keep cross-tenant reports off the OLTP shards.
10. Review a multi-tenant proposal for isolation, noisy neighbors, and a tested move of one tenant.

Prerequisites

• Lessons 28 and 29: a single primary, indexes matched to statements, and replication inside one cluster. Each shard in this lesson is one of those clusters.
• You can read a foreign key and you know it cannot reference a table in another database.
• Capacity Estimation and Workload Modeling supplies the write rate that makes one primary the bottleneck. If the number does not say that, do not shard.
• Java 21 and Spring AbstractRoutingDataSource, used in lesson 29 for replicas. Here the lookup key is a shard id.
• Citus is optional infrastructure. The functions named in this lesson are the documented Citus SQL API. Where you do not run Citus, the same key-design rules apply to application-level shards.

Detailed Core Concepts

Partitioning is one database

PostgreSQL declarative partitioning splits a logical table into child tables. The parent holds no rows. Range, list, and hash are the built-in methods.

sqlCopy
CREATE TABLE commerce.orders (
    tenant_id   uuid NOT NULL,
    id          bigint NOT NULL,
    placed_at   timestamptz NOT NULL,
    status      text NOT NULL,
    total_cents integer NOT NULL,
    PRIMARY KEY (tenant_id, id)
) PARTITION BY HASH (tenant_id);

CREATE TABLE commerce.orders_p0 PARTITION OF commerce.orders
    FOR VALUES WITH (MODULUS 8, REMAINDER 0);

The primary key includes tenant_id because PostgreSQL requires every unique constraint on a partitioned table to include the partition key. Without that, two partitions could each contain id = 10 and the parent could not enforce uniqueness without searching every child on every insert. The rule is the engine refusing a guarantee it cannot keep cheaply. A unique public_id that does not contain tenant_id cannot be declared on this parent. If clients need a globally unique public id, generate it so collisions are practically impossible (a UUID) and enforce uniqueness in a directory table, or include the tenant in the key the database checks.

enable_partition_pruning defaults on in current PostgreSQL. A query WHERE tenant_id = $1 on a hash-partitioned parent touches one partition when the planner can see the constant. A query that omits tenant_id touches every partition. The application that "falls back" to a scan of the parent is a fan-out. Partition pruning is an optimization for statements that carry the key. It is not a shard router with extra nodes.

Range partitions fit append-mostly time data:

sqlCopy
CREATE TABLE commerce.events (
    tenant_id uuid NOT NULL,
    happened_at timestamptz NOT NULL,
    payload jsonb NOT NULL,
    PRIMARY KEY (tenant_id, happened_at)
) PARTITION BY RANGE (happened_at);

You attach a month and later ALTER TABLE commerce.events DETACH PARTITION commerce.events_2024_01 CONCURRENTLY (concurrent detach is PostgreSQL 14+). Detach is how retention stays an metadata operation plus a drop, once queries no longer need the month. Range on happened_at puts this week's inserts into one child. That child is a hotspot on a single primary: one partition receives the whole write rate. Hash the tenant if the goal is spread across children. Range the time if the goal is dropping old months. A composite need (tenant spread and time drop) is often a hash partition of tenants with each child itself range-partitioned, or a shard per tenant group plus range partitions inside the shard. One method does not deliver both outcomes.

Indexes on the parent are created on each partition. They follow lesson 28's rules, repeated per child. Autovacuum and reindex can target one child. That operational win is real on one machine. The WAL of all children still hits one primary.

Sharding is many databases

A shard is a PostgreSQL primary that owns a subset of the key space, with its own replicas from lesson 29. The application, or a coordinator, chooses the shard. There is no parent table the planner can prune across servers. A transaction that updates tenant A's order and tenant B's invoice is two commits. Partial failure is a saga or an explicit repair, not ROLLBACK.

Vertical partitioning, by contrast, splits columns, not rows: payment tokens leave the hot order row and live in commerce.payments. That belongs to lesson 28's TOAST and width discussion. It does not add a writer. This lesson uses "partitioning" for horizontal splits unless it says vertical.

The shard key is the queries you refuse to scatter

Pick the key from the transaction that must be atomic and local.

TechLearningPro's academy is multi-tenant. A tenant is a company with many learners. Checkout for a learner inserts an order and lines and a seat count for that tenant. Those rows must share a transaction. The shard key is tenant_id. Learner id is a poor primary key for sharding if one learner's activity is a small part of a tenant invoice that must commit with it, and if two learners in the same company must be read together on the admin screen. Learner id can still be the leading index column inside the shard, as in lesson 28.

A good key has three properties at once:

1. Locality of the transaction. Rows written together share the key.
2. Spread of the write rate. The key's hot values do not all hash to one shard.
3. Presence on the query. Request-path statements include the key so the router picks one shard. A statement that cannot include it is a scatter and must be rare, asynchronous, or redesigned.

order_id alone fails locality: the line item insert has to discover the shard of an order id, which means a lookup, and a tenant admin query becomes "find every shard that might hold this tenant." placed_at fails spread: every new order is "now," so one range shard takes 100% of inserts while old shards sit idle. status fails cardinality: a handful of values, one of them paid, owns the table. tenant_id matches checkout and the admin screen. It fails when one tenant is half the platform. That tenant is a hotspot you isolate onto its own shard, which is why the mapping from tenant to shard should be data, not only a modulus.

Hotspots

A hotspot is a shard whose load stays unfair no matter how many other shards you add.

• Monotonic keys. Serial ids and insertion timestamps append to one end. Hash the key if you need spread. Do not hash away a key you needed for range deletes unless retention moved elsewhere.
• Low cardinality. country_code with most customers in one country is one shard and a diagram of sixteen shards.
• Celebrity tenants. Hash(tenant_id) is even across tenants and uneven across bytes. The tenant with five million learners occupies one shard's entire disk and WAL. More shards do not move that tenant. A dedicated shard does.
• Hot row inside a shard. A single counter row for "seats left" serializes transactions even when the shard is otherwise idle. Split the counter into buckets or hold the invariant in a single row on purpose and accept the queue. Sharding the rest of the table will not fix that row.

Measure hotspots with writes per shard, not with shard count. Sixteen shards with 90% of WAL on shard 3 is a three-shard system wearing a costume.

Citus, stated with the functions it has

Citus adds a coordinator and workers. You distribute a table by a column:

sqlCopy
SELECT create_distributed_table('commerce.orders', 'tenant_id');
SELECT create_distributed_table('commerce.order_lines', 'tenant_id');

Tables distributed on the same column, with the same shard count and colocation, keep matching tenant_id values on the same worker. The order and its lines commit in a local transaction on that worker. A join on tenant_id is a set of local joins, not a pull of every line to the coordinator.

Small tables that must sit next to every shard, such as a country list, are reference tables:

sqlCopy
SELECT create_reference_table('commerce.currencies');

Citus replicates a reference table to the workers. It is for small, rarely written dimensions. A large "reference" table repeats a write on every worker.

SELECT rebalance_table_shards('commerce.orders'); moves existing shards between workers to even disk or to fill a new worker. It does not by itself change how many shards a row's key hashes into, and it does not change the distribution column. Those two changes rewrite data. Treat them as migrations with a rollback story, not as a configuration toggle.

A query that filters tenant_id = $1 can be routed to one shard. A query that sums revenue across all tenants runs on every shard and merges at the coordinator. That merge is legitimate for a nightly job and hostile for a web request. The coordinator is also a PostgreSQL node with its own CPU. A fan-out query storm lands there.

Citus does not repeal the shard-key rules. A distribution column of placed_at hotspots the latest shard. A missing tenant_id predicate fan-outs. Cross-tenant uniqueness is still not a local unique index.

Resharding pain

You will reshard for one of four reasons: a shard is too big, a shard is too hot, you need more workers, or the key was wrong. Each one moves rows that are being written.

Application-level modulus (hash(tenant) % 16) makes "add a 17th shard" a rewrite of almost every row if you change the modulus. A lookup table (tenant_id to shard_id) makes a move a copy of one tenant. The lookup is a small strongly consistent map, often its own PostgreSQL primary with replicas. Cache it in the application and invalidate on move. The lookup is allowed to be boring. It is not allowed to be wrong, because a stale cache writes a tenant's new order to the old shard.

A tenant move, in order:

1. Mark the tenant read-only, or start a catch-up mechanism you have tested.
2. Copy rows to the destination shard.
3. Apply changes that happened during the copy. Logical replication filtered to the tenant, or a short write stall, is the usual tool. Dual writes to both shards are easy to start and hard to make idempotent.
4. Flip the lookup row in one transaction.
5. Read from the new shard only after the flip is visible to routers.
6. Keep the old rows until a lagging router cannot still write them. Then delete.

During the flip, lesson 29's read-your-writes problem returns inside one tenant: a router with a stale lookup reads the old shard. Version the lookup and refuse writes when the router's map version is behind. This is the actual cost. The shard count on the slide was the cheap part.

Identity columns and sequences are per database. Do not assume bigint ids are globally unique after sharding unless you encode the shard id in the high bits or use a UUID. Foreign keys to a table on another shard do not exist. Replace them with an identifier the application checks, and accept that the database will not. Cross-shard transactions are lesson 32. If you need them on the checkout path, the shard key is wrong, because checkout should have been local.

What stays a single database

Shard when one primary's write capacity or disk is the constraint you measured, and the workload has a key with locality. Stay on one database with partitions when the pain is retention, vacuum of old data, or a few fat indexes. Stay on one database with replicas when the pain is read QPS. Add a warehouse subscription when the pain is analytical SQL. Sharding checkout because a dashboard is slow is how you spend a year losing foreign keys you did not need to lose.

Architecture and Design Diagrams

mermaidCopy
flowchart TD
    api[Spring checkout]
    map[Tenant shard lookup]
    s0[Shard 0 primary plus replicas]
    s1[Shard 1 primary plus replicas]
    sN[Shard N dedicated large tenant]
    wh[Warehouse subscriber]
    api --> map
    map -->|tenant A| s0
    map -->|tenant B| s1
    map -->|celebrity tenant| sN
    s0 -->|logical publication of posted orders| wh
    s1 -->|logical publication of posted orders| wh

Each shard is a lesson 29 topology: one primary, a sync standby for payments if that shard holds payments, async replicas for history. The lookup is consulted before the connection is chosen. The warehouse is not a shard and is not a failover target.

mermaidCopy
sequenceDiagram
    participant App as Checkout
    participant Map as Shard map
    participant Shard as Tenant shard
    App->>Map: shard id for tenant
    Map-->>App: shard 3 map version 41
    App->>Shard: BEGIN order plus lines COMMIT
    Note over App,Shard: same tenant id so one local transaction
    App->>App: history reads shard 3 with LSN token

A cross-tenant finance query does not appear on this sequence. It reads the warehouse. Putting it on the checkout sequence is the design bug.

Technical Explanation

Pruning versus routing

On one PostgreSQL instance the planner eliminates partitions whose bounds cannot contain tenant_id. The session still connects to one host. A prepared statement that uses a generic plan might hide the constant and weaken pruning. Check EXPLAIN for Subplans Removed or for a single child scan. If you see every child, the predicate did not constrain the key, or the plan was generic. SET plan_cache_mode = force_custom_plan is a session experiment, not a product architecture.

Across shards nothing prunes automatically. The router must know the shard before it borrows a connection. If it does not, it queries every shard and merges. That is a scatter-gather. Latency becomes the slowest shard plus merge time. Availability becomes "every shard must answer" unless you define a partial result, which is a product decision. Checkout does not scatter. A tenant search box that omits tenant_id does, and it should not be on the request path.

Colocation and the join you are allowed to write

Inside one shard, or on colocated Citus shards, orders join order_lines on tenant_id and order_id is a normal join. The foreign key can exist because both tables live in the same database and the referenced key includes the partition or distribution column as required.

A join from orders to a global courses catalog is the reference-table case when the catalog is small, or a cache in the application when it is not. Replicating the entire course catalog onto every shard duplicates writes from the instructor studio. Prefer the application to read course title from the catalog service, and store course_id plus the captured price on the line, which lesson 28 already required for money. The line does not need a live foreign key to a catalog database in another shard. The captured price is the fact.

Lookup-based routing in Spring

javaCopy
public final class ShardRoutingDataSource extends AbstractRoutingDataSource {
    @Override
    protected Object determineCurrentLookupKey() {
        return ShardContext.requiredShardId();
    }
}

ShardContext is set from the authenticated tenant at the start of the request, after reading the map:

javaCopy
public record ShardPlacement(int shardId, long mapVersion) {}

public ShardPlacement locate(UUID tenantId) {
    return lookupJdbc.sql("""
            SELECT shard_id, map_version
            FROM directory.tenant_placement
            WHERE tenant_id = :tenantId
            """)
        .param("tenantId", tenantId)
        .query((rs, n) -> new ShardPlacement(rs.getInt("shard_id"), rs.getLong("map_version")))
        .single();
}

The directory database is small and replicated. Cache placements by tenant id with the map version. On a move, increment map_version in the same transaction as the shard_id change. Routers that present an old version to a shard that has been marked moved get a rejection and reload. Do not compute Math.floorMod(tenant.hashCode(), shardCount) as the only scheme if you expect celebrity tenants. Hashing is a default placement policy when you insert a new tenant into the directory. The directory remains the source of truth.

Every checkout statement includes tenant_id in the WHERE or the insert even after routing. Routing bugs then affect one tenant's performance, and a defense-in-depth row policy can still express tenant_id = current_setting('app.tenant_id')::uuid. Row-Level Security on a shard limits damage inside that database. It does not find a row you wrote to the wrong shard.

Moving a tenant

Copy with a server-side statement between databases is not a single SQL feature. Practical options you can operate:

• pg_dump of a filtered query is a blunt tool for a small tenant.
• A batch job with SELECT ... WHERE tenant_id = $1 AND id > $last on the source and inserts on the destination, followed by a short freeze, fits medium tenants.
• Logical replication of the shard, with the subscriber applying only that tenant, fits large tenants if you set replica identity and you have rehearsed conflict handling.

During the freeze, checkout for that tenant returns "maintenance" or queues. Keep the window measured. After the directory flip, run a count and a checksum of money columns per tenant on both sides before you delete the source. Money mismatches stop the delete. The old shard's replicas must not keep serving that tenant's history. Lesson 29's LSN token is per shard. A token from shard 3 is meaningless on shard 9. Clear it on move.

Failure and isolation

One shard down means those tenants fail and other tenants keep checking out. That partial failure is the availability argument for sharding, and it is also why a scatter query that requires all shards turns one shard's outage into a platform outage. Timeouts on cross-shard work must be shorter than the checkout timeout and must not hold checkout connections.

Noisy neighbors are the celebrity tenant in milder form: one tenant's import job saturates a shared shard. Isolation options are a dedicated shard, a lower priority for batch connections, or statement_timeout on the import role. Connection pools are per shard. A global pool divided by accident gives shard 0 the entire pool and the others timeouts. Size pools from the shard's QPS.

Backups are per shard. A restore of shard 2 must not restore shard 3's directory mapping. The directory backup and the shard backups need a documented order: you must know which map version matches which shard contents. Restoring a shard without the directory row that points at it loses the tenant from the application even though the bytes are there.

Practical Implementation or Design Walkthrough

Problem

One PostgreSQL primary serves every TechLearningPro tenant. Writes are the bottleneck in the capacity model. Reads are already on replicas. Tenants vary from a dozen learners to one company with millions. Checkout must remain one local transaction. A finance report needs all tenants and can be an hour stale.

Constraints

• Shard by tenant_id.
• New tenants go to the shard with the most free capacity, recorded in directory.tenant_placement.
• The largest tenant can be moved to an empty shard without renumbering the others.
• Orders and lines share a shard and a transaction.
• Global uniqueness of learner email is required at signup and is not the order shard key.
• Reports do not run on shard primaries.

Directory and shard DDL

sqlCopy
CREATE TABLE directory.tenant_placement (
    tenant_id   uuid PRIMARY KEY,
    shard_id    integer NOT NULL,
    map_version bigint NOT NULL,
    state       text NOT NULL CHECK (state IN ('active', 'moving', 'frozen'))
);

On each shard, hash partitioning is optional and local. It does not replace shards. Use it when a single shard's orders table needs monthly detach or smaller indexes. A small shard can start as a plain table:

sqlCopy
CREATE TABLE commerce.orders (
    tenant_id   uuid NOT NULL,
    id          bigint GENERATED ALWAYS AS IDENTITY,
    public_id   uuid NOT NULL,
    placed_at   timestamptz NOT NULL,
    total_cents integer NOT NULL CHECK (total_cents >= 0),
    PRIMARY KEY (tenant_id, id),
    UNIQUE (tenant_id, public_id)
);

UNIQUE (tenant_id, public_id) is enforceable because both columns are on this shard and, if you later partition by tenant_id, the key includes the partition column. It does not stop the same public_id on another shard. Generate public_id as a UUID. Do not use a sequence for a value you will show as globally unique unless the shard id is part of the value.

Learner emails live in the directory database, with a unique index on email, next to placement. Signup is a directory transaction. It is not a cross-shard unique index on every orders shard. The order row stores learner_id. It does not look up email on the shard.

Citus alternative for the same key

If the platform runs Citus instead of application routing, the directory can shrink and the coordinator routes:

sqlCopy
SELECT create_distributed_table('commerce.orders', 'tenant_id');
SELECT create_distributed_table('commerce.order_lines', 'tenant_id', colocate_with => 'commerce.orders');

The Spring service still sends tenant_id on every statement. You point JdbcClient at the coordinator. You do not open worker connections from the application unless you have a measured reason. rebalance_table_shards is how you fill a new worker. Moving a single celebrity tenant to its own node is an operation you rehearse with Citus shard moves. It is still a move, with a write window you must measure. Do not invent a second distribution column for that tenant inside the application and forget the coordinator.

Request path

1. Authenticate. Read tenant_id from the token, not from an untrusted body field alone.
2. Load placement. If state is frozen or moving, reject writes with a retryable status.
3. Set ShardContext and the session variable for row level security.
4. Run the lesson 28 insert of order and lines on that shard's primary, in one @Transactional method.
5. Store the LSN token for that shard only.
6. History reads use that shard's replicas under lesson 29's rule.

Validation

• Two tenants on different shards: a transaction on shard 0 cannot see shard 1's rows through the application.
• A checkout that omits tenant_id fails review: the SQL is rejected by a test that inspects the statement, and RLS rejects it if the setting is missing.
• Place 95% of synthetic write load on one tenant and show one shard's WAL grows. Move that tenant. Show the other shard's WAL drops and checkout for a quiet tenant continues during the move.
• Run a cross-tenant sum on a shard primary in a test and show it is incomplete. Run it on the warehouse fixture and show it matches the sum of shards.
• Kill one shard in a lab and show only its tenants fail checkout.

Errors

Wrong shard: the row is missing or RLS returns no data. The repair is a directory fix plus a copy of orphaned rows, which is why the checksum after a move exists. Frozen tenant: return 503 with Retry-After for writes. Directory down: checkout cannot route. Replicate the directory and keep a short cache, and refuse writes when the cache is older than a bound you set. Serving writes from a stale cache during a move double-writes the tenant. Reads can use a slightly stale map only if you would rather miss than split. For checkout, stale is worse than down.

Enterprise Real-World Scenarios

Scenario 1: Multi-tenant academy

Context. TechLearningPro sells the academy to companies. Most tenants have under a thousand learners. Two enterprises have imported their whole workforce. One primary's WAL hits the capacity model's ceiling at enrollment peaks. History is already on replicas.

Design. Directory maps tenant_id to shard_id. Ordinary tenants share shards, placed by free capacity at signup, not by hash % N alone. Each enterprise tenant gets a dedicated shard before its import. Orders and lines are local. Email uniqueness stays in the directory. Instructor revenue across tenants is a logical publication from each shard into a warehouse, with the analytical indexes lesson 28 refused on the OLTP table. Logical Replication is the feed. Shard primaries keep the payment sync standby from lesson 29.

Failure mode this design accepts. A shared shard outage pauses many small tenants at once. Dedicated tenants stay up. The status page names the shard. A design that scatters one checkout across shards would pause everyone when any shard failed. Checkout does not scatter.

Scenario 2: Time-series progress on one primary, sharded wrong

Context. The player writes progress events. Someone proposes shards on happened_at by day, sixteen shards, "so it scales."

What happens. Today's shard receives every insert. The other fifteen serve yesterday's reads. You operate sixteen clusters to get one writer's worth of capacity, and a learner's "last position" query, which is not time-bounded the way you hoped, fans out or misses. Retention was the real requirement.

Design that matches the requirement. One primary, range partitions by month, detach and drop months outside the retention window. If write rate later exceeds one primary, shard by tenant_id or learner_id depending on which id the update transaction already has, and range-partition inside the shard for retention. The monotonic key remains a partition key for detach. It does not become the shard key that is supposed to spread writes.

Best Practices

• Shard only after replicas and indexes have taken the load they can take, and a measurement says the primary writes are the limit.
• Choose the key from the transaction that must be atomic. For this academy that is tenant_id.
• Keep a placement table so one tenant can move. Use hashing as a placement policy, not as the only address.
• Include the shard key in primary keys and in request-path SQL.
• Put rows that commit together on the same shard. Colocate Citus tables with create_distributed_table and colocate_with.
• Send small dimensions through create_reference_table or through a cache, not through a cross-shard join on the web path.
• Keep global reports on a warehouse fed by logical replication.
• Encode globally unique ids as UUIDs or as values that contain the shard id. Do not expect sequences to agree across databases.
• Size connection pools, alerts, and backups per shard.
• Rehearse a tenant move, including a frozen write window and a money checksum, before a celebrity tenant forces one.
• Cap fan-out queries. A request that touches every shard needs an owner and a latency budget.
• Leave row-level security in place inside the shard.

Common Mistakes and Anti-Patterns

Partitioning called sharding. Hash partitions on one primary improve prune and maintenance. The WAL rate is unchanged. Capacity planning that counts partitions as write scale will miss the next enrollment week.

Shard key equals insertion time. All inserts hit the open range. You built idle shards. Use time for retention partitions inside a shard whose key actually spreads.

Modulus as the address. hash % 16 means a new shard count rewrites the world. A placement row per tenant makes the common move a single-tenant copy.

Cross-shard foreign keys by application hope. The line inserts on shard A and the order commits on shard B because the router refreshed mid-request. One database per request, set at the edge, and both inserts in one transaction.

Global unique constraint expected from local indexes. Each shard can enforce UNIQUE (tenant_id, public_id). None of them can see another shard's email. Put emails where signup already goes: the directory.

Scatter on the checkout path. "Query all shards and take the max" inside the payment request makes the slowest shard your latency and the failed shard your availability. Checkout uses the placement row and one connection.

Resharding as a weekend flag flip. Copy, catch-up, versioned directory flip, checksum, then delete. Skipping the freeze or the version check dual-writes the tenant. You will find out from the money checksum if you kept it, and from learners if you did not.

Troubleshooting and Design Review

Design review: What is the atomic transaction, and does the key cover every row it writes? What is the hottest tenant's share of WAL? How does one tenant move? Where is global uniqueness actually enforced? What happens to checkout when one shard is down? Where do cross-tenant queries run?

Interview Questions and Answers

Q1. What is the difference between a PostgreSQL partition and a shard?

A partition is a child table in the same database. The planner can prune it, and a transaction can write two partitions and roll both back. A shard is another database with its own primary, WAL, and failover. Writes scale when two shards take the load. You lose a single commit and any foreign key that would have crossed the boundary. Hash partitions on one instance are still one writer. Calling them shards in a capacity plan hides the bottleneck.

Q2. Why must a unique constraint on a partitioned table include the partition key?

Each partition is a separate table with its own indexes. PostgreSQL will not check uniqueness by scanning every partition on every insert. The unique index is local, so the key has to contain the column that decides the partition. Otherwise two children could store the same id. If you need a globally unique value that is not part of the partition key, generate a UUID or enforce the rule in a table that is not partitioned that way, such as the tenant directory.

Q3. How do you choose a shard key for a multi-tenant learning platform?

Use the column that checkout already needs in one transaction. Orders, lines, and the tenant's seat count share tenant_id, so that is the key. The admin's tenant screen is then one shard. Learner id shards a tenant's learners apart and makes the admin screen a fan-out. Timestamp shards send every insert to the latest shard. Confirm the hottest tenant's share. If it is most of the WAL, give that tenant a shard of its own instead of hashing harder.

Q4. What is a hotspot, and why doesn't adding shards always fix it?

A hotspot is a shard that keeps a disproportionate write or read rate. A monotonic key fills only the current range. A low-cardinality key collapses to a few buckets. A hashed celebrity tenant still occupies one bucket. New shards change the modulus or the worker count. They do not move the hot key unless you re-place that key. Measure writes per shard after the change. Idle shards are evidence, not capacity.

Q5. What does Citus colocation buy you, and what does rebalance_table_shards not do?

create_distributed_table on orders and order_lines with the same distribution column and colocate_with keeps one tenant's rows on the same worker, so the join and the transaction stay local. create_reference_table copies a small dimension onto every worker. rebalance_table_shards moves existing shards between workers to spread storage. It does not change the distribution column, and it is not a substitute for a predicate on tenant_id. A query without that predicate still fan-outs.

Q6. Why is resharding operationally hard?

Live rows are being updated while you copy them. A modulus change touches most rows. Even a single-tenant move needs a source of catch-up, a directory flip the routers see at the same version, and a plan for a stale router that writes the old shard after the flip. Sequences, foreign keys, and LSN tokens do not survive the move unchanged. You checksum business totals before deleting the source. The hard part is the dual-write window, not the CREATE TABLE on the empty shard.

Q7. How should Spring decide which database receives a checkout?

Resolve tenant_id from the authenticated session, read directory.tenant_placement, and set the AbstractRoutingDataSource key to that shard_id for the rest of the request. Insert order and lines through one @Transactional JdbcClient so they share a connection. Include tenant_id in the SQL anyway. Do not trust a client-supplied shard id. If placement state is frozen or moving, fail the write. Clear replica LSN pins when the shard id changes.

Q8. Where do you enforce that an email is unique if users are sharded by tenant?

Not with a unique index on every shard. Each index only sees its own rows. Enforce email uniqueness in the directory database that already owns signup, in the same transaction that allocates learner_id and the placement row. The orders shard stores learner_id and never claims to police email. If two services can create learners, only one of them is allowed to insert into that unique table.

Q9. A dashboard query joins every tenant's orders. Where does it run?

Not on the checkout primaries as a synchronous fan-out. Publish posted orders from each shard with logical replication into a warehouse, and run the dashboard there with the staleness you agreed. If the product requires a live total inside a tenant, that query is local to the tenant's shard and can run there. "Live across all tenants" is a different SLO, and it belongs in the capacity model as its own read path.

Q10. When would you stay with declarative partitioning and refuse to shard?

When the pain is dropping old months, vacuuming a huge child, or pruning a query by tenant on hardware that still has write headroom. Range detach and hash partitions solve those on one primary with foreign keys and one failover story intact. Shard when measured write throughput or disk on that primary cannot grow vertically or with better indexes, and a key with locality exists. Sharding to make an analytical query faster gives up constraints you still needed at checkout.

Q11. What fails first when one shard is unreachable?

Tenants mapped to that shard cannot check out or read their own history. Other shards continue, which is the isolation you wanted, if the application does not call every shard on the way. A gateway that aggregates all shards for a home page turns the outage global. Connection pools sized as one shared pool can also stall unrelated shards waiting for connections. Alerts and status pages should be per shard. The directory must still be up, or no tenant can be routed.

Hands-On Exercises

Exercise 1: Shard strategy for the academy

Problem. Design placement for TechLearningPro tenants. Peak checkout writes exceed one primary. One tenant is 40% of writes. Email uniqueness is required. Finance needs a daily all-tenant total that can be one hour late.

Tasks.

1. Name the shard key and the transaction that forced it.
2. Define the placement table and how a new tenant chooses a shard.
3. Show how the large tenant is isolated and later moved, including the freeze and the check that money matches.
4. Specify what runs on shards, what runs in the directory, and what runs in the warehouse.
5. State the learner-visible behavior when that tenant's shard is down, and when a different shard is down.

Deliverables. DDL for placement and for commerce.orders on a shard, the routing rule, and a move sequence.

Exercise 2: Partition, do not shard

Problem. Progress events need 14-month retention and the write rate fits one primary. A colleague proposes 30 shards by day of month.

Tasks.

1. Explain the hotspot that design creates.
2. Provide a range-partitioned table and the detach step for the oldest month.
3. Name the measurement that would later justify sharding, and the key you would use then.

Rubric.

Lesson Summary

Declarative partitioning splits a table inside one PostgreSQL database so the planner can prune and so you can detach old data. Unique constraints include the partition key. The primary is still one writer. Sharding splits that writer by a key that covers the local transaction. For the academy, the key is tenant_id, stored as a placement row so a large tenant can move without a modulus rewrite. Citus expresses the same key with create_distributed_table, colocation, reference tables, and rebalance_table_shards. Cross-shard uniqueness, foreign keys, and live all-tenant queries do not come along. Put emails in the directory and reports in a warehouse. Resharding is a copy, a versioned flip, and a checksum.

The next lesson names the consistency you still have inside one shard's primary, and the consistency you gave up when two shards can diverge. That is CAP Theorem and Consistency Models.

Further Reading

• PostgreSQL documentation: Table partitioning, partition pruning, and unique constraints on partitioned tables.
• Citus documentation: create_distributed_table, create_reference_table, colocation, and rebalance_table_shards. Use the version that matches the extension you run.
• TechLearningPro: Partitioning, Sharding Concepts, Logical Replication, Row-Level Security.
• Lessons in this module: Database Replication and Consistency, CAP Theorem and Consistency Models, Distributed Transactions and the Saga Pattern.