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;