from text2vec import SentenceModel 
# 自动下载并加载模型
model = SentenceModel('shibing624/text2vec-base-chinese')
sentence = '这里是你想编码的文本输入'
vec = model.encode(sentence)

CREATE EXTENSION vector;
CREATE TABLE sentences(id    BIGINT PRIMARY KEY,  -- 标识    txt   TEXT NOT NULL,       -- 文本    vec   VECTOR(768) NOT NULL -- 向量);

# !/usr/bin/env python3
from text2vec import SentenceModel
from psycopg2 import connect

model = SentenceModel('shibing624/text2vec-base-chinese')

def query(question, limit=64):
    vec = model.encode(question)  # 生成一个一次性的编码向量，默认查找最接近的64条记录
    item = 'ARRAY[' + ','.join([str(f) for f in vec.tolist()]) + ']::VECTOR(768)'
    cursor = connect('postgres:///').cursor()
    cursor.execute("""SELECT id, txt, vec <-> %s AS d FROM sentences ORDER BY 3 LIMIT %s;""" % (item, limit))
    for id, txt, distance in cursor.fetchall():
        print("%-6d [%.3f]\t%s" % (id, distance, txt))

-- 1M 个 1536 维向量，随机取 TOP1～50，余弦距离， 单核：插入与索引耗时均为5～6分钟，大小8GB左右。随机向量最近邻 Top1 召回：8ms
DROP TABLE IF EXISTS vtest; 
CREATE TABLE vtest ( id BIGINT, v  VECTOR(1536) ); 
TRUNCATE vtest;

INSERT INTO vtest SELECT i, random_array(1536)::vector(1536) FROM generate_series(1, 1000000) AS i;
CREATE INDEX ON vtest USING ivfflat (v vector_cosine_ops) WITH(lists = 1000);
WITH probe AS (SELECT random_array(1536)::VECTOR(1536) AS v) 
  SELECT id FROM vtest ORDER BY v <=> (SELECT v FROM probe) limit 1;


-- 简易SIFT ，1亿个128维向量，测试L2距离，召回1个最近向量， 5 ms， 召回最近100个向量：21ms
DROP TABLE IF EXISTS vtest;
CREATE TABLE vtest( id BIGINT, v  VECTOR(128) );
TRUNCATE vtest;

INSERT INTO vtest SELECT i, random_array(128)::vector(128) FROM generate_series(1, 100000000) AS i;
CREATE INDEX ON vtest USING ivfflat (v vector_l2_ops) WITH(lists = 10000);
WITH probe AS (SELECT random_array(128)::VECTOR(128) AS v) 
  SELECT id FROM vtest ORDER BY v <-> (SELECT v FROM probe) limit 1; -- LIMIT 100

-- SIFT 1M 数据集，128维embedding，使用ivfflat索引, L2距离，10K测试向量集。
DROP TABLE IF EXISTS sift_base;
CREATE TABLE sift_base  (id BIGINT PRIMARY KEY , v VECTOR(128));
DROP TABLE IF EXISTS sift_query; 
CREATE TABLE sift_query (id BIGINT PRIMARY KEY , v VECTOR(128));
CREATE INDEX ON sift_base USING ivfflat (v vector_l2_ops) WITH(lists = 1000);

-- 一次性寻找 sift_query 表中 10000 条向量在 sift_base 表中的最近邻 Top1: 单进程 18553ms / 10000 Q = 1.8ms
explain analyze SELECT q.id, s.id FROM sift_query q ,LATERAL (SELECT id FROM sift_base ORDER BY v <-> q.v limit 1) AS s; 

-- 单次随机查询耗时在 个位数毫秒
WITH probe AS (SELECT v AS query FROM sift_query WHERE id =  (random() * 999)::BIGINT LIMIT 1)
  SELECT id FROM sift_base ORDER BY v <-> (SELECT query FROM probe) LIMIT 1;

CREATE EXTENSION vector;
CREATE TABLE items (vec vector(2));
INSERT INTO items (vec) VALUES ('[1,1]'), ('[-2,-2]'), ('[-3,4]');
SELECT *, vec <=> '[0,1]' AS d FROM items ORDER BY 2 LIMIT 3;

CREATE TABLE app(name TEXT PRIMARY KEY); 
-- COPY app FROM '/tmp/app.csv';

