Citus 分散式 PostgreSQL：水平擴展到多節點叢集 | PostgreSQL

-- 建立普通表
CREATE TABLE orders (
    order_id    BIGINT NOT NULL,
    customer_id BIGINT NOT NULL,
    created_at  TIMESTAMPTZ NOT NULL,
    amount      NUMERIC(12,2),
    status      TEXT
);

-- 轉換為分散表，以 customer_id 為分片鍵
SELECT create_distributed_table('orders', 'customer_id');

-- Citus 自動建立 32 個 Shard（預設值），分布在所有 Worker 上
-- orders_102008, orders_102009, ... 每個都是 Worker 上的真實 PG 表

-- 建立參考表
CREATE TABLE countries (
    country_code CHAR(2) PRIMARY KEY,
    country_name TEXT NOT NULL,
    region       TEXT
);

SELECT create_reference_table('countries');

-- 分散表與參考表的 JOIN 在各 Worker 本地完成（零跨節點通訊）
SELECT o.order_id, o.amount, c.country_name
FROM orders o
JOIN countries c ON o.country_code = c.country_code
WHERE o.customer_id = 12345;
-- 路由到單一 Worker，JOIN 在本地完成

SELECT create_distributed_table('orders', 'customer_id');

SELECT create_distributed_table('events', 'created_at', 'append');

-- Router Query：定位到單一 Shard，效能等同單機 PG
SELECT * FROM orders WHERE customer_id = 12345;

INSERT INTO orders (order_id, customer_id, amount)
VALUES (9001, 12345, 99.99);

UPDATE orders SET status = 'shipped' WHERE customer_id = 12345;

-- Real-Time Query：平行掃描所有 Shard
SELECT status, COUNT(*), SUM(amount)
FROM orders
GROUP BY status;

-- 沒有分片鍵過濾，需掃描所有 Shard
SELECT * FROM orders WHERE amount > 1000;

-- 問題：orders 和 shipments 分片鍵不同，需大量跨節點資料移動
SELECT o.order_id, s.tracking_number
FROM orders o          -- 以 customer_id 分片
JOIN shipments s ON o.order_id = s.order_id;  -- 以 order_id 分片

-- 解決方案：加上分片鍵條件 + Subquery Pushdown
SELECT o.order_id, s.tracking_number
FROM orders o
JOIN (SELECT order_id, tracking_number FROM shipments) s
    ON o.order_id = s.order_id
WHERE o.customer_id = 12345;  -- 加上分片鍵

-- 以 tenant_id 為分片鍵，所有表共置
SELECT create_distributed_table('orders',    'tenant_id');
SELECT create_distributed_table('customers', 'tenant_id', colocate_with => 'orders');
SELECT create_distributed_table('products',  'tenant_id', colocate_with => 'orders');
SELECT create_distributed_table('invoices',  'tenant_id', colocate_with => 'orders');

-- 以下 JOIN 完全在本地執行，零跨節點通訊
SELECT c.name, o.order_id, o.amount, i.invoice_number
FROM customers c
JOIN orders o ON c.tenant_id = o.tenant_id AND c.customer_id = o.customer_id
JOIN invoices i ON i.tenant_id = o.tenant_id AND i.order_id = o.order_id
WHERE c.tenant_id = 'tenant_abc'
  AND c.customer_id = 12345;
-- 路由到單一 Worker，零跨節點通訊

-- 確認各表的 Colocation Group
SELECT logicalrelid AS table_name, colocationid
FROM pg_dist_partition
ORDER BY colocationid, logicalrelid;

-- 啟用 Schema-based Sharding
SET citus.enable_schema_based_sharding TO on;

-- 為每個租戶建立獨立 Schema
CREATE SCHEMA tenant_abc;
CREATE SCHEMA tenant_xyz;

-- 在 Schema 內建表（不需要呼叫 create_distributed_table！）
CREATE TABLE tenant_abc.orders (
    order_id    BIGINT PRIMARY KEY,
    amount      NUMERIC(12,2),
    created_at  TIMESTAMPTZ DEFAULT NOW()
);

