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

alias pt='patronictl -c /pg/bin/patroni.yml'

alias pt-up='sudo systemctl start patroni'     # launch patroni
alias pt-dw='sudo systemctl stop  patroni'     # stop patroni
alias pt-st='systemctl status patroni'         # report patroni status
alias pt-ps='ps aux | grep patroni'            # show patroni processes
alias pt-log='tail -f /pg/log/patroni.log'     # watch patroni logs

$ pt --help
Usage: patronictl [OPTIONS] COMMAND [ARGS]...

Options:
  -c, --config-file TEXT  Configuration file
  -d, --dcs TEXT          Use this DCS
  -k, --insecure          Allow connections to SSL sites without certs
  --help                  Show this message and exit.

Commands:
  configure    Create configuration file
  dsn          Generate a dsn for the provided member,...
  edit-config  Edit cluster configuration
  failover     Failover to a replica
  flush        Discard scheduled events
  history      Show the history of failovers/switchovers
  list         List the Patroni members for a given Patroni
  pause        Disable auto failover
  query        Query a Patroni PostgreSQL member
  reinit       Reinitialize cluster member
  reload       Reload cluster member configuration
  remove       Remove cluster from DCS
  restart      Restart cluster member
  resume       Resume auto failover
  scaffold     Create a structure for the cluster in DCS
  show-config  Show cluster configuration
  switchover   Switchover to a replica
  topology     Prints ASCII topology for given cluster
  version      Output version of patronictl command or a...

# run as postgres @ any member of cluster `pg-test`
$ pt failover
Candidate ['pg-test-2', 'pg-test-3'] []: pg-test-3
Current cluster topology
+ Cluster: pg-test (6886641621295638555) -----+----+-----------+-----------------+
| Member    | Host        | Role    | State   | TL | Lag in MB | Tags            |
+-----------+-------------+---------+---------+----+-----------+-----------------+
| pg-test-1 | 10.10.10.11 | Leader  | running |  1 |           | clonefrom: true |
| pg-test-2 | 10.10.10.12 | Replica | running |  1 |         0 | clonefrom: true |
| pg-test-3 | 10.10.10.13 | Replica | running |  1 |         0 | clonefrom: true |
+-----------+-------------+---------+---------+----+-----------+-----------------+
Are you sure you want to failover cluster pg-test, demoting current master pg-test-1? [y/N]: y
+ Cluster: pg-test (6886641621295638555) -----+----+-----------+-----------------+
| Member    | Host        | Role    | State   | TL | Lag in MB | Tags            |
+-----------+-------------+---------+---------+----+-----------+-----------------+
| pg-test-1 | 10.10.10.11 | Replica | running |  2 |         0 | clonefrom: true |
| pg-test-2 | 10.10.10.12 | Replica | running |  2 |         0 | clonefrom: true |
| pg-test-3 | 10.10.10.13 | Leader  | running |  2 |           | clonefrom: true |
+-----------+-------------+---------+---------+----+-----------+-----------------+

ALTER TABLE pgbench_accounts DROP CONSTRAINT pgbench_accounts_pkey ;

SELECT abalance FROM pgbench_accounts WHERE aid = $1

CREATE TABLE public.isd_station
(
    station    VARCHAR(12) PRIMARY KEY,
    usaf       VARCHAR(6) GENERATED ALWAYS AS (substring(station, 1, 6)) STORED,
    wban       VARCHAR(5) GENERATED ALWAYS AS (substring(station, 7, 5)) STORED,
    name       VARCHAR(32),
    country    VARCHAR(2),
    province   VARCHAR(2),
    icao       VARCHAR(4),
    location   GEOMETRY(POINT),
    longitude  NUMERIC GENERATED ALWAYS AS (Round(ST_X(location)::NUMERIC, 6)) STORED,
    latitude   NUMERIC GENERATED ALWAYS AS (Round(ST_Y(location)::NUMERIC, 6)) STORED,
    elevation  NUMERIC,
    period     daterange,
    begin_date DATE GENERATED ALWAYS AS (lower(period)) STORED,
    end_date   DATE GENERATED ALWAYS AS (upper(period)) STORED
);

CREATE TABLE public.isd_hourly
(
    station    VARCHAR(11) NOT NULL,
    ts         TIMESTAMP   NOT NULL,
    temp       NUMERIC(3, 1),
    dewp       NUMERIC(3, 1),
    slp        NUMERIC(5, 1),
    stp        NUMERIC(5, 1),
    vis        NUMERIC(6),
    wd_angle   NUMERIC(3),
    wd_speed   NUMERIC(4, 1),
    wd_gust    NUMERIC(4, 1),
    wd_code    VARCHAR(1),
    cld_height NUMERIC(5),
    cld_code   VARCHAR(2),
    sndp       NUMERIC(5, 1),
    prcp       NUMERIC(5, 1),
    prcp_hour  NUMERIC(2),
    prcp_code  VARCHAR(1),
    mw_code    VARCHAR(2),
    aw_code    VARCHAR(2),
    pw_code    VARCHAR(1),
    pw_hour    NUMERIC(2),
    data       JSONB
) PARTITION BY RANGE (ts);

CREATE TABLE public.isd_daily
(
    station     VARCHAR(12) NOT NULL,
    ts          DATE        NOT NULL,
    temp_mean   NUMERIC(3, 1),
    temp_min    NUMERIC(3, 1),
    temp_max    NUMERIC(3, 1),
    dewp_mean   NUMERIC(3, 1),
    slp_mean    NUMERIC(5, 1),
    stp_mean    NUMERIC(5, 1),
    vis_mean    NUMERIC(6),
    wdsp_mean   NUMERIC(4, 1),
    wdsp_max    NUMERIC(4, 1),
    gust        NUMERIC(4, 1),
    prcp_mean   NUMERIC(5, 1),
    prcp        NUMERIC(5, 1),
    sndp        NuMERIC(5, 1),
    is_foggy    BOOLEAN,
    is_rainy    BOOLEAN,
    is_snowy    BOOLEAN,
    is_hail     BOOLEAN,
    is_thunder  BOOLEAN,
    is_tornado  BOOLEAN,
    temp_count  SMALLINT,
    dewp_count  SMALLINT,
    slp_count   SMALLINT,
    stp_count   SMALLINT,
    wdsp_count  SMALLINT,
    visib_count SMALLINT,
    temp_min_f  BOOLEAN,
    temp_max_f  BOOLEAN,
    prcp_flag   CHAR,
    PRIMARY KEY (ts, station)
) PARTITION BY RANGE (ts);

# download, clean, reload latest hourly dataset
bin/get-isd-daily.sh
bin/load-isd-daily.sh

# download, clean, reload latest daily dataset
bin/get-isd-daily.sh
bin/load-isd-daily.sh

# recalculate latest partition of monthly and yearly
bin/refresh-latest.sh

NAME
	isdh -- Intergrated Surface Dataset Hourly Parser

SYNOPSIS
	isdh [-i <input|stdin>] [-o <output|st>] -p -d -c -v

DESCRIPTION
	The isdh program takes isd hourly (yearly tarball file) as input.
	And generate csv format as output

OPTIONS
	-i	<input>		input file, stdin by default
	-o	<output>	output file, stdout by default
	-p	<profpath>	pprof file path (disable by default)	
	-v                verbose progress report
	-d                de-duplicate rows (raw, ts-first, hour-first)
	-c                add comma separated extra columns