Citus does not scale PostgreSQL writes because a diagram contains more worker nodes. It scales the part of the workload whose data ownership, constraints, queries, and transactions all agree on the same distribution key.

Situation

RetailCo operates four retail brands and thousands of stores and merchant accounts. Its order platform grew from one PostgreSQL database into a 40-TB transaction system. Read replicas and table partitioning improved read capacity and lifecycle management, but checkout writes still pass through one writer.

The preceding RDS PostgreSQL and Citus reality check established that standard Amazon RDS for PostgreSQL cannot supply the Citus runtime. This article starts after that decision.

RetailCo is now evaluating Citus 13.0.1 on EC2 under five explicit design assumptions:

  • most checkout transactions belong to one tenant_id;
  • cross-tenant finance and risk reporting can leave the synchronous checkout path;
  • PostgreSQL behavior and extension control matter more than a fully managed database control plane;
  • a database platform team can operate PostgreSQL and Citus across multiple hosts;
  • no throughput or recovery claim will be approved without a lab result.

RetailCo is a reference architecture, not a claimed customer deployment. The 40-TB size, tenant counts, and transaction shapes are scenario inputs. Any measurements remain proposed until a reproducible lab produces raw output.

The intended production shape has one application entry point, a highly available coordinator tier, and worker groups distributed across Availability Zones:

flowchart TD
  App[RetailCo order services] --> Endpoint[database writer endpoint]
  Endpoint --> Pool[PgBouncer transaction pool]
  Pool --> CoordA[coordinator primary — zone A]
  CoordA --> CoordB[coordinator standby — zone B]
  CoordA --> WorkerA[worker primary A]
  CoordA --> WorkerB[worker primary B]
  CoordA --> WorkerC[worker primary C]
  WorkerA --> StandbyA[worker standby A]
  WorkerB --> StandbyB[worker standby B]
  WorkerC --> StandbyC[worker standby C]

The arrows from primary to standby represent PostgreSQL streaming replication. The application endpoint, promotion controller, and connection pool are separate responsibilities; PgBouncer does not elect a new database primary.

This diagram describes the failure domains, not the final capacity. The correct worker count, EC2 instance type, EBS configuration, shard count, and connection limits must come from the benchmark and recovery exercises.

The Problem

RetailCo could distribute its largest tables with a few SQL commands:

SELECT create_distributed_table('orders', 'tenant_id');
SELECT create_distributed_table('order_items', 'tenant_id');
SELECT create_distributed_table('payments', 'tenant_id');

That is necessary but insufficient. Four design errors can leave the platform slower, less reliable, or impossible to operate.

Error 1 — choosing a business label instead of an ownership key

brand_id has only four values. It creates four large ownership domains and makes one busy brand a permanent hot spot. Adding a fifth worker would not divide any of those four values.

tenant_id has much higher cardinality and appears in the dominant checkout path. Hashing many tenants across many shards gives Citus room to spread independent tenants across workers. It does not guarantee balance: a single large tenant can still dominate the shard and worker that own it.

Error 2 — distributing tables but not colocating the transaction

If orders is distributed by tenant_id while order_items is distributed by order_id, one checkout transaction can cross worker boundaries. Joins and foreign keys no longer align naturally, and each commit may require distributed coordination.

Citus describes colocation as placing the same distribution-key values for related tables on the same workers. In a row-based multi-tenant model, every tenant-owned table should normally carry the same distribution column, and tenant-local queries should filter by it: Citus concepts and Citus DDL reference.

Error 3 — assuming distributed SQL makes global rules free

Citus cannot enforce a primary key or unique constraint across independent shards unless the constraint includes the distribution column. Foreign keys between colocated distributed tables must include that column as well: constraints on distributed tables.

RetailCo must decide whether an email, order number, or idempotency key is unique globally or only inside one tenant. A database cannot preserve an unstated business rule.

Error 4 — treating EC2 as a managed database service

Citus on EC2 gives RetailCo package, extension, operating-system, storage, network, and failover control. It also gives RetailCo responsibility for all of them.

