6.2 列式扫描：裁剪、Pruning、向量化与延迟物化

分析员只需要三列和少量记录，执行器却读遍整张宽表；阿花决定把“少读数据”拆成可以测量的几种机制。

“谓词下推”经常被当作一个万能词，实际至少包含四件不同的事：减少 columns、用 metadata 排除 data units、在 reader 内执行 filter、延迟加载/物化其他 columns。只有把它们分开，才能解释 query 为什么仍读取很多 bytes。

四层减少工作

1. Projection pushdown / column pruning

query 只需要 item_type 和 value_cents，reader 不读取其他 column chunks。

SELECT item_type, SUM(value_cents)
FROM items
GROUP BY item_type;

这是最稳定的列式收益，但 nested schema、SELECT *、UDF 与 hidden columns 可能扩大 required set。

2. Partition pruning

dataset 路径或 catalog partition expression：

year=2025/month=01/part-....parquet
year=2025/month=02/part-....parquet

WHERE year=2025 AND month=2 可在打开 file 前排除其他 partitions。partition field 来自路径/catalog，不等于 file 内 column statistics。

高基数 partition 会制造大量小 files/directories/metadata；低基数又可能 pruning 不够。partition key 要匹配常见 filter 与 data lifecycle。

3. Metadata pruning

若 row group 的 value_cents.max < 50000，predicate value_cents >= 50000 可证明整个 row group 不匹配。

这依赖：

• statistics 存在且可信；
• comparator/type/NaN/NULL semantics compatible；
• reader 实现支持；
• predicate 可转换为 metadata condition；
• data clustering 让 min/max 足够窄。

min/max overlap 只能表示“可能匹配”，不能证明每行都匹配。统计缺失或被截断时通常只能保守读取。

4. Reader-side filter evaluation

候选 pages 解码后，reader 对 vector/batch 计算 predicate，产生 selection bitmap/vector。只有真正匹配 rows 进入后续 operator。

这仍会读取和解码 predicate columns。能否延迟读取 output-only columns 取决于 format、page alignment、reader 与 engine，不是所有 Parquet reader 都支持对任意 row positions 做低成本 gather。

PyArrow Dataset 示例

以下 API 需要当前 PyArrow，先在隔离环境安装并记录版本：

python3 -m pip install pyarrow

创建两个 row groups 的示例 file：

import pyarrow as pa
import pyarrow.parquet as pq


table = pa.table({
    "item_id": pa.array(range(1, 9), type=pa.int64()),
    "item_type": [
        "weapon", "armor", "consumable", "weapon",
        "armor", "weapon", "consumable", "armor",
    ],
    "value_cents": pa.array(
        [12000, 8000, 1500, 20000, 95000, 180000, 5000, 70000],
        type=pa.int64(),
    ),
    "description": ["x" * 200] * 8,
})

pq.write_table(
    table,
    "vault-items.parquet",
    row_group_size=4,
    compression="zstd",
)

只投影两列并过滤：

import pyarrow.dataset as ds


dataset = ds.dataset("vault-items.parquet", format="parquet")
result = dataset.to_table(
    columns=["item_type", "value_cents"],
    filter=ds.field("value_cents") >= 50000,
)

assert result.column_names == ["item_type", "value_cents"]
assert result.num_rows == 3
print(result)

PyArrow Dataset API 会做 projection 和 filter，并在可能时利用 partition/internal metadata；否则仍会在 loaded record batches 上过滤。不能从正确结果本身证明跳过了哪个 row group，需要结合 fragment metadata、scanner/log/trace 与 I/O counters。

官方 API：Apache Arrow Dataset。

检查 Parquet metadata

import pyarrow.parquet as pq


parquet_file = pq.ParquetFile("vault-items.parquet")
metadata = parquet_file.metadata

print("rows", metadata.num_rows)
print("row_groups", metadata.num_row_groups)

