Database Sharding (with PostgreSQL)

Horizontal partitioning of one dataset across many independent database servers: what it is, when it's the right call, how PostgreSQL does it (native partitioning, postgres_fdw, Citus, Aurora Limitless), and how to actually route/serve requests against it from Java/Kotlin/Spring. Practical, interview-presentable, opinionated. Pairs with the databases-and-data-storage, distributed-systems, locking, saga, and CDC guides in this folder.

Version anchor (July 2026): PostgreSQL 18 (released 25 Sep 2025), Citus 14 (AGPLv3), Apache ShardingSphere 5.5.x, Aurora PostgreSQL Limitless Database GA. Facts that move with versions are flagged inline.

The one idea (say this first in an interview)#

Sharding splits one logical dataset across many independent databases so that writes and storage scale horizontally instead of being capped by a single node. The shard key you choose is a near-permanent decision that decides, forever, which queries stay cheap (they hit one shard) and which become expensive distributed operations (scatter-gather, cross-shard joins, cross-shard transactions). You are not "scaling a database" — you are giving up the things a single node gives you for free (joins, foreign keys, unique constraints, ACID across the whole dataset) in exchange for scale, and then arranging the data so your hot path never needs those things across shards.

Everything else in this guide is a corollary:

• Routing exists because a request must be told which shard holds its data.
• Cross-shard queries are painful because the shard key aligned some queries to one node and left the rest to fan out.
• Cross-shard transactions are avoided because atomic commit across independent databases is slow and fragile — so you colocate, or you fall back to sagas.
• Rebalancing is hard because you originally hashed data to N nodes and now need N+1 without moving everything.

If you internalize "the shard key is the whole game, and sharding is a one-way door," you'll reason correctly about every sub-topic. The single most senior thing you can say about sharding is "and that's why I'd try hard not to do it yet."

1. Vocabulary, and the three things people conflate#

Three distinct techniques get muddled constantly. Getting them crisp is table-stakes for a senior conversation.

The relationships that matter:

• Sharding vs replication are orthogonal and usually combined. Replication makes copies for read-scaling/HA; sharding makes disjoint splits for write-scaling. In production each shard is itself a replicated cluster (a primary + replicas). "Shard" = the split; "replica" = the copy.
• Partitioning vs sharding is the one people fail. PostgreSQL's declarative partitioning (PARTITION BY RANGE/LIST/HASH) splits a table into partitions that, by default, all live on the same server. It buys you partition pruning, cheap DROP/DETACH of old data, and smaller indexes/vacuums — but it does not, on its own, add write throughput or break the single-node ceiling. Sharding means the pieces live on different machines. (You can combine them: postgres_fdw foreign-table partitions put partitions on remote servers — that's §9.2 — and Citus turns partitions into distributed shards — §9.3.)

Core terms you should use precisely:

• Shard — one independent database holding a disjoint slice of the data.
• Shard key / partition key / distribution column — the column(s) whose value decides which shard a row lands on. The decision (§3).
• Logical vs physical shard — a level of indirection: hash to a large fixed number of logical shards, then map logical→physical. Lets you add physical nodes without rehashing (§4, §8).
• Coordinator / router / query engine — a component that knows the shard map and dispatches a query to the right shard(s) and merges results (Citus coordinator, Aurora Limitless router, a Spring routing layer, a proxy like ShardingSphere).
• Colocation — deliberately placing related rows (e.g., a user and their orders) on the same shard by sharding them on the same key, so joins and transactions between them stay single-node. The most important optimization in the whole space.
• Reference / broadcast table — a small, rarely-written table (countries, currencies, feature flags) copied to every shard so it can be joined locally instead of fetched cross-shard.
• Scatter-gather (fan-out) — a query with no single-shard target: send it to all shards, then merge/sort/aggregate the partial results. Slow, and it loses global ordering/pagination guarantees.
• Resharding / rebalancing — moving data between shards when you add capacity or a shard gets hot.

2. When to shard — and the cheaper things to do first#

Sharding is the last rung of the scaling ladder because it's the only one that permanently takes away joins and transactions. Senior engineers are known by how hard they resist it.

The ladder — go down only as far as the load forces you:

1. Vertical scale. Bigger box (more RAM/CPU/NVMe). Boring, effective, buys years. A single modern Postgres primary handles tens of thousands of TPS and multiple TB comfortably. Do this first, unapologetically.
2. Read replicas (+ caching). If you're read-bound, replicas and a cache (see the databases guide) fix it without sharding. Replication does nothing for write load — know which one you have.
3. Offload and separate workloads. Move analytics/reporting off the OLTP box (CDC → a columnar store like ClickHouse/BigQuery — see the CDC guide), archive cold data, split the monolith DB by bounded context (see the DDD guide). Different problems masquerade as "we need to shard."
4. Table partitioning + retention. If the pain is a single enormous table (unbounded growth, slow vacuums, giant indexes), single-node declarative partitioning with time-based DROP PARTITION often removes the pain without distributing anything.
5. Then, and only then, shard — or, better, buy a system that shards for you (managed distributed SQL / Citus / Aurora Limitless), so the distributed-transaction and rebalancing machinery is someone else's problem.

Signals you have actually earned sharding (need several, not one):

