金仓数据库征文-金仓KES数据同步优化实践：逻辑解码与增量同步

# 检查WAL日志级别
ksql -U system -d testdb -c "SHOW wal_level;"
# 验证逻辑解码插件
ls $KINGBASE_HOME/lib/kingbase/decoding_plugins/

wal_level = logical        # 启用逻辑解码
max_replication_slots = 8  # 每个同步任务占用一个slot
max_wal_senders = 16       # 并发同步连接数

SELECT * FROM pg_create_logical_replication_slot(
    'kes_sync_slot', 
    'mpp_decoder'
);

./kb_dump_logical -h 10.1.1.10 -p 54321 -U sync_user \
  -d src_db -s kes_sync_slot -f ./changes.sql \
  --start-lsn 0/1A3B5C7 -v

-- 源端生成校验码
SELECT md5(array_agg(md5((t.*)::text)::text) 
FROM my_table t;

-- 目标端验证
SELECT kes_compare_hash(
    'md5_hash_value', 
    'public.my_table'
);

# 调整WAL发送器参数
wal_sender_timeout = 60s
wal_keep_segments = 1024

// JDBC批量写入示例
conn.setAutoCommit(false);
PreparedStatement pstmt = conn.prepareStatement(insertSQL);
for (DataRecord record : recordList) {
    pstmt.setObject(1, record.getValue());
    pstmt.addBatch();
    if (i % 5000 == 0) {
        pstmt.executeBatch();
        conn.commit();
    }
}

CREATE TRIGGER sync_version_trigger 
BEFORE UPDATE ON order_table
FOR EACH ROW EXECUTE FUNCTION 
    update_version_func();

class CheckpointManager:
    def save_lsn(self, slot_name, lsn):
        self.redis_client.hset(
            'sync_checkpoints', 
            slot_name, 
            lsn
        )

-- 主库创建级联副本  
SELECT * FROM pg_create_physical_replication_slot('bj_slot');  
ALTER SYSTEM SET synchronous_standby_names = 'sh_slot,bj_slot';  

# 触发手动切换  
repmgr standby switchover \
  --siblings-follow \
  --force

# prometheus.yml配置示例  
- job_name: 'kes_sync'  
  static_configs:  
    - targets: ['10.1.1.10:9187']  
  params:  
    db: [sync_monitor]  

# Logstash管道配置  
input { jdbc { jdbc_driver_library => "/opt/kes/odbc/lib/kingbase.so" } }  
filter { grok { match => { "message" => "%{TIMESTAMP_ISO8601:log_time} %{LOGLEVEL:level}" } } }