pg_duckpipe 深度解析：PostgreSQL 原生 HTAP 革命，单数据库搞定事务与分析

在数据库领域，事务型（OLTP）和分析型（OLAP）工作负载长期处于"分而治之"的状态。传统方案需要两套系统：MySQL/PostgreSQL 处理在线交易，ClickHouse/Doris 承载报表分析，中间通过 ETL 或 CDC 管道同步数据。这种架构不仅运维复杂，数据延迟也难以消除。

2026 年，一个名为 pg_duckpipe 的 PostgreSQL 扩展正在改变游戏规则。它通过 PostgreSQL 的 WAL（Write-Ahead Log）实时将行式存储同步到 DuckLake 列式表中，让你在单个数据库实例内同时运行事务和分析工作负载——无需 Kafka，无需 Debezium，无需外部编排器。

本文将从架构原理、核心特性、代码实战到生产部署，全面剖析这一 HTAP 领域的革命性方案。

一、为什么需要 HTAP？传统架构的痛点

1.1 Lambda 与 Kappa 架构的困境

在 pg_duckpipe 出现之前，企业处理混合负载主要有两条路径：

Lambda 架构：数据同时写入批处理层（Hadoop/Hive）和流处理层（Kafka+Storm），查询时合并两层结果。优点是容错性强，缺点是维护两套代码逻辑、数据一致性难以保证。

Kappa 架构：简化为纯流处理，所有数据通过 Kafka 传输。但流式系统的复杂性和运维成本依然存在，且对历史数据的重放能力有限。

这两种架构的共同问题是：数据必须离开数据库，经过漫长的管道才能用于分析。对于实时性要求高的业务（如风控、推荐、监控），秒级延迟都可能造成损失。

1.2 内置 HTAP 的探索

一些数据库开始尝试内置 HTAP 能力：

• TiDB：通过 TiKV（行存）+ TiFlash（列存）双存储引擎实现
• OceanBase：类似架构，通过 Raft 协议同步
• ClickHouse：支持有限的 UPDATE/DELETE，但并非真正的 OLTP

这些方案要么需要独立部署新数据库，要么在兼容性上有所妥协。而 PostgreSQL 用户的需求很直接：能否在我现有的 PostgreSQL 上，零改造获得分析能力？

pg_duckpipe 正是为此而生。

二、pg_duckpipe 核心架构

2.1 整体设计理念

pg_duckpipe 的核心思想是：利用 PostgreSQL 现有的 WAL 机制，实时捕获数据变更，写入 DuckLake 列式存储。

┌─────────────────────────────────────────────────────────────────┐
│                     PostgreSQL Instance                         │
│                                                                 │
│  ┌──────────────┐    WAL     ┌──────────────────────────────┐  │
│  │              │ ────────▶  │        pg_duckpipe           │  │
│  │  Heap Table  │            │  (Background Worker)         │  │
│  │  (Row Store) │            │                              │  │
│  │              │            │  ┌────────────────────────┐  │  │
│  └──────────────┘            │  │   DuckLake Table       │  │  │
│                              │  │   (Columnar/Parquet)   │  │  │
│                              │  └────────────────────────┘  │  │
│                              └──────────────────────────────┘  │
└─────────────────────────────────────────────────────────────────┘

关键优势：

1. 零外部依赖：不需要 Kafka、Debezium、Flink 等组件
2. 事务一致性：WAL 本身就是数据库持久化的核心，天然保证一致性
3. 透明集成：应用层无需修改，DuckLake 表自动跟随源表更新
4. 存储分离：DuckLake 可以存储在本地磁盘、S3、GCS 等对象存储上

2.2 DuckLake：列式存储的秘密武器

DuckLake 是 DuckDB 生态中的湖仓格式，具有以下特点：

• 列式存储：分析查询只读取需要的列，IO 放大降低 10-100 倍
• Parquet 格式：与 Spark、Trino、Snowflake 等生态无缝兼容
• 向量化执行：利用 CPU SIMD 指令，单核吞吐量提升 5-10 倍
• 压缩效率：相比行存，压缩比可达 5:1 到 10:1

