Migrating Existing Applications to a Distributed PostgreSQL (Citus) Cluster

Citus extends PostgreSQL with distributed sharding but requires schema changes and query updates for optimal performance.

Determine Distribution Strategy

Select a distribution key that identifies the tenant (e.g., store_id in a multi‑tenant e‑commerce schema). Classify tables:

Distributed tables already contain the distribution key.

Tables needing backfill have the key column added but contain NULL values.

Reference tables are small, lack the key, and are replicated on every node.

Local tables reside only on the coordinator and never join other tables.

Add Distribution Key to Source Tables

Alter tables that lack the key and back‑fill missing values. Example for the line_items table:

-- denormalize line_items by including store_id
ALTER TABLE line_items ADD COLUMN store_id uuid;

Back‑fill using a join:

UPDATE line_items
   SET store_id = orders.store_id
  FROM line_items
  INNER JOIN orders ON line_items.order_id = orders.order_id;

For large tables, use a batch function invoked by pg_cron to avoid overload:

CREATE FUNCTION backfill_batch()
RETURNS void LANGUAGE sql AS $$
  WITH batch AS (
    SELECT line_item_id, order_id
      FROM line_items
     WHERE store_id IS NULL
     LIMIT 10000
     FOR UPDATE SKIP LOCKED
  )
  UPDATE line_items AS li
     SET store_id = orders.store_id
    FROM batch, orders
   WHERE batch.line_item_id = li.line_item_id
     AND batch.order_id = orders.order_id;
$$;
SELECT cron.schedule('*/15 * * * *', 'SELECT backfill_batch()');

Prepare Development Citus Cluster

Dump the original schema and load it into a single‑node Citus test database:

# get schema from source db
pg_dump \
   --format=plain \
   --no-owner \
   --schema-only \
   --file=schema.sql \
   --schema=target_schema \
   postgres://user:pass@host:5432/db
# load schema into test db
psql postgres://user:pass@testhost:5432/db -f schema.sql

Modify primary and foreign keys to include the distribution column ( store_id) because Citus requires the distribution key in uniqueness constraints:

BEGIN;
ALTER TABLE products   DROP CONSTRAINT products_pkey CASCADE;
ALTER TABLE orders     DROP CONSTRAINT orders_pkey CASCADE;
ALTER TABLE line_items DROP CONSTRAINT line_items_pkey CASCADE;
ALTER TABLE products   ADD PRIMARY KEY (store_id, product_id);
ALTER TABLE orders     ADD PRIMARY KEY (store_id, order_id);
ALTER TABLE line_items ADD PRIMARY KEY (store_id, line_item_id);
ALTER TABLE line_items ADD CONSTRAINT line_items_store_fkey FOREIGN KEY (store_id) REFERENCES stores (store_id);
ALTER TABLE line_items ADD CONSTRAINT line_items_product_fkey FOREIGN KEY (store_id, product_id) REFERENCES products (store_id, product_id);
ALTER TABLE line_items ADD CONSTRAINT line_items_order_fkey FOREIGN KEY (store_id, order_id) REFERENCES orders (store_id, order_id);
COMMIT;

Update Application Queries

All write and read queries must filter by the tenant ID. Example before/after:

-- before
SELECT * FROM orders WHERE order_id = 123;
-- after
SELECT * FROM orders WHERE order_id = 123 AND store_id = 42; -- added tenant filter

Joins also need the tenant filter to stay on a single shard, e.g.:

SELECT sum(l.quantity)
  FROM line_items l
 INNER JOIN products p ON l.product_id = p.product_id AND l.store_id = p.store_id
 WHERE p.name = 'Awesome Wool Pants'
   AND l.store_id = '8c69aa0d-3f13-4440-86ca-443566c1fc75';

Enable Secure Connections

Clients must connect via SSL; Citus Cloud rejects unencrypted connections. See the Citus SSL documentation for configuration.

Detect Cross‑Shard Queries

Set citus.multi_task_query_log_level to 'error' during testing so queries that hit multiple shards raise an error. In production set it to 'log' to record occurrences:

ALTER DATABASE citus SET citus.multi_task_query_log_level = 'error';
-- production
ALTER DATABASE citus SET citus.multi_task_query_log_level = 'log';

Migrate Production Data

Choose a migration path based on downtime tolerance and data size.

Small Database Migration (pg_dump/pg_restore)

Dump the schema from the source database.

Load the schema into the Citus cluster and run create_distributed_table / create_reference_table statements.

Put the application in maintenance mode.

Dump data with pg_dump --format=custom --data-only.

Restore the dump into Citus with pg_restore.

Test the application and switch to the new database.

Large Database Migration (Citus Warp)

Use Citus Warp for online logical replication.

Copy the schema to the target Citus cluster.

Enable logical replication on the source (set wal_level = logical, max_replication_slots, max_wal_senders).

Allow network access from the Citus coordinator to the source (IP whitelist or VPC peering).

Open a support ticket; engineers create a replication slot, perform an initial dump, and start streaming changes.

Monitor WAL growth (recommend at least 100 GB free or 20 % of total disk).

Minimize schema changes during replication; if needed, pause Warp, apply changes on both sides, then resume.

When replication catches up, adjust sequence values manually, switch the application to the new database, and drop the replication slot.

Additional Technical References

pg_cron repository: https://github.com/citusdata/pg_cron

activerecord-multi-tenant repository: https://github.com/citusdata/activerecord-multi-tenant

Citus documentation for app type, data modeling, and security: https://docs.citusdata.com/en/v10.2/

Configuration parameter reference for citus.multi_task_query_log_level: https://docs.citusdata.com/en/v10.2/develop/api_guc.html#multi-task-logging