Building a Multi‑Tenant SaaS with Citus: An Official Distributed PostgreSQL Example

Background

In a SaaS application each tenant’s data is usually stored in a single relational database with a tenant identifier column. This model simplifies management and hardware utilization but a single PostgreSQL instance cannot scale beyond the capacity of one node. Citus extends PostgreSQL to a horizontally scalable cluster while preserving full SQL semantics. Minimal client‑side changes are required; the distribution column determines which worker stores each row.

Example Application – Ad Analytics

The example tracks companies, campaigns, ads, clicks and impressions. The source code is available at:

https://github.com/citusdata/citus-example-ad-analytics

Initial schema (simplified):

CREATE TABLE companies (
  id bigserial PRIMARY KEY,
  name text NOT NULL,
  image_url text,
  created_at timestamp without time zone NOT NULL,
  updated_at timestamp without time zone NOT NULL
);

CREATE TABLE campaigns (
  id bigserial PRIMARY KEY,
  company_id bigint REFERENCES companies (id),
  name text NOT NULL,
  cost_model text NOT NULL,
  state text NOT NULL,
  monthly_budget bigint,
  blacklisted_site_urls text[],
  created_at timestamp without time zone NOT NULL,
  updated_at timestamp without time zone NOT NULL
);

CREATE TABLE ads (
  id bigserial PRIMARY KEY,
  campaign_id bigint REFERENCES campaigns (id),
  name text NOT NULL,
  image_url text,
  target_url text,
  impressions_count bigint DEFAULT 0,
  clicks_count bigint DEFAULT 0,
  created_at timestamp without time zone NOT NULL,
  updated_at timestamp without time zone NOT NULL
);

CREATE TABLE clicks (
  id bigserial PRIMARY KEY,
  ad_id bigint REFERENCES ads (id),
  clicked_at timestamp without time zone NOT NULL,
  site_url text NOT NULL,
  cost_per_click_usd numeric(20,10),
  user_ip inet NOT NULL,
  user_data jsonb NOT NULL
);

CREATE TABLE impressions (
  id bigserial PRIMARY KEY,
  ad_id bigint REFERENCES ads (id),
  seen_at timestamp without time zone NOT NULL,
  site_url text NOT NULL,
  cost_per_impression_usd numeric(20,10),
  user_ip inet NOT NULL,
  user_data jsonb NOT NULL
);

Extending the Relational Model for Citus

Citus requires that primary keys and foreign keys include the distribution column so that constraint checks are local to a single worker. The chosen distribution column is company_id because queries are always scoped to a single tenant.

Modified schema with composite primary keys and foreign keys:

CREATE TABLE companies (
  id bigserial PRIMARY KEY,
  name text NOT NULL,
  image_url text,
  created_at timestamp without time zone NOT NULL,
  updated_at timestamp without time zone NOT NULL
);

CREATE TABLE campaigns (
  id bigserial,
  company_id bigint REFERENCES companies (id),
  name text NOT NULL,
  cost_model text NOT NULL,
  state text NOT NULL,
  monthly_budget bigint,
  blacklisted_site_urls text[],
  created_at timestamp without time zone NOT NULL,
  updated_at timestamp without time zone NOT NULL,
  PRIMARY KEY (company_id, id)
);

CREATE TABLE ads (
  id bigserial,
  company_id bigint,
  campaign_id bigint,
  name text NOT NULL,
  image_url text,
  target_url text,
  impressions_count bigint DEFAULT 0,
  clicks_count bigint DEFAULT 0,
  created_at timestamp without time zone NOT NULL,
  updated_at timestamp without time zone NOT NULL,
  PRIMARY KEY (company_id, id),
  FOREIGN KEY (company_id, campaign_id) REFERENCES campaigns (company_id, id)
);

CREATE TABLE clicks (
  id bigserial,
  company_id bigint,
  ad_id bigint,
  clicked_at timestamp without time zone NOT NULL,
  site_url text NOT NULL,
  cost_per_click_usd numeric(20,10),
  user_ip inet NOT NULL,
  user_data jsonb NOT NULL,
  PRIMARY KEY (company_id, id),
  FOREIGN KEY (company_id, ad_id) REFERENCES ads (company_id, id)
);

CREATE TABLE impressions (
  id bigserial,
  company_id bigint,
  ad_id bigint,
  seen_at timestamp without time zone NOT NULL,
  site_url text NOT NULL,
  cost_per_impression_usd numeric(20,10),
  user_ip inet NOT NULL,
  user_data jsonb NOT NULL,
  PRIMARY KEY (company_id, id),
  FOREIGN KEY (company_id, ad_id) REFERENCES ads (company_id, id)
);

For a full migration guide see the multi‑tenant schema migration documentation.

Creating Distributed Tables

After the schema is created, run the following on the coordinator to shard the tables:

SELECT create_distributed_table('companies',   'id');
SELECT create_distributed_table('campaigns',   'company_id');
SELECT create_distributed_table('ads',         'company_id');
SELECT create_distributed_table('clicks',      'company_id');
SELECT create_distributed_table('impressions', 'company_id');

The create_distributed_table function registers the table for sharding and creates the initial shards on workers.

Ingesting Sample Data

Download CSV files and load them with COPY (or \copy from psql).