-- DuckLake 表的物理结构示例
-- 每个分区对应一个 Parquet 文件
ducklake/
├── orders_ducklake/
│   ├── _ducklake_metadata.json
│   ├── part-001.parquet  -- 2026-04-01 数据
│   ├── part-002.parquet  -- 2026-04-02 数据
│   └── part-003.parquet  -- 2026-04-03 数据

2.3 WAL 捕获机制

pg_duckpipe 通过 PostgreSQL 的 pgoutput 逻辑解码插件捕获变更：

// 简化的 WAL 解码流程
static void
parse_wal_change(LogicalDecodingContext *ctx, ReorderBufferChange *change)
{
    // 1. 解析 RELATION 消息，获取表结构
    // 2. 解析 INSERT/UPDATE/DELETE 的 Tuple 数据
    // 3. 写入 DuckDB 的 buffer table（可溢出到磁盘）
    // 4. 定期 flush 到 Parquet 文件
}

关键设计点：

• 双缓冲机制：WAL 解码写入内存 buffer，后台线程批量 flush
• Exactly-Once 语义：通过 LSN 双层去重，保证崩溃恢复后无重复
• 背压控制：当 DuckLake 写入落后时，WAL 解码自动降速

三、核心特性深度剖析

3.1 透明查询路由：真正的 HTAP 体验

这是 pg_duckpipe 最具革命性的特性。你无需修改任何 SQL，系统会自动判断查询类型并路由到合适的存储引擎。

-- 启用自动路由
SET duckpipe.query_routing = 'auto';

-- 这是一个典型的 OLTP 查询（点查）
-- 自动路由到 Heap Table，延迟 < 1ms
SELECT * FROM orders WHERE order_id = 12345;

-- 这是一个典型的 OLAP 查询（聚合扫描）
-- 自动路由到 DuckLake 列存，吞吐量提升 10-50 倍
SELECT 
    customer_id,
    DATE_TRUNC('month', order_date) AS month,
    SUM(total_amount) AS revenue,
    COUNT(*) AS orders
FROM orders
WHERE order_date >= '2026-01-01'
GROUP BY customer_id, DATE_TRUNC('month', order_date);

路由模式详解：

auto 模式的判断逻辑：

// 简化的路由判断伪代码
bool should_route_to_ducklake(Query *query, Relation relation)
{
    // 1. 检查是否有主键等值条件
    if (has_pk_equality_condition(query)) {
        return false;  // 点查，走 Heap Table
    }
    
    // 2. 检查扫描行数估算
    if (estimated_rows < 1000) {
        return false;  // 小扫描，走 Heap Table
    }
    
    // 3. 检查是否有聚合操作
    if (has_aggregates(query)) {
        return true;   // 聚合，走 DuckLake
    }
    
    // 4. 检查扫描字段比例
    if (scan_columns_ratio < 0.3) {
        return true;   // 列裁剪明显，走列存
    }
    
    return false;
}

3.2 Append 同步模式：内置 SCD Type 2

传统的 CDC 系统通常采用 upsert 模式，保持目标表与源表一致。但很多场景需要保留历史版本：

• 审计追踪：谁在什么时候修改了什么数据
• 事件溯源：从变更日志重建任意时刻的状态
• 数据分析：分析数据变化趋势

pg_duckpipe 的 append 模式实现了开箱即用的 SCD Type 2（Slowly Changing Dimension Type 2）：

-- 添加表时指定 append 模式
SELECT duckpipe.add_table('public.customers', sync_mode => 'append');

-- 源表操作
INSERT INTO customers(id, name, email) VALUES (1, 'Alice', 'alice@old.com');
UPDATE customers SET email = 'alice@new.com' WHERE id = 1;
DELETE FROM customers WHERE id = 1;

-- DuckLake 目标表自动记录所有变更版本
SELECT * FROM customers_ducklake ORDER BY _duckpipe_lsn;

输出结果：

 id | name  | email          | _duckpipe_op | _duckpipe_lsn
