Citus

# macOS 上为 PostgresSQL 加 citus 扩展
# brew install citus
# echo "shared_preload_libraries = 'citus'" >> postgresql.conf

# Docker 单节点
docker run --rm -it \
  -p 5432:5432 \
  -e POSTGRES_PASSWORD=mypass \
  -v $PWD/data:/var/lib/postgresql/data \
  --name citus citusdata/citus:11-alpine

SELECT * FROM master_get_active_worker_nodes();

SELECT version();
SELECT citus_is_coordinator();

-- 分片表
SELECT create_distributed_table('orgs', 'id');
-- 在所有节点上都存在
SELECT create_reference_table('geo_ips');
-- 取消 colocated - 默认基于 分布列 的值进行 colocated
SELECT create_distributed_table('A', 'foo', colocate_with => 'none');
-- 取消
SELECT undistribute_table('table_name');

create table orgs
(
    id   bigserial primary key,
    name text not null
);

create table users
(
    id     bigserial,
    org_id bigint references orgs (id),
    name   text not null,
    primary key (org_id, id)
);

create table notes
(
    id      bigserial,
    org_id  bigint,
    user_id bigint,
    title   text not null,
    content text not null,
    primary key (org_id, id),
    foreign key (org_id, user_id)
        references users (org_id, id)
);
SELECT create_distributed_table('orgs', 'id');
SELECT create_distributed_table('users', 'org_id');
SELECT create_distributed_table('notes', 'org_id');

-- 准备数据
insert into orgs(name)
select 'org-' || num
from generate_series(1, 100) num;

insert into users(org_id, name)
select coalesce(nullif(num % 100, 0), 10), 'user-' || num
from generate_series(1, 10000) num;

insert into notes(org_id, user_id, title, content)
select coalesce(nullif(num % 100, 0), 10), coalesce(nullif(num % 10000, 0), 100), 'note-' || num, 'content-' || num
from generate_series(1, 1000000) num;

-- 重新平衡分片
SELECT rebalance_table_shards('notes');
-- 隔离租户分片 - 输出新的分片 ID
SELECT isolate_tenant_to_new_shard('orgs', 5, 'CASCADE');

-- 查看分片所处位置
SELECT shardid, nodename, nodeport
FROM pg_dist_placement AS placement,
     pg_dist_node AS node
WHERE placement.groupid = node.groupid
  AND node.noderole = 'primary'
--   AND shardid = 102105
ORDER BY shardid
;

-- 移动分片位置
-- SELECT citus_move_shard_placement(
--                102240,
--                'source_host', source_port,
--                'dest_host', dest_port);

-- 分片数量
show citus.shard_count;

-- relation=table/index
-- main fork - https://www.postgresql.org/docs/current/static/storage-file-layout.html
select citus_relation_size('orgs');
-- citus_relation_size + free space map + visibility map
select citus_table_size('orgs');
-- citus_table_size + indices
select citus_total_relation_size('orgs');

-- coordinator
-- 设置 worker 应该怎么连接 coordinator
SELECT citus_set_coordinator_host('coord.example.com', 5432);

-- 单节点只需要设置为 localhost，不再需要 add_node
SELECT citus_set_coordinator_host('localhost', 5432);

-- 添加节点
SELECT * from citus_add_node('worker-101', 5432);
SELECT * from citus_add_node('worker-102', 5432);
-- 删除节点
SELECT * from citus_remove_node('worker-102', 5432);
-- 添加冗余节点
select * from citus_add_secondary_node('new-node', 12345, 'primary-node', 12345);
-- 添加不启用
select * from citus_add_inactive_node('new-node', 12345);
-- 激活
select * from citus_activate_node('new-node', 12345);

-- 清空节点 - 移除节点准备
SELECT * from citus_drain_node('10.0.0.1', 5432);
-- 设置节点无 分片
SELECT * FROM citus_set_node_property('localhost', 5433, 'shouldhaveshards', false);
-- 重新平衡
SELECT * FROM rebalance_table_shards(drain_only := true);

SELECT * FROM citus_get_active_worker_nodes();
-- 健康检查
SELECT * FROM citus_check_cluster_node_health();

-- 创建恢复点
select citus_create_restore_point('名字');

; 基础配置
; ==========
citus.max_worker_nodes_tracked=2048
citus.cluster_name=
; always - 总是读 secondary_node
citus.use_secondary_nodes=never

citus.log_distributed_deadlock_detection=false
; -1 禁止
citus.distributed_deadlock_detection_factor=2

citus.node_connection_timeout=3000

; citus.node_conninfo
; 自己连自己时使用
; citus.local_hostname

citus.show_shards_for_app_name_prefixes=

; 查询统计
; ==========
; 单位 秒
citus.stat_statements_purge_interval=10
citus.stat_statements_max=50000
; none
citus.stat_statements_track=all

; Data Loading
; ============
; 1pc
citus.multi_shard_commit_protocol = 2pc
citus.shard_count = 32
citus.shard_max_size = 1GB

; 在节点激活时同步 reference 表
; 设置为 off 可增加 add_node 速度 - 在创建 shard 时同步
citus.replicate_reference_tables_on_activate = on

; Planner
; ============
citus.local_table_join_policy=auto
; 限制拉取行数
citus.limit_clause_row_fetch_count=-1
citus.count_distinct_error_rate=
; greedy, round-robin, first-replica
citus.task_assignment_policy=

; Intermediate Data Transfer
; ==========================
; pg >= 14 = true
citus.binary_worker_copy_format =
; 默认 1GB, 单位 KB, -1 不限制
citus.max_intermediate_result_size =

; DDL
; ==========================
citus.enable_ddl_propagation=true
; coordinator 必须通过 citus_add_node 注册自己
citus.enable_local_reference_table_foreign_keys=true