The coordinator holds cluster metadata and plans distributed work. Workers hold the distributed shards. Losing either tier without a tested promotion and recovery procedure is an outage, even when every EC2 instance and EBS volume still exists.

The architectural question is therefore not “Can Citus create shards?” It is:

Can RetailCo make its critical writes tenant-local while operating the coordinator, workers, replication, rebalancing, upgrades, backups, and recovery as one database system?

Build Around Tenant-Local Transactions

The recommended model is Citus row-based sharding with tenant_id as the common distribution column for the order write path. Citus also supports schema-based sharding, but row-based sharding is the better match here because RetailCo has many tenants sharing one schema and needs parallel cross-tenant analysis outside the checkout path: Citus sharding models.

Classify every table before distributing anything

Every table belongs in one of three groups:

Table classPlacementRetailCo examplesRule
Distributed and colocatedHash-sharded across workers by tenant_idtenants, customers, orders, order items, paymentsInclude tenant_id in keys and tenant-local queries
ReferenceOne logical shard copied to every workerbrands, order status codes, country codesKeep small and relatively stable
Coordinator-localStored only on the coordinatornarrowly scoped administration metadataDo not put large or checkout-critical data here

Reference tables are useful because every worker can join them locally. They are not free replicas for arbitrary tables. Citus uses two-phase commits when modifying reference tables, and every new worker needs their contents. A large or frequently updated product catalog may be a poor reference table even if every tenant reads it: Citus table types.

For RetailCo, brands and order_status_codes are safe reference-table candidates. A multi-million-row catalog with frequent price and inventory updates needs a separate ownership decision.

Put the ownership key into the schema

The following schema is deliberately explicit. Identifiers are supplied by the application or an approved ID service; the example does not assume a particular global sequence design.

CREATE TABLE brands (
    brand_id        bigint PRIMARY KEY,
    brand_name      text NOT NULL UNIQUE
);

CREATE TABLE order_status_codes (
    order_status    text PRIMARY KEY,
    is_terminal     boolean NOT NULL
);

CREATE TABLE tenants (
    tenant_id       bigint NOT NULL,
    brand_id        bigint NOT NULL,
    tenant_name     text NOT NULL,
    PRIMARY KEY (tenant_id),
    FOREIGN KEY (brand_id) REFERENCES brands (brand_id)
);

CREATE TABLE customers (
    tenant_id       bigint NOT NULL,
    customer_id     bigint NOT NULL,
    email           text NOT NULL,
    customer_name   text NOT NULL,
    PRIMARY KEY (tenant_id, customer_id),
    UNIQUE (tenant_id, email),
    FOREIGN KEY (tenant_id) REFERENCES tenants (tenant_id)
);

CREATE TABLE orders (
    tenant_id       bigint NOT NULL,
    order_id        bigint NOT NULL,
    customer_id     bigint NOT NULL,
    order_status    text NOT NULL,
    ordered_at      timestamptz NOT NULL,
    total_amount    numeric(18,2) NOT NULL,
    PRIMARY KEY (tenant_id, order_id),
    FOREIGN KEY (tenant_id, customer_id)
        REFERENCES customers (tenant_id, customer_id),
    FOREIGN KEY (order_status)
        REFERENCES order_status_codes (order_status)
);

CREATE TABLE order_items (
    tenant_id       bigint NOT NULL,
    order_id        bigint NOT NULL,
    line_number     integer NOT NULL,
    product_id      bigint NOT NULL,
    quantity        integer NOT NULL CHECK (quantity > 0),
    unit_price      numeric(18,2) NOT NULL,
    PRIMARY KEY (tenant_id, order_id, line_number),
    FOREIGN KEY (tenant_id, order_id)
        REFERENCES orders (tenant_id, order_id)
);

CREATE TABLE payments (
    tenant_id       bigint NOT NULL,
    payment_id      bigint NOT NULL,
    order_id        bigint NOT NULL,
    idempotency_key text NOT NULL,
    payment_state   text NOT NULL,
    amount          numeric(18,2) NOT NULL,
    updated_at      timestamptz NOT NULL,
    PRIMARY KEY (tenant_id, payment_id),
    UNIQUE (tenant_id, idempotency_key),
    FOREIGN KEY (tenant_id, order_id)
        REFERENCES orders (tenant_id, order_id)
);