----+-------+----------------+--------------+---------------
  1 | Alice | alice@old.com  | I            |      10485800
  1 | Alice | alice@new.com  | U            |      10486920
  1 | Alice | alice@new.com  | D            |      10487104

元数据字段：

• _duckpipe_op：操作类型（I=INSERT, U=UPDATE, D=DELETE）
• _duckpipe_lsn：WAL 日志序列号，天然有序

无主键表支持：

append 模式的另一个优势是支持无主键表：

-- 日志表通常没有主键
CREATE TABLE access_logs (
    request_time TIMESTAMP,
    client_ip INET,
    path TEXT,
    status_code INT
);

-- upsert 模式无法同步无主键表
-- append 模式完美支持
SELECT duckpipe.add_table('public.access_logs', sync_mode => 'append');

3.3 Fan-In 流式聚合：多源统一分析

现代企业通常有多个数据库实例（分库分表、多地域部署），分析时需要汇总全局数据。传统方案需要复杂的 ETL 或数据仓库同步。

pg_duckpipe 的 Fan-In 特性允许将多个源数据库的同一张表聚合到一张 DuckLake 表：

-- 创建同步组，连接不同的源数据库
SELECT duckpipe.create_group('us_prod',
    conninfo => 'host=us-prod.example.com port=5432 dbname=orders user=replicator password=xxx');
    
SELECT duckpipe.create_group('eu_prod',
    conninfo => 'host=eu-prod.example.com port=5432 dbname=orders user=replicator password=xxx');

-- 将两个数据库的 orders 表同步到同一张 DuckLake 表
SELECT duckpipe.add_table('public.orders', sync_group => 'us_prod');
SELECT duckpipe.add_table('public.orders', sync_group => 'eu_prod', fan_in => true);

-- 查询全局数据，_duckpipe_source 标识来源
SELECT 
    _duckpipe_source,
    DATE_TRUNC('day', order_date) AS day,
    COUNT(*) AS orders,
    SUM(total_amount) AS revenue
FROM orders_ducklake
WHERE order_date >= CURRENT_DATE - INTERVAL '7 days'
GROUP BY _duckpipe_source, DATE_TRUNC('day', order_date)
ORDER BY day, _duckpipe_source;

Fan-In 的智能设计：

1. Parquet 文件级分区：每个 _duckpipe_source 对应独立的 Parquet 文件，查询过滤时可直接跳过无关文件
2. 源级别隔离：DELETE、TRUNCATE、resync 操作按源隔离，互不影响
3. 水平扩展：添加新源不影响已有源的同步性能

3.4 分区表自动检测

PostgreSQL 分区表是处理大数据量的常用手段。pg_duckpipe 自动检测分区表结构，只需添加父表即可：

-- 创建按月分区的订单表
CREATE TABLE orders (
    order_id BIGSERIAL,
    customer_id INT,
    order_date DATE,
    total_amount DECIMAL(10,2)
) PARTITION BY RANGE (order_date);

CREATE TABLE orders_2026_01 PARTITION OF orders
    FOR VALUES FROM ('2026-01-01') TO ('2026-02-01');
CREATE TABLE orders_2026_02 PARTITION OF orders
    FOR VALUES FROM ('2026-02-01') TO ('2026-03-01');
CREATE TABLE orders_2026_03 PARTITION OF orders
    FOR VALUES FROM ('2026-03-01') TO ('2026-04-01');

-- 只需添加父表，所有子分区自动同步
SELECT duckpipe.add_table('public.orders');

-- DuckLake 目标是统一的视图，无需处理分区逻辑
SELECT COUNT(*) FROM orders_ducklake WHERE order_date >= '2026-01-15';

3.5 分层配置体系

不同表的同步需求可能不同：热表需要更频繁的 flush，冷表可以降低频率节省资源。pg_duckpipe 提供 4 层配置层级：

全局默认 → 组级覆盖 → 表级覆盖 → 硬编码默认

-- 设置全局默认（10秒 flush 一次）
SELECT duckpipe.set_config('flush_interval_ms', '10000');

-- 为高吞吐组覆盖（2秒 flush）
SELECT duckpipe.set_group_config('high_throughput', 'flush_interval_ms', '2000');

