citus

pig build pkg citus         # 构建 RPM / DEB 包

pig repo add pgsql -u          # 添加仓库并更新缓存

shared_preload_libraries = 'citus';

CREATE EXTENSION citus;

CREATE EXTENSION citus;

-- 将协调节点添加到集群
SELECT citus_set_coordinator_host('coord-host', 5432);
SELECT * FROM citus_add_node('worker-1', 5432);
SELECT * FROM citus_add_node('worker-2', 5432);

-- 验证集群状态
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)
);

-- 按 tenant_id 进行哈希分布（默认：32 个分片）
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');

-- 此连接被下推到各节点执行（无跨分片流量）
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');

-- 单分片查询（快速，只访问一个节点）
SELECT * FROM events WHERE tenant_id = 42;

-- 跨所有分片的并行聚合
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;

-- 协同定位连接：高效，无数据移动
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;

-- 添加新节点
SELECT * FROM citus_add_node('worker-3', 5432);

-- 移除节点（先将分片迁移到其他节点）
SELECT * FROM citus_drain_node('worker-1', 5432);
SELECT * FROM citus_remove_node('worker-1', 5432);

-- 临时禁用节点
SELECT * FROM citus_disable_node('worker-2', 5432);
SELECT * FROM citus_activate_node('worker-2', 5432);

-- 查看当前集群状态
SELECT * FROM citus_get_active_worker_nodes();

-- 再平衡所有分布式表
SELECT citus_rebalance_start();

-- 监控再平衡进度
SELECT * FROM citus_rebalance_status();

-- 再平衡特定表
SELECT rebalance_table_shards('events');

-- 查看分片分布
SELECT * FROM citus_shards;

-- 查看分片大小
SELECT table_name, shard_count, citus_total_relation_size(table_name::text)
FROM citus_tables;

-- 将特定分片迁移到另一个节点
SELECT citus_move_shard_placement(shard_id, 'source-host', 5432, 'dest-host', 5432);

-- 多分片查询时每个节点的并行连接数
SET citus.max_adaptive_executor_pool_size = 4;

-- 分片复制因子（默认 1；无流复制时可设为 2 以实现高可用）
SET citus.shard_replication_factor = 1;

-- 控制执行器行为
SET citus.multi_shard_modify_mode = 'parallel';   -- 或 'sequential'
SET citus.enable_repartition_joins = on;           -- 启用重分区连接

-- 任务分配策略
SET citus.task_assignment_policy = 'round-robin';  -- 或 'greedy'、'first-replica'

-- 记录分布式查询日志
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');

-- 共享查找表：复制到每个节点
CREATE TABLE plans (
    name TEXT PRIMARY KEY,
    max_users INT,
    max_projects INT
);
SELECT create_reference_table('plans');

-- 所有限定在单个租户范围内的连接都是协同定位的，查询速度快
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');

-- 实时汇总：跨分片并行执行
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;

-- 所有站点的热门页面（跨分片并行查询）
SELECT url, count(*) AS views
FROM page_views
WHERE view_time > now() - INTERVAL '24 hours'
GROUP BY url
ORDER BY views DESC
LIMIT 20;