The schema changes the meaning of uniqueness. UNIQUE (tenant_id, email) prevents duplicate emails inside one tenant, not across RetailCo. If the business requires global email uniqueness, it needs a globally coordinated registry or a different ownership rule.

Next, create reference tables first, distribute the parent tables before their dependents, and explicitly colocate the order graph:

SELECT create_reference_table('brands');
SELECT create_reference_table('order_status_codes');

-- Illustrative lab setting, not a universal production recommendation.
SET citus.shard_count = 128;

SELECT create_distributed_table('tenants', 'tenant_id');

SELECT create_distributed_table(
    'customers',
    'tenant_id',
    colocate_with => 'tenants'
);

SELECT create_distributed_table(
    'orders',
    'tenant_id',
    colocate_with => 'tenants'
);

SELECT create_distributed_table(
    'order_items',
    'tenant_id',
    colocate_with => 'tenants'
);

SELECT create_distributed_table(
    'payments',
    'tenant_id',
    colocate_with => 'tenants'
);

Citus defaults citus.shard_count to 32 for new hash-distributed tables: Citus configuration reference. The example uses 128 only to create enough placement units for a 1, 2, 4, and 8-worker lab.

Shard count is a tradeoff. Too few shards produce large movement units and cap placement flexibility. Too many add planning, metadata, connection, maintenance, and rebalance overhead. Citus can change an existing table’s shard count with alter_distributed_table, but that is still a data-reorganization operation—not permission to ignore the initial design: Citus cluster sizing.

For the scenario’s 40-TB transaction history, 32 equal shards would average about 1.25 TB before replicas and skew; 128 would average about 320 GB. Neither average proves that a shard is operationally safe to move, restore, vacuum, or keep within its storage latency target. Measure real shard sizes and tenant skew.

Make the checkout transaction route to one worker

A tenant-local checkout carries tenant_id into every statement:

BEGIN;

INSERT INTO orders (
    tenant_id,
    order_id,
    customer_id,
    order_status,
    ordered_at,
    total_amount
)
VALUES ($1, $2, $3, 'created', clock_timestamp(), $4);

INSERT INTO order_items (
    tenant_id,
    order_id,
    line_number,
    product_id,
    quantity,
    unit_price
)
VALUES
    ($1, $2, 1, $5, $6, $7),
    ($1, $2, 2, $8, $9, $10);

INSERT INTO payments (
    tenant_id,
    payment_id,
    order_id,
    idempotency_key,
    payment_state,
    amount,
    updated_at
)
VALUES ($1, $11, $2, $12, 'pending', $4, clock_timestamp());

COMMIT;

Because the tables are colocated on tenant_id, one tenant’s rows map to the same shard placements. The coordinator can route the tenant-local transaction to the worker that owns those placements.

flowchart TD
  Checkout[checkout with tenant identifier 1842] --> Coordinator[Citus coordinator]
  Coordinator --> Hash[hash tenant identifier]
  Hash --> Shard[owning colocated shard group]
  Shard --> WorkerB[worker B]
  WorkerB --> Orders[orders rows]
  WorkerB --> Items[order item rows]
  WorkerB --> Payments[payment state rows]

The application must preserve that locality. A query that omits tenant_id, joins on unrelated keys, or modifies several tenants can fan out to multiple shards. Citus splits multi-shard queries into tasks, opens connections to relevant workers, collects results through the coordinator, and returns the final result: Citus query processing.

Multi-shard SQL is supported, but it has different cost and failure behavior. Citus’s current DML documentation notes that updates or deletes affecting one shard execute on one worker, while multi-shard modifications can use two-phase commit for greater safety: Citus DML reference. Transaction locality should therefore be measured as a first-class production metric, not treated as a coding convention.

Use time partitioning for time, distribution for ownership

The 40-TB history still needs lifecycle boundaries. The distribution column should remain tenant_id; using a timestamp as the hash key would scatter time ranges and make tenant-local transactions cross shards.