-- 为特定表覆盖（批量阈值 50000 行）
SELECT duckpipe.set_table_config('public.orders', 'flush_batch_threshold', '50000');

常用配置参数：

3.6 Schema DDL 自动传播

表结构变更（DDL）是 CDC 系统的噩梦。pg_duckpipe 通过 WAL 中的 RELATION 消息自动检测和传播 Schema 变更：

-- 添加新列
ALTER TABLE orders ADD COLUMN discount DECIMAL(5,2);
-- DuckLake 目标表自动添加新列，后续数据写入新列

-- 重命名列
ALTER TABLE orders RENAME COLUMN discount TO discount_rate;
-- DuckLake 目标表自动重命名

-- 删除列
ALTER TABLE orders DROP COLUMN discount_rate;
-- DuckLake 目标表自动删除列

当前限制：

• ALTER COLUMN TYPE 被阻止（防止 Parquet 文件中类型不一致）
• 近期将支持安全 widening（如 INT → BIGINT）

3.7 可观测性：内置 Metrics

pg_duckpipe 提供 duckpipe.metrics() 函数，返回实时监控数据：

SELECT duckpipe.metrics();

输出示例：

{
  "tables": {
    "public.orders": {
      "queued_changes": 128,
      "total_queued_changes": 584320,
      "flush_count": 1247,
      "flush_duration_ms": 34500,
      "avg_row_bytes": 142,
      "is_backpressured": false
    },
    "public.customers": {
      "queued_changes": 45,
      "total_queued_changes": 23456,
      "flush_count": 892,
      "flush_duration_ms": 8200,
      "avg_row_bytes": 98,
      "is_backpressured": false
    }
  },
  "groups": {
    "default": {
      "total_queued_changes": 607776,
      "is_backpressured": false
    }
  }
}

关键指标解读：

• queued_changes：当前排队等待 flush 的变更数，持续增长说明写入落后
• flush_count / flush_duration_ms：可以计算平均 flush 耗时
• is_backpressured：是否触发背压，true 时 WAL 解码会降速

四、代码实战：从零搭建 HTAP 系统

4.1 环境准备

系统要求：

• PostgreSQL 17+（推荐 18.x）
• Linux / macOS（Windows 支持计划中）
• 足够的磁盘空间存储 Parquet 文件

安装 pg_duckpipe：

# 从源码编译
git clone https://github.com/relytcloud/pg_duckpipe.git
cd pg_duckpipe
make
make install

# 或者使用预编译包（部分平台）
# 下载对应 PostgreSQL 版本的 .so 文件

修改 postgresql.conf：

# 启用逻辑复制（pg_duckpipe 依赖此功能）
wal_level = logical
max_replication_slots = 10
max_wal_senders = 10

# 加载 pg_duckpipe 扩展
shared_preload_libraries = 'pg_duckpipe'

# pg_duckpipe 配置
pg_duckpipe.enabled = true
pg_duckpipe.ducklake_path = '/data/ducklake'  # DuckLake 存储路径

重启 PostgreSQL：

pg_ctl restart

4.2 创建扩展和同步表

-- 创建扩展
CREATE EXTENSION pg_duckpipe;

-- 创建测试表
CREATE TABLE products (
    id SERIAL PRIMARY KEY,
    name TEXT NOT NULL,
    category VARCHAR(50),
    price DECIMAL(10,2),
    stock INT,
    created_at TIMESTAMP DEFAULT NOW()
);

-- 插入测试数据
INSERT INTO products (name, category, price, stock) VALUES
    ('iPhone 18', 'Electronics', 999.00, 100),
    ('MacBook Pro M6', 'Electronics', 2499.00, 50),
    ('AirPods Pro 4', 'Electronics', 249.00, 500),
    ('Nike Air Max', 'Footwear', 180.00, 200),
    ('Levi''s 501', 'Apparel', 79.99, 300);

-- 添加表到同步
SELECT duckpipe.add_table('public.products');

-- 查看 DuckLake 目标表
SELECT * FROM products_ducklake;

4.3 验证实时同步