CREATE TABLE tenant_abc.customers (
    customer_id BIGINT PRIMARY KEY,
    name        TEXT NOT NULL,
    email       TEXT
);

-- Citus 自動將整個 tenant_abc schema 分配到某個 Worker
-- tenant_abc 內的所有表自動 Collocated！

# Ubuntu/Debian 安裝 Citus（以 PG 16 為例）
curl https://install.citusdata.com/community/deb.sh | sudo bash
sudo apt-get install -y postgresql-16-citus-12.1

# 設定 shared_preload_libraries（必須，需重啟）
echo "shared_preload_libraries = 'citus'" >> /etc/postgresql/16/main/postgresql.conf
sudo systemctl restart postgresql

-- 在每個節點（Coordinator 和所有 Worker）執行
CREATE EXTENSION citus;

-- 在 Coordinator 上新增 Worker 節點
SELECT citus_add_node('worker1-hostname', 5432);
SELECT citus_add_node('worker2-hostname', 5432);
SELECT citus_add_node('worker3-hostname', 5432);

-- 驗證節點狀態
SELECT * FROM citus_get_active_worker_nodes();

-- 查看完整節點清單
SELECT nodeid, nodename, nodeport, isactive, noderole
FROM pg_dist_node
ORDER BY nodeid;

-- 自動應用到所有 Shard
ALTER TABLE orders ADD COLUMN notes TEXT;
CREATE INDEX CONCURRENTLY idx_orders_status ON orders(status);

-- 外鍵約束必須包含分片鍵
ALTER TABLE orders ADD CONSTRAINT fk_customer
    FOREIGN KEY (tenant_id, customer_id)
    REFERENCES customers(tenant_id, customer_id);
-- 注意：orders 和 customers 必須 Colocated

-- 不允許的操作
UPDATE orders SET customer_id = 99999 WHERE order_id = 1;
-- ERROR: modifying the partition column of a row is not allowed

-- 查看分散表設定
SELECT logicalrelid AS table_name, partmethod, partkey, colocationid
FROM pg_dist_partition
ORDER BY logicalrelid;

-- 查看 Shard 分布
SELECT s.logicalrelid, s.shardid, sp.nodename, sp.nodeport
FROM pg_dist_shard s
JOIN pg_dist_shard_placement sp ON s.shardid = sp.shardid
WHERE s.logicalrelid = 'orders'::regclass
ORDER BY s.shardid;

-- 調整 Shard 數量（建表前設定）
SET citus.shard_count = 64;  -- 預設 32
SELECT create_distributed_table('large_table', 'user_id');

-- 查詢特定記錄在哪個 Worker
SELECT get_shard_id_for_distribution_column('orders', 12345);

-- 自動重新平衡
SELECT rebalance_table_shards('orders');

-- 手動搬移特定 Shard
SELECT citus_move_shard_placement(
    shardid,
    'source_worker', 5432,
    'target_worker', 5432
)
FROM pg_dist_shard_placement
WHERE nodename = 'source_worker'
LIMIT 5;

-- 查看重新平衡進度
SELECT * FROM get_rebalance_progress();

-- === Step 1: 建立分散表並共置 ===
SELECT create_distributed_table('tenants',   'tenant_id');
SELECT create_distributed_table('users',     'tenant_id', colocate_with => 'tenants');
SELECT create_distributed_table('orders',    'tenant_id', colocate_with => 'tenants');
SELECT create_distributed_table('products',  'tenant_id', colocate_with => 'tenants');

-- === Step 2: 建立參考表（跨租戶共用資料） ===
SELECT create_reference_table('payment_methods');
SELECT create_reference_table('shipping_zones');

-- === Step 3: 設定 Row Level Security ===
ALTER TABLE orders ENABLE ROW LEVEL SECURITY;
CREATE POLICY tenant_isolation ON orders
    USING (tenant_id = current_setting('app.current_tenant')::bigint);

-- === Step 4: 應用程式設定租戶 context ===
SET LOCAL app.current_tenant = '12345';
SELECT * FROM orders WHERE status = 'pending';
-- 自動過濾：只返回 tenant_id = 12345 的訂單