for rg_index in range(metadata.num_row_groups):
    row_group = metadata.row_group(rg_index)
    print("row_group", rg_index, "rows", row_group.num_rows)
    for col_index in range(row_group.num_columns):
        column = row_group.column(col_index)
        stats = column.statistics
        print(
            column.path_in_schema,
            "compressed", column.total_compressed_size,
            "min", None if stats is None else stats.min,
            "max", None if stats is None else stats.max,
        )

metadata inspection 只显示 writer 记录的信息。敏感 columns 的 min/max 也可能泄漏信息；格式/平台可能允许禁用或截断 statistics。

Vectorized execution

row-at-a-time volcano model 每次 operator 调用产生一行，函数调用、branch 和 poor cache locality 会成为 CPU 开销。vectorized engine 一次处理一个 batch/vector：

values:    [12000, 8000, 95000, 180000, ...]
predicate: value >= 50000
selection: [false, false, true, true, ...]
aggregate selected values in tight loop

优势：

• amortize virtual/function-call overhead；
• contiguous typed arrays 改善 cache/prefetch；
• compiler/runtime 更容易使用 SIMD；
• selection vector 避免立即拷贝完整 rows。

vectorization 不等于自动 SIMD，也不保证 batch 越大越好。大 batch 增加 latency/memory，branchy UDF、variable strings 与 decompression 仍可能限制 CPU。

Late materialization

early materialization 会较早把 columns 组合成 row/tuple；late materialization 尽量保留 column vectors + selection，直到 output/join 等需要完整 values 时才组合。

decode predicate columns
  -> evaluate filter -> selection vector
  -> aggregate/join using selected positions
  -> fetch/decode output-only columns when useful and supported
  -> materialize final result

但“最后按行号只读 name 中的几个 value”可能仍需：

• 定位包含这些 rows 的 pages；
• 读取压缩 page；
• 解压和解码 page prefix/dictionary；
• 处理 nested levels。

高 selectivity、fragmented positions 时，late gather 可能比顺序解码整批更贵。optimizer 要比较 selectivity 与 decode/gather cost。

Data skipping 与排序

若 value_cents 在 file 中随机分布，每个 row group 的 min/max 都覆盖宽范围，predicate 很难跳过。按常用 filter key 排序/聚簇可缩窄 ranges：

RG0 value: 0..9,999
RG1 value: 10,000..19,999
RG2 value: 20,000..29,999

但一个 physical order 不能同时优化所有 dimensions。按 date 排序有利于 date pruning，却可能不利于 tenant/item_type。可通过 partition + sort、data skipping index、multiple projections/materialized views 等平衡。

Small-file problem

大量 tiny Parquet files 会让：

• listing/catalog metadata 变大；
• object-store requests 增多；
• footer reads 与 scheduler tasks 占比上升；
• compression/dictionary scope 太小；
• query startup latency 上升。

compaction/coalescing 要保留 partition/order/statistics，并考虑并发 writers 和 failure atomicity。不能只把文件 concatenate。

Column store 与 OLTP 的真实边界

列式 main store 可通过 delta store、delete bitmap、primary-key index、MVCC、background merge 支持 updates；row store 也能借助 covering index、columnar replica 和 vectorized scan 做 analytics。

选择系统时比较 workload：

• point lookup 与 narrow updates；
• scan projection width/selectivity；
• ingest batch size 与 freshness；
• joins/grouping/sort；
• transaction/constraint needs；
• compaction/merge 与 mutation tail；
• concurrency、replication 与 recovery。

验收清单

• [ ] 不把 column pruning 算成 storage compression；
• [ ] 能区分 partition pruning、metadata pruning 与 row filtering；
• [ ] 明确 statistics 只能保守证明“不可能匹配”；
• [ ] 能解释 late materialization 为什么不等于任意行零成本随机读；
• [ ] 用实际 metadata/I/O 证明 skipping，而不是看结果猜；
• [ ] 报告 row-group size、sort/partition、codec 与 reader version。

本章小结

列式性能来自布局、encoding、metadata 与 execution 的配合：少读 columns，用 statistics 排除不可能匹配的 units，在 vectors 上过滤聚合，再尽量晚地 materialize。下一章将比较 hash index 与 external sort，它们分别利用 equality hashing 和有界内存的 run merge。