• Write throughput exceeds what one primary + tuning can absorb (you're I/O- or CPU-bound on the write path, and vertical headroom is gone or absurdly expensive).
• Dataset / working set no longer fits one machine's storage or RAM, so cache hit-rates collapse.
• A single table is so large that index maintenance, vacuum, and backup/restore times violate your SLOs even after partitioning.
• Connection or tenant isolation demands physical separation (a huge tenant needs its own capacity/blast-radius; data-residency requires per-region storage).

The trap — "premature sharding": designing for hypothetical Google-scale on startup-scale data, taking the one-way door and losing joins/transactions for a load a single node would have shrugged off. Fix: measure first; exhaust the ladder; if you truly need scale-out SQL, prefer a managed distributed store over hand-rolled sharding. Say this out loud in interviews — it signals judgment, not ignorance.

Why it's a one-way door. Once data is split by user_id, a query that isn't scoped by user_id has no home; a transaction spanning two users spans two databases; a UNIQUE(email) constraint no longer holds globally; JOINs across the split need a coordinator or denormalization. Un-sharding means a full data migration. You are buying scale with permanent complexity — make sure you're buying something you need.

3. The shard key — the one decision that matters#

If you take one thing into the interview: the shard key determines your ceiling, and it's effectively immutable. Pick it by looking at your hottest queries and your write distribution, not by what's "natural."

A good shard key is:

• High cardinality — many distinct values, so data (and load) can spread finely. status (5 values) is a catastrophe; user_id is fine.
• Uniformly accessed — no single value takes a wildly disproportionate share of traffic. This is the hot shard / celebrity / noisy-neighbor problem: shard a social graph by celebrity_id and one node melts while the rest idle.
• Query-aligned — the key appears in your hot-path queries, so those queries carry the key and route to one shard. This is the property that makes or breaks you: align the shard key with the dominant access pattern so the common query is single-shard, and let the rare query pay the scatter-gather tax.
• Stable (rarely changed) — changing a row's shard-key value means physically moving the row between shards. Pick something immutable (an ID), never something mutable (region, status, email).

The tension you must name: the key that spreads load evenly (random/high-entropy) is often not the key your queries filter by, and the key your queries filter by (a tenant) is often skewed. Sharding is the art of resolving that tension.

Techniques for resolving it:

• Composite / hierarchical keys. Shard by (tenant_id, entity_id): routes tenant-scoped queries to one shard, but a whale tenant's data can be spread across shards by the second component. Or hash tenant_id normally but carve whales out to dedicated shards (a hybrid directory — §4).
• Salting / bucketing hot keys. For a known celebrity, append a bucket: store under celebrity_id:0..k, spreading writes across k shards; reads fan out to the k buckets and merge. Trades a bounded scatter for killing the hotspot. (Same idea as salting a hot Kafka partition — the DB-sharding and Kafka-partitioning problems are the same problem.)
• Colocation as a first-class design goal. Choose keys so that entities you join and transact together share a shard. Classic: shard users by id and orders by user_id on the same hash space → a user and all their orders live together → their joins and transactions never leave one node. Design the schema so your aggregates (DDD sense) are shard-local.

Worked choices (be ready to defend one):

The senior move: state the dominant access pattern, pick the key that makes it single-shard, then immediately name what you just made expensive and how you'll handle it (offload to search/OLAP, denormalize, fan-out-on-write, accept bounded scatter). Interviewers are grading whether you see the cost, not whether you found a magic key — there is no key with no downside.

4. Sharding strategies — how a key maps to a shard#

Given a shard key, how do you compute the target shard? Four families, each with a failure mode.

Hash vs range is the core trade: hash gives you even load but destroys locality (range queries must scatter-gather); range gives you locality but invites hotspots. Most OLTP systems shard by hash of an entity id because point-by-key access dominates and even load matters more than range scans; time-series systems often range-partition by time because the queries are time-windows and old partitions get dropped.

The mod N problem and the two fixes. With hash(key) mod N, changing N (adding a node) remaps ~(N-1)/N of all keys — a near-total data reshuffle. Fixes:

1. Consistent hashing (with virtual nodes). Place shards on a hash ring; a key goes to the next shard clockwise. Adding/removing a node remaps only ~1/N of keys instead of nearly all. Virtual nodes (each physical shard owns many ring positions) smooth out distribution and make rebalancing granular. This is how Cassandra/DynamoDB/Riak rebalance without a full reshuffle.
2. Fixed logical shards + a lookup (the pragmatic favorite). Hash into a large, permanently-fixed number of logical shards (say 4096), then keep a small logical → physical map. Physical node count changes by editing the map and moving a few logical shards — the hash function never changes. This is Vitess's model, and the easiest to hand-roll correctly in an app. Pick logical-shard count once, generously; you can't grow it without rehashing.

Rule of thumb: never shard directly to physical node count. Shard to a stable space (a hash ring, or a fixed logical-shard count) and map that space to physical nodes. This is the difference between "adding a shard is a config change + background copy" and "adding a shard is a weekend outage."

5. Routing — how a request finds its shard#

Something must translate a request into "connect to shard k." Three places that logic can live:

• In the application (client-side routing). The app holds the shard map, computes resolve(shardKey) → shard, and picks the connection pool. Lowest latency (no extra hop), full control, but every service must embed the routing logic and the shard map, and cross-shard queries are your code's problem. This is what you build with Spring's AbstractRoutingDataSource (§10) or a library like ShardingSphere-JDBC.
• In a proxy / middleware (server-side routing). A wire-protocol proxy (ShardingSphere-Proxy, PgCat, a Vitess/Multigres vtgate-equivalent) speaks PostgreSQL to the app and fans out to shards behind it. The app thinks it's talking to one Postgres. Centralizes routing and cross-shard query handling; adds a network hop and an operational component to run and scale.
• In the database itself (a coordinator). Citus's coordinator and Aurora Limitless's routers are inside the database product: you send normal SQL to one endpoint and the engine routes, pushes down, and aggregates. Least app change, most "it's just Postgres," at the cost of adopting that engine.

The trade is always latency/control (client-side) vs transparency/central-logic (proxy or coordinator). For a Staff+ answer: hand-rolled client-side routing is fine when queries are ~always single-shard and keyed the same way; the moment you need transparent cross-shard SQL, joins, and distributed transactions, prefer a coordinator/proxy that already solved them over reinventing a query engine in your service layer.

Where does the shard key come from at request time? Usually one of: the authenticated principal (a tenant_id/account_id in the JWT), the URL path (/tenants/{id}/…), or the request body. You resolve it before you open a database connection or start a transaction (a real footgun in Spring — §10.1). If a request genuinely has no shard key (an admin "search all tenants" endpoint), that's a deliberate scatter-gather, and you should treat it as a separate, rate-limited, non-hot-path code route.

6. Cross-shard reads — scatter-gather and how to avoid it#

A query that doesn't carry the shard key has no single home. Your options, best to worst:

1. Don't. Re-shape the data so the query is single-shard or served elsewhere. The senior instinct. If "orders by user" is hot, shard by user and you're done. If "orders by product" is also needed, that's a different access pattern — serve it from a secondary store built by CDC (a search index, or an OLAP/columnar store), not by fanning out across OLTP shards. Cross-cutting reads belong in a read model (CQRS), not in scatter-gather on your write shards.

2. Global secondary index / denormalized lookup. Maintain a second sharded table keyed by the other attribute (e.g., product_id → order_ids), sharded by that attribute, updated on write. Turns a scatter into a two-step point lookup. You now own the consistency of that index (update it via outbox/CDC, not a dual write — see the outbox guide).

3. Scatter-gather, done deliberately. Send the query to all shards in parallel, merge results. Acceptable for genuinely rare, bounded queries. The hard parts:

• Fan out concurrently, not serially. Latency = the slowest shard, not the sum. Use a thread pool / virtual threads / coroutines (§10.3). Serial fan-out across 32 shards is a self-inflicted outage.
• Tail latency dominates. With 32 shards, your p99 becomes the p99 of the slowest of 32 responses — the more shards, the worse the tail. Set per-shard timeouts and decide whether partial results are acceptable.
• Global ordering & pagination break. ORDER BY created_at LIMIT 20 across shards means each shard returns its top-20, you merge and re-sort — and OFFSET deep-paging is effectively impossible (offset 10000 means pulling 10000 from every shard). Use keyset/seek pagination (WHERE created_at < :cursor ORDER BY created_at DESC LIMIT 20) so each shard filters by the same cursor and you merge bounded results.
• Aggregations must be re-aggregated. COUNT/SUM are additive (sum the per-shard results). AVG must become SUM/COUNT computed from partials. COUNT(DISTINCT …) and percentiles are not composable exactly — approximate (HyperLogLog, t-digest) or offload to OLAP.

4. Cross-shard joins — avoid outright. A join whose two sides live on different shards forces either shipping one side across the network or a coordinator broadcast. Prevent it by colocation (put the joined entities on the same shard via the same key) or reference tables (replicate the small side to every shard so the join is local). If neither applies, the join is telling you the two datasets want different shard keys — denormalize or split the query.

One-liner for the room: "Cross-shard reads are a symptom that a query doesn't match the shard key. I fix the data model (colocation, reference tables, a CDC-built read model) before I fix the query — and only fall back to a bounded, parallel, keyset-paginated scatter-gather for genuinely rare cross-cutting reads."

7. Cross-shard writes & transactions — the genuinely hard part#

Single-shard transactions are ordinary ACID — one Postgres, one BEGIN…COMMIT. Everything hard starts when one logical operation must change data on two shards (move money from user A on shard 1 to user B on shard 7).

Why not two-phase commit (2PC / XA)? It exists (PREPARE TRANSACTION in Postgres, XA in ShardingSphere), and it does give atomicity, but Staff engineers avoid it on the hot path because:

• It's blocking. Between PREPARE and COMMIT, participants hold locks. If the coordinator dies after PREPARE, participants are stuck in-doubt holding locks until it recovers — a classic availability hole (the "coordinator is a SPOF" problem).
• It's slow. Two network round-trips plus fsyncs across all participants, at the pace of the slowest one; throughput collapses under contention.
• It doesn't compose with modern stacks. It couples service availability, fights with connection pools, and interacts badly with replication failover.

So the playbook is, in order:

1. Design the transaction to be single-shard. Colocate the entities that must change together (same shard key → same shard → normal local ACID). The overwhelming majority of "cross-shard transactions" are avoidable by choosing the aggregate boundary as the shard boundary (DDD pays off here). This is the answer they want first.
2. If it's genuinely cross-shard, use a saga. Model the operation as a sequence of local transactions, one per shard, with compensating actions to undo on failure (see the saga guide). You trade atomicity+isolation for eventual consistency + idempotency + compensation. Debit A (local txn) → credit B (local txn) → on failure, compensate the debit. Make every step idempotent (dedupe by a business key) because retries are certain.
3. Only reach for 2PC when correctness truly demands cross-node atomic commit and the volume is low enough to eat the cost (rare, controlled financial operations), and prefer letting a coordinator that implements it (Citus, Aurora Limitless routers) own it rather than hand-rolling XA.

Two distributed-data problems sharding creates that people forget:

• Global uniqueness dies. A UNIQUE(email) constraint is per-shard; two shards can both accept alice@x.com. Fixes: make the unique attribute the shard key (all alice@x.com rows hash to one shard — works only if you look up by it), or maintain a separate uniqueness registry (a single table/service that allocates the value), or accept app-level check-then-insert with a reconciliation sweep. Name this in interviews — it's a favorite gotcha.
• No shared sequence for IDs. You can't use one Postgres BIGSERIAL across shards (it'd be a single-node bottleneck and a coordination point). Generate IDs that are globally unique without coordination:
  • Snowflake-style — timestamp | machineId | sequence packed into 64 bits: k-sortable, compact, no central counter.
  • UUIDv7 / ULID — time-ordered UUIDs: globally unique, client-generatable, and k-sortable so they preserve B-tree insert locality (unlike random UUIDv4, which shreds it). PostgreSQL 18 ships a native uuidv7() function; in the app, generate them at write time. This is the modern default.
  • Avoid random UUIDv4 as a primary key at scale (16 bytes, random insert order → index write amplification), and never a per-shard auto-increment you might later need to be globally unique.