# download datasets
for dataset in companies campaigns ads clicks impressions geo_ips; do
  curl -O https://examples.citusdata.com/mt_ref_arch/${dataset}.csv
done

# copy files into a Docker container (if using Docker)
for dataset in companies campaigns ads clicks impressions geo_ips; do
  docker cp ${dataset}.csv citus:.
done

# load data
\copy companies FROM 'companies.csv' WITH csv;
\copy campaigns FROM 'campaigns.csv' WITH csv;
\copy ads FROM 'ads.csv' WITH csv;
\copy clicks FROM 'clicks.csv' WITH csv;
\copy impressions FROM 'impressions.csv' WITH csv;

Application Integration

Queries that filter on company_id are routed to a single worker, preserving joins, transactions and window functions.

ORM examples:

# ActiveRecord (Ruby on Rails)
Impression.where(company_id: 5).count

# Django ORM
Impression.objects.filter(company_id=5).count()

Typical tenant‑scoped query and update:

SELECT name, cost_model, state, monthly_budget
FROM campaigns
WHERE company_id = 5
ORDER BY monthly_budget DESC
LIMIT 10;

UPDATE campaigns
SET monthly_budget = monthly_budget * 2
WHERE company_id = 5;

Transactional example:

BEGIN;
UPDATE campaigns SET monthly_budget = monthly_budget + 1000 WHERE company_id = 5 AND id = 40;
UPDATE campaigns SET monthly_budget = monthly_budget - 1000 WHERE company_id = 5 AND id = 41;
COMMIT;

Window function with join:

SELECT a.campaign_id,
       RANK() OVER (PARTITION BY a.campaign_id ORDER BY COUNT(*) DESC) AS rank,
       COUNT(*) AS n_impressions,
       a.id
FROM ads AS a
JOIN impressions AS i ON i.company_id = a.company_id AND i.ad_id = a.id
WHERE a.company_id = 5
GROUP BY a.campaign_id, a.id
ORDER BY a.campaign_id, n_impressions DESC;

Sharing Data Between Tenants

Static lookup tables (e.g., geo‑IP) can be defined as reference tables so they are copied to every worker, eliminating network hops.

CREATE TABLE geo_ips (
  addrs cidr NOT NULL PRIMARY KEY,
  latlon point NOT NULL CHECK (
    -90 <= latlon[0] AND latlon[0] <= 90 AND
    -180 <= latlon[1] AND latlon[1] <= 180)
);
CREATE INDEX ON geo_ips USING gist (addrs inet_ops);

Mark as reference table:

SELECT create_reference_table('geo_ips');

Load data:

\copy geo_ips FROM 'geo_ips.csv' WITH csv;

Join example:

SELECT c.id, clicked_at, latlon
FROM geo_ips, clicks c
WHERE addrs >> c.user_ip
  AND c.company_id = 5
  AND c.ad_id = 290;

Online Schema Changes

Standard PostgreSQL DDL commands propagate to all workers via a two‑phase commit. Example – adding a caption column to ads:

ALTER TABLE ads ADD COLUMN caption text;

Tenant‑Specific Custom Data

Use a JSONB column ( user_data in clicks) for flexible per‑tenant attributes.

Query example counting mobile vs. desktop visitors for company 5:

SELECT user_data->>'is_mobile' AS is_mobile,
       COUNT(*) AS count
FROM clicks
WHERE company_id = 5
GROUP BY user_data->>'is_mobile'
ORDER BY count DESC;

Partial index to speed up the same query for a single tenant:

CREATE INDEX click_user_data_is_mobile
ON clicks ((user_data->>'is_mobile'))
WHERE company_id = 5;

GIN index for generic JSONB key‑presence queries:

CREATE INDEX click_user_data ON clicks USING gin (user_data);

SELECT id FROM clicks WHERE user_data ? 'is_mobile' AND company_id = 5;

Scaling Hardware Resources

Add a new worker node (self‑hosted) and rebalance shards:

SELECT citus_add_node('new_worker_host', 5432);
SELECT rebalance_table_shards('companies');

Table co‑location ensures tables sharded on the same column stay on the same node, preserving join performance.

Dealing with Large Tenants

Tenant data often follows a Zipfian distribution; the largest tenant typically occupies a modest fraction of total storage (e.g., ~200 GB in a 10 TB dataset). To guarantee QoS, isolate large tenants onto dedicated nodes using Citus Enterprise tenant isolation.

Isolate tenant 5’s data into a new shard (cascade to related tables):

SELECT isolate_tenant_to_new_shard('companies', 5, 'CASCADE');

Move the shard to a chosen worker:

SELECT citus_move_shard_placement(
  102240,
  'source_host', source_port,
  'dest_host', dest_port
);

Next Steps

For migrating existing schemas see the multi‑tenant migration guide:

https://docs.citusdata.com/en/v10.2/develop/migration.html#transitioning-mt

For framework‑specific migrations consult the Ruby on Rails and Django guides:

https://docs.citusdata.com/en/v10.2/develop/migration_mt_ror.html#rails-migration

https://docs.citusdata.com/en/v10.2/develop/migration_mt_django.html#django-migration

Managed Citus clusters are available via Azure Database for PostgreSQL‑Hyperscale:

https://docs.microsoft.com/azure/postgresql/quickstart-create-hyperscale-portal