DuckDB 查询执行引擎深度实战：从 Vector/DataChunk 内存布局到推送式 Pipeline 与并行聚合的工程全解

DuckDB 向量化执行引擎深度实战：从 Vector/DataChunk 内存布局到推送式 Pipeline 与并行聚合的工程全解

一、为什么 DuckDB 值得单独拆开看

过去五年，OLAP 世界的注意力几乎全被分布式引擎吸走：ClickHouse、Dremio、Trino、Snowflake。但真实工程里有一类查询长期被这两极挤压——数据量在几 GB 到几百 GB 之间，跑在单机上，却又远超 pandas 能舒适承载的规模。你把数据扔进 Postgres 太慢，扔进 Spark 又重得离谱。

DuckDB 填补的正是这个"单机 OLAP 甜点区"。它不是一个简化版数据库，而是一台完整实现现代分析型执行技术的引擎：列式向量化执行、推送式 Pipeline、并行 hash 聚合、radix 分区 hash join、自适应过滤器重排、外存溢出。理解它的内部结构，等于把过去二十年列存研究的工程结论一次性过一遍。

二、数据模型的地基：Vector 与 DataChunk

DuckDB 的执行单元不是行，也不是列的全量数组，而是被切成固定大小的块：

• STANDARD_VECTOR_SIZE = 2048，即一个 Vector 默认承载 2048 个值；
• 若干列组成一个 DataChunk，DataChunk::SetCardinality(n) 声明该块有效行数；
• Vector 内部持有 data 指针、validity mask（位图）和一个 VectorBuffer 链表。

关键在于 Vector 不是只有一种形态。它是带标签的联合：

enum class VectorType : uint8_t {
  FLAT_VECTOR,        // 物理连续的原生数组
  CONSTANT_VECTOR,    // 整块只有一个值
  DICTIONARY_VECTOR,  // sel 索引 + 子向量
  SEQUENCE_VECTOR,    // start + increment，不占内存
  FSST_VECTOR         // 字符串压缩编码向量
};

这个设计是纯粹的缓存经济学。考虑 WHERE region = 'CN'：过滤后原 Vector 不需要复制，只需包一层 DICTIONARY_VECTOR，用 selection vector 指向幸存的下标。SEQUENCE_VECTOR 让 row_number() 或 generate_series 的基数列在不落盘的情况下参与后续算子，把一列的物化成本压到零。

统一向量格式：消灭笛卡尔组合爆炸

如果每个算子都要处理 5 种 Vector 类型的两两组合，代码规模会爆炸。DuckDB 的解法是 UnifiedVectorFormat：

struct UnifiedVectorFormat {
  const ValidityMask *validity;  // 统一的 NULL 位图指针
  const SelectionVector *sel;    // 可选的下标间接层
  void *data;                    // 实际数据指针
};

算子执行前调用 Vector::ToUnifiedFormat()，把任意 Vector 归一化为"数据指针 + NULL 位图 + 可选 sel"三元组。这样内核函数只需要面对极少数情形，而代价仅是一次间接寻址——在有分支预测和硬件预取的情况下，这远比分支到十几个特化路径便宜。

NULL 的处理也值得单独说。DuckDB 不用哨兵值，而是独立的 ValidityMask（一位一值）。这让 SUM(x) 可以在有 NULL 时先做一次 mask 检查，无 NULL 时走完全无分支的快速路径——AllValid() 判断是一次性的，而不是每行一次。

三、表达式执行：Selection Vector 与自适应过滤

表达式层有一个容易忽略的事实：过滤操作不改变数据，只改变选择集。

// ExpressionExecutor::Select 的语义
idx_t ExpressionExecutor::Select(DataChunk &input, SelectionVector &sel) {
  // 返回满足条件的行数，sel 中保存幸存下标
}

PhysicalTableScan 会把可下推的谓词拆成多个 filter，逐个对输入块执行 Select，每次都就地缩短 selection vector。结果是：第一个 filter 扫 2048 行，第二个可能只扫 40 行，第三个性只扫 3 行。数据一次都没被复制。

更进一步，DuckDB 实现了自适应过滤器重排（AdaptiveFilter）。引擎会给每个 filter 记录一个 1.5 秒的时间窗口，统计它在最近若干批块上的选择率，然后动态把选择率最高的 filter 排到最前面。这是一次很务实的妥协：优化器没有列的直方图时，与其猜，不如在执行中测。

-- 一个能观察到该行为的查询
SELECT * FROM events
WHERE user_id % 977 = 3        -- 便宜但选择性差
  AND payload LIKE '%kernel%'  -- 昂贵但选择性强

跑一段时间后，EXPLAIN ANALYZE 里 filter 的顺序会和 SQL 书写顺序不同。这不是 bug，是引擎在自我修正。

四、从火山模型到推送式 Pipeline

DuckDB 在 0.7.0 做了一次架构级重构：把经典的 pull-based 火山模型换成了 push-based Pipeline 模型。

火山模型的问题很典型：Next() 递归调用链意味着每个算子的状态机要靠栈帧维持，并行化时你很难回答"这个算子的哪一部分可以被多个线程同时驱动"。Pipeline 模型把算子链切成若干段，每段是一个 Pipeline：