Native PostgreSQL range partitioning can be layered on the time dimension where the workload and Citus version support the chosen table design:

tenant_id hash distribution
        ↓
independent worker ownership
        ↓
ordered_at range partitions inside the distributed design
        ↓
smaller retention and maintenance units

This preserves two separate purposes: Citus distribution assigns write compute and ownership, while PostgreSQL time partitioning supports retention, pruning, and maintenance. Test the exact DDL, foreign keys, indexes, and lifecycle operations before applying this combination to the production schema.

Plan for hot tenants before they become incidents

Hash distribution balances many tenant identifiers statistically. It does not divide the workload generated by one identifier.

flowchart TD
  Many[thousands of normal tenants] --> Shards[many shared shards]
  Large[one dominant tenant] --> Hot[one hot shard group]
  Shards --> Workers[balanced worker fleet]
  Hot --> OneWorker[one saturated worker]

Citus provides isolate_tenant_to_new_shard() to move one tenant out of a shared shard. With CASCADE, it isolates the tenant across colocated tables. The isolated shard can then be moved to a dedicated worker: Citus tenant isolation.

Isolation improves blast-radius and placement control. It does not split one tenant’s rows across several write owners. If one tenant alone exceeds one worker, RetailCo must subdivide that tenant by a stable lower-level ownership key, move the tenant to a dedicated architecture, or decompose the write domain.

Adding a worker does not move existing data

Registering a new worker adds it to Citus metadata and copies reference tables as required:

SELECT citus_add_node('worker-d.internal', 5432);

Existing distributed shards do not become balanced merely because the worker exists. Current Citus provides a plan-first background rebalance workflow:

SELECT *
FROM get_rebalance_table_shards_plan();

SELECT citus_rebalance_start();

The older rebalance_table_shards() function is deprecated as of Citus 11.2; current designs should use citus_rebalance_start(): Citus utility functions.

Citus Community Edition 11.0 and later supports nonblocking reads and writes during logical-replication-based shard movement, but movement still consumes source disk reads, destination writes, WAL, network bandwidth, CPU, and replication capacity: Citus cluster management. The next article in this series treats rebalancing as its own production operation.

Operate the Cluster as One Failure Domain

Coordinator availability

Applications normally connect to the coordinator. It stores the metadata describing nodes, shards, placements, and colocation groups, then plans work against workers. Citus documents PostgreSQL streaming replication as one coordinator-availability option: coordinator node failures.

A workable EC2 design needs:

  • synchronous or asynchronous replication selected from an explicit RPO and latency requirement;
  • automated health assessment and a controlled promotion mechanism;
  • an application endpoint that follows the promoted coordinator;
  • fencing so the previous primary cannot return as a writer;
  • metadata validation after promotion;
  • a tested failback procedure.

The coordinator should not become a hidden monolith. Keep large application-local tables off it, bound client sessions with PgBouncer, monitor coordinator CPU and memory, and measure the rows returned from workers for cross-tenant queries. Citus notes that multi-shard results still pass through the coordinator and that each session can create connection pools to workers: Citus query execution.

Worker availability

The first thing to establish is that Citus does not replicate hash-distributed shards by default. citus.shard_replication_factor defaults to 1, so each shard placement exists on exactly one worker. Losing a worker does not degrade the cluster; it removes the only copy of every tenant that worker owned. Verify this rather than assume it:

SHOW citus.shard_replication_factor;

-- Any distributed shard with a single placement has no in-Citus redundancy.
SELECT shardid, count(*) AS placements
FROM citus_shards
WHERE citus_table_type = 'distributed'
GROUP BY shardid
HAVING count(*) < 2
LIMIT 10;

Raising the replication factor is not the fix. Citus’s statement-based shard replication is intended for append-only workloads, not for a transactional checkout path, and it multiplies write work rather than providing failover. For heavy OLTP, current Citus documentation recommends PostgreSQL streaming replication of entire workers instead. Each worker primary therefore needs a standby and a promotion procedure: worker node failures.

The consequence for RetailCo’s capacity plan is concrete: an eight-worker cluster is sixteen PostgreSQL servers plus the coordinator pair, and the redundancy lives entirely in PostgreSQL replication that RetailCo operates — not in Citus.

