PG 数据库迁移

CREATE USER dbuser_admin SUPERUSER;              -- 超级用户用于创建发布与订阅
CREATE USER replicator REPLICATION BYPASSRLS;    -- 复制用户用于订阅变更

wal_level: logical                      # MANDATORY!	
max_worker_processes: 64                # default 8 -> 64, set to CPU CORE 64
max_parallel_workers: 32                # default 8 -> 32, limit by max_worker_processes
max_parallel_maintenance_workers: 16    # default 2 -> 16, limit by parallel worker
max_parallel_workers_per_gather: 0      # default 2 -> 0,  disable parallel query on OLTP instance
# max_parallel_workers_per_gather: 16   # default 2 -> 16, enable parallel query on OLAP instance

max_wal_senders: 24                     # 10 -> 24
max_replication_slots: 16               # 10 -> 16 
max_logical_replication_workers: 8      # 4 -> 8, 6 sync worker + 1~2 apply worker
max_sync_workers_per_subscription: 6    # 2 -> 6, 6 sync worker

PGSRC='postgres://dbuser_admin@10.10.10.10/meta'        # 源端发布者 (SU)
PGDST='postgres://dbuser_admin@10.10.10.11/test'        # 宿端订阅者 (SU)

pgbench -is64 ${PGSRC}

psql ${PGSRC} -qAXtw <<-EOF
DROP TABLE IF EXISTS pgbench_extras;
CREATE TABLE IF NOT EXISTS pgbench_extras
  (id BIGSERIAL PRIMARY KEY,v  TIMESTAMP NOT NULL UNIQUE);
EOF

SELECT quote_ident(nspname) || '.' || quote_ident(relname) AS name
FROM pg_class c JOIN pg_namespace n ON c.relnamespace = n.oid
WHERE relkind = 'r' AND nspname NOT IN ('pg_catalog', 'information_schema', 'monitor', 'repack', 'pg_toast')

SELECT quote_ident(nspname) || '.' || quote_ident(relname) AS name, con.ri AS keys,
       CASE relreplident WHEN 'd' THEN 'default' WHEN 'n' THEN 'nothing' WHEN 'f' THEN 'full' WHEN 'i' THEN 'index' END AS identity
FROM pg_class c JOIN pg_namespace n ON c.relnamespace = n.oid, LATERAL (SELECT array_agg(contype) AS ri FROM pg_constraint WHERE conrelid = c.oid) con
WHERE relkind = 'r' AND nspname NOT IN ('pg_catalog', 'information_schema', 'monitor', 'repack', 'pg_toast')
ORDER BY 2,3;

          name           | keys  | identity
-------------------------+-------+----------
 public.spatial_ref_sys  | {c,p} | default
 public.pgbench_accounts | {p}   | default
 public.pgbench_branches | {p}   | default
 public.pgbench_tellers  | {p}   | default
 public.pgbench_extras   | {p,u} | default
 public.pgbench_history  | NULL  | default

-- 一个例子：即使pgbench_extras上有主键，但也可以使用唯一索引作为身份标识
ALTER TABLE pgbench_extras REPLICA IDENTITY USING INDEX pgbench_extras_v_key;

ALTER TABLE pgbench_history REPLICA IDENTITY FULL;

          name           | keys  | identity
-------------------------+-------+----------
 public.spatial_ref_sys  | {c,p} | default
 public.pgbench_accounts | {p}   | default
 public.pgbench_branches | {p}   | default
 public.pgbench_tellers  | {p}   | default
 public.pgbench_extras   | {p,u} | index
 public.pgbench_history  | NULL  | full

pg_dump ${PGSRC} --schema-only -n public | psql ${PGDST}

pg_dump ${PGSRC} --schema-only -n public -t 'pgbench_*' | psql ${PGDST}

CREATE PUBLICATION name
    [ FOR TABLE [ ONLY ] table_name [ * ] [, ...]
      | FOR ALL TABLES ]
    [ WITH ( publication_parameter [= value] [, ... ] ) ]

CREATE PUBLICATION "pg_meta_pub" FOR ALL TABLES;

CREATE PUBLICATION "pg_meta_pub" FOR ALL TABLES 
WITH(publish = 'insert', publish_via_partition_root = true);

SELECT E'CREATE PUBLICATION pg_meta_pub FOR TABLE\n' ||
       string_agg(quote_ident(nspname) || '.' || quote_ident(relname), E',\n') || ';' AS sql
FROM pg_class c JOIN pg_namespace n ON c.relnamespace = n.oid
WHERE relkind = 'r' AND nspname NOT IN ('pg_catalog', 'information_schema', 'monitor', 'repack', 'pg_toast')
AND relname ~ 'pgbench'; -- 只复制表名形如 pgbench* 的表
\gexec    -- 在psql中执行上面命令生成的SQL语句

psql ${PGSRC} -Xtw <<-EOF
  CREATE PUBLICATION pg_meta_pub FOR TABLE
    public.pgbench_accounts,
    public.pgbench_branches,
    public.pgbench_tellers,
    public.pgbench_history,
    public.pgbench_extras;