Pipeline 0: TableScan -> Filter -> Projection  (Source)
Pipeline 1: HashAggregate Build                (Source -> Sink)
Pipeline 2: HashAggregate Scan -> Result       (Inter-Pipeline)

调度器把 Pipeline 包装成 Event，通过依赖计数器驱动：Source 产出的块被 PipelineExecutor 主动推给下游，直到 Source 耗尽或 Sink 无法接收。

这个改动的直接收益是并行度与算子解耦。同一张表的不同 RowGroup 可以被不同线程并行扫描，输出到同一条 Pipeline；而 HashAggregate 这类 Sink 算子可以声明"我支持 N 个 Local 状态 + 1 个 Global 状态"，调度器据此自动扩线程。

五、并行聚合：Local/Global 状态分离

这是 DuckDB 并行设计里最值得抄的一段。以 hash 聚合为例：

struct HashAggregateGlobalSinkState {  // 全局唯一，合并结果
  RadixPartitionedHashTable *ht;
  mutex lock;
};
struct HashAggregateLocalSinkState {   // 每线程一份，无锁写入
  LocalSinkState  *local_ht;
  TupleDataCollection *unprocessed;
};

每个线程把输入按 hash 的若干高位做 radix 分区，写入自己私有的哈希表；扫描阶段再由全局状态按分区合并。合并之所以高效，是因为分区后不同线程的同 key 一定落在同一分区，合并时可以做到分区级并行、分区内串行，锁粒度从"整个哈希表"降到"一个分区"。

GROUP BY 基数极大时，DuckDB 会切换到两层 radix：先按高位分区落盘，再逐分区加载做内存聚合。这就是它能跑"内存装不下的聚合"的原理。

六、实战：把引擎行为变成可调的参数

理解结构不是为了炫技，是为了在查询慢的时候知道该动哪个旋钮。下面是一套完整的生产级配置：

import duckdb

con = duckdb.connect()
con.execute("PRAGMA threads=8")                    # 与物理核对齐
con.execute("PRAGMA memory_limit='12GB'")          # 触发溢出的阈值
con.execute("PRAGMA temp_directory='/fast/nvme/spill'")  # 溢出目录必须在 SSD
con.execute("PRAGMA enable_progress_bar=true")
con.execute("SET enable_profiling='json'")
con.execute("SET profiling_output='/tmp/plan.json'")

con.execute("""
  CREATE VIEW logs AS
  SELECT * FROM read_parquet(
    's3://bucket/logs/**/*.parquet',
    hive_partitioning = true
  )
""")

# 关键：先让引擎统计列存 zone map，再做下推决策
con.execute("ANALYZE")

df = con.execute("""
  SELECT dt, region,
         count(*) AS cnt,
         quantile_cont(latency_ms, 0.99) AS p99
  FROM logs
  WHERE dt BETWEEN '2026-09-01' AND '2026-09-30'
  GROUP BY dt, region
  ORDER BY p99 DESC
  LIMIT 20
""").df()

几个实战观点：

1. memory_limit 不要设成物理内存的 100%。 DuckDB 的缓冲区管理依赖预留空间做溢出决策，设满会导致 OOM 而不是优雅降级。经验值是物理内存的 70-75%。
2. temp_directory 的位置决定溢出代价。 默认是数据库文件同目录，如果你跑的是 :memory: 实例，默认落到系统临时目录。生产环境必须显式指向 NVMe。
3. ANALYZE 在列存场景下性价比极高。 DuckDB 的 Parquet reader 会读 zone map（每列的 min/max），有统计信息后过滤器能整块跳过 RowGroup。这一步常常带来数量级差异。
4. 观察 EXPLAIN ANALYZE 里的 Rows Scanned vs Rows Filtered。 如果两者接近，说明下推没生效，检查谓词是否作用在表达式而非裸列上。

-- 定位 Pipeline 级瓶颈
EXPLAIN ANALYZE
SELECT region, sum(amount) FROM sales
GROUP BY region ORDER BY 2 DESC;

输出中每个算子会带 Timing 和 Cardinality。如果 HASH_GROUP_BY 的 cardinality 远大于输出行数，说明分组基数被低估，考虑先用更高选择性的谓词剪枝。

七、什么时候不该用 DuckDB

说清楚边界比吹能力更有价值。

• 高并发短查询：DuckDB 的并行模型是"一条查询吃满所有线程"，为吞吐设计。上百 QPS 的 serving 场景它会把 CPU 打满，这是设计取舍而非缺陷。
• 需要在线事务写：它是 append-oriented 的分析引擎，单行更新走的是删除+重写的路径。
• 数据远超单机：超过几 TB 就该考虑分布式，或者用 DuckDB 做分区级分批处理。
• 需要严格的 MVCC 隔离级别：这是分析引擎，不是事务型数据库。

八、结语

DuckDB 真正的技术价值不在于"嵌入式"这个部署形态，而在于它把过去二十年列存研究的结论——向量化执行、selection vector 复用、radix 分区并行、自适应优化——压缩进了一个几十 MB 的库里，并且全部默认开启。

读它的执行引擎，你实际上在读一部现代 OLAP 的工程教科书。而它最有启发性的一点或许是：在没有统计信息时，不要猜，去测。自适应过滤器重排就是这句话最直接的实现。