The coordinator tracks workers by hostname and port. A failover design must either preserve the registered address through a stable endpoint or update Citus metadata with citus_update_node() when the address changes: Citus utility functions.

Do not stop the game day after promoting PostgreSQL. Verify:

  1. the coordinator reaches the promoted worker;
  2. tenant-local reads and writes route correctly;
  3. standby replication is rebuilt;
  4. the failed node is fenced;
  5. monitoring and backups follow the new primary;
  6. the application sees the expected error and retry behavior during the transition.

Security and version consistency

Workers accept database traffic from the coordinator and, for some operations, from other workers. Place database nodes in private subnets, restrict security groups to required application and inter-node paths, use TLS with hostname verification for inter-node connections, and manage database credentials outside instance images.

Citus supports sslmode=verify-full in citus.node_conninfo: Citus configuration reference. Certificates, DNS names, failover endpoints, and citus.local_hostname must agree.

Keep PostgreSQL and Citus versions consistent across the cluster. Citus validates extension code and SQL-object versions because loading mismatched versions can cause errors or crashes: Citus version checks. An upgrade is a coordinated cluster operation, not an independent package update on whichever worker is convenient.

Backup and recovery

A distributed backup is not proven because every EC2 instance has a snapshot. Recovery must recreate a mutually consistent coordinator and worker state, roles, extensions, configuration, reference tables, shard placements, and WAL history.

This article does not prescribe an untested backup design. The minimum acceptance test is a restore into an isolated environment followed by metadata checks, tenant-local checksum or aggregate reconciliation, cross-table constraint checks, and application replay. The dedicated Citus backup and recovery article will compare logical, physical, snapshot, and PITR approaches in detail.

In Practice

Context: Citus 13.0.1 documents row-based sharding as a shared-schema model where each tenant-owned table and tenant-local query carries a common distribution column.

Action: Generate a query inventory from pg_stat_statements, application traces, and schema dependencies. Classify each critical statement as single-tenant, cross-tenant, or missing tenant context before changing the schema.

Result: The architecture review can calculate transaction locality and identify queries that would fan out or lose supported constraint patterns.

Learning: A promising shard key is not enough; the application must consistently supply it.

Context: Citus requires primary and unique keys on distributed tables to include the distribution column. Foreign keys between colocated distributed tables must include it as well.

Action: Redesign RetailCo’s keys as tenant-scoped composites before the migration. Record each rule that remains global and give it an explicit coordination mechanism.

Result: The schema expresses the ownership boundary Citus can enforce locally instead of relying on cross-worker uniqueness checks that do not exist.

Learning: The distribution key changes the business meaning of a constraint.

Context: Citus routes single-shard work to one worker and decomposes multi-shard work into tasks whose results return through the coordinator.

Action: Tag lab scripts and production telemetry with tenant_id, shard count, task count, worker connections, and whether each transaction touched one or several workers. Use citus_stat_statements, citus_stat_tenants, PostgreSQL statistics, and infrastructure metrics with their documented collection overhead understood: Citus query statistics.

Result: The team can attribute latency to tenant skew, fan-out, coordinator pressure, worker saturation, storage, or locks instead of reporting only cluster-wide CPU.

Learning: Aggregate TPS can hide the tenant and worker that determine the real ceiling.

Proposed 1, 2, 4, and 8-worker benchmark

There are no empirical results yet. The lab must answer two different scaling questions:

  1. Strong scaling: With the dataset and offered workload fixed, how much faster does the system become as workers are added?
  2. Weak scaling: If data and offered load grow with worker count, can latency and utilization remain within the same target envelope?

Use identical worker instance types, PostgreSQL and Citus versions, EBS layouts, kernel settings, durability settings, autovacuum policy, shard count, schema, and query mix. Run load generators on separate hosts so the client does not become the bottleneck.

