Distributing PostgreSQL work

PostgreSQL can divide a table into partitions, expose another server’s tables through SQL, or run as one of several databases that own different customers. Citus adds distributed planning and shard management to PostgreSQL. These choices move different work and leave different responsibilities with the application.

Start with one request. Mina’s order history needs her orders and their lines. Keeping those records together lets one database answer it. A purchase also changes the stock shared with Tariq, so distributing their histories does not make their purchases independent.

Sharding and distributed SQL established that boundary. Here we follow it through PostgreSQL mechanisms: find an owner, run local work, inspect what crosses a foreign-table connection, and move a customer group without letting both copies accept writes. The placement drawings are illustrative. The SQL checks and the deliberately paused native experiment have narrower, stated limits.

Native table partitioning

A native partitioned table remains one logical table. With monthly order partitions, an October predicate can exclude other months before looking for rows. That can reduce work inside one PostgreSQL instance; it does not give the instance another server’s processor or memory.

A replica instead holds another copy of the same history and can serve eligible reads. A shard owns a disjoint portion of the data and its independent work. Partitions, copies and owners therefore answer different capacity questions even when each divides what a reader sees.

If the problem is excluding old months or retiring a whole eligible month, begin with Table partitioning and data lifecycle. If it is read traffic, consider the freshness contract and routing in Replicas, lag and read routing. Sharding becomes useful when independent stored data or work needs more than one server.

PostgreSQL also permits foreign tables as partitions, which changes the ordinary local-storage arrangement. Their data can live elsewhere. That requires care with partition bounds and brings the foreign wrapper’s execution and transaction behaviour. Declaring a remote partition does not supply automatic shard balancing or a distributed purchase protocol.

Application-managed shards

With application-managed shards, the application chooses the database connection. Reuse the parent article’s placement: customer group A contains Mina, B contains Tariq and C contains Edda. Each group keeps its customers, orders, lines and payments together. P owns the shared products. Groups can contain many customers; these letters are illustrative assignments, not a hash function.

The router might consult a durable directory or compute a bucket from a customer key and look up that bucket’s current owner. A hash that changes whenever a server is added would remap data the application has not moved. Keep the logical group separate from its current database address.

Finding the right owner

Mina’s history request carries customer_id = 1, which selects A. The query can then find her newest orders through A’s local indexes. Routing finds the database; an index finds rows inside it.

A customer history can reach one owner because its request supplies the customer key. An order-only lookup needs another way to find that group; a purchase may also need the shared stock owner. Request information and invariant scope determine which owners participate.

An order number alone need not reveal the customer group. A maintained directory can map 103 to A, but its entries must be created consistently with orders and updated with movement. A stale or missing entry needs a defined response. Searching every owner is another design: each can use its own order-ID index, but the application must combine the answers and handle an unavailable owner. A local index on each database does not create a global routing index.

Each database also brings connection limits, schema versions, statistics, backups and a recovery procedure. A rollout that upgrades A’s schema but leaves B’s behind must still serve compatible application requests. Restoring rows without the matching routing and request-outcome records can recover a different application state from the one customers were told about.

Keeping rules within reach

Return independently to before purchase 103. A must create Mina’s pending order, its P7 and P3 lines, and the durable result for request purchase-103. P must guard the last mug, reducing P7 from 1 to 0, and reduce P3 from 12 to 10. The line value is 29.00 and no payment is created.

Committing A, then P, permits a failure between them. Reversing the order permits the opposite partial result. A transaction on each connection is insufficient to give the purchase one atomic outcome. The design must also preserve global order identity and the references from lines and payments, wherever those checks live.

PostgreSQL’s PREPARE TRANSACTION can retain prepared work for a later commit or rollback. It is a building block for an external transaction manager that durably records and recovers the joint decision. It is not a replacement for that manager. Prepared transactions can keep locks while the decision is unresolved. Transactions across partitions develops this protocol and its failure cases; the placement here tells us why it is needed.

Co-location in Citus

Citus supplies a SQL entry point that uses distribution metadata to plan work on PostgreSQL workers. For a customer-distributed layout, a distribution column chooses each row’s shard. Co-location keeps corresponding shards of related tables together, so matching customer records can be joined on a worker.

In a candidate customer-distributed copy, adding a customer column to lines gives the planner the routing key and lets it participate in the reference. The corresponding shard ranges and placements must agree; matching column names alone cannot establish locality.

The corresponding local join includes the distribution column:

SELECT o.order_id, SUM(i.quantity * i.unit_price) AS line_value
FROM orders_dist AS o
JOIN order_items_dist AS i
  ON i.customer_id = o.customer_id
 AND i.order_id = o.order_id
WHERE o.customer_id = 1
GROUP BY o.order_id;

For separately prepared candidate tables, the documented distribution calls make that intended co-location explicit. This is Citus SQL, not ordinary PostgreSQL, and was not executed in the article’s SQL fixture:

SELECT create_distributed_table(
  'orders_dist', 'customer_id', colocate_with => 'none');
SELECT create_distributed_table(
  'order_items_dist', 'customer_id',
  colocate_with => 'orders_dist');