-- 插入新数据
INSERT INTO products (name, category, price, stock) VALUES
    ('Galaxy S26', 'Electronics', 899.00, 150);

-- 等待 flush（默认 5 秒）
SELECT pg_sleep(6);

-- 验证同步
SELECT * FROM products_ducklake WHERE name = 'Galaxy S26';

-- 更新数据
UPDATE products SET price = 849.00 WHERE name = 'Galaxy S26';

-- 验证同步
SELECT pg_sleep(6);
SELECT * FROM products_ducklake WHERE name = 'Galaxy S26';

4.4 测试透明查询路由

-- 启用自动路由
SET duckpipe.query_routing = 'auto';

-- 点查（走 Heap Table）
EXPLAIN (ANALYZE, BUFFERS) 
SELECT * FROM products WHERE id = 1;
-- 注意输出中的 Index Scan，确认走的是源表

-- 聚合查询（走 DuckLake）
EXPLAIN (ANALYZE, BUFFERS)
SELECT category, COUNT(*), SUM(price * stock) AS inventory_value
FROM products
GROUP BY category;
-- 注意输出中的 DuckLake 扫描

4.5 配置 S3 存储

生产环境通常将 DuckLake 存储在对象存储上：

-- 配置 S3 访问
SELECT duckpipe.set_config('ducklake_storage', 's3');
SELECT duckpipe.set_config('s3_endpoint', 'https://s3.amazonaws.com');
SELECT duckpipe.set_config('s3_bucket', 'my-ducklake-bucket');
SELECT duckpipe.set_config('s3_access_key', 'AKIAIOSFODNN7EXAMPLE');
SELECT duckpipe.set_config('s3_secret_key', 'wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY');
SELECT duckpipe.set_config('s3_region', 'us-east-1');

-- 后续创建的 DuckLake 表会自动存储到 S3
SELECT duckpipe.add_table('public.orders');

五、性能优化与最佳实践

5.1 性能基准测试

在 1000 万行订单数据上的测试结果：

关键结论：

• 点查和 OLTP 场景应使用 auto 模式，保持走 Heap Table
• 分析查询在 DuckLake 上有 30-60 倍加速
• 列存对宽表（100+ 列）的优化更明显，可达 100 倍以上

5.2 Flush 策略调优

-- 高频小批量写入（如订单系统）
SELECT duckpipe.set_table_config('public.orders', 
    'flush_interval_ms', '2000');      -- 2秒 flush
SELECT duckpipe.set_table_config('public.orders', 
    'flush_batch_threshold', '5000');  -- 5000 行触发

-- 低频大批量写入（如日志）
SELECT duckpipe.set_table_config('public.access_logs', 
    'flush_interval_ms', '30000');     -- 30秒 flush
SELECT duckpipe.set_table_config('public.access_logs', 
    'flush_batch_threshold', '100000'); -- 10万行触发

5.3 存储优化

Parquet 文件大小：

过小的文件会增加元数据开销，过大的文件影响查询并行度。建议控制在 128MB-512MB 之间。

-- 通过 flush_byte_threshold 控制文件大小
SELECT duckpipe.set_config('flush_byte_threshold', '268435456');  -- 256MB

压缩配置：

-- DuckLake 支持 zstd 和 gzip 压缩
SELECT duckpipe.set_config('compression', 'zstd');
SELECT duckpipe.set_config('compression_level', '3');  -- 1-22，越高压缩率越高但越慢

5.4 监控与告警

建议监控以下指标并设置告警：

-- 创建监控视图
CREATE VIEW duckpipe_health AS
SELECT 
    t.table_name,
    t.queued_changes,
    t.total_queued_changes,
    t.flush_count,
    CASE 
        WHEN t.is_backpressured THEN 'BACKPRESSURED'
        WHEN t.queued_changes > 100000 THEN 'HIGH_LATENCY'
        ELSE 'HEALTHY'
    END AS status
FROM jsonb_to_recordset(
    (SELECT duckpipe.metrics()::jsonb -> 'tables')
) AS t(table_name TEXT, queued_changes BIGINT, total_queued_changes BIGINT, 
       flush_count BIGINT, is_backpressured BOOLEAN);