EOF

$ psql ${PGSRC} -Xxwc 'table pg_publication;'
-[ RECORD 1 ]+------------
oid          | 24679
pubname      | pg_meta_pub
pubowner     | 10
puballtables | f
pubinsert    | t
pubupdate    | t
pubdelete    | t
pubtruncate  | t
pubviaroot   | f

$ psql ${PGSRC} -Xwc 'table pg_publication_tables;'
   pubname   | schemaname |    tablename
-------------+------------+------------------
 pg_meta_pub | public     | pgbench_history
 pg_meta_pub | public     | pgbench_tellers
 pg_meta_pub | public     | pgbench_accounts
 pg_meta_pub | public     | pgbench_branches
 pg_meta_pub | public     | pgbench_extras

CREATE SUBSCRIPTION subscription_name
    CONNECTION 'conninfo'
    PUBLICATION publication_name [, ...]
    [ WITH ( subscription_parameter [= value] [, ... ] ) ]

psql ${PGDST} -Xtw <<-EOF
    CREATE SUBSCRIPTION "pg_test_sub" 
      CONNECTION 'host=10.10.10.10 user=replicator dbname=meta' 
      PUBLICATION "pg_meta_pub";
EOF

$ psql ${PGDST} -Xxwc 'TABLE pg_subscription;'
-[ RECORD 1 ]---+---------------------------------------------
oid             | 20759
subdbid         | 19351
subname         | pg_test_sub
subowner        | 16390
subenabled      | t
subconninfo     | host=10.10.10.10 user=replicator dbname=meta
subslotname     | pg_test_sub
subsynccommit   | off
subpublications | {pg_meta_pub}

$ psql ${PGDST} -Xwc 'table pg_subscription_rel;'
 srsubid | srrelid | srsubstate |  srsublsn
---------+---------+------------+------------
   20759 |   20742 | r          | 0/B0BC1FB8
   20759 |   20734 | r          | 0/B0BC20B0
   20759 |   20737 | r          | 0/B0BC20B0
   20759 |   20745 | r          | 0/B0BC20B0
   20759 |   20731 | r          | 0/B0BC20B0

psql ${PGDST} -Xxw <<-'EOF'
    SELECT subname, json_object_agg(srsubstate, cnt) FROM
    pg_subscription s JOIN
      (SELECT srsubid, srsubstate, count(*) AS cnt FROM pg_subscription_rel 
       GROUP BY srsubid, srsubstate) sr
    ON s.oid = sr.srsubid GROUP BY subname;
EOF

   subname   | json_object_agg
-------------+-----------------
 pg_test_sub | { "r" : 5 }

psql ${PGSRC} -Xxwc 'CHECKPOINT;ANALYZE;CHECKPOINT;'
psql ${PGSRC} -Xxwc 'CHECKPOINT;ANALYZE;CHECKPOINT;'

psql ${PGSRC} -Xxw <<-EOF
    SELECT application_name AS name,
           pg_current_wal_lsn() AS lsn,
           pg_current_wal_lsn() - replay_lsn AS lag 
    FROM pg_stat_replication;
EOF

-[ RECORD 1 ]-----
name | pg_test_sub
lsn  | 0/B0C24918
lag  | 0

psql ${PGDST} -qAXtwc 'ALTER SUBSCRIPTION pg_test_sub DISABLE;'

# 在宿端创建发布：pg_test_pub
psql ${PGDST} -Xtw <<-EOF
  CREATE PUBLICATION pg_test_pub FOR TABLE
    public.pgbench_accounts,
    public.pgbench_branches,
    public.pgbench_tellers,
    public.pgbench_history,
    public.pgbench_extras;
  TABLE pg_publication;
EOF

# 在源端创建订阅
psql ${PGSRC} -Xtw <<-EOF
    CREATE SUBSCRIPTION "pg_meta_sub" 
      CONNECTION 'host=10.10.10.11 user=replicator dbname=test' 
      PUBLICATION "pg_test_pub";
    TABLE pg_subscription;
EOF

# 清空源端所有相关表（危险），等待/或者不等待同步完成
psql ${PGSRC} -Xtw <<-EOF
  TRUNCATE TABLE
    public.pgbench_accounts,
    public.pgbench_branches,
    public.pgbench_tellers,
    public.pgbench_history,
    public.pgbench_extras;
  TABLE pg_publication;
EOF

PGSRC='postgres://dbuser_admin@10.10.10.10/meta'        # 源端发布者 (SU)
PGDST='postgres://dbuser_admin@10.10.10.11/test'        # 宿端订阅者 (SU)

-- 查询订阅端，生成的用于同步SEQUENCE的shell命令
psql ${PGDST} -qAXtw <<-'EOF'
    SELECT 'pg_dump ${PGSRC} -a ' ||
    string_agg('-t ' || quote_ident(schemaname) || '.' || quote_ident(sequencename), ' ') ||
    ' | grep setval | psql -qAXtw ${PGDST}'
    FROM pg_sequences;
EOF