WorkloadPurposeExpected routing shape
Tenant-local checkoutTest the intended architectureOne colocated shard group per transaction
Tenant-local order updateMeasure update and lock behaviorOne worker
Cross-tenant operational reportMeasure fan-out and coordinator mergeMany workers
Cross-tenant modificationMeasure distributed commit and failure behaviorMany shards and workers
Hot-tenant mixExpose skew hidden by averagesOne worker receives disproportionate load
Reference-table updateMeasure replicated coordinationAll relevant workers
Rebalance under checkout loadMeasure capacity-addition interferenceSource, target, WAL, and network paths

Use custom pgbench scripts that preserve RetailCo’s transaction boundaries rather than the built-in TPC-B-like schema. PostgreSQL documents -f for custom scripts, -c for clients, -j for client threads, -T for duration, -l for per-transaction logs, and -r for statement-level reporting: PostgreSQL pgbench.

A reproducible run should include:

  • a documented data-generation seed and tenant-size distribution;
  • at least one warm-up period followed by a sustained measurement period;
  • at least three repetitions per configuration;
  • open-loop rate-controlled tests as well as maximum-throughput tests;
  • P50, P95, P99, maximum latency, TPS, failures, retries, and schedule lag;
  • coordinator and per-worker CPU, memory, connections, WAL, disk latency, throughput, network, locks, and autovacuum activity;
  • raw logs and configuration snapshots stored with the result.

Run the matrix at 1, 2, 4, and 8 worker primaries. For strong scaling, rebalance the same dataset across the target worker count. For weak scaling, grow tenants and offered load proportionally while keeping the per-worker data and load target approximately constant.

Do not call the result linear because one data point doubled. Report efficiency explicitly:

speedup = baseline runtime ÷ measured runtime

scaling efficiency = measured speedup ÷ worker-count multiplier

Also separate the steady-state result from the capacity-addition result. A cluster can show excellent tenant-local TPS after rebalancing while taking too long or consuming too much headroom to move shards safely.

Failure matrix

Performance without failure behavior is not a production evaluation.

FailureExerciseRequired evidence
Coordinator process stopsPromote standby and move endpointRTO, rejected transactions, fencing, metadata validation
Worker primary stopsPromote worker standbyAffected tenant errors, routing restoration, replica rebuild
One Availability Zone is isolatedObserve coordinator and worker dependenciesSurviving quorum assumptions, endpoint behavior, blast radius
Network latency rises between coordinator and workerHold offered load constantP95 and P99, connection buildup, timeouts, retry amplification
Rebalance is interruptedStop source or target during movementJob state, source integrity, retry or abort procedure
Restore to a point in timeRecover into isolated networkCoordinator-worker consistency, tenant reconciliation, measured RPO and RTO

Where It Breaks

Design choiceGood fitWhere it breaksMitigation or alternative
Row-based sharding by tenant_idMany tenants and mostly tenant-local transactionsCritical transactions omit or cross tenant boundariesRedesign boundary, application sharding, or decomposition
Four-value brand_id distributionRarely appropriate for scale-outLow cardinality and brand skew cap parallel ownershipUse higher-cardinality tenant ownership
Reference tablesSmall, shared, relatively stable dataLarge or frequently updated tables replicate work everywhereDistribute, decompose, or serve through another domain
Coordinator-local tablesSmall administration dataCheckout or large data recreates a coordinator write ceilingDistribute or remove from critical path
More workersEnough movable shards and broad tenant loadEmpty new node, rebalance cost, hot tenant remains hotPlan and test citus_rebalance_start()
Tenant isolationOne large tenant needs dedicated placementOne tenant exceeds one workerSub-shard tenant or use dedicated architecture
Citus on EC2Need extension control and have strong DB operationsTeam cannot own HA, backup, upgrade, monitoring, and recoveryAurora Limitless or application-sharded managed databases

RetailCo should reject Citus for this workload if any of these statements is true:

  1. The majority of critical writes cannot be scoped to one stable tenant identifier.
  2. Global uniqueness and cross-tenant foreign keys are pervasive and cannot be redesigned.
  3. A single tenant regularly requires more write capacity than one worker can provide.
  4. Cross-tenant analytical queries must run synchronously against raw OLTP rows at high concurrency.
  5. The organization cannot provide on-call ownership for PostgreSQL, Citus, Linux, storage, networking, failover, backup, restore, and upgrades.
  6. The migration cannot backfill and validate tenant_id across every tenant-owned row.
  7. The benchmark shows coordinator pressure, shard skew, or rebalance interference before reaching the required safety margin.