One-liner: "I make the transaction single-shard by colocating on the aggregate boundary; if it must cross shards it becomes a saga of local transactions with idempotent compensations, never hot-path 2PC; and I hand global uniqueness and ID generation to the design (shard-key-as-unique, UUIDv7/Snowflake) because a sharded DB has no global constraints or sequences."

8. Rebalancing & resharding — the operational reality#

You will get the shard count or key distribution wrong, or simply grow. Moving data between shards while serving live traffic is the hardest operational part of sharding, and interviewers probe whether you've thought past day one.

Why it's hard: with naïve hash mod N, going from N to N+1 shards remaps almost every key — you'd have to move nearly the whole dataset. The §4 designs (consistent hashing, or fixed logical shards + a logical→physical map) exist precisely so that adding capacity moves only a slice.

The online resharding playbook (zero-downtime):

1. Provision the new shard(s) and decide which logical shards / hash ranges move.
2. Backfill — copy the affected rows to their new home in the background (bulk copy, or Postgres logical replication streaming a subset). The source keeps serving.
3. Dual-write / catch-up — apply ongoing changes to both old and new until the new copy is caught up (logical replication does this continuously; or the app dual-writes during the window).
4. Cutover — flip the logical→physical map (or ring) so reads/writes for those keys go to the new shard. Keep it brief and atomic per range; fence writers so a key is never writable in two places at once (see the locking guide on fencing).
5. Verify & clean up — reconcile row counts/checksums, then drop the moved data from the old shard.

