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。

sql

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：

text

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 示例

下面使用 Python 3.12 与 PyArrow 25.0.1。以下命令适用于 Linux/macOS，创建隔离环境并记录版本；后面的 Python 代码用 .venv/bin/python 执行：

bash

python3 -m venv .venv
.venv/bin/python -m pip install pyarrow==25.0.1
.venv/bin/python -c "import pyarrow; print(pyarrow.__version__)"

在专用练习目录创建两个 row groups 的示例文件。写入会替换同名 catalog-items.parquet，不要指向现有数据：

python

import pyarrow as pa
import pyarrow.parquet as pq

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

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

只投影两列并过滤：

python

import pyarrow.dataset as ds

dataset = ds.dataset("catalog-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

python

import pyarrow.parquet as pq

parquet_file = pq.ParquetFile("catalog-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,
        )

初始文件应有 8 行、2 个行组，每组 4 行。value_cents 的两个范围分别为 [1500, 20000] 和 [5000, 180000]：第一组不可能满足条件，第二组中也只有三行匹配。

用下面的 API 核对基于统计信息保留下来的候选行组：

python

import pyarrow.dataset as ds

fragment = next(ds.dataset(
    "catalog-items.parquet", format="parquet"
).get_fragments())
candidates = fragment.split_by_row_group(
    filter=ds.field("value_cents") >= 50000
)
kept_ids = [group.id for part in candidates for group in part.row_groups]
assert kept_ids == [1]
print("candidate_row_groups", kept_ids)

candidate_row_groups [1] 证明这个元数据筛选只保留第二组；它不等于一次设备 I/O 测量，预读和字节范围合并仍可能读取额外字节。API 行为见 ParquetFileFragment。

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：

text

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 时才组合。

text

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 定义为通过筛选的行数占比。匹配比例高或位置很分散时，late gather 可能比顺序解码整批更贵。optimizer 要比较 selectivity 与 decode/gather cost。

Data skipping 与排序

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

text

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。

第1章逻辑

第2章集合论

第3章证明方法

第4章递归、归纳与递推

第5章关系与函数

第6章组合数学

第7章图论

第8章概率与统计

第9章数论

第10章线性代数

第11章微积分与优化

第12章信息论与数值计算

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

四层减少工作

1. Projection pushdown / column pruning

2. Partition pruning

3. Metadata pruning

4. Reader-side filter evaluation

PyArrow Dataset 示例

检查 Parquet metadata

Vectorized execution

Late materialization

Data skipping 与排序

Small-file problem

Column store 与 OLTP 的真实边界

验收清单

本章小结