Sharded Postgres: Query Locality Is the Architecture Contract

A billing screen asks for one company’s unpaid invoices. A finance export asks for every company’s unpaid invoices. The SQL may differ by one predicate. In a sharded database, those two requests can have different capacity limits, consistency guarantees, and failure modes.

That is the distinction I would put at the centre of a sharding review. Not how many nodes the database supports, but which product operations remain local as the database grows.

PlanetScale’s walkthrough of a sharded Postgres query provides the starting point for this article. It describes Neki’s router, distributed planning, sidecar connection pools, and execution on PostgreSQL shards. The architectural questions below use a separate billing example; they are not claims about capabilities every sharded PostgreSQL product provides.

Start with the operation, then choose the key #

Consider a SaaS product where users inspect invoices inside one tenant. A candidate relational model is:

CREATE TABLE accounts (
    tenant_id bigint NOT NULL,
    account_id bigint NOT NULL,
    name text NOT NULL,
    PRIMARY KEY (tenant_id, account_id)
);

CREATE TABLE invoices (
    tenant_id bigint NOT NULL,
    invoice_id bigint NOT NULL,
    account_id bigint NOT NULL,
    amount_due numeric(14, 2) NOT NULL,
    due_at timestamptz NOT NULL,
    PRIMARY KEY (tenant_id, invoice_id),
    FOREIGN KEY (tenant_id, account_id)
        REFERENCES accounts (tenant_id, account_id)
);

CREATE INDEX invoices_tenant_due_idx
    ON invoices (tenant_id, due_at);

This is ordinary PostgreSQL DDL, not a command to distribute the tables. Distribution requires the chosen platform’s configuration. Verify its support for foreign keys and uniqueness constraints before adopting this schema.

The intended placement rule is that both tables use tenant_id, with compatible hashing and shard placement. A tenant’s account and invoice rows should then share a location. Matching column names alone do not establish that guarantee. Citus’s distribution-column guidance explains tenant-based distribution and why related tables should be colocated.

The corresponding application query is:

PREPARE overdue_invoices (bigint, timestamptz) AS
SELECT i.invoice_id, a.name, i.amount_due, i.due_at
FROM invoices AS i
JOIN accounts AS a
  ON a.tenant_id = i.tenant_id
 AND a.account_id = i.account_id
WHERE i.tenant_id = $1
  AND i.due_at < $2
ORDER BY i.due_at, i.invoice_id
LIMIT 100;

The tenant predicate offers an opportunity to route to one shard. The join includes the tenant boundary explicitly. The composite keys prevent a local account identifier from silently referring to another tenant’s account.

None of that authorizes access. The application must derive or validate the tenant against the authenticated principal. A caller-controlled tenant ID is a routing input, not a security policy.

Local joins and local requests are different #

There are three useful execution shapes:

Tenant-scoped request
  application → router → one shard → local join

Global request, colocated tables
  application → router → all relevant shards
                         each performs a local join
                     ← combined results

Global request, unrelated placement
  application → distributed execution
                fetch or redistribute matching inputs
                join across placement boundaries
              ← results

These are conceptual diagrams, not a promise about a particular optimizer.

Colocation can eliminate movement of join inputs without eliminating fan-out. Remove the tenant filter from the example and the database still needs to find overdue invoices across tenants. A locally efficient join does not make that request a single-shard operation.

PlanetScale’s example makes this concrete: changing the distribution relationship lets shards perform a join that previously ran in the router. Its router-side hash join can spill to disk when its memory budget is exceeded. Those are implementation details worth understanding, not universal descriptions of all distributed SQL engines. Other plans may broadcast a small input, repartition data, or perform targeted lookups. Source walkthrough.

My review question is therefore precise: for each important query, where are rows filtered, where do they move, and where does state accumulate?

Estimate the work hidden behind one request #

Here is an illustrative capacity model, not a benchmark. Assume a report fetches 25,000 intermediate rows from each of 24 shards, averaging 160 transmitted bytes per row:

Intermediate data per report
  = shards × rows per shard × bytes per row
  = 24 × 25,000 × 160
  = 96,000,000 bytes, or 96 MB

At five reports per second
  = 480 MB/s of intermediate row traffic

That estimate excludes framing, retransmission, and other queries. It says nothing about the size of the final response. A tiny chart can require a large distributed scan.

If the router holds a join input, estimate resident memory separately. Encoded row width is not an adequate estimate of hash-table overhead, allocation behaviour, or concurrent operator state. Measure peak memory under representative concurrency, including spill and cancellation.

For a stage that needs every participating shard, completion time is approximately bounded by the slowest participant, plus coordination and merge work. As a deliberately simplified probability model, if each shard independently meets a deadline with probability p, all n meet it with probability p^n. With p = 0.99 and n = 24, that is about 0.786.