Don't hand-roll this if you can help it. This is exactly the machinery a good sharding product gives you: Citus has rebalance_table_shards() and citus_move_shard_placement() (it uses logical replication under the hood and moves shards with minimal disruption); Vitess/Multigres have Reshard workflows; Aurora Limitless resizes shard groups for you. Hand-rolled resharding is a multi-week project with a data-loss failure mode — a strong argument for buying rather than building.

Design so you rarely reshard: over-provision logical shards up front (cheap — many logical shards can share a physical node) so growth is "spread the same logical shards over more physical nodes," not "re-key the dataset." Adding a node should be a rebalance, never a rehash.

9. PostgreSQL specifically — five ways to "shard," from not-really to fully-managed#

PostgreSQL gives you a spectrum. The crucial framing: PostgreSQL core does not ship multi-node horizontal sharding. It ships single-node partitioning and the plumbing (postgres_fdw, logical replication) that extensions and products build real sharding on top of. Know where each option sits.

9.1 Declarative partitioning — single-node, and not sharding#

Since PostgreSQL 10 (mature and improved through 18), you can split one table into partitions by RANGE, LIST, or HASH. All partitions live on the same server by default. It buys manageability, not scale-out.

What to know for the room:

• Partition pruning lets the planner skip partitions when the query filters on the partition key — the main performance win. Filter by the partition key or you scan every partition.
• Every unique/primary-key constraint must include the partition key columns — a real modeling constraint (you can't have a globally-unique id independent of started_at above without extra work).
• It does not scale writes across machines. If someone says "we sharded with Postgres partitioning," they've usually just organized one big table on one box. That's often exactly right — but it isn't sharding. It's also the substrate the next options distribute.

9.2 postgres_fdw + partitioning — real cross-node, hand-built, sharp edges#

Combine declarative partitioning with the foreign-data-wrapper: make each partition a foreign table pointing at a different server. Now partitions live on different machines — genuinely sharded storage, using only core Postgres.

Reality check — this is plumbing, not a distributed database:

• With enable_partitionwise_join/enable_partitionwise_aggregate on, some joins/aggregates push down to the shard holding the data; a query filtered by user_id prunes to one foreign server. But complex cross-shard queries pull data back to the coordinating node and process locally — slow.
• No global snapshot and no automatic cross-node atomic commit. A transaction touching two foreign servers is not atomic out of the box (there's ongoing work on FDW 2PC/atomic commit, but don't assume it in an interview). You get sharded storage, not distributed transactions.
• PostgreSQL 18 added SCRAM passthrough auth for postgres_fdw/dblink, which makes securing the inter-node connections less awful — a nice, current detail to drop.
• Verdict: educational and occasionally right for read-mostly geo-partitioning, but for real workloads you want Citus or a managed product rather than babysitting FDW sharding yourself.

9.3 Citus — the real distributed-PostgreSQL extension (the strong default)#

Citus (an extension, not a fork; AGPLv3; Citus 14 supports PostgreSQL 16–18; managed as Azure Cosmos DB for PostgreSQL) turns a cluster into distributed Postgres: a coordinator node holds metadata and routes; worker nodes hold the shards. You keep talking SQL to the coordinator.

Why it's usually the right PostgreSQL answer when you truly need scale-out OLTP:

• Colocation and reference tables are first-class — the two things that keep joins/transactions single-node are built in, not hand-rolled.
• Distributed transactions across workers are handled by the coordinator (2PC internally) so single-statement and colocated multi-row writes are atomic without you touching XA.
• Online rebalancing (rebalance_table_shards, citus_move_shard_placement) solves the §8 nightmare for you.
• It's still Postgres — extensions, psql, most SQL, the ecosystem. Your app largely doesn't know it's sharded (the coordinator is the router).
• Watch-outs: the coordinator can become a bottleneck/SPOF (mitigated by HA and, for some workloads, Citus MX letting workers accept queries directly); queries that don't include the distribution column or ignore colocation still fan out; some Postgres features are restricted on distributed tables. Model for colocation from day one.

9.4 Managed / cloud — let someone else run the coordinator#

• Amazon Aurora PostgreSQL Limitless Database (GA). A DB shard group of routers (accept connections, route, run distributed transactions, aggregate results) in front of shards (Aurora Postgres instances). You declare table types: sharded tables (hash shard key, spread across shards), reference tables (copied to every shard for local joins), and standard tables (single shard). Same colocation-and-reference model as Citus, fully managed, elastic. Cross-shard consistency and scaling are the service's problem.
• Azure Cosmos DB for PostgreSQL — managed Citus. Same mental model as §9.3 without running the cluster.
• Google AlloyDB / Cloud SQL — scale-up + read pooling; for true horizontal SQL, Google's answer is Spanner (below).

9.5 "Sharding you don't manage" — distributed SQL & proxies#

• Distributed SQL (CockroachDB, YugabyteDB, Google Spanner, TiDB). These are sharded internally (ranges/tablets over a Raft/Paxos consensus layer) but present a single logical SQL database with cross-shard ACID transactions and strong consistency. CockroachDB and YugabyteDB speak the PostgreSQL wire protocol, so a Spring app connects with the normal Postgres driver. You trade single-node latency (consensus round-trips — recall PACELC from the databases guide: these are PC/EC) and operational weight for never hand-managing a shard key or a resharding. For many teams that "genuinely need scale-out SQL," this beats hand-rolled Postgres sharding. Say so.
• Proxies / middleware. Apache ShardingSphere-Proxy (speaks the Postgres/MySQL protocol, applies sharding rules behind it), PgCat (Rust pooler with sharding + load-balancing + failover), and the emerging Multigres ("Vitess for Postgres" from Supabase/Sugu Sougoumarane — announced 2025, v0.1 alpha released June 2026, not production-ready (early releases do HA/pooling; sharding not yet shipped); Vitess itself is MySQL). Proxies centralize routing without adopting a new database, at the cost of a component to run.

PostgreSQL decision table (memorize the shape):

10. Handling requests with a sharded DB in Java / Kotlin / Spring#

This is the part interviewers actually make you write. The core pattern for application-level sharding in Spring is a AbstractRoutingDataSource that picks one of N per-shard connection pools based on a shard id carried on the current thread. Everything below builds on that, plus the traps that separate a senior answer from a broken one.

10.1 The routing DataSource + per-shard pools#

Carry the resolved shard id on the thread, and let a routing DataSource select the matching pool when a connection is acquired.

Java

Kotlin

The #1 timing trap (they will probe this). AbstractRoutingDataSource resolves the key lazily, when a Connection is acquired — and @Transactional acquires the connection at transaction start and binds it for the whole transaction. So ShardContext.set(...) must run before the @Transactional method is entered. Set it in a filter/interceptor (below), or in a non-transactional outer method that then calls the transactional one. Set it inside a @Transactional method and you've already bound the wrong (or a null) shard. And you can't switch shards mid-transaction — one transaction = one shard, by construction. This is the same caveat flagged in the databases guide, and it's a favorite "why is this null / why did it hit the wrong shard" bug.

10.2 Resolving the shard key per request and setting the context#

Resolve the shard from the key, using the fixed-logical-shards indirection so you can rebalance without rehashing (§4).

Java

Set the context before any transactional work — in an interceptor, cleared on completion so a shard id never leaks onto a pooled request thread:

Kotlin

When the shard key is only in the request body (not the JWT/path), an interceptor is too early. Resolve and set the context in a thin non-transactional façade that reads the key, calls ShardContext.set(...), then delegates to the @Transactional service — keeping the "set before the transaction" rule intact. (Consider a stable hash from a library, e.g. Guava Hashing.murmur3_32_fixed(), instead of the inline mixer if you prefer a dependency to bit-twiddling.)

10.3 Cross-shard scatter-gather in the app#

For the rare query with no shard key, fan out to every shard in parallel, then merge. Crucially, do not rely on the routing DataSource + ShardContext here: ThreadLocal does not propagate to the worker threads (a trap straight from the concurrency guide). Inject the per-shard Map<Integer, DataSource> and hit each pool directly.

Java (virtual threads — stable since JDK 21)

Kotlin (coroutines — coroutineScope gives structured concurrency; Dispatchers.IO for blocking JDBC)

Points to say out loud: latency is the slowest shard (fan out concurrently, never in a loop); a per-shard timeout bounds the tail; keyset pagination keeps each shard's work to limit rows; additive aggregates (COUNT/SUM) re-aggregate trivially, AVG→SUM/COUNT, and DISTINCT/percentiles need approximation or an OLAP offload. If you're writing scatter-gather often, your shard key is wrong or that query belongs in a CDC-built read model.

10.4 Or don't hand-roll it: ShardingSphere-JDBC#

Apache ShardingSphere-JDBC (5.5.x) makes the routing, cross-shard merge/sort/limit, distributed IDs, and read/write-splitting declarative — your repositories stay plain Spring Data / JdbcTemplate, and the library rewrites and routes the SQL.

Hand-rolled vs ShardingSphere-JDBC vs ShardingSphere-Proxy: hand-roll when access is ~always single-key and you want zero extra deps and total control; use ShardingSphere-JDBC when you want declarative rules, transparent cross-shard SQL, and distributed IDs without writing a query engine; use ShardingSphere-Proxy (or PgCat, or a Citus/Aurora-Limitless coordinator) when non-Java clients must share the same sharded database.

10.5 IDs and uniqueness (no shared sequence exists)#

Use UUIDv7/ULID or Snowflake; never a per-shard BIGSERIAL you might later need globally unique. For a globally-unique business attribute (e.g. email), make it the shard key, or front it with a single allocator/registry (§7) — a sharded DB has no cross-shard UNIQUE.

10.6 Transactions, migrations, and the connection-count bomb#

• Keep every @Transactional single-shard. The context is set before the method; the routed connection is bound for the whole transaction. Never try to touch two shards' data in one @Transactional — it all flows to the one bound connection (wrong shard) or needs XA (avoid). Cross-shard writes are sagas of separate, single-shard transactions (§7).
• Migrate every shard. Spring Boot's single-DataSource Flyway auto-config can't migrate N shards — disable it (spring.flyway.enabled=false) and loop:

• The connection-count bomb. Each app instance now holds one pool per shard: total Postgres backends ≈ poolSize × shardCount × appInstances. 10 × 4 shards × 20 instances = 800 backends — Postgres (process-per-connection) falls over long before that. Put PgBouncer (transaction pooling) in front of each shard and keep per-shard pools small. This is the sharding-specific reason "just raise the pool size" is the wrong reflex.

11. Operating a sharded fleet#

The costs that don't show up in a whiteboard design but sink real deployments:

• Schema migrations run N times. Every shard must reach the same schema. Use the per-shard migration loop (§10.6) and the expand/contract (parallel change) discipline from the schema-evolution guide — additive change, deploy, backfill, then contract — so a migration mid-rollout never breaks a shard whose app is a version behind.
• Backups are per-shard, and there is no free global snapshot. Each shard gets its own automated snapshots + WAL archiving for PITR. A point-in-time-consistent restore across shards is genuinely hard (the shards' clocks/WAL positions don't line up); you either accept per-shard consistency + idempotent replay to reconcile, or use a product that coordinates it. Know your RPO/RTO per shard and restore-test them.
• Monitoring exists to catch skew. Per-shard dashboards for QPS, CPU, storage, connections, replication lag, and p99 — the entire point is to see the hot shard early and split it before it melts. Tag every metric, log, and trace with the shard_id (ties to the trace-context guide) so a latency spike is attributable to a shard, not a mystery.
• Capacity = logical-shard headroom. Track per-shard growth and keep enough spare physical capacity that a rebalance is routine, not an emergency. The failure mode is discovering you need to reshard while a shard is already saturated.

Problems & how to solve them#

Hot shard / celebrity / noisy neighbor (one node melts, others idle).

• Root cause: low-cardinality or skewed shard key, a monotonic key routing all new writes to one shard, or one whale tenant on a shared shard.
• Fix: higher-cardinality or composite key; salt/bucket the hot key across k shards; carve whales onto dedicated shards via a directory override; for monotonic keys use hash not range. Detect it with per-shard dashboards, not customer complaints.

Every other query is a cross-shard scatter-gather.

• Root cause: the shard key is aligned to one access pattern but the app has several; queries don't carry the key.
• Fix: re-shape data for the dominant pattern and serve the others from a CDC-built read model (search index / OLAP), a global secondary index (a second table sharded by the other attribute), colocation, or reference tables — not by fanning out across OLTP shards.

Adding a node reshuffles almost all the data.

• Root cause: hash(key) mod N bound directly to physical node count.
• Fix: shard to a stable space — consistent hashing with virtual nodes, or a fixed number of logical shards mapped to physical nodes — so growth moves ~1/N of data, not all of it.

A cross-shard transaction half-committed (money left A, never reached B).

• Root cause: tried to update two shards without atomic commit (or a 2PC coordinator died in-doubt).
• Fix: colocate so the txn is single-shard (best); if truly cross-shard, model a saga of local transactions with idempotent compensations; reserve 2PC for rare, low-volume, correctness-critical writes and let a coordinator own it. Never hot-path XA.

Two shards both accepted the same "unique" email/username.

• Root cause: UNIQUE is per-shard; there's no global constraint.
• Fix: make the unique attribute the shard key, or front it with a single allocator/uniqueness registry, or check-then-insert with a reconciliation sweep. Decide before launch.

OFFSET 100000 LIMIT 20 across shards is unusably slow.

• Root cause: deep offset means pulling 100020 rows from every shard, then discarding.
• Fix: keyset/seek pagination — pass the last-seen sort cursor, WHERE (sort_col) < :cursor ORDER BY sort_col DESC LIMIT 20 on each shard, merge. Constant work per page.

Postgres is out of connections / falling over.

• Root cause: one pool per shard per app instance → poolSize × shardCount × instances backends; Postgres is process-per-connection.
• Fix: PgBouncer (transaction pooling) in front of each shard; small per-shard pools; watch total backend count as a first-class metric.

Wrong shard / null lookup key / data on the wrong node.

• Root cause: ShardContext set inside the @Transactional method (too late — connection already bound), never set, or not cleared so a pooled thread inherited a stale shard id.
• Fix: set the context in a filter/interceptor (or non-transactional façade) before the transaction; setLenientFallback(false) so unset = loud failure; always clear() in afterCompletion/finally.

Rebalancing caused an outage or lost writes.

• Root cause: hand-rolled move with no fencing — a key was writable on both old and new shard during cutover.
• Fix: the online playbook (backfill → catch-up via logical replication → brief fenced cutover → verify), or let Citus/Vitess/Aurora Limitless do the shard move. Fence writers (fencing tokens — locking guide) so a key is never live in two places.

Reference/broadcast table went stale on some shards.

• Root cause: the "copied to every shard" table was updated on one node only.
• Fix: update reference tables through the coordinator (Citus/Aurora do this), or propagate via the outbox/CDC pipeline; keep them genuinely small and low-write.

Premature sharding.

• Root cause: sharded startup-scale data for hypothetical scale; paid joins/transactions for load one box would have handled.
• Fix: vertical scale + replicas + partitioning + offload first; measure; if you truly need scale-out SQL, prefer managed distributed SQL over hand-rolled sharding.

Interview framing — how to sound senior (Staff+ practical)#

Lead with restraint, then decisiveness. "Before I shard I'd confirm I'm actually write-bound and out of vertical + replica + partitioning + offload headroom, because sharding is a one-way door that costs me joins, foreign keys, unique constraints, and cross-shard transactions. If I've earned it, here's how I'd do it." This ordering alone reads as Staff.

Tell the shard-key story as a trade, not a lookup. "I'd shard by <key> because the dominant access pattern — <query> — carries it and stays single-shard. That immediately makes <other pattern> cross-shard, so I'll serve that from <colocation / reference table / CDC read model / bounded scatter-gather>." Naming the cost is the signal. "There is no key with no downside" is a sentence worth saying.

Volunteer the hard parts before you're asked: colocation on the aggregate boundary so transactions stay local; sagas + idempotency instead of 2PC for the genuinely cross-shard write; UUIDv7/Snowflake because there's no shared sequence; the unique-constraint problem; keyset pagination; a logical-shard indirection so resharding is a rebalance not a rehash; PgBouncer for the connection math.

Prefer buying the engine. "For real scale-out Postgres I'd reach for Citus (colocation, reference tables, online rebalancing built in) or a managed option (Aurora Limitless, Azure Cosmos for PostgreSQL); if I need strong cross-shard transactions with minimal shard-thinking, CockroachDB/YugabyteDB speak the Postgres wire protocol and hide sharding entirely. I'd hand-roll AbstractRoutingDataSource routing only when access is ~always single-key." This shows you know the landscape and value your team's time.

Ground it in your domain (EV charging / roaming — DCS/OCPI). "For a charging/roaming platform ingesting sessions + CDRs at scale, I'd shard by CPO or account, not session_id: CPO/account-scoped dashboards stay single-shard, and I'd colocate a session with its CDRs and lifecycle events on the same shard key so a session's whole lifecycle is one local transaction. A giant CPO is the celebrity, so I'd carve whales onto dedicated shards. This mirrors what we already do in Kafka — partitioning by sessionId is the same idea, and sharding the DB by the same key keeps a key's data and its event stream aligned. Cross-CPO analytics never scatter-gathers the OLTP shards — it's a CDC → OLAP read model (CQRS). And because our workflows are single-owner in Temporal, per-session mutual exclusion comes from ownership, sidestepping distributed locks entirely." That paragraph ties sharding to five other guides and your real system — exactly the synthesis Staff interviews reward.

Traps that mark you as junior — don't step in them:

• Calling single-node partitioning "sharding." (They're different; partitioning doesn't scale writes across machines.)
• Reaching for 2PC/XA as the default for cross-shard writes.
• Sharding on a mutable or low-cardinality key (status, region, country).
• Claiming "exactly-once" cross-shard anything (it's at-least-once + idempotency).
• Forgetting that a global UNIQUE and a global sequence no longer exist.
• Binding the shard function to physical node count (mod N) with no rebalancing story.
• Serving cross-cutting reads by scatter-gathering the OLTP shards instead of a read model.

The 20-second summary: "Sharding scales writes/storage by splitting data across independent databases. The shard key is a near-permanent bet: align it with the dominant access pattern so the hot path is single-shard, colocate what must transact together, and push everything else (cross-cutting reads, analytics) to read models built by CDC. Avoid cross-shard 2PC — use sagas + idempotency. Use a stable hash into fixed logical shards so you can rebalance without rehashing. And honestly, prefer Citus / Aurora Limitless / distributed SQL over hand-rolling it — the value is in the shard-key and data-model decisions, not in reinventing a query router."

Cheat-sheet#

Three techniques (don't conflate): replication = copies (read scaling/HA) · partitioning = one table split on one node (manageability) · sharding = data split across nodes (write/storage scaling). Usually combined: each shard is itself replicated.

Shard-key checklist: high cardinality · uniform access (no celebrity) · query-aligned (hot path = single-shard) · stable/immutable. Enemies: hot shards, cross-shard joins/txns, no global unique/sequence.

Strategy → use → danger: hash → even load, point gets → kills ranges, mod N reshuffle · range → range/time scans → hotspots on monotonic keys · directory → flexibility, whale carve-out → lookup is a dependency · geo/entity → residency, isolation → skew.

Never hash mod (physicalNodes). Hash to a stable space: consistent hashing + vnodes, or fixed logical shards → physical map. Rebalance = remap, not rehash.

Cross-shard reads: avoid via colocation / reference tables / CDC read model / global secondary index. If you must: parallel fan-out, per-shard timeout, keyset (not offset) pagination, re-aggregate (AVG=SUM/COUNT; DISTINCT/percentiles → approx/OLAP).

Cross-shard writes: single-shard by colocation first → saga + idempotent compensation → 2PC only rarely and coordinator-owned. No hot-path XA.

IDs/uniqueness: UUIDv7/ULID or Snowflake (no shared sequence); PG18 has native uuidv7(). Global unique = shard-key-as-unique or a registry.

PostgreSQL ladder: partitioning (1 node) → replicas → Citus / Aurora Limitless (managed) → distributed SQL (Cockroach/Yugabyte/Spanner, sharding hidden) → hand-rolled Spring routing / ShardingSphere. postgres_fdw+partitioning = plumbing, not a distributed DB.

Spring routing gotchas: set ShardContext before @Transactional (connection binds at txn start) · one txn = one shard · setLenientFallback(false) · clear() on completion · ThreadLocal doesn't cross threads → scatter-gather hits per-shard DataSources directly · connections ≈ pool × shards × instances → PgBouncer · migrate every shard in a loop.

Self-test (can you answer these cold?)#

1. Distinguish replication, single-node partitioning, and sharding — what does each scale, and what does each not?
2. Why is PostgreSQL declarative partitioning not sharding, and what is it actually good for?
3. Name the four properties of a good shard key. Give a plausible key that violates each.
4. You shard a multi-tenant SaaS by tenant_id and one tenant is 50× the rest. What breaks, and what are your options?
5. Why does hash(key) mod N make adding a node a nightmare, and what are the two standard fixes?
6. What is colocation, and how do you configure it in Citus so a user and their orders join locally?
7. A request must debit user A (shard 1) and credit user B (shard 7). Walk through why not 2PC, and what you do instead.
8. Global UNIQUE(email) and a global auto-increment id both stop working after sharding. Why, and how do you handle each?
9. In Spring, exactly when does AbstractRoutingDataSource resolve the shard, and why must ShardContext be set before an @Transactional method? What's the symptom if you get it wrong?
10. Why must scatter-gather not use the routing DataSource + ThreadLocal, and how do you implement it correctly with virtual threads or coroutines?
11. Why is OFFSET 1000000 LIMIT 20 catastrophic across shards, and what's the constant-work alternative?
12. Compare hand-rolled Spring routing, ShardingSphere-JDBC, ShardingSphere-Proxy, Citus, and CockroachDB/YugabyteDB — when would you pick each?
13. Do the connection math for 4 shards, pool size 10, 30 app instances. What's the fix?
14. Outline a zero-downtime resharding from 4 → 8 shards. Where does fencing matter, and why prefer logical shards?
15. For an EV-charging/roaming platform, choose a shard key and defend it: which queries become single-shard, which become cross-shard, and where does each cross-shard need go instead?

Lineage & sources (verified July 2026, current-fact checks): PostgreSQL 18 released 25 Sep 2025 (release notes, press kit); declarative partitioning docs (Table Partitioning). Citus 14, AGPLv3, PG 16–18, coordinator/worker + create_distributed_table/reference tables/colocation (Citus FAQ, GitHub, Microsoft Learn). Aurora PostgreSQL Limitless Database GA — routers/shards/shard groups, sharded/reference/standard tables (architecture, GA announcement). Apache ShardingSphere 5.5.x (project, 5.5.3 release). Multigres "Vitess for Postgres," v0.1 alpha released June 2026, announced 2025 (announcement, v0.1 alpha). Distributed-transaction, ID, and consistency framing cross-referenced with this folder's databases-and-data-storage, distributed-systems, saga, locking, CDC, and schema-evolution guides. Foundational sharding/consistent-hashing/rebalancing concepts are standard distributed-systems material (stable across versions); tool capabilities and version numbers are the version-dependent facts — re-verify against current release notes before quoting them in an interview.*