Compared with application routing, Citus takes responsibility for finding shards and distributing supported query work. The worker still runs PostgreSQL plans. SQL that crosses unrelated shard boundaries can require data movement, multiple tasks or encounter feature restrictions. Check the actual query on the selected Citus release rather than inferring unrestricted distributed PostgreSQL from a local join.

Distribution keys and constraints

For the hash-distributed tables considered here, Citus requires primary and unique keys to include the distribution column. Foreign keys between co-located distributed tables must include it too. A distributed table may also reference a reference table; a reference table cannot reference a distributed table in the reverse direction.

That makes these candidate keys natural for local enforcement:

-- Parts of a candidate schema, not an accepted shop migration:
-- orders_dist:
PRIMARY KEY (customer_id, order_id)

-- order_items_dist:
PRIMARY KEY (customer_id, order_id, line_no)
FOREIGN KEY (customer_id, order_id)
  REFERENCES orders_dist (customer_id, order_id)

But the shop’s existing rule is stronger: order_id alone identifies an order globally. A pair key accepts both (1, 103) and (2, 103). These are a counterexample, not two valid shop orders. Repeating (1, 103) is rejected; repeating only 103 under another customer is not. An ID-only foreign key also lacks a unique parent to reference.

Choose an actual enforcement design for that global rule, or deliberately change the identity contract and every dependant. An ID generator alone does not reject an explicit duplicate. The candidate copy demonstrates locality; it has not settled the shop’s migration. Payments, request identities and any order directory need the same scrutiny as lines.

Shared reference tables

A small product catalogue projection containing IDs and names can be useful on every worker. Citus reference tables are managed replicated tables for such shared data. A worker can join its local customer rows to a catalogue copy without fetching each name from a separate owner. Maintaining those copies makes writes a coordinated operation; the documentation describes two-phase commit for reference-table modifications.

A copied catalogue is different from an independent stock allowance. Keeping the original shared stock authority means a purchase can still cross owners even when its history and catalogue-name join are local. Putting stock into a managed reference table would still require coordinated updates. Giving each worker stock 1 would instead create several opportunities to sell the same last mug.

Foreign tables and remote work

postgres_fdw lets a PostgreSQL database expose another PostgreSQL server’s tables as foreign tables. The caller writes local SQL; the wrapper sends suitable work to the configured remote server and brings results back. The server definition and user mapping identify the remote connection. They do not decide which owner ought to hold customer 1.

On the querying PostgreSQL instance, remote.orders is a foreign table mapped to remote server S’s public.orders. A history request can send its customer predicate to S, then sort the returned rows locally. The plan identifies which side performs each operation.

EXPLAIN (VERBOSE, COSTS OFF)
SELECT order_id, status
FROM remote.orders
WHERE customer_id = 1
ORDER BY order_id;

The native plan shows a local sort above the foreign scan. The customer filter travels; the requested ordering is done after the rows return:

Sort
  Sort Key: orders.order_id
  → Foreign Scan on remote.orders
      Remote SQL: SELECT order_id, status FROM public.orders
                  WHERE ((customer_id = 1))

A second native query joins S’s orders and lines and sums their values. Its plan instead contains Relations: Aggregate on ((remote.orders o) INNER JOIN (remote.order_items i)): both join and grouping ran on S, returning 101: 39.50 and 103: 29.00 before the querying server’s local sort. The input tables did not have to cross the connection to be joined locally.

Filtering before transfer can save rows crossing the connection; selecting only needed columns can save width. A join between compatible foreign tables on the same foreign server can also run remotely when the planner chooses it. A local function it cannot send, different user mappings or tables on different servers can change that boundary. Read the actual Remote SQL and the local plan, rather than assuming every SQL operation travels.

This is useful for accessing separately owned data, gradual integration or a deliberately bounded remote query. It does not maintain an order directory, allocate stock or move a shard when a worker fills.

The transaction boundary

The wrapper opens corresponding remote transactions and ends them with the local transaction. In PostgreSQL 18, their isolation is Repeatable Read unless the local transaction is Serializable, in which case it is Serializable. Repeated scans on one such remote connection can therefore share its snapshot even under a local Read Committed transaction.

Separate remote servers do not thereby acquire one common snapshot. Nor does a local BEGIN prove an atomic purchase across them: PostgreSQL 18’s postgres_fdw cannot prepare its remote transactions for two-phase commit. Its transaction handling is insufficient evidence for the original all-or-nothing order-and-stock contract. Test the complete failure protocol before making that promise.

Skew and queries across workers

A customer key can exclude owners from one history lookup. The October paid-order report has no such equality. It needs every customer owner that can contribute. Co-location still helps: each owner joins its orders and lines locally and can reduce them before sending results.

For the small paid-order report, A can return Mina’s total 39.50 and B Tariq’s 40.00; Edda’s group contributes no paid order. The entry point combines those complete customer totals and sorts them.

Because one owner contains all of each customer’s orders, its grouped result is complete for that customer. If the layout splits a customer across owners, the entry point must sum that customer’s partials before sorting. Sending every qualifying line instead would move more rows. A different join key can require moving inputs before the join as well.