-- === Step 5: 管理員跨租戶查詢 ===
RESET app.current_tenant;
SELECT tenant_id, COUNT(*) AS order_count, SUM(amount) AS total
FROM orders
GROUP BY tenant_id
ORDER BY total DESC;

-- 建立時序分散表（先分散後分割）
CREATE TABLE metrics (
    device_id   BIGINT NOT NULL,
    recorded_at TIMESTAMPTZ NOT NULL,
    metric_name TEXT NOT NULL,
    value       DOUBLE PRECISION,
    tags        JSONB
) PARTITION BY RANGE (recorded_at);

-- 將父表設為分散表
SELECT create_distributed_table('metrics', 'device_id');

-- 建立時間分割子表
CREATE TABLE metrics_2026_05 PARTITION OF metrics
    FOR VALUES FROM ('2026-05-01') TO ('2026-06-01');

CREATE TABLE metrics_2026_06 PARTITION OF metrics
    FOR VALUES FROM ('2026-06-01') TO ('2026-07-01');

-- 查詢效率極高：Partition Pruning + Shard Pruning 雙重裁剪
SELECT AVG(value)
FROM metrics
WHERE device_id = 1001                                    -- Shard Pruning
  AND recorded_at BETWEEN '2026-05-01' AND '2026-05-31';  -- Partition Pruning
-- 只訪問 metrics_2026_05 分割，且只在 device_id=1001 所在的 Worker 上執行

-- Citus 支援分散式交易（透過兩階段提交 2PC）
BEGIN;
INSERT INTO orders (tenant_id, order_id, amount)
VALUES (100, 9001, 150.00);
INSERT INTO order_items (tenant_id, order_id, product_id, qty)
VALUES (100, 9001, 5, 2);
COMMIT;
-- 若 Colocated，此交易在單一 Worker 本地執行（效能最佳）

-- 涉及多個 Worker 時自動使用 2PC
BEGIN;
UPDATE orders SET status = 'cancelled'
WHERE order_id IN (1, 2, 3, 4, 5);
-- 若這些 order_id 分布在不同 Worker，Citus 自動 2PC
COMMIT;

-- 1. 啟用 Repartition Joins
SET citus.enable_repartition_joins = on;

-- 2. 調整平行查詢
SET citus.max_adaptive_executor_pool_size = 16;

-- 3. 分析查詢計畫，識別跨節點資料移動
EXPLAIN (ANALYZE, VERBOSE)
SELECT customer_id, SUM(amount)
FROM orders
GROUP BY customer_id
HAVING SUM(amount) > 10000;
-- 觀察是否有 Repartition 或 Broadcast 步驟

-- 4. Worker 內部平行度（各 Worker 的 postgresql.conf）
-- max_parallel_workers_per_gather = 4

-- 5. 複合索引策略（分片鍵必須在索引中）
CREATE INDEX idx_orders_customer_created
ON orders (customer_id, created_at DESC);

-- 6. Coordinator 前端建議使用 PgBouncer 做連線池

-- 查看分散式查詢的執行計畫
EXPLAIN SELECT COUNT(*) FROM orders WHERE customer_id = 12345;

-- 查看 Coordinator 活躍連線
SELECT * FROM citus_dist_stat_activity;

-- 查看 Worker 任務執行
SELECT * FROM citus_worker_stat_activity;

-- 查看各 Shard 資料大小
SELECT table_name, shard_id, shard_size
FROM citus_shards
WHERE table_name = 'orders'::regclass
ORDER BY shard_size DESC;

-- 慢查詢分析
SELECT * FROM citus_stat_statements
ORDER BY total_time DESC LIMIT 20;

-- 節點健康狀態
SELECT nodename, nodeport, isactive, shouldhaveshards
FROM pg_dist_node;

-- 查看各節點健康
SELECT nodename, nodeport, isactive FROM pg_dist_node;

-- Worker 故障後標記為不活躍
SELECT citus_disable_node('failed_worker', 5432);

-- 恢復後重新啟用
SELECT citus_activate_node('recovered_worker', 5432);