Rome travel guide, rome italy map rome tourist attractions directions to colosseum, vatican museum, offline ATAC city rome bus tram underground train maps, 罗马地图,罗马地铁,罗马火车,罗马旅行指南"""
Urban Pics - 游戏俚语词典
世界经典童话故事大全(6到12岁少年儿童睡前故事英语亲子软件) 2 - 高级版
星征服者
客房控制系统
Santa ME! - 易圣诞老人,小精灵快乐的脸效果！

SELECT * FROM app WHERE name LIKE '%支付宝%';

CREATE INDEX ON app(name COLLATE "C");          -- 后模糊
CREATE INDEX ON app(reverse(name) COLLATE "C"); -- 前模糊

-- 使用trgm操作符提取关键词素，并建立gist索引
CREATE INDEX ON app USING gist (name gist_trgm_ops);

SELECT name, similarity(name, '支付宝') AS sim FROM app 
WHERE name % '支付宝'  ORDER BY 2 DESC;

         name          |     sim
-----------------------+------------
 支付宝 - 让生活更简单 | 0.36363637
 支付搜                | 0.33333334
 支付社                | 0.33333334
 支付啦                | 0.33333334
(4 rows)

Time: 231.872 ms

Sort  (cost=177.20..177.57 rows=151 width=29) (actual time=251.969..251.970 rows=4 loops=1)
"  Sort Key: (similarity(name, '支付宝'::text)) DESC"
  Sort Method: quicksort  Memory: 25kB
  ->  Index Scan using app_name_idx1 on app  (cost=0.41..171.73 rows=151 width=29) (actual time=145.414..251.956 rows=4 loops=1)
        Index Cond: (name % '支付宝'::text)
Planning Time: 2.331 ms
Execution Time: 252.011 ms

-- 创建一个分词函数
CREATE OR REPLACE FUNCTION tokens12(text) returns text[] as $$....$$;

-- 基于该分词函数创建表达式索引
CREATE INDEX ON app USING GIN(tokens12(name));

-- 使用关键词进行复杂的定制查询（关键词数组操作）
SELECT * from app where split_to_chars(name) && ARRAY['天气'];

-- 使用关键词进行复杂的定制查询（tsquery操作）
SELECT * from app where to_tsvector123(name) @@ 'BTC &! 钱包 & ! 交易 '::tsquery;

-- 创建分词函数，将字符串拆为单字，双字组成的词素数组
CREATE OR REPLACE FUNCTION tokens12(text) returns text[] AS $$
DECLARE
    res TEXT[];
BEGIN
    SELECT regexp_split_to_array($1, '') INTO res;
    FOR i in 1..length($1) - 1 LOOP
            res := array_append(res, substring($1, i, 2));
    END LOOP;
    RETURN res;
END;
$$ LANGUAGE plpgsql STRICT PARALLEL SAFE IMMUTABLE;

SELECT tokens2('艾米莉的埃及历险记');
-- {艾米,米莉,莉的,的埃,埃及,及历,历险,险记}

SELECT tokens2('艾米莉');
-- {艾米,米莉}

SELECT * FROM app WHERE tokens2(name) @> tokens2('艾米莉');
 美味餐厅 - 艾米莉的圣诞颂歌
 美味餐厅 - 艾米莉的瓶中信笺
 小清新艾米莉
 艾米莉的埃及历险记
 艾米莉的极地大冒险
 艾米莉的万圣节历险记
 6rows / 0.38ms

SELECT * FROM app WHERE tokens2(name) && tokens2('艾米莉');

 AR艾米互动故事-智慧妈妈必备
 Amy and train 艾米和小火车
 米莉·马洛塔的涂色探索
 给利伴_艾米罗公司旗下专业购物返利网
 艾米团购
 记忆游戏 - 米莉和泰迪
 (56 row ) / 0.4 ms

-- 包含关键词 微信、红包，但不包含 ‘支付’ (1ms | 11 rows)
SELECT * FROM app WHERE tokens2(name) @> ARRAY['微信','红包'] 
AND NOT tokens2(name) @> ARRAY['支付'];

-- 包含关键词 微信 的应用，按照L式编辑距离排序 ( 1.1 ms | 10 rows)
-- create extension fuzzystrmatch;
SELECT name, levenshtein(name, '微信') AS d 
FROM app WHERE tokens12(name) @> ARRAY['微信'] 
ORDER BY 2 LIMIT 10;

 微信           | 0
 微信读书       | 2
 微信趣图       | 2
 微信加密       | 2
 企业微信       | 2
 微信通助手     | 3
 微信彩色消息   | 4
 艺术微信平台网 | 5
 涂鸦画板- 微信 | 6
 手写板for微信  | 6

CREATE OR REPLACE FUNCTION cjk_to_tsvector(_src text) RETURNS tsvector AS $$
DECLARE
    res TEXT[]:= show_trgm(_src);
    cjk TEXT; -- 中日韩连续文本段
BEGIN
    FOR cjk IN SELECT unnest(i) FROM regexp_matches(_src,'[\u4E00-\u9FCC\u3400-\u4DBF\u20000-\u2A6D6\u2A700-\u2B81F\u2E80-\u2FDF\uF900-\uFA6D\u2F800-\u2FA1B]+','g') regex(i) LOOP
            FOR i in 1..length(cjk) - 1 LOOP
                    res := array_append(res, substring(cjk, i, 2));
                END LOOP; -- 将每个中日韩连续文本段两字词语加入列表
        END LOOP;
    return array_to_tsvector(res);
end
$$ LANGUAGE PlPgSQL PARALLEL SAFE COST 100 STRICT IMMUTABLE;


-- 如果需要使用标签数组的方式，可以使用此函数。
CREATE OR REPLACE FUNCTION cjk_to_array(_src text) RETURNS TEXT[] AS $$
BEGIN
    RETURN tsvector_to_array(cjk_to_tsvector(_src));
END
$$ LANGUAGE PlPgSQL PARALLEL SAFE COST 100 STRICT IMMUTABLE;

-- 创建分词专用函数索引
CREATE INDEX ON app USING GIN(cjk_to_array(name));

CREATE OR REPLACE FUNCTION to_tsvector123(src text) RETURNS tsvector AS $$
DECLARE
    res TEXT[];
    n INTEGER:= length(src);
begin
    SELECT regexp_split_to_array(src, '') INTO res;
    FOR i in 1..n - 2 LOOP res := array_append(res, substring(src, i, 2));res := array_append(res, substring(src, i, 3)); END LOOP;
    res := array_append(res, substring(src, n-1, 2));
    SELECT array_agg(distinct i) INTO res FROM (SELECT i FROM unnest(res) r(i) EXCEPT SELECT * FROM (VALUES(' '),('，'),('的'),('。'),('-'),('.')) c ) d; -- optional (normalize)
    RETURN array_to_tsvector(res);
end
$$ LANGUAGE PlPgSQL PARALLEL SAFE COST 100 STRICT IMMUTABLE;

-- 使用自定义分词函数，创建函数表达式索引
CREATE INDEX ON app USING GIN(to_tsvector123(name));

-- 包含 '学英语' 和 '雅思'
SELECT * from app where to_tsvector123(name) @@ '学英语 & 雅思'::tsquery;

-- 所有关于 'BTC' 但不含'钱包' '交易'字样的应用
SELECT * from app where to_tsvector123(name) @@ 'BTC &! 钱包 & ! 交易 '::tsquery;

package main

import (
	"fmt"
	"net"
	"time"

	"github.com/jackc/pgx/pgproto3"
)

func GetFrontend(address string) *pgproto3.Frontend {
	conn, _ := (&net.Dialer{KeepAlive: 5 * time.Minute}).Dial("tcp4", address)
	frontend, _ := pgproto3.NewFrontend(conn, conn)
	return frontend
}

func main() {
	frontend := GetFrontend("127.0.0.1:5432")

	// 建立连接
	startupMsg := &pgproto3.StartupMessage{
		ProtocolVersion: pgproto3.ProtocolVersionNumber,
		Parameters:      map[string]string{"user": "vonng"},
	}
	frontend.Send(startupMsg)

	// 启动过程，收到ReadyForQuery消息代表启动过程结束
	for {
		msg, _ := frontend.Receive()
		fmt.Printf("%T %v\n", msg, msg)
		if _, ok := msg.(*pgproto3.ReadyForQuery); ok {
			fmt.Println("[STARTUP] connection established")
			break
		}
	}

	// 简单查询协议
	simpleQueryMsg := &pgproto3.Query{String: `SELECT 1 as a;`}
	frontend.Send(simpleQueryMsg)
	// 收到CommandComplete消息代表查询结束
	for {
		msg, _ := frontend.Receive()
		fmt.Printf("%T %v\n", msg, msg)
		if _, ok := msg.(*pgproto3.CommandComplete); ok {
			fmt.Println("[QUERY] query complete")
			break
		}
	}
}

*pgproto3.Authentication &{0 [0 0 0 0] [] []}
*pgproto3.ParameterStatus &{application_name }
*pgproto3.ParameterStatus &{client_encoding UTF8}
*pgproto3.ParameterStatus &{DateStyle ISO, MDY}
*pgproto3.ParameterStatus &{integer_datetimes on}
*pgproto3.ParameterStatus &{IntervalStyle postgres}
*pgproto3.ParameterStatus &{is_superuser on}
*pgproto3.ParameterStatus &{server_encoding UTF8}
*pgproto3.ParameterStatus &{server_version 11.3}
*pgproto3.ParameterStatus &{session_authorization vonng}
*pgproto3.ParameterStatus &{standard_conforming_strings on}
*pgproto3.ParameterStatus &{TimeZone PRC}
*pgproto3.BackendKeyData &{35703 345830596}
*pgproto3.ReadyForQuery &{73}
[STARTUP] connection established
*pgproto3.RowDescription &{[{a 0 0 23 4 -1 0}]}
*pgproto3.DataRow &{[[49]]}
*pgproto3.CommandComplete &{SELECT 1}
[QUERY] query complete

package main

import (
	"io"
	"net"
	"strings"
	"time"

	"github.com/jackc/pgx/pgproto3"
)

type ProxyServer struct {
	UpstreamAddr string
	ListenAddr   string
	Listener     net.Listener
	Dialer       net.Dialer
}

func NewProxyServer(listenAddr, upstreamAddr string) *ProxyServer {
	ln, _ := net.Listen(`tcp4`, listenAddr)
	return &ProxyServer{
		ListenAddr:   listenAddr,
		UpstreamAddr: upstreamAddr,
		Listener:     ln,
		Dialer:       net.Dialer{KeepAlive: 1 * time.Minute},
	}
}

func (ps *ProxyServer) Serve() error {
	for {
		conn, err := ps.Listener.Accept()
		if err != nil {
			panic(err)
		}
		go ps.ServeOne(conn)
	}
}

func (ps *ProxyServer) ServeOne(clientConn net.Conn) error {
	backend, _ := pgproto3.NewBackend(clientConn, clientConn)
	startupMsg, err := backend.ReceiveStartupMessage()
	if err != nil && strings.Contains(err.Error(), "ssl") {
		if _, err := clientConn.Write([]byte(`N`)); err != nil {
			panic(err)
		}
		// ssl is not welcome, now receive real startup msg
		startupMsg, err = backend.ReceiveStartupMessage()
		if err != nil {
			panic(err)
		}
	}

	serverConn, _ := ps.Dialer.Dial(`tcp4`, ps.UpstreamAddr)
	frontend, _ := pgproto3.NewFrontend(serverConn, serverConn)
	frontend.Send(startupMsg)

	errChan := make(chan error, 2)
	go func() {
		_, err := io.Copy(clientConn, serverConn)
		errChan <- err
	}()
	go func() {
		_, err := io.Copy(serverConn, clientConn)
		errChan <- err
	}()

	return <-errChan
}

func main() {
	proxy := NewProxyServer("127.0.0.1:5433", "127.0.0.1:5432")
	proxy.Serve()
}

$ psql postgres://127.0.0.1:5433/data?sslmode=disable -c 'SELECT * FROM pg_stat_activity LIMIT 1;'

[B2F] *pgproto3.ParameterStatus &{application_name psql}
[B2F] *pgproto3.ParameterStatus &{client_encoding UTF8}
[B2F] *pgproto3.ParameterStatus &{DateStyle ISO, MDY}
[B2F] *pgproto3.ParameterStatus &{integer_datetimes on}
[B2F] *pgproto3.ParameterStatus &{IntervalStyle postgres}
[B2F] *pgproto3.ParameterStatus &{is_superuser on}
[B2F] *pgproto3.ParameterStatus &{server_encoding UTF8}
[B2F] *pgproto3.ParameterStatus &{server_version 11.3}
[B2F] *pgproto3.ParameterStatus &{session_authorization vonng}
[B2F] *pgproto3.ParameterStatus &{standard_conforming_strings on}
[B2F] *pgproto3.ParameterStatus &{TimeZone PRC}
[B2F] *pgproto3.BackendKeyData &{41588 1354047533}
[B2F] *pgproto3.ReadyForQuery &{73}
[F2B] *pgproto3.Query &{SELECT * FROM pg_stat_activity LIMIT 1;}
[B2F] *pgproto3.RowDescription &{[{datid 11750 1 26 4 -1 0} {datname 11750 2 19 64 -1 0} {pid 11750 3 23 4 -1 0} {usesysid 11750 4 26 4 -1 0} {usename 11750 5 19 64 -1 0} {application_name 11750 6 25 -1 -1 0} {client_addr 11750 7 869 -1 -1 0} {client_hostname 11750 8 25 -1 -1 0} {client_port 11750 9 23 4 -1 0} {backend_start 11750 10 1184 8 -1 0} {xact_start 11750 11 1184 8 -1 0} {query_start 11750 12 1184 8 -1 0} {state_change 11750 13 1184 8 -1 0} {wait_event_type 11750 14 25 -1 -1 0} {wait_event 11750 15 25 -1 -1 0} {state 11750 16 25 -1 -1 0} {backend_xid 11750 17 28 4 -1 0} {backend_xmin 11750 18 28 4 -1 0} {query 11750 19 25 -1 -1 0} {backend_type 11750 20 25 -1 -1 0}]}
[B2F] *pgproto3.DataRow &{[[] [] [52 56 55 52] [] [] [] [] [] [] [50 48 49 57 45 48 53 45 49 56 32 50 48 58 52 56 58 49 57 46 51 50 55 50 54 55 43 48 56] [] [] [] [65 99 116 105 118 105 116 121] [65 117 116 111 86 97 99 117 117 109 77 97 105 110] [] [] [] [] [97 117 116 111 118 97 99 117 117 109 32 108 97 117 110 99 104 101 114]]}
[B2F] *pgproto3.CommandComplete &{SELECT 1}
[B2F] *pgproto3.ReadyForQuery &{73}
[F2B] *pgproto3.Terminate &{}

CREATE TABLE t (k INTEGER PRIMARY KEY, v int);
TRUNCATE t; INSERT INTO t VALUES (1,10), (2,20);

begin; set transaction isolation level read committed; -- T1
begin; set transaction isolation level read committed; -- T2

update t set v = 11 where k = 1; -- T1
update t set v = 12 where k = 1; -- T2, BLOCKS
update t set v = 21 where k = 2; -- T1

commit; -- T1. This unblocks T2
select * from t; -- T1. Shows 1 => 11, 2 => 21
update t set v = 22 where k = 2; -- T2


commit; -- T2
select * from test; -- either. Shows 1 => 12, 2 => 22

 relname | locktype | virtualtransaction |  pid  |       mode       | granted | fastpath
---------+----------+--------------------+-------+------------------+---------+----------
 t_pkey  | relation | 4/578              | 37670 | RowExclusiveLock | t       | t
 t       | relation | 4/578              | 37670 | RowExclusiveLock | t       | t

 relname | locktype | virtualtransaction |  pid  |       mode       | granted | fastpath
---------+----------+--------------------+-------+------------------+---------+----------
 t_pkey  | relation | 4/578              | 37670 | RowExclusiveLock | t       | t
 t       | relation | 4/578              | 37670 | RowExclusiveLock | t       | t
 t_pkey  | relation | 6/494              | 37672 | RowExclusiveLock | t       | t
 t       | relation | 6/494              | 37672 | RowExclusiveLock | t       | t
 t       | tuple    | 6/494              | 37672 | ExclusiveLock    | t       | f

 relname | locktype | virtualtransaction |  pid  |       mode       | granted | fastpath
---------+----------+--------------------+-------+------------------+---------+----------
 t_pkey  | relation | 4/578              | 37670 | RowExclusiveLock | t       | t
 t       | relation | 4/578              | 37670 | RowExclusiveLock | t       | t
 t_pkey  | relation | 6/494              | 37672 | RowExclusiveLock | t       | t
 t       | relation | 6/494              | 37672 | RowExclusiveLock | t       | t
 t       | tuple    | 6/494              | 37672 | ExclusiveLock    | t       | f

create table test (id int primary key, value int);
insert into test (id, value) values (1, 10), (2, 20);

select current_setting('transaction_isolation');

begin; set transaction isolation level read committed; -- T1
begin; set transaction isolation level read committed; -- T2
update test set value = 11 where id = 1; -- T1
update test set value = 12 where id = 1; -- T2, BLOCKS
update test set value = 21 where id = 2; -- T1
commit; -- T1. This unblocks T2
select * from test; -- T1. Shows 1 => 11, 2 => 21
update test set value = 22 where id = 2; -- T2
commit; -- T2
select * from test; -- either. Shows 1 => 12, 2 => 22

begin; set transaction isolation level read committed; -- T1
begin; set transaction isolation level read committed; -- T2
update test set value = 101 where id = 1; -- T1
select * from test; -- T2. Still shows 1 => 10
abort;  -- T1
select * from test; -- T2. Still shows 1 => 10
commit; -- T2

begin; set transaction isolation level read committed; -- T1
begin; set transaction isolation level read committed; -- T2
update test set value = 101 where id = 1; -- T1
select * from test; -- T2. Still shows 1 => 10
update test set value = 11 where id = 1; -- T1
commit; -- T1
select * from test; -- T2. Now shows 1 => 11
commit; -- T2

begin; set transaction isolation level read committed; -- T1
begin; set transaction isolation level read committed; -- T2
update test set value = 11 where id = 1; -- T1
update test set value = 22 where id = 2; -- T2
select * from test where id = 2; -- T1. Still shows 2 => 20
select * from test where id = 1; -- T2. Still shows 1 => 10
commit; -- T1
commit; -- T2

begin; set transaction isolation level read committed; -- T1
begin; set transaction isolation level read committed; -- T2
begin; set transaction isolation level read committed; -- T3
update test set value = 11 where id = 1; -- T1
update test set value = 19 where id = 2; -- T1
update test set value = 12 where id = 1; -- T2. BLOCKS
commit; -- T1. This unblocks T2
select * from test where id = 1; -- T3. Shows 1 => 11
update test set value = 18 where id = 2; -- T2
select * from test where id = 2; -- T3. Shows 2 => 19
commit; -- T2
select * from test where id = 2; -- T3. Shows 2 => 18
select * from test where id = 1; -- T3. Shows 1 => 12
commit; -- T3

begin; set transaction isolation level read committed; -- T1
begin; set transaction isolation level read committed; -- T2
select * from test where value = 30; -- T1. Returns nothing
insert into test (id, value) values(3, 30); -- T2
commit; -- T2
select * from test where value % 3 = 0; -- T1. Returns the newly inserted row
commit; -- T1

begin; set transaction isolation level repeatable read; -- T1
begin; set transaction isolation level repeatable read; -- T2
select * from test where value = 30; -- T1. Returns nothing
insert into test (id, value) values(3, 30); -- T2
commit; -- T2
select * from test where value % 3 = 0; -- T1. Still returns nothing
commit; -- T1

begin; set transaction isolation level read committed; -- T1
begin; set transaction isolation level read committed; -- T2
update test set value = value + 10; -- T1
delete from test where value = 20;  -- T2, BLOCKS
commit; -- T1. This unblocks T2
select * from test where value = 20; -- T2, returns 1 => 20 (despite ostensibly having been deleted)
commit; -- T2

begin; set transaction isolation level repeatable read; -- T1
begin; set transaction isolation level repeatable read; -- T2
update test set value = value + 10; -- T1
delete from test where value = 20;  -- T2, BLOCKS
commit; -- T1. T2 now prints out "ERROR: could not serialize access due to concurrent update"
abort;  -- T2. There's nothing else we can do, this transaction has failed

begin; set transaction isolation level read committed; -- T1
begin; set transaction isolation level read committed; -- T2
select * from test where id = 1; -- T1
select * from test where id = 1; -- T2
update test set value = 11 where id = 1; -- T1
update test set value = 11 where id = 1; -- T2, BLOCKS
commit; -- T1. This unblocks T2, so T1's update is overwritten
commit; -- T2

begin; set transaction isolation level repeatable read; -- T1
begin; set transaction isolation level repeatable read; -- T2
select * from test where id = 1; -- T1
select * from test where id = 1; -- T2
update test set value = 11 where id = 1; -- T1
update test set value = 11 where id = 1; -- T2, BLOCKS
commit; -- T1. T2 now prints out "ERROR: could not serialize access due to concurrent update"
abort;  -- T2. There's nothing else we can do, this transaction has failed

begin; set transaction isolation level read committed; -- T1
begin; set transaction isolation level read committed; -- T2
select * from test where id = 1; -- T1. Shows 1 => 10
select * from test where id = 1; -- T2
select * from test where id = 2; -- T2
update test set value = 12 where id = 1; -- T2
update test set value = 18 where id = 2; -- T2
commit; -- T2
select * from test where id = 2; -- T1. Shows 2 => 18
commit; -- T1

begin; set transaction isolation level repeatable read; -- T1
begin; set transaction isolation level repeatable read; -- T2
select * from test where id = 1; -- T1. Shows 1 => 10
select * from test where id = 1; -- T2
select * from test where id = 2; -- T2
update test set value = 12 where id = 1; -- T2
update test set value = 18 where id = 2; -- T2
commit; -- T2
select * from test where id = 2; -- T1. Shows 2 => 20
commit; -- T1

begin; set transaction isolation level repeatable read; -- T1
begin; set transaction isolation level repeatable read; -- T2
select * from test where value % 5 = 0; -- T1
update test set value = 12 where value = 10; -- T2
commit; -- T2
select * from test where value % 3 = 0; -- T1. Returns nothing
commit; -- T1

begin; set transaction isolation level repeatable read; -- T1
begin; set transaction isolation level repeatable read; -- T2
select * from test where id = 1; -- T1. Shows 1 => 10
select * from test; -- T2
update test set value = 12 where id = 1; -- T2
update test set value = 18 where id = 2; -- T2
commit; -- T2
delete from test where value = 20; -- T1. Prints "ERROR: could not serialize access due to concurrent update"
abort; -- T1. There's nothing else we can do, this transaction has failed

begin; set transaction isolation level repeatable read; -- T1
begin; set transaction isolation level repeatable read; -- T2
select * from test where id in (1,2); -- T1
select * from test where id in (1,2); -- T2
update test set value = 11 where id = 1; -- T1
update test set value = 21 where id = 2; -- T2
commit; -- T1
commit; -- T2

begin; set transaction isolation level serializable; -- T1
begin; set transaction isolation level serializable; -- T2
select * from test where id in (1,2); -- T1
select * from test where id in (1,2); -- T2
update test set value = 11 where id = 1; -- T1
update test set value = 21 where id = 2; -- T2
commit; -- T1
commit; -- T2. Prints out "ERROR: could not serialize access due to read/write dependencies among transactions"

begin; set transaction isolation level repeatable read; -- T1
begin; set transaction isolation level repeatable read; -- T2
select * from test where value % 3 = 0; -- T1
select * from test where value % 3 = 0; -- T2
insert into test (id, value) values(3, 30); -- T1
insert into test (id, value) values(4, 42); -- T2
commit; -- T1
commit; -- T2
select * from test where value % 3 = 0; -- Either. Returns 3 => 30, 4 => 42

begin; set transaction isolation level serializable; -- T1
begin; set transaction isolation level serializable; -- T2
select * from test where value % 3 = 0; -- T1
select * from test where value % 3 = 0; -- T2
insert into test (id, value) values(3, 30); -- T1
insert into test (id, value) values(4, 42); -- T2
commit; -- T1
commit; -- T2. Prints out "ERROR: could not serialize access due to read/write dependencies among transactions"

begin; set transaction isolation level serializable; -- T1
select * from test; -- T1. Shows 1 => 10, 2 => 20
begin; set transaction isolation level serializable; -- T2
update test set value = value + 5 where id = 2; -- T2
commit; -- T2
begin; set transaction isolation level serializable; -- T3
select * from test; -- T3. Shows 1 => 10, 2 => 25
commit; -- T3
update test set value = 0 where id = 1; -- T1. Prints out "ERROR: could not serialize access due to read/write dependencies among transactions"
abort; -- T1. There's nothing else we can do, this transaction has failed

rmgr: Btree       len (rec/tot):     64/    64, tx:       1342, lsn: 2D/AAFFC9F0, prev 2D/AAFFC810, desc: INSERT_LEAF off 126, blkref #0: rel 1663/3101882/3105398 blk 4
rmgr: Heap        len (rec/tot):    485/   485, tx:       1342, lsn: 2D/AAFFCA30, prev 2D/AAFFC9F0, desc: INSERT off 10, blkref #0: rel 1663/3101882/3105391 blk 139

INSERT INTO public.test (id, data) VALUES (14, 'hoho');

{
  "change": [
    {
      "kind": "update",
      "schema": "public",
      "table": "test",
      "columnnames": ["id", "data" ],
      "columntypes": [ "integer", "text" ],
      "columnvalues": [ 1, "hoho"],
      "oldkeys": { "keynames": [ "id"],
        "keytypes": ["integer" ],
        "keyvalues": [1]
      }
    }
  ]
}

DROP TABLE IF EXISTS users;
CREATE TABLE users(id SERIAL PRIMARY KEY, name TEXT);

INSERT INTO users VALUES (100, 'Vonng');
INSERT INTO users VALUES (101, 'Xiao Wang');
DELETE FROM users WHERE id = 100;
UPDATE users SET name = 'Lao Wang' WHERE id = 101;

TABLE pg_replication_slots; -- 查
pg_create_logical_replication_slot(slot_name name, plugin name) -- 增
pg_drop_replication_slot(slot_name name) -- 删

pg_logical_slot_get_changes(slot_name name, ...)  -- 消费掉
pg_logical_slot_peek_changes(slot_name name, ...) -- 只查看不消费

-- 创建一个复制槽test_slot，使用系统自带的测试解码插件test_decoding，解码插件会在后面介绍
SELECT * FROM pg_create_logical_replication_slot('test_slot', 'test_decoding');

-- 重放上面的建表与增删改操作
-- DROP TABLE | CREATE TABLE | INSERT 1 | INSERT 1 | DELETE 1 | UPDATE 1

-- 读取复制槽test_slot中未消费的最新的变更事件流
SELECT * FROM  pg_logical_slot_get_changes('test_slot', NULL, NULL);
    lsn    | xid |                                data
-----------+-----+--------------------------------------------------------------------
 0/167C7E8 | 569 | BEGIN 569
 0/169F6F8 | 569 | COMMIT 569
 0/169F6F8 | 570 | BEGIN 570
 0/169F6F8 | 570 | table public.users: INSERT: id[integer]:100 name[text]:'Vonng'
 0/169F810 | 570 | COMMIT 570
 0/169F810 | 571 | BEGIN 571
 0/169F810 | 571 | table public.users: INSERT: id[integer]:101 name[text]:'Xiao Wang'
 0/169F8C8 | 571 | COMMIT 571
 0/169F8C8 | 572 | BEGIN 572
 0/169F8C8 | 572 | table public.users: DELETE: id[integer]:100
 0/169F938 | 572 | COMMIT 572
 0/169F970 | 573 | BEGIN 573
 0/169F970 | 573 | table public.users: UPDATE: id[integer]:101 name[text]:'Lao Wang'
 0/169F9F0 | 573 | COMMIT 573

-- 清理掉创建的复制槽
SELECT pg_drop_replication_slot('test_slot');

# 启动一个CDC客户端，连接数据库postgres，创建名为test_slot的槽，使用test_decoding解码插件，标准输出
pg_recvlogical \
	-d postgres \
	--create-slot --if-not-exists --slot=test_slot \
	--plugin=test_decoding \
	--start -f -

# 开启另一个会话，重放上面的建表与增删改操作
# DROP TABLE | CREATE TABLE | INSERT 1 | INSERT 1 | DELETE 1 | UPDATE 1

# pg_recvlogical输出结果
BEGIN 585
COMMIT 585
BEGIN 586
table public.users: INSERT: id[integer]:100 name[text]:'Vonng'
COMMIT 586
BEGIN 587
table public.users: INSERT: id[integer]:101 name[text]:'Xiao Wang'
COMMIT 587
BEGIN 588
table public.users: DELETE: id[integer]:100
COMMIT 588
BEGIN 589
table public.users: UPDATE: id[integer]:101 name[text]:'Lao Wang'
COMMIT 589

# 清理：删除创建的复制槽
pg_recvlogical -d postgres --drop-slot --slot=test_slot

BEGIN {事务标识}
table {模式名}.{表名} {命令INSERT|UPDATE|DELETE}  {列名}[{类型}]:{取值} ...
COMMIT {事务标识}

yum install wal2json11
apt install postgresql-11-wal2json

git clone https://github.com/michaelpq/pg_plugins && cd pg_plugins/decoder_raw
make && sudo make install

pg_recvlogical -d postgres --drop-slot --slot=test_slot
pg_recvlogical -d postgres --create-slot --if-not-exists --slot=test_slot \
	--plugin=wal2json --start -f -

{"change":[]}
{"change":[{"kind":"insert","schema":"public","table":"users","columnnames":["id","name"],"columntypes":["integer","text"],"columnvalues":[100,"Vonng"]}]}
{"change":[{"kind":"insert","schema":"public","table":"users","columnnames":["id","name"],"columntypes":["integer","text"],"columnvalues":[101,"Xiao Wang"]}]}
{"change":[{"kind":"delete","schema":"public","table":"users","oldkeys":{"keynames":["id"],"keytypes":["integer"],"keyvalues":[100]}}]}
{"change":[{"kind":"update","schema":"public","table":"users","columnnames":["id","name"],"columntypes":["integer","text"],"columnvalues":[101,"Lao Wang"],"oldkeys":{"keynames":["id"],"keytypes":["integer"],"keyvalues":[101]}}]}

pg_recvlogical -d postgres --drop-slot --slot=test_slot
pg_recvlogical -d postgres --create-slot --if-not-exists --slot=test_slot \
	--plugin=decoder_raw --start -f -

INSERT INTO public.users (id, name) VALUES (100, 'Vonng');
INSERT INTO public.users (id, name) VALUES (101, 'Xiao Wang');
DELETE FROM public.users WHERE id = 100;
UPDATE public.users SET id = 101, name = 'Lao Wang' WHERE id = 101;

pg_recvlogical -d <new_master_url> --slot=test_slot --plugin=decoder_raw --start -f - |
psql <old_master_url>

$ psql 'postgres://localhost:5432/postgres?replication=on&application_name=mocker'

vonng=# table pg_stat_replication ;
-[ RECORD 1 ]----+-----------------------------
pid              | 7218
usesysid         | 10
usename          | vonng
application_name | mocker
client_addr      | ::1
client_hostname  |
client_port      | 53420

dsn := "postgres://localhost:5432/postgres?application_name=cdc"
plugin := "test_decoding"
slot := "test_slot"

go run main.go postgres:///postgres?application_name=cdc test_decoding test_slot

package main

import (
	"log"
	"os"
	"time"

	"context"
	"github.com/jackc/pgx"
)

type Subscriber struct {
	URL    string
	Slot   string
	Plugin string
	Conn   *pgx.ReplicationConn
	LSN    uint64
}

// Connect 会建立到服务器的复制连接，区别在于自动添加了replication=on|1|yes|dbname参数
func (s *Subscriber) Connect() {
	connConfig, _ := pgx.ParseURI(s.URL)
	s.Conn, _ = pgx.ReplicationConnect(connConfig)
}

// ReportProgress 会向主库汇报写盘，刷盘，应用的进度坐标（消费者偏移量）
func (s *Subscriber) ReportProgress() {
	status, _ := pgx.NewStandbyStatus(s.LSN)
	s.Conn.SendStandbyStatus(status)
}

// CreateReplicationSlot 会创建逻辑复制槽，并使用给定的解码插件
func (s *Subscriber) CreateReplicationSlot() {
	if consistPoint, snapshotName, err := s.Conn.CreateReplicationSlotEx(s.Slot, s.Plugin); err != nil {
		log.Fatalf("fail to create replication slot: %s", err.Error())
	} else {
		log.Printf("create replication slot %s with plugin %s : consist snapshot: %s, snapshot name: %s",
			s.Slot, s.Plugin, consistPoint, snapshotName)
		s.LSN, _ = pgx.ParseLSN(consistPoint)
	}
}

// StartReplication 会启动逻辑复制（服务器会开始发送事件消息）
func (s *Subscriber) StartReplication() {
	if err := s.Conn.StartReplication(s.Slot, 0, -1); err != nil {
		log.Fatalf("fail to start replication on slot %s : %s", s.Slot, err.Error())
	}
}

// DropReplicationSlot 会使用临时普通连接删除复制槽（如果存在）,注意如果复制连接正在使用这个槽是没法删的。
func (s *Subscriber) DropReplicationSlot() {
	connConfig, _ := pgx.ParseURI(s.URL)
	conn, _ := pgx.Connect(connConfig)
	var slotExists bool
	conn.QueryRow(`SELECT EXISTS(SELECT 1 FROM pg_replication_slots WHERE slot_name = $1)`, s.Slot).Scan(&slotExists)
	if slotExists {
		if s.Conn != nil {
			s.Conn.Close()
		}
		conn.Exec("SELECT pg_drop_replication_slot($1)", s.Slot)
		log.Printf("drop replication slot %s", s.Slot)
	}
}

// Subscribe 开始订阅变更事件，主消息循环
func (s *Subscriber) Subscribe() {
	var message *pgx.ReplicationMessage
	for {
		// 等待一条消息, 消息有可能是真的消息，也可能只是心跳包
		message, _ = s.Conn.WaitForReplicationMessage(context.Background())
		if message.WalMessage != nil {
			DoSomething(message.WalMessage) // 如果是真的消息就消费它
			if message.WalMessage.WalStart > s.LSN { // 消费完后更新消费进度，并向主库汇报
				s.LSN = message.WalMessage.WalStart + uint64(len(message.WalMessage.WalData))
				s.ReportProgress()
			}
		}
		// 如果是心跳包消息，按照协议，需要检查服务器是否要求回送进度。
		if message.ServerHeartbeat != nil && message.ServerHeartbeat.ReplyRequested == 1 {
			s.ReportProgress() // 如果服务器心跳包要求回送进度，则汇报进度
		}
	}
}

// 实际消费消息的函数，这里只是把消息打印出来，也可以写入Redis，写入Kafka，更新统计信息，发送邮件等
func DoSomething(message *pgx.WalMessage) {
	log.Printf("[LSN] %s [Payload] %s", 
             pgx.FormatLSN(message.WalStart), string(message.WalData))
}

// 如果使用JSON解码插件，这里是用于Decode的Schema
type Payload struct {
	Change []struct {
		Kind         string        `json:"kind"`
		Schema       string        `json:"schema"`
		Table        string        `json:"table"`
		ColumnNames  []string      `json:"columnnames"`
		ColumnTypes  []string      `json:"columntypes"`
		ColumnValues []interface{} `json:"columnvalues"`
		OldKeys      struct {
			KeyNames  []string      `json:"keynames"`
			KeyTypes  []string      `json:"keytypes"`
			KeyValues []interface{} `json:"keyvalues"`
		} `json:"oldkeys"`
	} `json:"change"`
}

func main() {
	dsn := "postgres://localhost:5432/postgres?application_name=cdc"
	plugin := "test_decoding"
	slot := "test_slot"
	if len(os.Args) > 1 {
		dsn = os.Args[1]
	}
	if len(os.Args) > 2 {
		plugin = os.Args[2]
	}
	if len(os.Args) > 3 {
		slot = os.Args[3]
	}

	subscriber := &Subscriber{
		URL:    dsn,
		Slot:   slot,
		Plugin: plugin,
	}                                // 创建新的CDC客户端
	subscriber.DropReplicationSlot() // 如果存在，清理掉遗留的Slot

	subscriber.Connect()                   // 建立复制连接
	defer subscriber.DropReplicationSlot() // 程序中止前清理掉复制槽
	subscriber.CreateReplicationSlot()     // 创建复制槽
	subscriber.StartReplication()          // 开始接收变更流
	go func() {
		for {
			time.Sleep(5 * time.Second)
			subscriber.ReportProgress()
		}
	}()                                    // 协程2每5秒地向主库汇报进度
	subscriber.Subscribe()                 // 主消息循环
}

postgres=# table pg_stat_replication; -- 查看当前从库
-[ RECORD 1 ]----+------------------------------
pid              | 14082
usesysid         | 10
usename          | vonng
application_name | cdc
client_addr      | 10.1.1.95
client_hostname  |
client_port      | 56609
backend_start    | 2019-05-19 13:14:34.606014+08
backend_xmin     |
state            | streaming
sent_lsn         | 2D/AB269AB8     -- 服务端已经发送的消息坐标
write_lsn        | 2D/AB269AB8     -- 客户端已经执行完写入的消息坐标
flush_lsn        | 2D/AB269AB8     -- 客户端已经刷盘的消息坐标（不会丢失）
replay_lsn       | 2D/AB269AB8     -- 客户端已经应用的消息坐标（已经生效）
write_lag        |
flush_lag        |
replay_lag       |
sync_priority    | 0
sync_state       | async

postgres=# table pg_replication_slots;  -- 查看当前复制槽
-[ RECORD 1 ]-------+------------
slot_name           | test
plugin              | decoder_raw
slot_type           | logical
datoid              | 13382
database            | postgres
temporary           | f
active              | t
active_pid          | 14082
xmin                |
catalog_xmin        | 1371
restart_lsn         | 2D/AB269A80       -- 下次客户端重连时将从这里开始重放
confirmed_flush_lsn | 2D/AB269AB8       -- 客户端确认完成的消息进度

LogicalDecodeStartupCB startup_cb;
LogicalDecodeBeginCB begin_cb;
LogicalDecodeChangeCB change_cb;
LogicalDecodeTruncateCB truncate_cb;
LogicalDecodeCommitCB commit_cb;
LogicalDecodeMessageCB message_cb;
LogicalDecodeFilterByOriginCB filter_by_origin_cb;
LogicalDecodeShutdownCB shutdown_cb;