pg_dump ${PGSRC} -a -t public.pgbench_extras_id_seq | grep setval | psql -qAXtw ${PGDST}

psql ${PGDST} -qAXtw <<-'EOF'
  SELECT pg_catalog.setval('public.pgbench_extras_id_seq', (SELECT max(id) + 1000 FROM pgbench_extras));
EOF

function compare_relation(){
	local relname=$1
	local identity=${2-'id'}
	psql ${3-${PGSRC}} -AXtwc "SELECT count(*) AS cnt, max($identity) AS max, min($identity) AS min FROM ${relname};"
	psql ${4-${PGDST}} -AXtwc "SELECT count(*) AS cnt, max($identity) AS max, min($identity) AS min FROM ${relname};"
}
compare_relation pgbench_accounts aid
compare_relation pgbench_branches bid
compare_relation pgbench_history  tid
compare_relation pgbench_tellers  tid

function compare_relation() {
    local src_url=${1}
    local dst_url=${2}
    local relname=${3}
    res1=$(psql "${src_url}" -AXtwc "SELECT count(*) AS cnt FROM ${relname};")
    res2=$(psql "${dst_url}" -AXtwc "SELECT count(*) AS cnt FROM ${relname};")
    if [[ "${res1}" == "${res2}" ]]; then
        echo -e "[ok] ${relname}\t\t\t${res1}\t${res2}"
    else
        echo -e "[xx] ${relname}\t\t\t${res1}\t${res2}"
    fi
}

function compare_all() {
    local src_url=${1}
    local dst_url=${2}
    tables=$(psql ${src_url} -AXtwc "SELECT quote_ident(nspname) || '.' || quote_ident(relname) AS name FROM pg_class c JOIN pg_namespace n ON c.relnamespace = n.oid WHERE relkind = 'r' AND nspname NOT IN ('pg_catalog', 'information_schema', 'monitor', 'repack', 'pg_toast')")
    for tbl in $tables; do
        result=$(compare_relation "${src_url}" "${dst_url}" ${tbl})
        echo ${result}
    done
}

compare_all ${PGSRC} ${PGDST}

psql ${PGSRC} -qwXtc "SELECT schemaname || '.' || sequencename AS name, last_value AS v FROM pg_sequences;"
psql ${PGDST} -qwXtc "SELECT schemaname || '.' || sequencename AS name, last_value AS v FROM pg_sequences;"

# 删除订阅侧的 订阅
psql ${PGDST} -qAXtw <<-'EOF'
    ALTER SUBSCRIPTION pg_test_sub DISABLE;
    DROP SUBSCRIPTION pg_test_sub;
EOF

# 删除发布侧的 发布
psql ${PGSRC} -qAXtw <<-'EOF'
    DROP PUBLICATION pg_meta_sub;
EOF

activate                          # 激活迁移上下文，注册环境变量
check-replica-identity            # 准备阶段：检查源集群所有表是否都具有复制身份（主键，或非空唯一候选键）
check-replica-identity-solution   # 准备阶段：针对没有合理复制身份表，生成修复SQL语句
check-special-object              # 准备阶段：检查物化视图，复合类型等特殊对象
compare                           # 比较：对源宿集群中的表进行快速比较（行数计算）
copy-schema                       # 存量迁移：将源集群中的模式复制到宿集群中（可以幂等执行）
create-pub                        # 存量迁移：在源集群中创建发布
create-sub                        # 存量迁移：在宿集群中创建订阅，建立源宿集群之间的逻辑复制
progress                          # 存量迁移：打印逻辑复制的进度
copy-seq                          # 存量/增量迁移：将源集群中的序列号复制到宿集群中（可以幂等执行，在切换时需要再次执行）
next-seq                          # 切换时刻：将宿集群的所有序列号紧急步进1000，以避免主键冲突。
remove-sub                        # 移除宿集群中的逻辑订阅

pg-meta-1	10.10.10.10  --> pg-test-1	10.10.10.11 (10.10.10.12,10.10.10.13)

#--------------------------------------------------------------#
#                   MIGRATION CONTEXT                          #
#--------------------------------------------------------------#

# src cluster (the old cluster)
src_cls: pg-meta                       # src cluster name
src_db: meta                           # src database name
src_ip: 10.10.10.10                    # ip address of src cluster primary
src_list: [ ]                          # ip address list of src cluster members (non-primary)

#--------------------------------------------------------------#
# dst cluster (the new cluster)
dst_cls: pg-test                       # dst cluster name
dst_db: test                           # dst database name
dst_ip: 10.10.10.11                    # dst cluster leader ip addressh
dst_list: [ 10.10.10.12, 10.10.10.13 ] # dst cluster members (non-primary)

# dst cluster access information
dst_dns: pg-test                       # dst cluster dns records
dst_vip: 10.10.10.3                    # dst cluster vip records

#--------------------------------------------------------------#
# credential (assume .pgpass viable)
pg_admin_username: dbuser_dba          # superuser @ both side
pg_replicatoin_username: replicator    # repl user @ src to be used
migration_context_dir: ~/migration     # this dir will be created
#--------------------------------------------------------------#