; 执行配置
; ==========================
citus.all_modifications_commutative=
; off,debug,log,notice,warning,error
citus.multi_task_query_log_level=
; local - 广播 SET LOCAL
citus.propagate_set_commands=none
; join 非分布列
citus.enable_repartition_joins=false
; INSERT INTO … SELECT
citus.enable_repartitioned_insert_select=
citus.enable_binary_protocol = true
; = max_connections
; -1 禁用
citus.max_shared_pool_size =
citus.max_adaptive_executor_pool_size = 16
; 单位 ms
citus.executor_slow_start_interval = 10
citus.max_cached_conns_per_worker = 1
citus.force_max_query_parallelization =
; Explain
; ==========================
citus.explain_all_tasks=false
; taskId
citus.explain_analyze_sort_method = execution-time

set local citus.propagate_set_commands = 'local';
set local app.tenant.id = 123;

BEGIN;
CREATE EXTENSION citus;
-- add Docker flag to node metadata
UPDATE pg_dist_node_metadata SET metadata=jsonb_insert(metadata, '{docker}', 'true');
COMMIT;

#!/bin/bash
set -e

until test -f /healthcheck/manager-ready; do
  echo >&2 "Manager is not ready - sleeping"
  sleep 1
done

echo >&2 "Manager is up - starting worker"

# exec gosu postgres "/usr/local/bin/docker-entrypoint.sh" "postgres"

# AlpineLinux
su-exec postgres "/usr/local/bin/docker-entrypoint.sh" "postgres"

create role test with login password 'test';
insert into pg_dist_authinfo(nodeid, rolename, authinfo)
values (0, 'test', 'password=test');
set role test;
select * from test;

SELECT run_command_on_workers($cmd$
 SELECT current_database() db;
$cmd$);

-- 在所有 worker 节点执行
SELECT run_command_on_workers($cmd$
 create database test;
$cmd$);

-- 默认 5 分钟
ALTER DATABASE citus
  SET statement_timeout TO 300000;
SELECT run_command_on_workers($cmd$
  ALTER DATABASE citus
    SET statement_timeout TO 300000;
$cmd$);


-- 单独事务修改
BEGIN;
SET LOCAL statement_timeout TO 300000;
-- ...
COMMIT;

-- 租户所处分片
SELECT shardid, shardstate, shardlength, nodename, nodeport, placementid
  FROM pg_dist_placement AS placement,
       pg_dist_node AS node
 WHERE placement.groupid = node.groupid
   AND node.noderole = 'primary'
   AND shardid = (
     SELECT get_shard_id_for_distribution_column('stores', 4)
   );

-- 表的分布列
SELECT column_to_column_name(logicalrelid, partkey) AS dist_col_name
FROM pg_dist_partition
WHERE logicalrelid='products'::regclass;

-- 检测 锁
SELECT * FROM citus_lock_waits;

-- 分片大小
SELECT shardid, table_name, shard_size
FROM citus_shards
WHERE table_name = 'my_table';

-- 所有分布式表的大小
SELECT table_name, table_size
FROM citus_tables;


-- 判断未使用的索引
SELECT *
FROM run_command_on_shards('my_distributed_table', $cmd$
  SELECT array_agg(a) as infos
  FROM (
    SELECT (
      schemaname || '.' || relname || '##' || indexrelname || '##'
                 || pg_size_pretty(pg_relation_size(i.indexrelid))::text
                 || '##' || idx_scan::text
    ) AS a
    FROM  pg_stat_user_indexes ui
    JOIN  pg_index i
    ON    ui.indexrelid = i.indexrelid
    WHERE NOT indisunique
    AND   idx_scan < 50
    AND   pg_relation_size(relid) > 5 * 8192
    AND   (schemaname || '.' || relname)::regclass = '%s'::regclass
    ORDER BY
      pg_relation_size(i.indexrelid) / NULLIF(idx_scan, 0) DESC nulls first,
      pg_relation_size(i.indexrelid) DESC
  ) sub
$cmd$);

-- 客户端连接数
SELECT state, count(*)
FROM pg_stat_activity
GROUP BY state;

-- 系统活跃查询
SELECT global_pid, query, state
FROM citus_stat_activity
WHERE state != 'idle';

-- 查询等待原因
SELECT wait_event || ':' || wait_event_type AS type, count(*) AS number_of_occurences
FROM pg_stat_activity
WHERE state != 'idle'
GROUP BY wait_event, wait_event_type
ORDER BY number_of_occurences DESC;

-- 索引命中率
-- on coordinator
SELECT 100 * (sum(idx_blks_hit) - sum(idx_blks_read)) / sum(idx_blks_hit) AS index_hit_rate
  FROM pg_statio_user_indexes;

-- on workers
SELECT nodename, result as index_hit_rate
FROM run_command_on_workers($cmd$
  SELECT 100 * (sum(idx_blks_hit) - sum(idx_blks_read)) / sum(idx_blks_hit) AS index_hit_rate
    FROM pg_statio_user_indexes;
$cmd$);


-- 缓存命中率
-- on coordinator
SELECT
  sum(heap_blks_read) AS heap_read,
  sum(heap_blks_hit)  AS heap_hit,
  100 * sum(heap_blks_hit) / (sum(heap_blks_hit) + sum(heap_blks_read)) AS cache_hit_rate
FROM
  pg_statio_user_tables;

-- on workers
SELECT nodename, result as cache_hit_rate
FROM run_command_on_workers($cmd$
  SELECT
    100 * sum(heap_blks_hit) / (sum(heap_blks_hit) + sum(heap_blks_read)) AS cache_hit_rate
  FROM
    pg_statio_user_tables;
$cmd$);