citus

pig build pkg citus;		# build rpm/deb

pig repo add pgsql -u   # add both repo and update cache

pig install citus;		# install via package name, for the active PG version

pig install citus -v 18;   # install for PG 18
pig install citus -v 17;   # install for PG 17
pig install citus -v 16;   # install for PG 16

shared_preload_libraries = 'citus';

CREATE EXTENSION citus;

CREATE EXTENSION citus;

-- Add worker nodes to the cluster
SELECT citus_set_coordinator_host('coord-host', 5432);
SELECT * FROM citus_add_node('worker-1', 5432);
SELECT * FROM citus_add_node('worker-2', 5432);

-- Verify the cluster
SELECT * FROM citus_get_active_worker_nodes();

CREATE TABLE events (
    tenant_id   INT,
    event_id    BIGSERIAL,
    event_time  TIMESTAMPTZ DEFAULT now(),
    event_type  TEXT,
    payload     JSONB,
    PRIMARY KEY (tenant_id, event_id)
);

-- Hash-distribute by tenant_id (default: 32 shards)
SELECT create_distributed_table('events', 'tenant_id');

SELECT create_distributed_table('events', 'tenant_id', shard_count := 64);

CREATE TABLE countries (
    code CHAR(2) PRIMARY KEY,
    name TEXT NOT NULL
);

SELECT create_reference_table('countries');

CREATE TABLE tenants (
    id   INT PRIMARY KEY,
    name TEXT
);
SELECT create_distributed_table('tenants', 'id');

CREATE TABLE orders (
    tenant_id  INT REFERENCES tenants(id),
    order_id   BIGSERIAL,
    amount     NUMERIC,
    PRIMARY KEY (tenant_id, order_id)
);
SELECT create_distributed_table('orders', 'tenant_id');

-- This join is pushed down to each node (no cross-shard traffic)
SELECT t.name, sum(o.amount)
FROM tenants t JOIN orders o ON t.id = o.tenant_id
GROUP BY t.name;

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

-- Single-shard query (fast, touches one node)
SELECT * FROM events WHERE tenant_id = 42;

-- Parallel aggregation across all shards
SELECT event_type, count(*), avg(payload->>'duration')::numeric
FROM events
WHERE event_time > now() - INTERVAL '1 hour'
GROUP BY event_type
ORDER BY count DESC
LIMIT 10;

-- Colocated join: efficient, no data movement
SELECT e.*, o.amount
FROM events e JOIN orders o
    ON e.tenant_id = o.tenant_id
WHERE e.tenant_id = 42;

SELECT e.*, c.name AS country_name
FROM events e JOIN countries c ON e.payload->>'country' = c.code;

-- Add a new node
SELECT * FROM citus_add_node('worker-3', 5432);

-- Remove a node (moves shards to other nodes first)
SELECT * FROM citus_drain_node('worker-1', 5432);
SELECT * FROM citus_remove_node('worker-1', 5432);

-- Disable a node temporarily
SELECT * FROM citus_disable_node('worker-2', 5432);
SELECT * FROM citus_activate_node('worker-2', 5432);

-- View current cluster state
SELECT * FROM citus_get_active_worker_nodes();

-- Rebalance all distributed tables
SELECT citus_rebalance_start();

-- Monitor rebalance progress
SELECT * FROM citus_rebalance_status();

-- Rebalance a specific table
SELECT rebalance_table_shards('events');

-- View shard placements
SELECT * FROM citus_shards;

-- View shard sizes
SELECT table_name, shard_count, citus_total_relation_size(table_name::text)
FROM citus_tables;

-- Move a specific shard to another node
SELECT citus_move_shard_placement(shard_id, 'source-host', 5432, 'dest-host', 5432);

-- Number of parallel connections per node for multi-shard queries
SET citus.max_adaptive_executor_pool_size = 4;

-- Shard replication factor (default 1; set to 2 for HA without streaming replication)
SET citus.shard_replication_factor = 1;

-- Control executor behavior
SET citus.multi_shard_modify_mode = 'parallel';   -- or 'sequential'
SET citus.enable_repartition_joins = on;           -- enable repartitioned joins

-- Task assignment policy
SET citus.task_assignment_policy = 'round-robin';  -- or 'greedy', 'first-replica'

-- Log distributed queries
SET citus.log_multi_join_order = on;
SET citus.explain_all_tasks = on;

CREATE TABLE tenants (
    tenant_id   INT PRIMARY KEY,
    name        TEXT,
    plan        TEXT DEFAULT 'free',
    created_at  TIMESTAMPTZ DEFAULT now()
);
SELECT create_distributed_table('tenants', 'tenant_id');

CREATE TABLE users (
    tenant_id   INT,
    user_id     BIGSERIAL,
    email       TEXT,
    PRIMARY KEY (tenant_id, user_id)
);
SELECT create_distributed_table('users', 'tenant_id');

CREATE TABLE projects (
    tenant_id   INT,
    project_id  BIGSERIAL,
    name        TEXT,
    owner_id    BIGINT,
    PRIMARY KEY (tenant_id, project_id)
);
SELECT create_distributed_table('projects', 'tenant_id');

-- Shared lookup: replicated to every node
CREATE TABLE plans (
    name TEXT PRIMARY KEY,
    max_users INT,
    max_projects INT
);
SELECT create_reference_table('plans');

-- All joins scoped to a tenant are colocated and fast
SELECT u.email, p.name AS project
FROM users u
JOIN projects p ON u.tenant_id = p.tenant_id AND u.user_id = p.owner_id
WHERE u.tenant_id = 7;

CREATE TABLE page_views (
    site_id      INT,
    url          TEXT,
    view_time    TIMESTAMPTZ DEFAULT now(),
    user_agent   TEXT,
    country      CHAR(2)
);
SELECT create_distributed_table('page_views', 'site_id');

-- Real-time rollup: parallelized across shards
SELECT
    date_trunc('minute', view_time) AS minute,
    count(*)                        AS views,
    count(DISTINCT country)         AS countries
FROM page_views
WHERE site_id = 1
  AND view_time > now() - INTERVAL '1 hour'
GROUP BY minute
ORDER BY minute DESC;

-- Top pages across all sites (cross-shard parallel query)
SELECT url, count(*) AS views
FROM page_views
WHERE view_time > now() - INTERVAL '24 hours'
GROUP BY url
ORDER BY views DESC
LIMIT 20;