-- 查询健康状态
SELECT * FROM duckpipe_health;

六、与其他方案对比

pg_duckpipe vs pg_duckdb：

两者都是 PostgreSQL + DuckDB 的集成方案，但定位不同：

• pg_duckdb：将 DuckDB 嵌入 PostgreSQL 进程，用于加速即席分析查询，数据仍存储在 Heap Table
• pg_duckpipe：真正的 CDC 管道，将数据复制到独立的 DuckLake 列存，支持多源聚合、历史版本追踪

选择建议：

• 只需要加速分析查询 → pg_duckdb
• 需要 HTAP、多源聚合、审计追踪 → pg_duckpipe

七、生产部署建议

7.1 高可用架构

┌─────────────────────────────────────────────────────────────┐
│                      Application Layer                      │
└─────────────────────┬───────────────────────────────────────┘
                      │
┌─────────────────────▼───────────────────────────────────────┐
│                   HAProxy / PgBouncer                        │
└─────────┬───────────────────────────────────┬───────────────┘
          │                                   │
┌─────────▼─────────┐               ┌─────────▼─────────┐
│   PostgreSQL      │               │   PostgreSQL      │
│   Primary         │ ── WAL ─────▶ │   Replica         │
│   + pg_duckpipe   │               │   + pg_duckpipe   │
└─────────┬─────────┘               └─────────┬─────────┘
          │                                   │
┌─────────▼─────────────────────────────────▼───────────────┐
│              Shared DuckLake Storage (S3/NFS)              │
└───────────────────────────────────────────────────────────┘

关键点：

• Primary 和 Replica 都部署 pg_duckpipe
• DuckLake 存储使用共享存储（S3 或 NFS）
• 故障切换时无需重新同步历史数据

7.2 容量规划

存储空间：

• DuckLake 数据量 ≈ 源表大小 × 压缩比（通常 0.2-0.3）
• 预留 2 倍空间用于 Compaction 和临时文件

内存：

• pg_duckpipe 进程约占用 200-500MB
• Buffer 空间可配置（默认 256MB 可溢出）

CPU：

• WAL 解码是单线程，建议每 10000 TPS 预留 1 核
• DuckDB 执行可并行，建议预留 2-4 核

7.3 运维 Checklist

每日检查：

• duckpipe.metrics() 中 queued_changes 是否正常
• 磁盘空间是否充足
• 错误日志中是否有 pg_duckpipe 相关报错

每周检查：

• 同步延迟趋势
• Parquet 文件大小分布
• 查询路由命中率

每月检查：

• 评估是否需要 Compaction
• 版本更新与安全补丁
• 性能基准测试对比

八、未来展望

pg_duckpipe 团队正在积极开发以下特性：

1. JSONB → VARIANT 映射：将 PostgreSQL 的 jsonb 列映射到 DuckDB 的 VARIANT 类型，提升半结构化数据分析能力

2. 自动 Compaction：合并小 Parquet 文件，优化查询性能

3. 数据保留策略：基于时间的自动数据清理

4. PostgreSQL 16 及更早版本支持：扩大兼容范围

5. ALTER COLUMN TYPE 支持：安全处理类型变更

总结

pg_duckpipe 代表了数据库架构的一次重要演进：让 OLTP 和 OLAP 在同一个数据库实例内和谐共存。它不是简单地嵌入一个分析引擎，而是构建了一套完整的 CDC 管道、列式存储、智能路由系统。

对于 PostgreSQL 用户来说，这意味着：

• 零迁移成本：在现有数据库上开启 HTAP 能力
• 零运维负担：没有额外的 Kafka、Debezium 需要维护
• 零查询改造：透明路由让应用无感知

如果你的业务正在经历数据分析的瓶颈，pg_duckpipe 值得一试。它可能是你一直在寻找的那个"让 PostgreSQL 变得更强大"的工具。

项目地址：https://github.com/relytcloud/pg_duckpipe
官方文档：https://pgducklake.select/

字数：约 6500 字