The existing 40-TB table makes migration a larger risk than cluster creation. Citus can distribute a nonempty coordinator-local table, but its documentation states that writes are blocked while rows move: distributing coordinator data. That is not a credible one-command plan for this scenario.

RetailCo needs a staged migration:

flowchart TD
  Inventory[schema and query inventory] --> Backfill[add and backfill tenant identifier]
  Backfill --> Validate[validate keys and tenant ownership]
  Validate --> Build[build empty distributed schema]
  Build --> Copy[bulk copy historical data by tenant ranges]
  Copy --> Capture[capture ongoing source changes]
  Capture --> Reconcile[reconcile counts hashes and business totals]
  Reconcile --> Canary[route canary tenants]
  Canary --> Cutover[move remaining tenants]
  Cutover --> Rollback[retain bounded rollback path]

Every step needs an abort condition. “The copy command completed” is not proof that orders, items, payments, and tenant ownership agree.

What to Do Next

For this RetailCo scenario, Citus row-based sharding on EC2 is the recommended self-managed proof of concept, not yet an approved production migration.

The design should use tenant_id across customers, orders, order items, payments, and every checkout query; small stable lookups should become reference tables; large shared domains should remain outside that shortcut. The coordinator and every worker need PostgreSQL streaming-replication standbys, controlled promotion, stable addressing, fencing, and tested restore procedures.

Production approval requires the following sequence:

  1. Calculate transaction locality from real queries and traces. Establish what percentage of checkout commits are single-tenant.
  2. Produce a table-classification ledger: distributed, reference, coordinator-local, or outside the order domain.
  3. Redesign primary, unique, and foreign keys so tenant-owned constraints include tenant_id.
  4. Build a Citus 13.0.1 lab using the same PostgreSQL version, extensions, schema, durability, TLS, and connection path intended for production.
  5. Run the 1, 2, 4, and 8-worker strong- and weak-scaling matrix without inventing or extrapolating results.
  6. Introduce hot tenants, cross-tenant SQL, worker failure, coordinator failure, network delay, rebalance, and restore while the workload is active.
  7. Select worker and shard counts from measured headroom, movement time, recovery time, and skew—not from a reference diagram.
  8. Migrate by verifiable tenant slices with continuous change capture, reconciliation, canaries, and rollback.

Good fit: thousands of tenants, shared schema, high-cardinality tenant_id, mostly tenant-local writes, and a team prepared to operate distributed PostgreSQL.

Poor fit: global relational constraints, frequent cross-tenant transactions, one tenant larger than one worker, or a strong requirement for provider-managed database operations.

Main advantage: independent workers can own different tenant writes while preserving PostgreSQL joins and transactions inside a colocated tenant boundary.

Main limitation: one distribution column becomes a hard architectural boundary, and Citus does not remove coordinator, hot-tenant, migration, or distributed-operations costs.

Operational complexity: high. RetailCo owns the full EC2, PostgreSQL, Citus, replication, failover, security, monitoring, backup, restore, upgrade, and rebalance lifecycle.

Migration complexity: very high for 40 TB. Treat the move as an online data-platform migration, not an extension installation.

  • Problem: RetailCo’s order writer has reached a sustained ceiling, but distributing tables without transaction locality would replace one bottleneck with cross-worker coordination.
  • Solution: use Citus row-based sharding on tenant_id, explicitly colocate the order graph, replicate small reference data, and operate coordinator and worker HA as one system.
  • Proof: Citus documentation shows that the distribution column controls shard placement, colocation keeps related tenant rows together, single-shard work runs on one worker, and multi-shard work returns through the coordinator. The proposed lab must now verify RetailCo’s workload against those mechanics.
  • Action: complete the transaction-locality and table-classification ledgers before provisioning production-sized EC2 instances. If the data model cannot make checkout tenant-local, stop the Citus migration and choose another ownership boundary.