This is not a production availability forecast: shard delays can be correlated, and real plans have multiple stages. It demonstrates why a healthy per-shard percentile is insufficient evidence for a healthy fan-out request.

Put global reporting in a separate workload budget. For interactive traffic, carry a bounded request deadline through routing, pool wait, execution, and result delivery.

PostgreSQL compatibility needs a test matrix #

Accepting a PostgreSQL connection does not prove every driver behaviour is supported.

The PostgreSQL extended query protocol separates parsing, binding, and execution; portals track execution state. A parameterized statement’s shard-key value may arrive during binding, rather than with its SQL text. That creates a concrete routing and testing requirement.

Test the driver configuration you actually deploy: named prepared statements, repeated binds for different tenants, transaction errors, cancellation, result formats, and connection reuse. Run the same tests during failover, not just against an idle cluster.

Pooling has its own compatibility boundary. PgBouncer’s feature matrix distinguishes session and transaction pooling and documents limitations around session-dependent features. Do not assume another proxy inherits either its guarantees or its restrictions. Ask which state is pinned, restored, rejected, or reset, and demonstrate that a reused backend cannot inherit the previous borrower’s identity.

Pooling is a concurrency budget, not free capacity #

Suppose 12 routers each maintain up to eight backend connections per shard. A shard can receive up to 96 pooled connections from those routers, before administrative and replication connections are considered. Across 24 shards, that configuration permits 2,304 application-side backend connections.

A shared pool near each shard can decouple router count from the backend pool size. It cannot remove the execution work. Excess demand moves into waiting, rejection, or timeouts.

For an initial estimate, use average busy connections ≈ arrival rate × average connection-hold time, with consistent units and stable conditions. At 400 operations/s and 20 ms average hold time, the mean is eight busy connections. Eight is not automatically a safe pool size: variance, locks, bursts, and transaction duration require headroom and measurement.

Record pool-wait time separately from execution time. Otherwise an apparently slow query may mostly be waiting for permission to start.

A global read needs an explicit consistency contract #

An invoice transfer between accounts is a useful thought experiment. If related records are read from different shards at different moments, can the report combine states that never existed together?

PostgreSQL documents statement snapshots at Read Committed and transaction snapshots at Repeatable Read in its isolation reference. Independently starting transactions on multiple PostgreSQL instances does not itself establish a shared global snapshot. The distributed layer must define any stronger guarantee.

Ask the platform what a cross-shard read promises, how distributed writes commit, and what happens during topology changes. For financial reconciliation, do not substitute “each shard uses MVCC” for an answer.

If a report permits lag, say so in the product contract. A dedicated analytical copy with an explicit freshness timestamp can be a better fit than an expensive request whose freshness users merely assume. This connects to read-replica freshness contracts.

Failure policy belongs beside the query plan #

The following is a proposed application policy, not a description of default PostgreSQL behaviour:

Returning some streamed rows is not proof that a query completed successfully. A consumer performing reconciliation must distinguish rows received from a successfully completed result. Read-only retries also need a defined snapshot or restart policy if downstream processing has already consumed part of the stream.

For writes, exactly-once boundaries remain relevant even when the transport changes from a message broker to SQL.

The acceptance test is a workload, not a node count #

Before migration, classify query fingerprints into single-shard, bounded fan-out, and global work. For each class, define acceptable latency, concurrency, intermediate bytes, and failure behaviour. Trace the parent request and its shard operations together.

Then test conditions that invalidate the attractive average:

• One tenant owns a disproportionate share of traffic or data.
• One shard slows down while the others remain healthy.
• A report spills while interactive requests arrive.
• A caller disconnects after receiving the first result batch.
• Prepared statements are reused across tenants and topology changes.
• A shard moves while reads and writes continue.

Tenant-based distribution trades simple locality for a possible large-tenant hotspot. Splitting a large tenant may relieve that hotspot while breaking the original locality assumption. Treat that change as a revised data model and operating contract, not a transparent capacity upgrade.

The decision I would want documented is short: which measured single-node limit requires sharding, which operations remain local, which global operations receive separate budgets, and what correctness tests must survive a failure.

More shards are useful when they add capacity to the work the product actually performs. They are much less useful when every important request still asks the entire cluster to cooperate.

References #

• PlanetScale: The lifecycle of a sharded Postgres query — the implementation walkthrough that prompted this analysis.
• Citus: Choosing a distribution column — workload locality and distribution choices.
• PostgreSQL: Protocol message flow — statements, portals, and execution messages.
• PostgreSQL: Transaction isolation — snapshot semantics within PostgreSQL.
• PgBouncer: Features — pooling-mode compatibility boundaries.