Fan-out can reduce elapsed time through parallel execution while adding messages and combining work. A complete report waits for every required owner. One slow or unavailable owner can determine its outcome; silently omitting that owner produces a different report. The snapshot contract remains a separate question, developed in Time, snapshots and distributed reads.

The collection’s separate generated workload makes another limit visible. Customer 1001 has 84,000 of 120,000 generated orders:

Hashing the customer key keeps all 84,000 together. More buckets do not divide equal key values. Moving that group to another worker can move its stored data and load, but cannot split the group. Distributing by order instead could spread its independent order work while making the customer history fan out.

Those counts establish data skew, not measured traffic or elapsed time. A customer with much old history may be quiet; one P7 stock row can be busy. Measure bytes, requests, writes and contention by key. Dividing the stock row’s callers across workers does not give them independent authority over the same stock.

Moving a shard safely

Suppose group A is to move from PostgreSQL S to D. Its customer, order, line, payment and durable request-outcome records must arrive together. Product stock remains with P. A maintained order directory and other incoming references must continue to lead to the same group at its new owner.

First take a deliberately paused approach. A pause makes the copy’s boundary easier to establish; it does not remove the authority problem. Keep the destination closed until the source has stopped accepting group writes and the new routing decision is durably recorded.

Close the source before handing over

The admission gate must cover every mutation of A: application calls, jobs, imports, administrative paths and new references that could change the copied scope. Stopping the main router alone does not stop an old connection or an already admitted transaction.

One bounded implementation uses a group gate row. Every admitted mutation checks that row while holding a shared row lock until its transaction ends. Closing takes a conflicting update lock, waits for admitted work, then commits the closed state. Later attempts must observe the closed gate before doing group work. The runtime role must be unable to bypass the guarded path. Isolation, function boundaries and lock order are part of this protocol, so substituting an earlier cached flag would not preserve it.

After drain, copy and compare the complete closed group. Counts alone can match while values differ. Validate keys, relationships, exact request payloads and terminal results, plus any sequences, large objects or external records the actual design requires. A source write after this check would invalidate the copy, which is why the gate stays closed.

Persist the route to D before admitting its writes. The controller must recover from each recorded stage. Before closure, S remains the owner and D stays closed. After closure but before the routing decision, it can resume copying from the closed source. After the decision, it completes or repairs the D handover while keeping S closed. A timeout or controller restart must not guess that both may reopen. Protect and recover the routing authority itself; a lost or independently writable directory would undermine the handover.

The native experiment uses three disposable PostgreSQL 18.6 instances, each with a separate data directory and Unix socket inside one isolated container. It observes an admitted transaction holding the shared gate lock while closure waits, then compares the closed-source copy exactly. After recording D as owner and opening it, the tested source path rejects a stale request. D recovers purchase-103 → 103; a changed payload under that identity is rejected.

This verifies that bounded gate and retry path. It does not test a complete purchase, bypass-proof production administration, controller crash recovery or independent-machine failure. The fixture’s runtime has only the guarded result-recovery function. The illustrated full handover also requires the wider scope and recovery rules described above.

The counterexample is direct: change routing while S remains open, and an old client can still pass S’s gate. A successful copy and a new route do not close the old authority. At cutover, fail or pause affected work when ownership cannot be established; availability does not justify allowing both owners to proceed.

Catch-up, return and managed movement

An online variant keeps the source accepting writes during most of the transfer. It additionally needs a consistent snapshot aligned with a change stream, enough retained history to catch up, and an apply process that preserves the relevant schema and transaction semantics. It still needs a final protected boundary to drain old writes, finish applying them and transfer authority.

Native logical replication can supply parts of that machinery, but its supported objects and limits must match the group. PostgreSQL logical replication and change feeds examines that mechanism; Rebalancing and cluster change develops catch-up and the capacity it consumes. A whole-system or version migration has a different scope in Upgrading and migrating PostgreSQL.

Citus provides supported shard-movement and rebalancing functions that move co-located shards together and update placement metadata. Its documented transfer modes include logical movement and blocking writes. Use the chosen release’s machinery and prerequisites, inspect completion and final placements, then exercise real requests. A submitted rebalance request or finished base copy is not the same evidence. No Citus move was run for this article.

Once D accepts new writes, returning is another transfer. Reconcile those writes or send them back through an equally protected reverse handover. Changing the route back to the old S copy would discard the very work that made D the owner.

Choosing and checking the design

Choose from the limiting work and the rules it must preserve. Several mechanisms can coexist, but each should earn a particular job.

Compare the same history, report and purchase contract in each candidate. Include an order-ID-only request, the frequent customer key, a missing worker and a stale route during movement. Check correct rows and ordering before measuring latency. Inspect local plans, remote SQL or distributed tasks to see which work was excluded, which stayed local and which rows crossed connections.

Then rehearse operation: compatible schema changes on every owner, backups of data and metadata, restoration to a coherent application state, route recovery and the return boundary after new writes. The useful result is a design that places independent work where capacity is available while preserving its declared rules. More PostgreSQL processes alone establish neither.