Velox Parquet SelectiveReader：过滤下推、延迟物化与 I/O 调度

Velox 的 Parquet SelectiveReader 是 Hive connector 把 Parquet 文件读成 Velox 向量的一条执行路径。上层交给它文件 split、需要的列和过滤条件；它组织列 reader，先筛选行，再为需要的行生成列值，最后交回 RowVector。这里的“selective”有两个含义：列可以少读，列中的值也可以按行集合选择性解码。它同时连接查询表达式、Parquet 编码和底层 I/O，不能只从某一个 ColumnReader 类来理解。

这条路径的入口是 HiveDataSource。当前实现把公共文件扫描流程放在 FileDataSource 和 FileSplitReader；Hive 子类负责分区值、合成列、bucket conversion 等语义。往下，ParquetRowReader 决定读哪个 RowGroup、推进多少行，StructColumnReader 协调各列的过滤与输出，叶子 reader 和 PageReader 完成编码解码，BufferedInput 则决定何时取得文件字节。

下面先跟踪一条查询从 Hive 到返回 batch 的完整过程，再展开嵌套数据、跨页状态、预取与异常。理解这条路径需要始终区分三种东西：查询所需的值、reader 当前选择的行号、存储实际读取的字节。返回一行不等于只读一行的字节；一个 LazyVector 也不意味着对应的压缩数据还没有进入内存。

代码基准：2026-09-29 获取的 Velox origin/main，432a2e73b2c1。本文分析普通 CPU Hive → Parquet 路径。源码摘录固定到此版本；SQL 和行号算例用于解释接口，未作为完整 Velox 程序编译。原分享的例子、行业背景和 PageReader 完整函数保留在对应章节与附录。

阅读路线

1. 1. 从 HiveDataSource 看整条读取链
2. 2. 一条查询怎样变成读取方案
3. 4. ScanSpec：把查询要求编译到列树上
4. 5. addSplit：把查询方案绑定到一个文件范围
5. 6. Parquet 布局与 footer：查询要求落到哪些字节
6. 7. 从 schema 到 reader tree：类型、列下标与所有权
7. 8. RowGroup 选择、预取窗口与组内游标
8. 9. next：同一批行怎样经过两轮过滤
9. 10. 先读哪一列：ScanSpec 重排与完整 nation 算例
10. 11. read 与 getValues：先确定选择，再对齐列结果
11. 12. 从整数 reader 到 ColumnVisitor：过滤进入解码循环
12. 13. ExprSet 与 LazyVector：把上层选择传回 payload reader
13. 14. 嵌套读取：从顶层行到 rep/def levels 与元素范围
14. 15. PageReader：跨页坐标、字典转换与 null bitmap
15. 16. BufferedInput：列的读取需求怎样变成 I/O 计划
16. 17. DirectCoalescedLoad：分配、共享与成功发布
17. 18. 消费线程与预取线程：等待、成功和失败分别发生在哪里
18. 19. Split preload、接管与 reader 的生命周期
19. 20. 用哪些指标判断过滤、延迟物化与预取的收益
20. 21. 从 Spark、native reader 与 Photon 看接口边界
21. 22. 设计取舍：选择信息如何跨越计算与存储边界
22. 23. 附录：PageReader 核心函数与状态对照
23. 24. 源码与参考资料

执行流程与相关实现核对于 2026-09-29，Velox 源码版本为 48883e8521b2。下文源码节选、历史资料和宿主集成引用各自注明版本；教学输入用于解释状态变化，未作为性能基准运行。

1. 从 HiveDataSource 看整条读取链

TableScan 不直接打开 Parquet 文件。它先从 Task 取得 connector split，通过 HiveConnector::createDataSource() 创建 HiveDataSource；后者的构造函数委托给 FileDataSource。第一次创建时编译查询需要的读取方案，之后每个 split 创建自己的 split reader。DataSource 是一次扫描的上下文，split reader 是当前文件范围的上下文，两者生命周期不同。

图 1 · 查询要求向下传递，文件字节向上变成 RowVector　打开原图

层	本层拿到什么	本层决定什么	交给下一层 / 返回什么
TableScan / HiveConnector	TableHandle、ColumnHandle assignments、outputType、Task split	创建 DataSource，推进 split 和 batch	addSplit；next(size, future)
HiveDataSource / FileDataSource	查询 schema、过滤表达式、分区 / 文件信息	编译 ScanSpec 和 remaining ExprSet，整理最终输出	HiveSplitReader；带 outputType 的 RowVector
HiveSplitReader / FileSplitReader	一个 split，共享 ScanSpec 和 I/O 上下文	文件打开、schema 适配、整 split 检查	ParquetReader 与 ParquetRowReader
ParquetRowReader / ReaderBase	footer、requested type、split range	有效 RowGroup、预取窗口、组内行游标	根 StructColumnReader 的一次 next
Struct / leaf ColumnReader	ScanSpec、batch offset、RowSet	按过滤顺序缩小 rows，构造或延迟输出列	通过的 rows、values/nulls、LazyVector
ParquetData / PageReader	ColumnChunk stream、编码与 levels 元数据	跨 page 定位、解压、解码、字典与 null 状态	decoder 提供给 visitor 的值
BufferedInput / InputStream / ReadFile	文件 Region、访问历史、读取预算	缓存路径、合并、预取、同步补读	有明确所有权的字节 buffer

图中的主调用链是一条同步 C++ 调用栈；后台 split preload 和数据预取是另外的执行活动。Driver 自身可在线程池上被调度，遇到算子级 future 时可以让出线程，但不能据此推断这条栈中的每个文件读取都是异步接口。TableScan 的 split preload与下文 DirectInputStream 的等待方式需要分别分析。

2. 一条查询怎样变成读取方案

用 nation 的前 11 行作为数据，增加一个值来自 split 的分区列 ds。假设文件中的 nationkey 与 regionkey 是 BIGINT，name/comment 是 VARCHAR；ds 为 VARCHAR 分区值。主例的 WHERE 同时包含分区条件、可直接用于 reader 的整数范围和需要表达式求值的两列加法。

贯穿全文的查询示例

SELECT name AS country, comment, ds
FROM nation
WHERE ds = '2026-09-01'
  AND nationkey BETWEEN 0 AND 3
  AND nationkey + regionkey >= 4;

2.1 TableHandle 与 assignments 分别描述什么

HiveTableHandle 继承文件表句柄的公共语义，保存表的数据列、subfieldFilters、remainingFilter 等；ColumnHandle 描述某个字段的实际名字、类型和列种类。TableScan 的 outputType 使用执行计划中的输出名字，assignments 把这些名字映射到 column handle。因此 country 是上层输出字段，name 才是文件中要寻找的字段，不能把二者直接拿来按名字匹配。

图 2 · 输出 schema、reader schema 与文件 schema 的三个边界　打开原图

名称 / 对象	主例中的内容	使用位置
outputType_	ROW(country VARCHAR, comment VARCHAR, ds VARCHAR)	TableScan 与下游算子看到的结果 schema
assignments	country → name handle；comment → comment handle；ds → partition handle	FileDataSource 构造阶段把别名还原为列名
readerOutputType_	name、comment、ds，再补入 nationkey、regionkey	reader 返回给 remaining ExprSet 的内部 RowVector；补入字段的顺序由 distinctFields 遍历确定
ScanSpec	name/comment 投影；ds 常量与过滤；nationkey 范围过滤且保留值；regionkey 供表达式读取	描述每个字段的读取要求，比最终投影列集合更丰富
file schema / TypeWithId	footer 中实际存在的列、Parquet 类型、schema id 与 leaf column ordinal	映射 ColumnChunk；不会因 SQL 别名或分区字段自动增加物理列

还要区分一个容易混淆的参数：RowReaderOptions::requestedType 并非直接等于 readerOutputType_。FileSplitReader::getAdaptedRowType 从 baseReader 的文件字段名出发，调用 HiveSplitReader::adaptColumns 调整字段目标类型，再把这个 row type 传给 createRowReader。readerOutputType_ 描述 connector 需要的内部结果列；adapted row type 则给列 reader 提供文件字段的目标类型视图。这样 ScanSpec 中非投影的 filter-only 字段也能找到其类型。

这三份 schema 都通过 Velox 的 Type / RowType 表达逻辑类型；ParquetTypeWithId 额外记录物理存储和层级信息。它们不是三套 C++ struct，也不是把一列从“逻辑类型”强转成“物理类型”。INT32/INT64 等文件存储形式、DATE/DECIMAL 等逻辑含义，以及 reader 的 C++ 解码宽度，在更下层才关联起来。

2.2 只过滤的列，什么时候也必须成为 reader 的输出列

对于 SELECT name WHERE nationkey BETWEEN 0 AND 3，nationkey 只需产生通过行号，ScanSpec 可以保留它作为 filter-only child，不需要把其值放进最终 RowVector。但主例的加法还需要 nationkey 的值，FileDataSource 会将其补入 readerOutputType，让 reader 在过滤时保留这些值。regionkey 同样被补入。它们对 SQL 是隐藏列，对 connector 内部 ExprSet 却是输入列。

构造阶段补齐 remaining filter 的依赖列和 subfields · FileDataSource.cpp:169

  if (remainingFilter) {
    remainingFilterExprSet_ = expressionEvaluator_->compile(remainingFilter);
    auto& remainingFilterExpr = remainingFilterExprSet_->expr(0);
    folly::F14FastMap<std::string, column_index_t> columnNames;
    for (int i = 0; i < readColumnNames.size(); ++i) {
      columnNames[readColumnNames[i]] = i;
    }
    // Capture top-level column names referenced by the remaining filter.
    // These columns must be loaded eagerly (not lazily) so the filter
    // can evaluate before lazy columns are accessed.
    folly::F14FastSet<std::string> remainingFilterColumns;
    for (auto& input : remainingFilterExpr->distinctFields()) {
      remainingFilterColumns.insert(input->field());
      auto it = columnNames.find(input->field());
      if (it != columnNames.end()) {
        if (shouldEagerlyMaterialize(*remainingFilterExpr, *input)) {
          multiReferencedFields_.push_back(it->second);
        }
        continue;
      }
      // Remaining filter may reference columns that are not used otherwise,
      // e.g. are not being projected out and are not used in range filters.
      // Make sure to add these columns to readerOutputType_.
      readColumnNames.push_back(input->field());
      readColumnTypes.push_back(input->type());
    }
    remainingFilterColumns_ = std::move(remainingFilterColumns);
    remainingFilterSubfields_ = remainingFilterExpr->extractSubfields();
    if (VLOG_IS_ON(1)) {
      VLOG(1) << fmt::format(
          "Extracted subfields from remaining filter: [{}]",
          fmt::join(remainingFilterSubfields_, ", "));
    }
    for (auto& subfield : remainingFilterSubfields_) {
      const auto& name = getColumnName(subfield);
      auto it = subfields_.find(name);
      if (it != subfields_.end()) {
        // Some subfields of the column are already projected out, we append the
        // remainingFilter subfield
        it->second.push_back(&subfield);
      } else if (columnNames.count(name) == 0) {
        // remainingFilter subfield's column is not projected out, we add the
        // column and append the subfield
        subfields_[name].push_back(&subfield);
      }
    }
  }

代码先收集已经投影的物理列名，再遍历编译后表达式的 distinctFields；缺失字段追加到 readColumnNames/readColumnTypes。requiredSubfields 也必须与 remainingFilterSubfields 合并：例如只投影 s.a、却用 s.b 过滤，不能在 schema pruning 时把 s.b 删除。投影裁剪的单位是“本次查询所有消费者需要的字段”，不只是 SELECT 列表。

readerProducedType_ 是公共基类为 extraction pushdown 等变换预留的结果 schema。当前 ParquetColumnReader::build 明确检查 extractionType 必须为 kNone，因此不能把公共代码中存在的 transform 分支写成这条 Parquet 主线必定支持的能力。

3. 一次扫描怎样从 11 行得到 1 行结果

沿上一节的 SQL 走完一批数据，再展开 ScanSpec、RowGroup、ColumnReader 与 I/O。这里的 remaining filter 是 nationkey + regionkey >= 4；后面的双 reader-filter nation 示例则使用 regionkey=0，它们刻意展示两种过滤放置方式，不能合并成同一个条件。

步骤	行集合与结果	哪一层负责
文件与分区	ds=2026-09-01 满足常量条件，选定一个有效 RowGroup	HiveSplitReader 适配 schema；FileSplitReader / ParquetRowReader 组织文件与行组。
请求 batch	从当前 RowGroup 推进前 11 行，候选 row set=[0..10]	ParquetRowReader::next 决定扫描跨度，未承诺返回 11 行结果。
reader filter	nationkey BETWEEN 0 AND 3 留下 [0,1,2,3]	StructColumnReader 按 ScanSpec 访问过滤列，叶子 reader / visitor 筛选值。
reader 产出	生成四行内部 RowVector；regionkey 对应 [0,1,1,1]	已读列整理到最终 reader row set；符合条件的其他列可生成 LazyVector。
remaining expression	[0,1,2,3] + [0,1,1,1] → [0,2,3,4]，只选择下标 3	FileDataSource 的 ExprSet 处理跨列条件；结果下标属于四行内部向量。
payload 与投影	ColumnLoader 将下标 3 映射回 outputRows[3]=3	读取或复用 name/comment，按 assignments 输出 country=CANADA，并附 ds 常量。
交还执行框架	scanned=11，reader result=4，final result=1	TableScan 交出 RowVector；若后续缺 split / 数据，Driver 才按阻塞协议等待。

此时可能已经读取比最终结果多得多的压缩字节：footer、RowGroup 元数据、预取合并的区间与字典都有自己的粒度。选择性解码节省的是被跳过值的处理或物化；是否同时减少物理读取，要看 Deferred IO、coalesced load、缓存命中和实际列访问，不能从 final result=1 推出只读了一行字节。

源码连接：两轮过滤与输出、扫描推进、lazy 选择映射。后面把这条路径逐层展开，原 nation 双列 filter 算例、跨页与嵌套算例分别保留。

4. ScanSpec：把查询要求编译到列树上

ScanSpec 是连接 Hive 查询语义与 DWIO 列 reader 的共享对象。它既不是 Parquet footer，也不是单纯的过滤表达式树。每个节点对应字段或 subfield，记录过滤器、是否产出值、结果 channel、常量值、读取下标，以及过滤计时和选择率。创建 reader 后，它还承载跨 batch 的自适应状态。

图 3 · 投影、过滤依赖与 split 常量如何汇合成 ScanSpec　打开原图

4.1 先拆能下推的条件，再编译 remaining ExprSet

FileDataSource 首先复制 table handle 的 subfieldFilters，再调用 extractFiltersFromRemainingFilter。这个函数尝试把叶子表达式转换为 SubfieldFilter；同一字段已有 filter 时合并。AND 中可提取的部分可以独立下推；OR 则必须满足可表示性约束，例如各分支都能完整转成同一 subfield 的 filter，再合成 OR filter，不能把 a=1 OR b=2 拆成两列同时成立。NOT 通过 negated 状态递归处理；识别出的采样表达式可以转成 sampleRate/RandomSkip。

主例中 nationkey 的区间可用 BigintRange 表达，nationkey+regionkey≥4 仍是 ExprSet。关键边界是 reader 的 filter 抽象能否表达该条件，不是“单列一律下推、多列一律不下推”。提取代码见 HiveConnectorUtil.cpp:761。

4.2 makeScanSpec 的两轮列处理

makeScanSpec 先处理 readerOutputType 中要产出的字段，为它们设置 channel，并递归添加全部或必要的 subfields。随后处理只在 subfieldFilters 出现、没有被产出的字段：从 table 的 dataColumns 找到类型，在 ScanSpec 中添加非投影节点。最后把各个 filter 安装到准确的 subfield 节点。这就是“filter-only 列没有输出 channel，也仍有 ColumnReader”的来源。

没有被投影的过滤列也要进入 ScanSpec · HiveConnectorUtil.cpp:461

  // Now process the columns that will not be projected out.
  if (!filterSubfields.empty()) {
    VELOX_CHECK_NOT_NULL(dataColumns);
    for (auto& [fieldName, subfields] : filterSubfields) {
      for (auto* subfield : subfields) {
        subfieldSpecs.push_back({subfield, true});
      }
      auto& type = dataColumns->findChild(fieldName);
      auto* fieldSpec = spec->getOrCreateChild(fieldName);
      addSubfields(*type, subfieldSpecs, 1, pool, *fieldSpec);
      processFieldSpec(dataColumns, type, *fieldSpec);
      subfieldSpecs.clear();
    }
  }

分区列与 synthesized info column 在这里有专门处理：它们不是普通文件 leaf。ds 的具体值要到 addSplit 之后才能知道，所以查询级 ScanSpec 先描述要求，split 适配阶段再安装 constantValue。多个 split 的 ds 可能不同，旧常量和缓存不能直接沿用。

4.3 MetadataFilter 与动态过滤更新

remaining expression 不等于只能在读完值后发挥作用。FileDataSource 会尝试构造 MetadataFilter，把可以从统计推导的布尔结构连接到 ScanSpec 的 metadata filters；RowGroup 阶段收集各叶子的不匹配位图，再按表达式组合。但 metadata filter 只证明某个组“不可能有匹配行”，并不替代表达式的逐行精确判断。缺少可靠统计时必须保留候选。

运行中 TableScan 可以接收来自其他算子的动态 filter。FileDataSource::addDynamicFilter 按 output channel 找节点，更新 filter，递归 resetCachedValues，并让当前 split reader resetFilterCaches。这里清除的包括需要重新判断的缓存状态；不能从这个函数推断已经排好的 rowGroupIds 会立即全部重新计算，也不能将已经发出的预取撤销视作动态过滤的保证。

动态过滤的安装与缓存失效 · FileDataSource.cpp:521

void FileDataSource::addDynamicFilter(
    column_index_t outputChannel,
    const std::shared_ptr<common::Filter>& filter) {
  auto& fieldSpec = scanSpec_->getChildByChannel(outputChannel);
  fieldSpec.setFilter(filter);
  scanSpec_->resetCachedValues(true);
  if (splitReader_) {
    splitReader_->resetFilterCaches();
  }
}

5. addSplit：把查询方案绑定到一个文件范围

构造 DataSource 时并不知道下一个 split 的文件元数据、实际分区值以及旧文件缺了哪些列。addSplit 负责补齐这些事实。FileDataSource 先检查旧 split 已被读完，再通过虚函数 createSplitReader() 进入 HiveDataSource：prepareSplit() 处理 Hive 级准备，接着构造 HiveSplitReader。之后的公共调用重新回到 FileDataSource 配置 options 和调用 split reader 的 prepareSplit。

图 4 · addSplit 的跨类调用：同名 prepareSplit 处于两个不同层次　打开原图

Hive split 描述文件路径、字节范围以及分区值等信息。当前公共逻辑在 FileDataSource / FileSplitReader，HiveDataSource / HiveSplitReader 在此基础上处理 Hive 特有内容。FileDataSource 收到 split 后创建 split reader，配置 options，然后调用 prepareSplit。

源码摘录 · FileDataSource.cpp:410

void FileDataSource::addSplit(std::shared_ptr<ConnectorSplit> split) {
  VELOX_CHECK_NULL(
      split_,
      "Previous split has not been processed yet. Call next to process the split.");
  split_ = checkedPointerCast<FileConnectorSplit>(split);

  VLOG(1) << "Adding split " << split_->toString();

  if (splitReader_) {
    splitReader_.reset();
  }

  splitReader_ = createSplitReader();

  // Split reader subclasses may need to use the reader options in prepareSplit
  // so we initialize it beforehand.
  splitReader_->configureReaderOptions(randomSkip_);
  splitReader_->setRemainingFilterColumns(remainingFilterColumns_);
  splitReader_->prepareSplit(metadataFilter_, runtimeStats_);
  readerOutputType_ = splitReader_->readerOutputType();
}

prepareSplit 的主流程很短，但每一步都有不同职责：先通过格式工厂创建 ParquetReader；ReaderBase 读取 footer 并建立文件 schema；接着把查询 schema 与文件 schema 对齐，处理常量列和缺失列；检查整个 split 是否已能排除；最后创建 ParquetRowReader。原分享中的 SplitReader 在当前版本对应下面的 FileSplitReader。

源码摘录 · FileSplitReader.cpp:137

void FileSplitReader::prepareSplit(
    std::shared_ptr<common::MetadataFilter> metadataFilter,
    dwio::common::RuntimeStats& runtimeStats,
    const folly::F14FastMap<std::string, std::string>& fileReadOps) {
  createReader(fileReadOps);
  if (emptySplit_) {
    return;
  }
  auto rowType = getAdaptedRowType();

  if (checkIfSplitIsEmpty(runtimeStats)) {
    VELOX_CHECK(emptySplit_);
    return;
  }

  createRowReader(std::move(metadataFilter), std::move(rowType), std::nullopt);
}

• 分区列：值来自 split 的 partition keys，写入 ScanSpec 的 constantValue，不需要去文件查找同名 ColumnChunk。
• 旧文件中缺失的字段：按 requested type 构造 null 常量或遵守对应的 schema evolution 策略。null 的意义取决于 filter，不能直接把缺失列当作整组无效。
• Info column：文件大小、修改时间等由 split/connector 提供，同样可以合成为常量。
• 输出与只用于过滤的列：Reader 的 requested schema 可能包含最终 SQL 投影之外的字段，因为 remaining filter 也要使用它们。

5.1 HiveSplitReader::adaptColumns 的四种情况

字段情况	适配动作	对读取的影响
partition key	从当前 Hive split 的 partitionKeys 解析值并安装常量	ds 不需要 ColumnChunk；常量过滤可排除 split
info column	从 split 信息取得路径、大小等值并构造成常量	不用访问同名文件列
文件缺失的普通列	按查询 / 表类型设置 null constant	schema evolution 下保持结果类型和 null 语义
文件存在的普通列	清除前一 split 的常量；协调 file type 与 requested type	构建正确的 reader，按支持范围进行类型适配

实际调用的是 HiveSplitReader 的 adaptColumns override；只读 FileSplitReader 基类会漏掉 Hive 的 info column 处理。适配后 resetCachedValues(false)，防止从上一文件继承的常量判定污染当前文件。嵌套字段的缺失与 struct nullness 由后续 Parquet struct reader 继续处理，不能把顶层补 null 规则简单套到整棵树。HiveSplitReader::validateSynthesizedColumnFilters 是对已有 split 信息的校验，失败会抛出检查异常；它也不能被当作普通的“条件不匹配就跳过文件”分支。

Hive 的 bucket conversion 也是 split 语义：表与分区 bucket 数不同时，需要额外读取 bucket 列，必要时扩展 readerOutputType、重建 ScanSpec，并 moveAdaptationFrom 保留已有反馈；HiveSplitReader::next 在普通 reader 输出后计算 bucketToKeep 并 copyRanges。这层过滤位于 remaining ExprSet 之前。普通非 bucket 查询不经过该分支。rowId 列则由 row index 与文件 / 分区等常量组合，不能当作普通 leaf column。

5.2 打开文件、footer 读取与整 split 排除

FileSplitReader::createReader 通过 FileHandleFactory 获得 ReadFile 和句柄信息，用 BufferedInputBuilder 选择 I/O 实现，再按 split.fileFormat 从 ReaderFactory 创建 ParquetReader。文件不存在时是否忽略由配置决定；不属于“任意读取异常都可吞掉”。之后 getAdaptedRowType 调用上述适配逻辑，checkIfSplitIsEmpty 综合不存在 / 空文件 / 文件级统计过滤，最后才创建 RowReader。

文件级统计检查与后面的 RowGroup 过滤是两个调用层次。在 Parquet 中能否从 Reader 得到足够的文件级统计取决于该格式接口；不能把 checkIfSplitIsEmpty 的存在等同于所有列都有精确文件 min/max。分区常量、缺失列 null 过滤以及零行判断则有各自的证据来源。

5.3 ParquetReader 和 ParquetRowReader 分开保存什么

图 5 · ParquetReader、ParquetRowReader 与共享 ReaderBase 的持有关系　打开原图

ParquetReader 包装一个 shared_ptr<ReaderBase>，提供 schema、row count、statistics 以及 createRowReader 等接口。ParquetRowReader::Impl 同样共享 ReaderBase，但它有自己的 options、RowGroup 游标和 columnReader_。columnReader_ 是 unique_ptr<SelectiveColumnReader>，实际根节点由工厂创建为 StructColumnReader。

源码摘录 · ParquetReader.cpp:1623

    columnReader_ = ParquetColumnReader::build(
        columnReaderOptions_,
        requestedType_,
        readerBase_->schemaWithId(), // Id is schema id
        params,
        *options_.scanSpec());
    columnReader_->setIsTopLevel();

“RowReader”表示一次返回一批逻辑行，不表示内部逐行解码。真正的读取仍沿着列 reader tree 进行。最外层 RowVector 与 Arrow RecordBatch 在“多个等长 child columns 带一份 schema”这一点相近，但 Velox child 还可能采用 constant、dictionary、lazy 等编码。

shared_ptr<ReaderBase> 说明元数据和 BufferedInput 的生命周期由两个 facade 共同维持，并不意味着可在多个线程上任意并发消费同一个 reader。RowReader 的组游标、ColumnReader 的 values/nulls/readOffset 和 PageReader 的 decoder 都是有状态对象；共享所有权与并发安全是不同的保证。

6. Parquet 布局与 footer：查询要求落到哪些字节

到了 createReader，Hive 的字段名和类型必须落到真实文件布局。ParquetReader 的 ReaderBase 先读取尾部：小文件可按 filePreloadThreshold 与 footerSpeculativeIoSize 的较大值整文件预载；大文件先读一个尾部窗口，检查 PAR1 和 footer length，窗口不够时再补读缺少的 metadata。这一步解释了为什么即使最终无数据行通过，也可能已经产生 footer I/O，甚至已经载入整个小文件。

6.1 RowGroup、ColumnChunk 与 footer

Parquet 可以看作文件级的 PAX 风格组织：先按行水平切成 RowGroup，再把每个 RowGroup 内各叶子列的数据分别放进 ColumnChunk。一个包含 A、B、C 三个叶子列和两个 RowGroup 的文件，共有六个 ColumnChunk。每个 chunk 有一个或多个 page；字典编码时可能先有 dictionary page。不同列的 page 大小和行边界不要求对齐。

图 6 · 同一组行按列分布，随后对应文件中的连续 chunk 区域　打开原图

footer 保存 schema、RowGroup 行数、ColumnChunk 位置与长度、codec、encoding、统计等元数据。reader 可以在解码数据前用这些信息选择范围。简单文件还在首尾使用 PAR1 标识，尾部记录 footer 长度；图中聚焦 chunk 与 footer，并未按真实字节比例画所有格式字段。

6.2 Data Page V1 与 V2 的压缩边界

图 7 · V1 压缩页体，V2 保留未压缩的 RL/DL，只对 values 区域处理压缩　打开原图

逻辑布局示意；不是 C++ 数据结构

[ PageHeader: Thrift, variable length ]
[ repetition levels, if present ]
[ definition levels, if present ]
[ encoded values ]

V1 在启用 codec 时，对页体中的 levels 与 encoded values 整体压缩。V2 在 header 中给出 repetition_levels_byte_length 和 definition_levels_byte_length，levels 本身不压缩；values 是否压缩由 is_compressed 决定。当前 prepareDataPageV2() 把缺省 is_compressed 按 true 处理，并验证 levelsSize 不超过 compressed/uncompressed page size。分析读取成本时，需要按 page version 分开计算 levels 与 values 的处理。

“page 内只需要两行”通常无法让通用压缩器只解压两行对应的字节。reader 需要先获得对应压缩块，再在 encoded values 中跳过不需要的项。不过，value decode、结果写入以及后面列的读取仍可以少做。这正是 page 裁剪与 row 裁剪应该分别讨论的原因。

footer 解码后的 Thrift 对象也有内存成本。ReaderBase 按阈值估算并 reportExternalAllocation；RowGroup 过滤释放不读组的 ColumnChunk metadata 时相应 releaseThriftBytes，析构释放剩余报告值。原始压缩文件大小、解码值 buffer 大小和 footer 内存大小，必须分开看。

7. 从 schema 到 reader tree：类型、列下标与所有权

文件 schema 决定可读的物理 leaf，ScanSpec 决定本次扫描需要哪些字段，两者共同决定 reader tree。ParquetColumnReader::build 接收 requestedType、fileType 和 ScanSpec：按照文件 TypeKind 选择具体 reader，再由 reader 处理目标类型。比如整数列可在 getValues 时按 requested type 输出，DECIMAL 还可能有 scale 扩大；这不是将原始 Parquet 字节直接解释为任意 C++ 目标类型。

文件 schema、查询 requested type、运行时 Vector 和 reader tree 有关联，但不是同一个对象。requested type 使用 Velox Type 描述逻辑语义；ParquetTypeWithId 还保留物理 schema、列号、父子路径和 level 信息；Vector 表示输出数据和编码；ColumnReader 维护解码与扫描状态。

图 8 · ROW、MAP 及其叶子，在 Type、Vector 与 reader 侧的对应关系　打开原图

以 ROW(c0 BIGINT, c1 MAP(VARCHAR, DOUBLE)) 为例，顶层 RowVector 直接有两个 child：c0 的整数 vector 和 c1 的 MapVector。MapVector 自己有 offsets、sizes 和 nulls，再持有 key/value vectors。因此总共三条 leaf values 流，但绝不能说整棵树只有三个 Vector。reader 一侧对应 root StructColumnReader、IntegerColumnReader、MapColumnReader，map 下再有 StringColumnReader 与 FloatingPointColumnReader<double,double>。

图 9 · 整数、struct、list、map 的继承链，箭头指向直接基类　打开原图

工厂 ParquetColumnReader::build 根据 fileType 的 Velox kind 选择具体 reader，再结合 requestedType 处理读取结果类型。整数走 IntegerColumnReader；REAL 可按 requested type 输出 float 或 double；ARRAY/MAP/ROW 走相应复杂 reader；timestamp 还要区分文件 INT64/INT96。C++ 值类型由 decoder/visitor 模板承接，和顶层 logical Type 的职责不同。

StructColumnReader 构造时使用 ScanSpec::stableChildren() 建立 children_，为每个真实 reader 设置 subscript。之后 ScanSpec::children() 可以因过滤重排而改变顺序；reader 在 children_ 中的位置和输出 channel 都不能跟着随意变。channel 指输出列位置，subscript 指实际 reader 位置，遍历顺序指当前读取优先级，必须分清。

复杂节点并不总有自己的物理值 stream。嵌套 struct 的 nullability 需要从 descendant leaf 的 RL/DL 推导；本版本还有 synthetic rep/def source：当逻辑投影 child 不足以提供结构信息时，选择一个物理叶子专门读取 levels。源码按完整 metadata、读取字节、解压字节、numValues 等成本选择候选。因此 reader tree 与“查询投影的物理叶子”也不是无条件一一对应。

构建 StructColumnReader 时按 stableChildren 创建子 reader，并把 children_ 下标写进每个 childSpec.subscript。运行时 ScanSpec.children() 可以为了过滤成本重排，但 subscript 仍指向同一 reader，channel 仍指向同一结果列。需要分别记住“调用顺序”“reader 地址”“输出位置”这三个坐标。

8. RowGroup 选择、预取窗口与组内游标

RowReader 构造完 reader tree 后，立即执行 filterRowGroups；只要有有效组，就调用 advanceToNextRowGroup。也就是说，第一组的 I/O 规划通常发生在 addSplit/prepareSplit 中，不必等第一次 next。若 split 已在后台 preload，这些工作也会提前发生。

图 10 · 有效 RowGroup 列表与随消费推进的输入窗口　打开原图

当前过滤入口是 ParquetRowReader::Impl::filterRowGroups()。它先用 split 的 [offset,limit) 和 RowGroup 文件起始 offset 排除不属于此 split 的组，再排除空组；随后调用 columnReader_->filterRowGroups 评估列统计，并把 MetadataFilter 的组合结果纳入排除位图。RowReader 决定有效组列表，ReaderBase 随后负责输入调度，二者职责不同。

split 是文件字节范围；RowGroup 是否由它处理根据文件位置归属判断，不是随意把一个 RowGroup 在行中间切开。跳过的组不进入 rowGroupIds_。统计只允许排除“不可能通过”的组；缺失或不足以判断的统计必须保留候选。

源码摘录 · ParquetReader.cpp:1525

void ReaderBase::scheduleRowGroups(
    const std::vector<uint32_t>& rowGroupIds,
    int32_t currentGroup,
    StructColumnReader& reader) {
  auto numRowGroupsToLoad = std::min(
      options_.prefetchRowGroups() + 1,
      static_cast<int64_t>(rowGroupIds.size() - currentGroup));
  for (auto i = 0; i < numRowGroupsToLoad; i++) {
    auto thisGroup = rowGroupIds[currentGroup + i];
    if (!inputs_[thisGroup]) {
      inputs_[thisGroup] = reader.loadRowGroup(thisGroup, input_);
    }
  }

  if (currentGroup >= 1) {
    inputs_.erase(rowGroupIds[currentGroup - 1]);
  }
}

preload 窗口由当前有效组开始，最多取 prefetchRowGroups+1 个有效组，而不是原始连续 group id。loadRowGroup 发现输入已 buffered 时复用原 input；否则 clone 一个 BufferedInput，递归 enqueue streams 后调用 load(STRIPE)。因此“每个 RowGroup 一定对应全新的 DirectBufferedInput”也太绝对：实际 BufferedInput 类型和已缓冲情况会改变路径。

进入新组时根 reader seekToRowGroup，child readers 同步切换。ParquetData 将已登记的 chunk stream 移交给新 PageReader；PageReader 生命周期跨越本组的多次 batch read，不是每个 batch 重新创建。窗口推进后释放上一组在 inputs_ 中的引用；仍被其他对象持有的数据遵守各自 shared ownership。

源码摘录 · ParquetData.cpp:109

void ParquetData::enqueueRowGroup(
    uint32_t index,
    dwio::common::BufferedInput& input) {
  auto chunk = fileMetaDataPtr_.rowGroup(index).columnChunk(type_->column());
  streams_.resize(fileMetaDataPtr_.numRowGroups());
  VELOX_CHECK(
      chunk.hasMetadata(),
      "ColumnMetaData does not exist for schema Id ",
      type_->column());
  ;

  uint64_t chunkReadOffset = chunk.dataPageOffset();
  if (chunk.hasDictionaryPageOffset() && chunk.dictionaryPageOffset() >= 4) {
    // this assumes the data pages follow the dict pages directly.
    chunkReadOffset = chunk.dictionaryPageOffset();
  }

  auto id = dwio::common::StreamIdentifier(type_->column());
  streams_[index] = input.enqueue({chunkReadOffset, chunk.readSize()}, &id);
}

Region 的起点优先选择 dictionary page offset，否则选择 data page offset；长度来自 chunk.readSize()。这解释了为什么后续 page 内的少量行选择，不会自动撤回已经按整个 chunk region 规划的预取。

8.1 next(size) 只推进当前 RowGroup 的一段

nextReadSize(size) 取请求大小和当前组剩余行数的较小值。next 将这个 rowsToRead 交给根 columnReader，随后 currentRowInGroup += rowsToRead，并返回扫描推进量。下一次调用发现当前组耗尽，才调度下一个有效组。一个 batch 的有效输出可以远小于 rowsToRead，但游标仍必须越过所有已扫描的输入行。

ParquetRowReader::Impl 的组内推进 · ParquetReader.cpp:1745

  uint64_t next(
      uint64_t size,
      velox::VectorPtr& result,
      const dwio::common::Mutation* mutation) {
    auto rowsToRead = nextReadSize(size);
    if (rowsToRead == kAtEnd) {
      return 0;
    }
    VELOX_DCHECK_GT(rowsToRead, 0);
    columnReader_->setCurrentRowNumber(nextRowNumber());
    if (!options_.rowNumberColumnInfo().has_value()) {
      columnReader_->next(rowsToRead, result, mutation);
    } else {
      readWithRowNumber(
          columnReader_,
          options_,
          nextRowNumber(),
          rowsToRead,
          mutation,
          result);
    }

    currentRowInGroup_ += rowsToRead;
    return rowsToRead;
  }

有效组列表和 firstRowOfRowGroup 还保存原始文件行号关系。过滤掉前面的组不会把后面组的文件行号重编号；这对 row number / row index 合成列尤其重要。被跳过组的 metadata 清理保留第 0 组，因为 estimatedRowSize 还可能利用它。

9. next：同一批行怎样经过两轮过滤

现在回到主例。取 11 行输入，nationkey 为 0…10，regionkey 为 [0,1,1,1,4,0,3,3,2,2,4]；该 split 的 ds 满足分区条件。假设 batch 从 RowGroup 行偏移 0 开始，整数范围先留下输入行 [0,1,2,3]，其余行不再参与后续 regionkey 解码。

图 11 · 主例的完整 batch：11 行扫描 → 4 行 reader 输出 → 1 行查询输出　打开原图

位置	行号 / 值	此时哪些列已读
根 reader.next(11)	batch RowSet = [0…10]	构造 dense input rows；尚未要求所有 payload
nationkey filter	outputRows = [0,1,2,3]	保留 nationkey=[0,1,2,3]，因为 remaining 仍需它
getValues	内部 RowVector 长度 4	nationkey 已物化；regionkey、name、comment 可为 lazy；ds 为 constant
ExprSet 求和	regionkey=[0,1,1,1]；和=[0,2,3,4]	remaining 加载 regionkey，留下内部位置 [3]
wrapChild / 返回	country、comment 的外层选择 indices=[3]	只返回 SQL 输出列；ds 仍为同值 constant
下游访问 payload	outputRows[3] = 原 batch row 3	ColumnLoader 读取 CANADA 及其 comment

为了突出接口关系，表中列出的是 reader 允许保留的 lazy 状态；表达式策略、ValueHook 或下游访问可能提早触发加载。确定不变的是：remaining filter 看到的行号是 reader 返回向量的位置，payload loader 最终需要的行号是原 batch 的输入行号。两者必须经过父 reader 的 outputRows 映射。

9.1 四种 next 结果不能混淆

层 / 返回值	意义	TableScan 下一步
RowReader 返回 rowsScanned > 0，result.size() 可为 0	扫描游标已经前进；本批可能全部被过滤	继续读取同一个 split
FileDataSource 返回空 RowVector	reader filter 或 remaining filter 没有留下行	没有数据 batch 可交付，但 split 尚未结束
FileDataSource 返回 optional 内的 nullptr	本 split 完成，resetSplit 已执行	标记 needNewSplit，完成当前 split 统计
DataSource 接口返回 nullopt	该 connector 正在等待 future，暂无结果	设置 kWaitForConnector；按 Driver blocking 协议等待

最后一项是 DataSource 接口能力，当前 FileDataSource::next 的形参写成 ContinueFuture& /*future*/，没有使用它。不能把 DirectInputStream::waitFuture.wait() 解释成这里返回 nullopt。把这两种等待混为一谈，会同时误读调度行为和 I/O latency 指标。

FileDataSource：先看扫描是否结束，再处理 remaining filter · FileDataSource.cpp:458

  const auto rowsScanned = splitReader_->next(size, output_);
  completedRows_ += rowsScanned;
  if (rowsScanned == 0) {
    splitReader_->updateRuntimeStats(runtimeStats_);
    resetSplit();
    return nullptr;
  }

  VELOX_CHECK(
      !output_->mayHaveNulls(), "Top-level row vector cannot have nulls");
  auto rowsRemaining = output_->size();
  if (rowsRemaining == 0) {
    // no rows passed the pushed down filters.
    return getEmptyOutput();
  }

  auto rowVector = std::dynamic_pointer_cast<RowVector>(output_);

  // In case there is a remaining filter that excludes some but not all
  // rows, collect the indices of the passing rows. If there is no filter,
  // or it passes on all rows, leave this as null and let exec::wrap skip
  // wrapping the results.
  BufferPtr remainingIndices;
  filterRows_.resize(rowVector->size());

  if (remainingFilterExprSet_) {
    rowsRemaining = evaluateRemainingFilter(rowVector);
    VELOX_CHECK_LE(rowsRemaining, rowsScanned);
    if (rowsRemaining == 0) {
      // No rows passed the remaining filter.
      return getEmptyOutput();
    }

    if (rowsRemaining < rowVector->size()) {
      // Some, but not all rows passed the remaining filter.
      remainingIndices = filterEvalCtx_.selectedIndices;
    }
  }

最终 outputColumns 只取 outputType_->size() 个字段，按 remainingIndices 调用 wrapChild，必要时执行 column postprocessor，再用 outputType_ 构造 RowVector。这一步恢复 country 别名并丢掉 nationkey/regionkey 这些内部输入列；通常可用 dictionary 包装表达选择，不必立即复制全部列值。

10. 先读哪一列：ScanSpec 重排与完整 nation 算例

主例先从一个范围 filter 入手；多个 filter 时，哪一列先读会影响后续工作量。下面保留原分享的完整 nation 数据和双条件例子，使用同一组输入展示 reader 内的选择如何逐层收缩。这里暂时换成两个都可下推的条件，便于把 reader 的 read/getValues 与上一节的 remaining ExprSet 区分开。

10.1 一个节点同时描述查询意图和运行时反馈

ScanSpec 信息	读取时的用途
fieldName / children	按名字找到文件字段，递归处理复杂类型
projectOut / channel	是否输出、放入输出 RowVector 的哪一列
constantValue / columnType	分区值、缺失字段、row index 或合成列
filter / hasFilter()	字段本身或后代是否含能影响输出行数的 filter
subscript	从重排后的 spec 找回 children_ 中的 reader
selectivity	累计输入、输出与过滤耗时，用于次序适应

hasFilter() 不是简单检查 filter_ 非空。一个 struct 的自身 filter_ 可能为空，但其 child 有 filter，仍会影响 parent 的行数。array element/map key 的子筛选又未必改变容器输出行数，ScanSpec 对这类路径有专门判断。把它概括成“scalar 总比 complex 优先”会丢失这些语义。

10.2 用丢掉一行的代价排序，而不是只看选择率

源码摘录 · SelectivityInfo.h:31

  float timeToDropValue() const {
    if (numIn_ == numOut_) {
      return timeClocks_;
    }
    return timeClocks_ / static_cast<float>(numIn_ - numOut_);
  }

假设 filter A 花 1000 clocks 丢 900 行，约 1.11 clocks/drop；filter B 花 100 clocks 丢 10 行，是 10 clocks/drop。B 单次总耗时较低，但 A 更适合提前缩小后续输入。没有丢行时，当前实现返回累计 timeClocks_，并不是除以零或返回无限大。这个指标是局部自适应启发式，不是完整代价优化器，也不显式建模后续列压缩比和远程延迟。

源码摘录 · ScanSpec.cpp:65

bool ScanSpec::compareTimeToDropValue(
    const std::shared_ptr<ScanSpec>& left,
    const std::shared_ptr<ScanSpec>& right) {
  if (left->hasFilter() && right->hasFilter()) {
    if (!disableStatsBasedFilterReorder_ &&
        (left->selectivity_.numIn() || right->selectivity_.numIn())) {
      return left->selectivity_.timeToDropValue() <
          right->selectivity_.timeToDropValue();
    }
    // Integer filters are before other filters if there is no
    // history data.
    if (left->filter_ && right->filter_) {
      if (left->filter_->kind() == right->filter_->kind()) {
        return left->fieldName_ < right->fieldName_;
      }
      return left->filter_->kind() < right->filter_->kind();
    }
    // If hasFilter() is true but 'filter_' is nullptr, we have a filter
    // on complex type members. The simple type filter goes first.
    if (left->filter_) {
      return true;
    }
    if (right->filter_) {
      return false;
    }
    return left->fieldName_ < right->fieldName_;
  }

  if (left->hasFilter()) {
    return true;
  }
  if (right->hasFilter()) {
    return false;
  }
  return left->fieldName_ < right->fieldName_;
}

uint64_t ScanSpec::newRead() {
  // NOTE: in case of split preload, a new split might see zero reads but
  // non-empty filter stats. Hence we need to avoid stats triggered filter
  // reordering even on the first read if 'disableStatsBasedFilterReorder_' is
  // set.
  if (numReads_ == 0 ||
      (!disableStatsBasedFilterReorder_ &&
       !std::is_sorted(
           children_.cbegin(),
           children_.cend(),
           [this](
               const std::shared_ptr<ScanSpec>& left,
               const std::shared_ptr<ScanSpec>& right) {
             return compareTimeToDropValue(left, right);
           }))) {
    reorder();
  }
  return ++numReads_;
}

第一次 newRead 会排序；有统计时，若现有 children 不再有序，会再次排序。disableStatsBasedFilterReorder 禁止统计触发的适应。无历史时，两个有 filter 的字段按 FilterKind、simple filter 与 complex child filter、字段名排序；有 filter 的字段在无 filter 字段前。同 kind 的 nationkey/regionkey 因字段名得到确定的初始次序。

Struct reader 在带 filter 的 child 外放 SelectivityTimer，记录本次 activeRows 数和输出数，并从计时里扣掉 reader 初始化耗时，减少字典准备等一次性开销对稳态过滤成本的干扰。动态 filter 到达时会重置 filter cache，ScanSpec adaptation 也可以跨 split 移交。

10.3 nation 示例：把数据、计划与 RowSet 对齐

保留原分享的 schema 与 plan；RowSet 下标从 0 开始

nationkey: BIGINT
name:      VARCHAR
regionkey: BIGINT
comment:   VARCHAR

TableScan[0][
  table: hive_table,
  range filters: [
    (nationkey, BigintRange: [0,3] no nulls),
    (regionkey, BigintRange: [0,0] no nulls)
  ]
] -> nationkey, name, regionkey, comment

batch row / nationkey	name	regionkey	comment：原截图可见前缀
0	ALGERIA	0	haggle. carefully final deposits detect slyly agai
1	ARGENTINA	1	al foxes promise slyly according to the regular accoun…
2	BRAZIL	1	y alongside of the pending deposits. carefully special…
3	CANADA	1	eas hang ironic, silent packages. slyly regular packag…
4	EGYPT	4	y above the carefully unusual theodolites. final dugou…
5	ETHIOPIA	0	ven packages wake quickly. regu
6	FRANCE	3	refully final requests. regular, ironi
7	GERMANY	3	l platelets. regular accounts x-ray: unusual, regular…
8	INDIA	2	ss excuses cajole slyly across the packages. deposits…
9	INDONESIA	2	slyly express asymptotes. regular deposits haggle sly…
10	IRAN	4	efully alongside of the slyly final dependencies.

图 12 · nationkey 留下 rows=[0,1,2,3]，regionkey 再留下 row 0　打开原图

第一轮扫描 nationkey，输入 11 行，留下 [0,1,2,3]；第二轮 regionkey 只访问这四行，值分别是 [0,1,1,1]，只留下 row 0。row 5 虽然 regionkey=0，但它已经被 nationkey 排除，不会重新出现。于是 name/comment 的最终选择只需要 ALGERIA 这一行。图中的 batch row 与 nationkey 恰好相同，只是数据特点，不是 reader 的通用假设。

ScanSpec 初始顺序：nationkey, name, regionkey, comment
无历史时重排结果：nationkey, regionkey, comment, name
reader child 下标：保持构造时的映射
输出 channel：     仍是查询要求的顺序

主例中 nationkey 是带 filter 且要保留值的列；regionkey 没有 reader filter，所以不因参与 remaining expression 就自动加入这套 filter 重排。它是否在表达式求值时加载由 ExprSet 的需求决定。ScanSpec 优化的是 reader 内的过滤次序，不能替代表达式层的求值策略。

11. read 与 getValues：先确定选择，再对齐列结果

继续看原双条件例，其 plan 同时投影 nationkey、name、regionkey、comment，所以两个过滤列也要保留值。nationkey 先留四行，regionkey 再留一行；早读列已经保存的四个值不能直接作为最终结果。Struct reader 用 read/getValues 将“做过滤并暂存”与“按最终选择生成输出”分开。若只投影 name/comment，两个整数列可以是 filter-only，不应再宣称它们一定保存了待输出的 values。

图 13 · 早读列暂存四个值，后读列继续缩小 RowSet，最后再 compact　打开原图

根 reader 的 next 先构造本 batch 的 dense rows，调用 read，再把最终 outputRows 交给 getValues。这两个阶段的边界是确定剩余行与生成对齐的输出。read 并不是“先把所有列完整读完”，payload 正是在这里被允许跳过的。

read() 的 child 遍历 · SelectiveStructColumnReader.cpp:472

  const auto& childSpecs = scanSpec_->children();
  // When using name-based mapping, empty child specs are valid
  // (e.g., when all struct fields are renamed/deleted).
  if (columnReaderOptions_.columnMappingMode_ != ColumnMappingMode::kName) {
    VELOX_CHECK(!childSpecs.empty());
  }
  for (size_t i = 0; i < childSpecs.size(); ++i) {
    const auto& childSpec = childSpecs[i];

    if (childSpec->deltaUpdate()) {
      // Will make LazyVector.
      continue;
    }

    if (isChildConstant(*childSpec)) {
      if (!testFilterOnConstant(*childSpec)) {
        activeRows = {};
        break;
      }
      continue;
    }

    if (!childSpec->readFromFile()) {
      VELOX_CHECK(!childSpec->hasFilter());
      continue;
    }

    const auto fieldIndex = childSpec->subscript();
    auto* reader = children_.at(fieldIndex);
    if (reader->isTopLevel() && childSpec->projectOut() &&
        !childSpec->hasFilter() && generateLazyChildren_) {
      // Will make a LazyVector (with or without transform).
      continue;
    }

    advanceFieldReader(reader, offset);
    if (childSpec->hasFilter()) {
      {
        SelectivityTimer timer(childSpec->selectivity(), activeRows.size());

        reader->resetInitTimeClocks();
        reader->readWithTiming(offset, activeRows, structNulls);

        // Exclude initialization time.
        timer.subtract(reader->initTimeClocks());

        activeRows = reader->outputRows();
        childSpec->selectivity().addOutput(activeRows.size());
      }
      if (activeRows.empty()) {
        break;
      }
    } else {
      reader->readWithTiming(offset, activeRows, structNulls);
    }
  }

进入 child 循环之前还会处理 deleted rows/random skip 等 mutation 和 struct 的 is-null/is-not-null filter。常量字段无需文件 reader，但 filter 可使整个 activeRows 为空；非文件合成字段也直接跳过。普通字段只有同时满足 reader isTopLevel、projectOut、没有 filter、generateLazyChildren 时，才跳过本次读并准备后续 LazyVector。不是看到“无 filter”就总能 lazy。

每个带 filter 的 child 接收当前 activeRows，完成 read 后用 child->outputRows() 更新它；一旦变空立即停止后续列读取。没有 filter 且不能 lazy 的 child 按当前 activeRows 读取。最后记录 parent nulls、lazyVectorReadOffset_ 和本 batch 的 readOffset。Parquet 的 advanceFieldReader 是 no-op，因为所有 readers 一起切 RowGroup；不能把通用 DWIO 的按需跨组行为强加给 Parquet。

11.1 早读列为什么还需要 compact

nationkey read 保存了原 rows=[0,1,2,3] 的四个值；regionkey 的 filter 最后只留下 row 0。getValues([0]) 时，nationkey 必须从已读结果中选出对应值并同步 nulls。否则 nationkey 是四行、regionkey 是一行，RowVector 就失去列之间的对齐。其他数据例中 row indices 可能稀疏且与值不同，所以 compact 是按行映射，不是直接截取前几个整数。

getValues() 的立即物化与 lazy 分支 · SelectiveStructColumnReader.cpp:726

    if (childSpec->hasFilter() || !children_[index]->isTopLevel() ||
        !generateLazyChildren_) {
      children_[index]->getValues(rows, &childResult);
      continue;
    }

    // LazyVector result.
    setOutputRowsForLazy(rows);
    // When the child has a transform (e.g., extraction pushdown), the lazy
    // vector type is the transform's output type, not the file column type.
    auto lazyType =
        (childSpec->hasTransform() && childSpec->transformOutputType())
        ? childSpec->transformOutputType()
        : resultRow->type()->childAt(channel);
    setLazyField(
        makeColumnLoader(index), lazyType, rows.size(), pool_, childResult);
  }

getValues 还处理常量列、missing child、row number、composite field 等输出；只有普通 projected 字段才进入相应 reader 或 lazy loader。通用基类中还存在 transform/delta-update 分支，但此处 Parquet 工厂明确拒绝 extraction pushdown，不能因为基类有代码就声称格式全部支持。

这不是两遍全量扫描。read 过程中已经对需要的列执行真实解码和 filter，getValues 复用这些结果并 compact；未读取的 projected child 可以在这时创建 LazyVector。主例还多一层 remaining filter，所以上层拿到 getValues 的结果后，仍可能继续缩小 payload 的加载集合。

12. 从整数 reader 到 ColumnVisitor：过滤进入解码循环

沿 nationkey 继续往下，范围 filter 最终不应成为每个值都要穿过的一长串虚函数。公共 SelectiveIntegerColumnReader 将 FilterKind、值宽度、dense/sparse 与提取策略分派成模板实例，Parquet 层再接到实际 page decoder。下面的代码展示这个分派边界。

图 14 · FilterKind、dense/sparse、C++ 值宽度与提取策略的分派　打开原图

IntegerColumnReader::read 先按文件整数宽度 prepareRead，再进入 SelectiveIntegerColumnReader::readCommon。模板参数 kEncodingHasNulls 告诉公共层 encoding 可能含空值；dense/sparse 由 rows 的形态决定。下面保留完整 readCommon，可以看到四个常见组合和 ValueHook 分支，避免用一个固定 true 模板实例冒充所有情况。

源码摘录 · SelectiveIntegerColumnReader.h:253

template <typename Reader, bool kEncodingHasNulls>
void SelectiveIntegerColumnReader::readCommon(const RowSet& rows) {
  const bool isDense = rows.back() == rows.size() - 1;
  auto* filter = scanSpec_->filter() ? scanSpec_->filter() : &alwaysTrue();
  if (scanSpec_->keepValues()) {
    if (scanSpec_->valueHook()) {
      if (isDense) {
        processValueHook<Reader, true>(rows, scanSpec_->valueHook());
      } else {
        processValueHook<Reader, false>(rows, scanSpec_->valueHook());
      }
    } else {
      if (isDense) {
        processFilter<Reader, true, kEncodingHasNulls>(
            filter, ExtractToReader(this), rows);
      } else {
        processFilter<Reader, false, kEncodingHasNulls>(
            filter, ExtractToReader(this), rows);
      }
    }
  } else {
    if (isDense) {
      processFilter<Reader, true, kEncodingHasNulls>(
          filter, DropValues(), rows);
    } else {
      processFilter<Reader, false, kEncodingHasNulls>(
          filter, DropValues(), rows);
    }
  }
}

keepValues=true 时通常通过 ExtractToReader 保存通过值；只做过滤则使用 DropValues，减少不必要的值写入。有 ValueHook 时走 processValueHook，例如某些聚合可直接接收读取结果。PageReader::readWithVisitor 对 hasFilter 与 hasHook 同时为真使用 static_assert 限制，不能把所有优化无条件叠加。

processFilter 对 FilterKind 做一次分派，选择 common::BigintRange 等具体类型；readHelper 再根据 valueSize_ 选择 int16_t/int32_t/int64_t/int128_t 的 ColumnVisitor。模板使编译器看到具体 filter、提取策略和 dense 标志，减少每个值上的虚调用与动态分支。

processFilter 的 BigintRange 分支 · SelectiveIntegerColumnReader.h:163

    case velox::common::FilterKind::kBigintRange:
      static_cast<Reader*>(this)
          ->template readHelper<Reader, velox::common::BigintRange, isDense>(
              filter, rows, extractValues);
      break;

readHelper 的 int64_t 分支 · SelectiveIntegerColumnReader.h:101

    case 8:
      reinterpret_cast<Reader*>(this)->Reader::readWithVisitor(
          rows,
          ColumnVisitor<int64_t, TFilter, ExtractValues, isDense>(
              *static_cast<const TFilter*>(filter), this, rows, extractValues));
      break;

ParquetData 把 visitor 转交 PageReader，PageReader 再根据 page encoding 选择 decoder。逐值解释可以理解为 decode 一个值后调用 visitor.process；实际 hot path 还可能成批 decode/filter，字典 filter cache 也能复用同一 dictionary value 的判断。整数 bulk 路径不是所有 encoding 都有：当前 IntegerColumnReader 对 delta binary packed、部分 decimal 表示有显式限制。SIMD 优化必须落到具体路径，不能写成“所有 SubfieldFilter 都 SIMD”。

整数 reader 使用 fileType 确定 prepareRead 的宽度，在 getValues 中依据 requestedType 形成目标 Vector。字符串 reader 则还要处理 StringView 和字符串 buffer 生命周期。两个路径都遵守同一个 RowSet 契约，但不能因为接口相同就假设每种 encoding 都有一样的批量化、零拷贝或 SIMD 能力。

13. ExprSet 与 LazyVector：把上层选择传回 payload reader

主例在 reader 中只有 nationkey 范围 filter；加法留给 FileDataSource。reader 输出四行后，ExprSet 按这四个逻辑位置访问 regionkey，得到 [0,2,3,4] 的和。仅位置 3 通过，于是 name/comment 外面可以包一层 indices=[3] 的 dictionary；下游第一次访问时才请求这个位置。下面把这件事推广到稀疏 reader 输出。

13.1 单列 filter 和表达式层 filter 的边界

SubfieldFilter 是 reader 能识别的字段或 subfield 条件，例如整数范围、字符串范围、null 检查等。复杂函数、不能提取成这些条件的单列表达式，或 c0+c1>10 这样的多列计算，交给 remaining ExprSet。并不是“单列一定下推，多列一定不下推”：FileDataSource 先调用 extractFiltersFromRemainingFilter，把能提取的部分放回 filters_，其余才编译为表达式。

payload 是不参与当前 predicate 的投影列。LazyVector 是带逻辑长度和 VectorLoader 的延迟表示，不保证文件 I/O 尚未发生：可能已经 prefetch 了压缩数据，只是尚未 decode 成 values。区分“延迟 stream I/O”和“延迟 value materialization”十分关键。

13.2 Parquet 的真实 lazy 条件，不由 remainingFilterColumns 直接决定

FileDataSource 确实收集 remainingFilterColumns_ 并传到 RowReaderOptions，但在本文版本中，直接使用 remainingFilterColumns() 排除 lazy I/O 列的是 SelectiveNimbleRowReader::computeLazyIoColumns()。它控制 Nimble 的 stream I/O 延迟集合，也不等于所有格式的统一 eager-vector 开关。Parquet 的 StructColumnReader 没有消费这个集合；其 lazy 条件是read/getValues 一节的源码中的 top-level/projectOut/filter/generateLazyChildren。

因此，Parquet reader 返回时，remaining filter 引用的列仍可能是 LazyVector。ExprSet 求值会加载实际需要的字段；对多次引用或 shouldEagerlyMaterialize 判断出的字段，FileDataSource 在 evaluateRemainingFilter 前显式 ensureLoadedRows，避免同一个 lazy child 在多个表达式分支中被不一致地局部加载。过滤需要但 SQL 不投影的字段会补进 readerOutputType，求值后再去掉。

源码摘录 · FileDataSource.cpp:631

vector_size_t FileDataSource::evaluateRemainingFilter(RowVectorPtr& rowVector) {
  for (auto fieldIndex : multiReferencedFields_) {
    LazyVector::ensureLoadedRows(
        rowVector->childAt(fieldIndex),
        filterRows_,
        filterLazyDecoded_,
        filterLazyBaseRows_);
  }
  CpuWallTiming filterTiming;
  vector_size_t rowsRemaining{0};
  {
    CpuWallTimer timer(filterTiming);
    expressionEvaluator_->evaluate(
        remainingFilterExprSet_.get(), filterRows_, *rowVector, filterResult_);
    rowsRemaining = exec::processFilterResults(
        filterResult_, filterRows_, filterEvalCtx_, pool_);
  }
  totalRemainingFilterTime_.fetch_add(
      filterTiming.wallNanos, std::memory_order_relaxed);
  totalRemainingFilterCpuTime_.fetch_add(
      filterTiming.cpuNanos, std::memory_order_relaxed);
  return rowsRemaining;
}

13.3 dictionary indices 怎样变成下层 RowSet

图 15 · remaining indices 与 outputRows 组合，映射回 payload reader 的原始行号　打开原图

独立看一个四行输出例子：reader filter 留下原 batch rows=[2,5,8,9]，输出 RowVector 的逻辑位置是 [0,1,2,3]。remaining filter 留下输出位置 [1,3]，FileDataSource 用 selectedIndices=[1,3] 调用 wrapChild。这个 dictionary 可以包在 LazyVector 外面；下游需要 payload 时，加载选择透过 dictionary 到达 lazy 的逻辑位置。

ColumnLoader 把逻辑位置转换为原 batch 行号 · ColumnLoader.cpp:41

  if (rows.size() == outputRows.size()) {
    // All the rows planned at creation are accessed.
    effectiveRows = outputRows;
  } else {
    // rows is a set of indices into outputRows. There has been a
    // selection between creation and loading.
    selectedRows.resize(rows.size());
    VELOX_DCHECK(!selectedRows.empty());
    for (auto i = 0; i < rows.size(); ++i) {
      selectedRows[i] = outputRows[rows[i]];
    }
    effectiveRows = selectedRows;
  }

如果访问全部 planned rows，loader 可直接复用父 reader 的 outputRows；访问子集时，对每个 rows[i] 取 outputRows[rows[i]]。本例 [1,3] 映射回 [5,9]，这才是交给 payload fieldReader 的 RowSet。加载后若需要保持原 lazy vector 的位置空间，scatter 再用 dictionary 把紧凑结果放回对应逻辑位置。

13.4 lazy 的生命周期约束

ColumnLoader 保存父 reader、field reader、创建版本和本批偏移。实际加载首先校验 version==structReader->numReads()，失败信息是 “Loading LazyVector after the enclosing reader has moved”。保留一个旧 batch 的未加载 LazyVector，并让 reader 继续 next，不能被视为无限期可用的快照。buffer 共享和 reader 游标生命周期是两个不同问题。

父 reader 的 lazyVectorReadOffset 用于定位该 batch；parent nulls 和 nested loadable rows 也要传递。loader 会先调用 formatData.loadLazyInputStreams()，然后 fieldReader.readWithTiming(effectiveRows)，最后 getValues 或 ValueHook。接口中存在延迟 stream 钩子，不代表 Parquet 必然在此首次发 I/O；Parquet streams 通常早在 RowGroup scheduling 时注册。

对于 shouldEagerlyMaterialize，代码实际检查表达式是否保证在不增大的 selection 上求参，以及引用该字段的输入是否包含条件分支。multiReferencedFields_ 这个名字不能简单解释成“凡引用两次就全部 eager”；当前构造代码仅对已在输出 readColumnNames 中的字段记录该预加载动作。对所有场景的判断应以条件和调用点为准。

现有测试：TableScan 的普通投影输出确实可以是 LazyVector · ParquetTableScanTest.cpp:496

TEST_F(ParquetTableScanTest, lazy) {
  auto filePath = getExampleFilePath("sample.parquet");
  auto schema = ROW({"a", "b"}, {BIGINT(), DOUBLE()});
  CursorParameters params;
  params.copyResult = false;
  params.planNode = PlanBuilder().tableScan(schema).planNode();
  auto cursor = TaskCursor::create(params);
  cursor->task()->addSplit("0", exec::Split(makeSplit(filePath)));
  cursor->task()->noMoreSplits("0");
  int rows = 0;
  while (cursor->moveNext()) {
    auto* result = cursor->current()->asUnchecked<RowVector>();
    ASSERT_TRUE(result->childAt(0)->isLazy());
    ASSERT_TRUE(result->childAt(1)->isLazy());
    rows += result->size();
  }
  ASSERT_EQ(rows, 20);
  ASSERT_TRUE(waitForTaskCompletion(cursor->task().get()));
}

14. 嵌套读取：从顶层行到 rep/def levels 与元素范围

前面的 nation 是标量列，一条顶层行对应一个 scalar 位置。Hive 查询一旦投影 ARRAY、MAP、ROW，就不能继续把 leaf decoder 的位置当成顶层行号。Parquet 的 physical leaf streams 还必须表达集合长度、元素边界和每一层的 null。先把这层数据模型展开，再回到 PageReader 的坐标。

14.1 层级来自 schema path

保留原分享的 AddressBook 示例。这里 contacts 是 repeated group，因此逻辑上是 ARRAY(ROW(name VARCHAR, blockID BIGINT))，而非单个 struct。每条 leaf column 保存沿自身路径的 levels 和实际值：owner、contacts.name、contacts.blockID。

Parquet schema 示例：旧式 repeated group；UTF8 标注字符串语义

message AddressBook {
  required binary owner (UTF8);
  repeated group contacts {
    required binary name (UTF8);
    optional int64 blockID;
  }
}

图 16 · 从 AddressBook 根到各个叶子的 schema 路径　打开原图

叶子列	路径	maxDL	maxRL
owner	required owner	0	0
contacts.name	repeated contacts → required name	1	1
contacts.blockID	repeated contacts → optional blockID	2	1

maxRL 是路径中 repeated 节点数；maxDL 是 optional 与 repeated 节点数。required 不增加 DL。definition level 说明路径定义到了哪一层，repetition level 说明当前条目与前一个条目在哪层重复结构中相接。对本例，RL=0 开始一个新的 AddressBook，RL=1 表示同一 AddressBook 的下一个 contact；这条解释不能直接套到有多层 repeated 的 schema。

列	DL	含义
contacts.name	0	没有 contact 元素：空列表 marker
contacts.name	1	contact 存在，required name 有值
contacts.blockID	0	没有 contact 元素：空列表 marker
contacts.blockID	1	contact 存在，但 optional blockID 为 null
contacts.blockID	2	blockID 有实际值

14.2 三条逻辑记录，展开后并不是三个 leaf values

Row	owner	contacts
1	A	[]
2	B	[{name:n1, blockID:null}]
3	C	[{name:n2, blockID:10}, {name:n3, blockID:20}]

图 17 · levels、leaf values 与重建的 offsets / sizes　打开原图

owner 的 Row	Value	DL	RL
1	A	0	0
2	B	0	0
3	C	0	0

contacts.name 的 Row	Value / marker	DL	RL
1	空列表 marker；不存 value	0	0
2	n1	1	0
3	n2	1	0
3	n3	1	1

contacts.blockID 的 Row	Value / marker	DL	RL
1	空列表 marker；不存 value	0	0
2	null；不存 value	1	0
3	10	2	0
3	20	2	1

第一行 contacts=[]，所以 name/blockID 两条叶子流都需要表达“这一条 parent 没有元素”，但没有相应 encoded value。第二行 blockID=null，同样不存 leaf value，却需要 DL=1 区分“元素存在但字段为空”。第三行的第一个 contact 用 RL=0 开始新 parent，第二个用 RL=1 延续这个 parent。owner 的 maxDL/maxRL 均为零，level stream 可以省略，概念上相当于每项都是 0。

重建后 contacts 的 sizes=[0,1,2]、offsets=[0,0,1]。name child 有 [n1,n2,n3]，blockID child 的逻辑值为 [null,10,20]；其 values buffer 的 null 位置不代表一定保存了有效整数。三个 parent rows、四个 level entries、两个实际 blockID values，是三种不同计数。

14.3 现代 LIST 为什么不能复用同一组 DL 数字

对比 schema：明确列出 nullable list 和 nullable element

optional group contacts (LIST) {
  repeated group list {
    optional group element {
      required binary name (UTF8);
      optional int64 blockID;
    }
  }
}

在这份现代 LIST 示例中，到 name 的 maxDL=3，到 blockID 的 maxDL=4，maxRL 仍是 1。DL=0 可以表达 null list，DL=1 表达 empty list，DL=2 表达 null element；再往下才是 element 内的字段状态。换一份 element required 的 schema，数值又会变化。2026-09-26 基准中的 legacy nested schema 修正涉及 maxDefine 传播，说明 reader 必须按真实路径计算，不能从 SQL ARRAY 类型直接猜 levels。

Parquet 的 ListColumnReader / MapColumnReader 通过共同的 repeated-reader 逻辑取得 lengths/nulls，再将所需父行转换为元素范围，交给子 reader。getValues 最后重建 offsets、sizes 和 element vectors。投影裁剪可能只保留 contacts.blockID，但 contacts 的边界仍然需要 rep/def 信息；不能只读取 non-null 的 blockID 值就复原原始记录。

15. PageReader：跨页坐标、字典转换与 null bitmap

到 PageReader 时，输入不是 SQL 表达式，而是 visitor 指定的一组目标位置，以及本 ColumnChunk 的 stream 和编码状态。一次 visitor 可能跨多个 page，page 内也可能包含未选择的位置。PageReader 要同时维持三个一致性条件：decoder 在正确位置、输出值连续排列、输出 nulls 与 outputRows 指向同一批结果。

15.1 RowSet 不是一个固定的文件绝对坐标

PageReader 中 visitorRows 是当前 visit 的相对行号，visitBase_ 由 startVisit 从 firstUnvisited_ 取得。对普通标量列，可以把两者之和理解为当前 ColumnChunk 对应的行位置；重复嵌套列的 leaf level/value 数不等于顶层 RowVector 行数，必须先经结构层转换。原附录把所有 visitorRows 直接叫“全局绝对行号”，会混淆层级。

rowsForPage 首先判断 visitor 是否已消费完；然后计算下一目标 rowZero。如果它超出当前 page，seekToPage 逐步解析 PageHeader 并移动流。遇到 numRowsInPage_==0 表示 chunk 末尾，函数返回 false。已知 page 行范围且目标在后面时，可 skip 该页数据；重复列可能先需要读 levels 才能知道边界，跳过不是无条件的。

图 18 · 从 batch rows 到 Page B decoder-local rows，再恢复 outputRows　打开原图

15.2 一个具体的跨页坐标算例

设 visitBase=1000，visitorRows=[5,8,210,215]。Page A 覆盖 [1000,1200)，Page B 覆盖 [1200,1400)。page 切片把前两项交给 A，后两项交给 B。到了 B，pageOffset=1200−1000=200，rowNumberBias=210，先 skip(10)，让 decoder 位于绝对行 1210，再用 [0,5] 表示两个目标。filter 产生的输出行号加回 210 后，才恢复成本 batch 的 [210,215]。

还有一个完全不同的 numValuesBias：它表示之前 page 已经向 values buffer 写入了多少项，用来定位当前 page 结果的追加位置。输入行号的 bias 与输出 buffer 的 bias 不能互换。最后 firstUnvisited 更新为最后一个目标绝对位置加一，即本例 1216。

rowsForPage 的 view 快路径与 SIMD rebias · PageReader.cpp:1230

  // If the page did not change and this is the first call, we can return a view
  // on the original visitor rows.
  if (rowOfPage_ == initialRowOfPage_ && currentVisitorRow_ == 0) {
    nulls =
        readNulls(visitorRows_[numToVisit - 1] + 1, reader.nullsInReadRange());
    rowNumberBias_ = 0;
    rows = folly::Range<const vector_size_t*>(visitorRows_, numToVisit);
  } else {
    // We scale row numbers to be relative to first on this page.
    auto pageOffset = rowOfPage_ - visitBase_;
    rowNumberBias_ = visitorRows_[currentVisitorRow_];
    skip(rowNumberBias_ - pageOffset);
    // The decoder is positioned at 'visitorRows_[currentVisitorRow_']'
    // We copy the rows to visit with a bias, so that the first to visit has
    // offset 0.
    rowsCopy_->resize(numToVisit);
    auto copy = rowsCopy_->data();
    // Subtract 'rowNumberBias_' from the rows to visit on this page.
    // 'copy' has a writable tail of SIMD width, so no special case for end of
    // loop.
    for (auto i = 0; i < numToVisit; i += xsimd::batch<int32_t>::size) {
      auto numbers = xsimd::batch<int32_t>::load_unaligned(
                         &visitorRows_[i + currentVisitorRow_]) -
          rowNumberBias_;
      numbers.store_unaligned(copy);
      copy += xsimd::batch<int32_t>::size;
    }
    nulls = readNulls(rowsCopy_->back() + 1, reader.nullsInReadRange());
    rows = folly::Range<const vector_size_t*>(
        rowsCopy_->data(), rowsCopy_->size());
  }

第一次调用且 page 没变时，代码可以直接返回原 visitorRows 的 view，无需分配 rowsCopy_。通用路径使用 xsimd 批量减 bias；rowsCopy_ 的 raw buffer 为 SIMD 宽度预留可写尾部，不能把这个循环复制到长度恰好等于 n 的普通数组上。readNulls 的范围是最后一个 local row + 1，包含目标之间的间隔，以保持 decoder 的 null/value 计数同步。

15.3 dictionary/direct 切换与 filter cache

同一 ColumnChunk 的 data pages 可以先 dictionary 后 plain。进入 dictionary page 时，scanState 指向当前字典，并在有 filter 时建立 filter cache。切到 direct page 前，如果先前积累的结果还是 dictionary ids，就要 dedictionarize。代码临时挂上已累计的 null bitmap，避免把 null 位置的未定义字典索引当作有效值读取；转换完成后清空 dictionary state，再准备新页 nulls。

15.4 多页 null bitmap 不能按输入长度盲目拼接

readWithVisitor 每页先记录 numValuesBeforePage，设置 visitor 的 rows 和 output bias，再 callDecoder。跨页时，BitConcatenation 按实际产出的值数拼接 nulls。没有 null 的页追加 ones（Velox 位图中 1 表示非空）；returnReaderNulls 的快路径直接取 reader 的输入 nulls；普通过滤/compact 路径取 mutableNulls，起点还要看 nullsFromFastPath 是否从零写入。

如果前一批输出仍共享 multiPageNulls_，代码先放弃复用旧 buffer，避免修改已交付的数据。delta binary packed decoder 耗尽时还校验内部指针到达页尾。最后把拼接结果 setNulls 回 reader，并用 rowNumberBias 修正通过行号。完整的两个函数保留在附录，正文逐一解释其状态。

PageHeader 中的长度来自文件，不能在边界检查前直接相加或传给解压器。当前 V1/V2 路径对负长度、level 区域越界、解压大小不匹配等设置检查；ParquetPageReaderTest 覆盖 corruptDefineLengthV1、corruptLevelLengthsV2、snappyUncompressedSizeMismatch 等情形。这些保护与 SIMD 尾部 padding 一样，都属于高性能 reader 的正确性前提。

16. BufferedInput：列的读取需求怎样变成 I/O 计划

值层主线已经闭合：RowSet 进入 decoder，过滤结果返回 Vector。现在沿 PageReader 持有的 SeekableInputStream 追踪字节来源。一个 ColumnChunk 的 stream 不一定独占一次读取：同一 RowGroup 中不同列的 Regions 会先汇集到 BufferedInput，再按距离、大小和历史访问率规划。

先限定这里的 I/O 路径。默认 std::unique_ptr<dwio::common::BufferedInput> createBufferedInput( 在 ConnectorQueryCtx 提供 cache 时创建 CachedBufferedInput；没有 cache 的 Parquet 路径创建 DirectBufferedInput。Local Cache 属于前一分支，可结合 AsyncDataCache 与配置的 SSD cache 使用；它不是每一个 Parquet reader 都自带的缓存。下面具体分析 Direct 分支，不能把其中每次 ReadFile 读取描述成一次缓存查找。

图 19 · 两个 Region 合并时，gap 成为额外读取量　打开原图

16.1 enqueue 注册，load 规划

DirectBufferedInput::enqueue 创建 DirectInputStream，记录 Region、TrackingId 和 stream 指针，返回流对象而不立即读取。zero-length region 返回空数组流；whole-file preload 情况直接复用 preload data，通常不再登记 coalesced request。开始新一轮 enqueue 时，旧 coalescedLoads 和 stream-to-load 映射会清理，旧结果不能被当成永久索引。

load 把本轮 requests_ 移走，确保后续 enqueue 不再混入当前计划。具有 sequentialFile 标识或空 tracking id 的 request 走 prefetchAnyway；其他请求通过 ScanTracker 的历史 reference/read 信息估计 adjustedReadPct。两类分别排序，先处理 prefetchable 组；moveCoalesced 还会把已落在这些合并区间中的其他请求吸收进去，然后再规划 sparse 组。

源码摘录 · DirectBufferedInput.cpp:99

void DirectBufferedInput::load(const LogType /*unused*/) {
  // After load, new requests cannot be merged into pre-load ones.
  auto requests = std::move(requests_);
  std::vector<LoadRequest*> storageLoad[2];
  for (auto& request : requests) {
    cache::TrackingData trackingData;
    const bool prefetchAnyway = request.trackingId.empty() ||
        request.trackingId.id() == StreamIdentifier::sequentialFile().id_;
    if (!prefetchAnyway && tracker_) {
      trackingData = tracker_->trackingData(request.trackingId);
    }
    const int loadIndex =
        (prefetchAnyway || isPrefetchablePct(adjustedReadPct(trackingData)))
        ? 1
        : 0;
    storageLoad[loadIndex].push_back(&request);
  }
  std::sort(storageLoad[1].begin(), storageLoad[1].end(), lessThan);
  std::sort(storageLoad[0].begin(), storageLoad[0].end(), lessThan);
  std::vector<int32_t> groupEnds[2];
  groupEnds[1] = groupRequests(storageLoad[1], true);
  moveCoalesced(
      storageLoad[1],
      groupEnds[1],
      storageLoad[0],
      [](auto* request) { return request->region.offset; },
      [](auto* request) {
        return request->region.offset + request->region.length;
      });
  groupEnds[0] = groupRequests(storageLoad[0], false);
  readRegions(storageLoad[1], true, groupEnds[1]);
  readRegions(storageLoad[0], false, groupEnds[0]);
}

16.2 两个预算决定“合并到多大”

maxCoalesceDistance 控制相邻请求允许跨过多远的 gap。prefetch 类采用 maxCoalesceBytes 追求较大的吞吐读取；sparse 类采用 loadQuantum 限制过读。单个 request 计入分组的大小还可能按 quantum 截断，不是简单把 region.length 直接求和后画一个大矩形。图中的 A+B 只是在距离与字节预算都满足时的算例。

readRegion 为一组合并请求创建 DirectCoalescedLoad，并把每个 stream 指向该 load。只有 prefetch 且 executor_ 存在时，readRegions 才提交后台任务。一个 non-prefetch 的单独 request 没有合并价值，不建立 load，stream 第一次 Next 时同步读取；多个 sparse requests 可以共享 planned load，由实际消费线程触发。

ScanTracker 的“是否经常被读取”与 ScanSpec 的“哪个 filter 更划算”是两套反馈：前者优化物理预取，后者优化逻辑列过滤顺序。两者有间接关系，却不共享一个所谓全局 cost model。

主例中 name/comment 虽然只有一行需要物化，其 chunk stream 仍可能早已登记，甚至被并入积极预取的范围。所以“LazyVector 省掉 10/11 的 payload values”不能直接换算成“少读 10/11 的远端字节”。真正少发哪些 I/O，需要继续看 stream 是否被消费、coalescing 是否跨 gap、whole-file preload 和 cache 是否命中。

17. DirectCoalescedLoad：分配、共享与成功发布

I/O 计划还需要落实到内存：一组 request 的数据是分别分配、共同分配，还是适合用小字节数组？这里的目标不只是降低 malloc 次数，还包括避免每个小 region 各自按页取整导致的浪费，同时保证 DirectInputStream 拿到的数据在使用期间有效。

DirectCoalescedLoad 表示一个合并加载任务，最终调用 ReadFile::read(ranges, offset)。ranges 可跨 MemoryPool 的多个连续 runs，也可包含 gap 占位；一次这层调用不保证等于一次操作系统 read，更不等于一次本地磁盘寻址。具体 ReadFile 可由远程存储等后端实现。

图 20 · 三种 buffer 表示及其交付给 DirectInputStream 的方式　打开原图

17.1 tiny、owned pages 与 shared allocation

computeLoadSizes 确定各 request 实际 requestBytes，普通大 request 分配页时会发生 page-rounding。shared allocation 受配置开关和累计过度分配是否超过最小 allocation run 的阈值影响；不能简单写成“大 request 一律 shared”。尾部 request 常按 loadQuantum 截断，位于合并组中间且要连接后续请求的 region 则可能一次完整读取，并检查 int32_t 容量上界。

• tinyData：小请求使用连续字节数组；数据被移交给对应 stream。
• ownedData：MemoryPool 分配非连续 pages；getData 转移 Allocation，Next 按其中连续 run 暴露数据。
• sharedData：slice 属于 load 内的共享 allocation；stream 保存 sharedPtr，并用 sharedHolder 持有 load，维持 slice 生命周期。

完全相同 offset/length 的 duplicate request 不重复生成底层读取。shared 路径下消费者可以共用同一只读 slice；非 shared 路径为不同消费者准备副本。getData 对已移交的 buffer 标记 consumed，避免再次移动同一 buffer；shared slice 则无需移动。这比“stream 只是一个指针，不拥有数据”复杂得多。

17.2 只有成功状态才能交付 buffer

成功发布与重复 stream 的数据交付 · DirectBufferedInput.cpp:525

LoadedBuffer DirectCoalescedLoad::getData(uint64_t offset) {
  // A failed read may have partially filled the request buffers. Only publish
  // them after the complete coalesced load has succeeded.
  if (state() != CoalescedLoad::State::kLoaded) {
    return {};
  }
  auto it = std::lower_bound(
      requests_.begin(),
      requests_.end(),
      offset,
      [](const auto& request, uint64_t targetOffset) {
        return request.region.offset < targetOffset;
      });
  if (it == requests_.end() || it->region.offset != offset) {
    return {};
  }
  if (it->buffer.sharedData != nullptr) {
    // Duplicates backed by the shared allocation share one read-only slice;
    // nothing is moved.
    return LoadedBuffer{
        .requestBytes = it->buffer.requestBytes,
        .sharedData = it->buffer.sharedData,
    };
  }
  // Skip buffers already handed to earlier duplicate streams.
  while (it != requests_.end() && it->region.offset == offset &&
         it->bufferConsumed) {
    ++it;
  }
  if (it == requests_.end() || it->region.offset != offset) {
    return {};
  }
  it->bufferConsumed = true;
  return std::move(it->buffer);
}

getData 首先检查 state()==kLoaded；失败或取消时返回空 LoadedBuffer，即使底层已经写入部分 bytes 也不交付。这条发布条件与 SharedPromise 是否 ready 分离，是后文异常路径正确性的关键。流离开已加载范围时会清理借用 slice，再走同步 quantum 读取。

这里的“页”指 MemoryPool 分配页，与 Parquet Data Page 不是同一对象；一个压缩 Data Page 可以跨多个内存 run，一个 allocation 也可以装多个 request。Next 暴露的连续区间不能直接当作一个完整的 Parquet page，PageReader 的 refill 必须处理跨 stream chunk 的内容。

18. 消费线程与预取线程：等待、成功和失败分别发生在哪里

下面画线程而不是只画对象。同一 DirectCoalescedLoad 的 loadOrFuture 可以被 I/O executor 或消费线程首先调用；谁把 Planned 改成 Loading，谁负责在锁外执行 loadData。已在执行的线程不会因为另一个 consumer 到来就转交工作。

图 21 · 预取线程先取得 load 时，查询线程同步等待 future　打开原图

18.1 先到的线程取得 Planned → Loading

loadOrFuture 在 mutex 下检查状态。Loaded/Cancelled 直接返回 true；如果已经 Loading，wait==nullptr 的调用者直接返回 false，提供 wait 的调用者获得 SharedPromise future 后返回 false；只有看到 Planned 的线程把状态改成 Loading，然后在锁外执行 loadData。这个线程可以是 I/O executor，也可以是先消费到此处的查询线程。

DirectInputStream::loadPosition 首次通过 bufferedInput_->coalescedLoad(this) 取出与当前 stream 关联的 load；映射项随之移除。若另一个线程正在加载，它会执行 waitFuture.wait()，当前调用线程停在这里。不能把它画成 Operator 把 future 交给 Driver 然后 off-thread，因为这条函数没有做这件事。

loadPosition 的同步等待与 buffer 接收 · DirectInputStream.cpp:219

    if (load != nullptr) {
      folly::SemiFuture<bool> waitFuture(false);
      uint64_t loadUs = 0;
      {
        MicrosecondWallTimer timer(&loadUs);
        if (!load->loadOrFuture(&waitFuture)) {
          waitFuture.wait();
        }
        loadedRegion_.offset = region_.offset;
        auto loaded = load->getData(region_.offset);
        loadedRegion_.length = loaded.requestBytes;
        loadedData_.set(std::move(loaded), load);
      }
      ioStats_->queryThreadIoLatencyUs().increment(loadUs);
      // DirectCoalescedLoad always reads from remote storage, not SSD.
      ioStats_->coalescedStorageLoadLatencyUs().increment(loadUs);

18.2 已规划数据不覆盖当前位置时

load 完成后，stream 通过 getData(region.offset) 接收 requestBytes 和一种 buffer 表示。当前位置若超出 loadedRegion，或当前 stream 没有 coalesced load，便从当前位置按 loadQuantum 计算下一段，调用 loadSync。Next 最后把当前连续 run 的指针和长度交给 PageReader，跨 run 时继续推进。whole-file preloaded 路径则先于这些逻辑，直接取已加载字节的 view。

18.3 Future ready 不代表成功，也不传递这个 load 的原始异常

图 22 · CoalescedLoad 的成功、取消与等待者通知　打开原图

源码摘录 · AsyncDataCache.cpp:390

bool CoalescedLoad::loadOrFuture(
    folly::SemiFuture<bool>* wait,
    bool ssdSavable) {
  {
    common::testutil::TestValue::adjust(
        "facebook::velox::cache::CoalescedLoad::loadOrFuture", this);
    std::lock_guard<std::mutex> l(mutex_);
    if (state_ == State::kCancelled || state_ == State::kLoaded) {
      return true;
    }
    if (state_ == State::kLoading) {
      if (wait == nullptr) {
        return false;
      }
      if (promise_ == nullptr) {
        promise_ = std::make_unique<folly::SharedPromise<bool>>();
      }
      *wait = promise_->getSemiFuture();
      return false;
    }

    VELOX_CHECK_EQ(State::kPlanned, state_);
    state_ = State::kLoading;
  }

  // Outside of 'mutex_'.
  common::testutil::TestValue::adjust(
      "facebook::velox::cache::CoalescedLoad::loadOrFuture::loading", this);
  try {
    const auto pins = loadData(/*prefetch=*/wait == nullptr);
    for (const auto& pin : pins) {
      auto* entry = pin.checkedEntry();
      VELOX_CHECK(entry->key().fileNum.hasValue());
      VELOX_CHECK(entry->isExclusive());
      entry->setExclusiveToShared(ssdSavable);
    }
    setEndState(State::kLoaded);
  } catch (std::exception&) {
    try {
      setEndState(State::kCancelled);
    } catch (std::exception&) {
      // May not throw from inside catch.
    }
    throw;
  }
  return true;
}

void CoalescedLoad::setEndState(State endState) {
  std::unique_ptr<folly::SharedPromise<bool>> promise;
  {
    std::lock_guard<std::mutex> l(mutex_);
    state_ = endState;
    promise.swap(promise_);
  }
  if (promise != nullptr) {
    promise->setValue(true);
  }
}

loadData 抛 std::exception 时，执行线程把状态改成 kCancelled，再重抛异常。setEndState 将 SharedPromise 移出 mutex 保护范围后 setValue(true)，成功与取消都会唤醒等待者；这里没有 setException。因此 wait 的意义是“先前的加载结束了”，不是“读取成功”或“收到原始 I/O 异常”。

等待者随后 getData 看到非 Loaded 得到空 buffer，DirectInputStream 会落到 loadSync，重新在消费路径读取；若该读取也失败，异常在当前查询调用栈中出现。预取任务本身的异常还受 executor 的任务执行方式影响，不能声称当前 SharedPromise 自动把后台异常传递给所有 query waiters。读取到的部分 buffer 始终受 getData 的状态检查保护。

reset 会 cancel 当前 loads，并清空 requests 与映射。这里的 cancel 调用 setEndState(kCancelled) 并通知等待者，不会抢占底层 ReadFile 的进行中调用。setEndState 也没有校验旧状态；如果 cancel 与已经执行的 loadData 交错，后者完成时仍可再写入 kLoaded。因此图中描述的是一次 load 的正常成功/异常路径，不能把 kCancelled 理解成对所有并发交错都不可退出的终态。后台闭包的 AsyncLoadHolder 保持 load 和 MemoryPool 引用，也是保证异步任务运行时对象有效的组成部分。

与此处不同，TableScan 的 split preload 使用 AsyncSource。AsyncSource 会保存 exception_ptr，并在消费者 move() 时 rethrow；DirectCoalescedLoad 的 SharedPromise 则仅作结束通知。这两种异常通道不能因为都出现 future 就合并为同一套机制。下一节把 split 层的所有权与接管过程单独列出。

19. Split preload、接管与 reader 的生命周期

TableScan::checkPreload 在配置允许、connector 支持且有 ioExecutor 时安装 split preloader。preload 创建 AsyncSource<DataSource>，捕获 output type、table handle、columns、connector、独立 ConnectorQueryCtx、Task 和 split；executor 调用 prepare，实际工作仍是 createDataSource + addSplit。它预制的是一个准备好的 DataSource，不只是预先读取一个裸 buffer。

图 23 · 后台准备下一 split，消费线程接管已有 DataSource 状态　打开原图

19.1 AsyncSource 如何把准备失败交给消费方

AsyncSource 的 mutex 保护 Init/Making/Prepared/Failed 等状态。prepare 取得 itemMaker 后在锁外执行；move 如果先到且仍是 Init，可以自己完成准备，如果已经 Making 则等待。如果 itemMaker 失败，makeItem 保存异常并通知等待者，move 检查 exception_ 后重新抛出。与 CoalescedLoad 的结束通知相比，AsyncSource 明确承担“把制作结果或原始异常交给消费者”的职责。

后台闭包捕获 Task 的 shared_ptr，是为了在准备可能晚于调度活动结束时保持所需 memory pools 存活，并在开始及关键阶段检查任务是否取消。取消不能理解为抢占底层已经开始的 ReadFile 调用。getSplit 取得 preparedDataSource 为空时检查 Task cancellation；其余错误仍应进入查询错误处理，而不是被当成一个空 split。

19.2 setFromDataSource 接管的不是一个裸 reader 指针

接管预制 split，同时保留扫描反馈和 I/O 统计 · FileDataSource.cpp:593

void FileDataSource::setFromDataSource(
    std::unique_ptr<DataSource> sourceUnique) {
  auto source = dynamic_cast<FileDataSource*>(sourceUnique.get());
  VELOX_CHECK_NOT_NULL(source, "Bad DataSource type");

  split_ = std::move(source->split_);
  runtimeStats_.skippedSplits += source->runtimeStats_.skippedSplits;
  runtimeStats_.processedSplits += source->runtimeStats_.processedSplits;
  runtimeStats_.skippedSplitBytes += source->runtimeStats_.skippedSplitBytes;
  readerOutputType_ = std::move(source->readerOutputType_);
  readerProducedType_ = std::move(source->readerProducedType_);
  extractionColumns_ = std::move(source->extractionColumns_);
  source->scanSpec_->moveAdaptationFrom(*scanSpec_);
  scanSpec_ = std::move(source->scanSpec_);
  metadataFilter_ = std::move(source->metadataFilter_);
  splitReader_ = std::move(source->splitReader_);
  splitReader_->setConnectorQueryCtx(connectorQueryCtx_);
  // New io will be accounted on the stats of 'source'. Add the existing
  // balance to that.
  source->dataIoStats_->merge(*dataIoStats_);
  dataIoStats_ = std::move(source->dataIoStats_);
  source->metadataIoStats_->merge(*metadataIoStats_);
  metadataIoStats_ = std::move(source->metadataIoStats_);
  source->ioStats_->merge(*ioStats_);
  ioStats_ = std::move(source->ioStats_);
}

旧扫描累积的 filter selectivity 通过 moveAdaptationFrom 合入新 ScanSpec，然后接管 split、readerOutputType、metadataFilter 和 splitReader。setConnectorQueryCtx 把 reader 重绑定到消费方的查询上下文；I/O stats 合并后采用预制 source 的统计对象，因为已启动的 I/O 还会继续向那里记账。只移动 splitReader 而不处理这些状态，会丢失反馈、统计或上下文关联。

19.3 几个必须维持的生命周期约束

对象	有效范围	容易误解的地方
DataSource / ScanSpec	一条 TableScan 的多次 split/batch，preload 接管会替换部分对象	保留的是可迁移的 adaptation，不是永远复用同一个 splitReader
splitReader / ParquetRowReader	当前文件 split 的扫描状态	resetSplit 后暂留 reader；下次 addSplit 销毁旧 splitReader
PageReader / decoder	某个 leaf 的当前 RowGroup，跨多次 batch 使用	seekToRowGroup 才安装新 PageReader，不是每批都重建
未加载的 LazyVector	依赖创建它的父 reader 版本和 batch offset	reader 前进后不能当作独立文件快照随时加载
异步 load / MemoryPool	后台闭包和 stream 共享引用覆盖实际使用期	释放调度表中的引用不等于立即取消执行或销毁数据

整 split 完成时 FileDataSource::resetSplit 清掉 split_，调用 splitReader->resetSplit，暂时保留 readers 以持有 adaptation；下一个 addSplit 才 reset 旧 splitReader。跨 split 的反馈存在于共享 ScanSpec 等对象中。无论是否 preload，旧 batch 的 lazy 生命周期约束仍需由执行框架和消费者遵守。

20. 用哪些指标判断过滤、延迟物化与预取的收益

主例的三个数字可以用作指标阅读的起点：扫描 11 行，reader 输出 4 行，remaining 输出 1 行。它们分别对应游标推进、reader filter 选择率和表达式选择率；不能只拿最后的 1/11 解释每一层成本。

图 24 · 从 metadata 到表达式层的收益与边界　打开原图

RowGroup 统计可以避免进入整组数据，但本来就已缓存/预载的数据不一定有新的网络请求可省。PageReader 的 target-row skip 要看能否知道页边界以及 levels 是否必须读取；page 内 filter 则主要减少结果写入和后续列的 decode。remaining filter 之后的 LazyVector 还能避免生成被过滤行的 payload values。

观察项	可解释的问题	不能直接推出的结论
scanned rows / output rows	过滤选择率与扫描推进量	输出减少 99% 不代表 storage bytes 同比减少
prefetch bytes / reference-read history	预取是否过于积极	prefetch 多不自动意味着慢，可能隐藏延迟
raw bytes / overread bytes	有效请求与 gap 开销	合并一次 ReadFile 调用不代表底层只有一次请求
queryThreadIoLatencyUs	消费线程同步读或等待的时间	与 Driver blocked time 不能直接混为同一指标
remaining filter CPU/wall time	表达式层过滤开销	不包括全部 reader decode 与后台 I/O
buffer ownership / retained pages	内存占用与共享时长	LazyVector 小并不代表预取 buffer 也小

本版本的 I/O runtime metrics 由 FileDataSource 汇总 DWIO 统计，还可由 ReadFile 层的 storageReadBytes 覆盖估算值。对优化做结论时，应该说明观察的是哪一层计数，并同时报告数据布局、过滤率、缓存状态和并发度。本文的数字数组是推演例子，没有把它们当作实测吞吐基准。

20.1 源码与测试怎样交叉核对

问题	本文核对的实现	相关已有测试（仅阅读，未运行整套 Velox）
输出是否可以 lazy	SelectiveStructColumnReader::getValues、ColumnLoader	ParquetTableScanTest.lazy
range filter 是否保持各列对齐	IntegerColumnReader、Struct read/getValues	ParquetReaderTest.readSampleBigintRangeFilter / intMultipleFilters
预取窗口是否随消费前进	ReaderBase::scheduleRowGroups	ParquetReaderTest.prefetchRowGroups
没有逻辑 child 的 struct 如何取 levels	StructColumnReader::ensureSyntheticRepDefSource	ParquetReaderTest.cheapestRepDefSource / legacyRepDefSource
V1/V2 page 边界与损坏输入	PageReader::prepareDataPageV1/V2、refill	ParquetPageReaderTest 的 corrupt* / refillSpansMultipleStreamChunks
footer 统计内存如何释放	ReaderBase / filterRowGroups	ParquetReaderTest.thriftMemoryReleasedForSkippedRowGroups

本文中的 nation 选择、remaining indices 组合、AddressBook levels 和跨页 bias 算例另做了独立算术检查。这验证的是解释中的数据关系；它不替代实际文件、codec、缓存、线程配置下的性能测量。

21. 从 Spark、native reader 与 Photon 看接口边界

读完 Velox 的完整路径，再回看行业方案，比较点就更具体：谁产生 selection，selection 能传到哪一层，payload 什么时候物化，JVM 与 native 之间如何交换向量。下面保留原分享的 Spark / native reader / Photon 背景，但将公开接口与 Velox 的具体实现分开。

Velox 原生 Parquet reader 沿用了 DWRF selective reader 的许多抽象：ScanSpec 表达读取要求，ColumnReader 接受 RowSet，ColumnVisitor 融合过滤与提取，VectorLoader 把计算层的访问需求带回读取层。原分享提及早期 DuckDB 路径的性能差距；没有同一数据、查询和版本的基准，不能把“只有 DWRF 的 50%”当作当前通用结论。原生 reader 的设计动机和社区讨论可见文末资料。

图 25 · 从先物化后过滤，到 predicate 先产生 selection　打开原图

Spark 社区的 LazyColumnVector 提案把一个问题提得很直接：如果 payload 未被访问，就不应先物化它。但实现不能停在 ColumnVector 的 wrapper：vectorized reader 要改变列访问次序，selection 要到达 parquet-mr 的 decoder，嵌套类型和 dictionary page 也要理解同一组行坐标。图中的“过滤在前”意味着先读 predicate 列并求值，不是在没有读取 predicate 数据时凭空执行过滤。

图 26 · JVM column reader 通过指针与 native reader 共享堆外状态和向量　打开原图

DBR/native reader 的公开材料展示了另一种组织方式：JVM vectorized reader 请求各列，native reader 接收堆外状态与 vector 指针，解码 predicate 列并生成 mask，再据此减少 payload 解码。Photon 使用原生向量化执行，但具体的预取、缓存和 page 选择策略应以公开材料为界，不能由示意图推断其闭源实现与 Velox 相同。原分享中有关作者去向和提案未完成原因的个人归因，不作为技术结论。

图 27 · 公开 Photon 示例计划中的 Adapter 与 Transition 边界　打开原图

在原分享所引用的 Photon 计划图中，JVM 侧的 FileScan 经 Adapter 进入 PhotonFilter、PhotonShuffle，随后通过 Transition 回到 JVM 的 ToObject。它展示的是原生执行片段如何接入已有计划，而不是声称整条链中的每个节点都被替换。native reader 的堆外 state、data/dictionary IDs 和 nulls vectors 又是更低一层的数据接口：Generated code 请求 batch 后，仍需要能够正确解释这些布局和有效行 mask。

Velox 把 selection 分成 reader 内和表达式层两段：SubfieldFilter 可以在单个字段的读取过程中判断；remaining filter 由 ExprSet 求值；两段得到的选择结果都可能继续减少 payload 的解码。后文将具体追踪它们之间的坐标转换。

22. 设计取舍：选择信息如何跨越计算与存储边界

这些实现面对的不是一个单独的“更快解码”问题，而是多个约束的协调：过滤条件直到运行时才知道，文件以压缩页组织，下游可能只访问一部分值，I/O 又希望用较大的请求隐藏延迟。设计的价值应从它如何化解这些矛盾来评估。

22.1 把行集合做成接口，跨过解码与计算边界

RowSet 很窄，却足以连接多个优化：过滤列只需交出通过行号，后续 reader 不必理解上一个 filter 的具体表达式；Lazy loader 只需把输出 indices 映射回父 reader 的 rows，就能把计算层的筛选带回 Parquet decoder。好处是模块职责清晰，代价是每一层必须维护一致的坐标、null 和生命周期。

22.2 两阶段给自适应读取顺序留出空间

若每读一列就立即固定最终输出，后续 filter 每次继续丢行，都要回头改已经产出的 vectors。read/getValues 分开后，过滤顺序可以变化，先读值只在最终 RowSet 确定后 compact，payload 还能继续 lazy。代价是 reader 需要暂存 values、outputRows 和 nulls，不能把解码对象看成无状态函数。

22.3 公共抽象与模板专用路径并存

格式无关类把 projection、filter、null、RowSet 和 Vector 组装共用起来；格式实现处理 encoding、page 和 levels。模板把 filter kind、值宽度与 dense 情形变成具体实例，减少热循环分支。代价是调用链更深、编译实例更多，阅读时容易把通用基类支持的能力误判为每种格式都支持。本文特别区分 Parquet 与 Nimble 的 lazy I/O 就是这个边界。

22.4 自适应预取不等于全局最优

ScanTracker 根据历史访问率避免总是预取所有 payload，coalescing 用少量 overread 换取更少的请求和更好的吞吐。但过滤相关性、数据分布变化以及不同列的访问热点会让历史失真。ScanSpec 的 timeToDropValue 同样只反映局部过滤代价。正确的工程做法是保留可观测统计和退路，而不是宣称一个 heuristic 在所有查询上都优于其他顺序。

这套 reader 最有价值的地方，是让“先筛哪些行、何时读哪些列、从哪里取得哪些字节、何时交付哪些 buffer”可以分别优化，又通过明确的 RowSet 和生命周期约束接起来。也正因为这些约束存在，解释性能必须落到具体路径，不能只列 LazyVector、SIMD、prefetch 等功能名。

22.5 把文件变化留在 split 边界，把查询意图保留在 ScanSpec

分区值、缺失字段、bucket conversion 都随文件变化；输出 schema 和过滤目标则属于查询。Hive/File 的分层让两类信息在 addSplit 时汇合，Reader/RowReader 的分层又把文件元数据和扫描游标分开。代价是继承链中出现多个 prepareSplit、两份 output type 和可移动的预制状态。理解时应沿真实调用点确认职责，不能仅凭相似类名推断。

22.6 延迟值物化和积极 I/O 可以同时存在

压缩页与远程请求有固定成本；完全按最终行号发细粒度请求往往得不偿失。Velox 允许字节先预取、值后解码，再用访问历史抑制长期不用的 stream。收益是吞吐和计算节省可以独立获得，代价是预取内存、gap 过读和命中预测误差。LazyVector 的价值首先是避免不需要的 values 与下游工作，额外节省多少存储 I/O 需要由实际路径证明。

把具体输入走完后，可以看到三个独立但相关的优化目标：减少候选行、减少值的解码或物化、减少存储字节。它们分别由过滤与 RowSet、read/getValues 与 LazyVector、BufferedInput 与 cache/preload 控制。评估时保留三层计数，才能判断一个改动只是把工作推迟，还是确实减少了总工作；异常和 reader 生命周期则是这些优化成立之前必须满足的条件。

23. 附录：PageReader 核心函数与状态对照

原分享附录对 rowsForPage / readWithVisitor 做了逐步注释。这里保留其全部技术步骤，并给出基准版本的完整函数：终止检查、seek、dictionary 变换、lower_bound 切片、快路径、SIMD rebias、null 准备、计数推进、跨页拼接和 outputRows 修正均可在同一段源码中查到。

状态 / 动作	读取前后负责什么	关键边界
visitBase_ + visitorRows_	本次 visit 的目标位置	重复列不可直接当顶层 RowVector 行号
rowOfPage_ / numRowsInPage_	当前 page 覆盖范围	页行数可能需先解 levels
rowNumberBias_	decoder-local rows 回到 batch 坐标	不同于输出 buffer 的 numValuesBias
currentVisitorRow_	已经消费的请求行数	不等于已解码所有 leaf values 数
scanState.dictionary	字典值、filter cache 与 raw state	转 direct 前可能需 dedictionarize
nullConcatenation_	跨 page 的最终 validity bits	1=非空，拼接长度按实际结果项数
firstUnvisited_	下一轮 skip / visit 基准	最后一个访问位置 +1
rowsCopy_	rebias 后临时行号	SIMD 尾部 padding 是内存前提

23.1 A. rowsForPage：page 切片与输入坐标

源码摘录 · PageReader.cpp:1163

bool PageReader::rowsForPage(
    dwio::common::SelectiveColumnReader& reader,
    bool hasFilter,
    bool mayProduceNulls,
    folly::Range<const vector_size_t*>& rows,
    const uint64_t* FOLLY_NULLABLE& nulls) {
  if (currentVisitorRow_ == numVisitorRows_) {
    return false;
  }
  int32_t numToVisit;
  // Check if the first row to go to is in the current page. If not, seek to the
  // page that contains the row.
  auto rowZero = visitBase_ + visitorRows_[currentVisitorRow_];
  if (rowZero >= rowOfPage_ + numRowsInPage_) {
    seekToPage(rowZero);
    // If seekToPage set numRowsInPage_=0, we've reached the end of the chunk
    if (numRowsInPage_ == 0) {
      return false;
    }
    if (hasChunkRepDefs_) {
      numLeafNullsConsumed_ = rowOfPage_;
    }
  }
  auto& scanState = reader.scanState();
  if (isDictionary()) {
    if (scanState.dictionary.values != dictionary_.values) {
      scanState.dictionary = dictionary_;
      if (hasFilter) {
        makeFilterCache(scanState);
      }
      scanState.updateRawState();
    }
  } else {
    if (scanState.dictionary.values) {
      // If there are previous pages in the current read, nulls read
      // from them are in 'nullConcatenation_' Put this into the
      // reader for the time of dedictionarizing so we don't read
      // undefined dictionary indices.
      if (mayProduceNulls && reader.numValues()) {
        reader.setNulls(nullConcatenation_.buffer());
      }
      reader.dedictionarize();
      // The nulls across all pages are in nullConcatenation_. Clear
      // the nulls and let the prepareNulls below reserve nulls for
      // the new page.
      reader.setNulls(nullptr);
      scanState.dictionary.clear();
    }
  }

  // Then check how many of the rows to visit are on the same page as the
  // current one.
  int32_t firstOnNextPage = rowOfPage_ + numRowsInPage_ - visitBase_;
  if (firstOnNextPage > visitorRows_[numVisitorRows_ - 1]) {
    // All the remaining rows are on this page.
    numToVisit = numVisitorRows_ - currentVisitorRow_;
  } else {
    // Find the last row in the rows to visit that is on this page.
    auto rangeLeft = folly::Range<const int32_t*>(
        visitorRows_ + currentVisitorRow_,
        numVisitorRows_ - currentVisitorRow_);
    auto it =
        std::lower_bound(rangeLeft.begin(), rangeLeft.end(), firstOnNextPage);
    assert(it != rangeLeft.end());
    assert(it != rangeLeft.begin());
    numToVisit = it - (visitorRows_ + currentVisitorRow_);
  }
  // If the page did not change and this is the first call, we can return a view
  // on the original visitor rows.
  if (rowOfPage_ == initialRowOfPage_ && currentVisitorRow_ == 0) {
    nulls =
        readNulls(visitorRows_[numToVisit - 1] + 1, reader.nullsInReadRange());
    rowNumberBias_ = 0;
    rows = folly::Range<const vector_size_t*>(visitorRows_, numToVisit);
  } else {
    // We scale row numbers to be relative to first on this page.
    auto pageOffset = rowOfPage_ - visitBase_;
    rowNumberBias_ = visitorRows_[currentVisitorRow_];
    skip(rowNumberBias_ - pageOffset);
    // The decoder is positioned at 'visitorRows_[currentVisitorRow_']'
    // We copy the rows to visit with a bias, so that the first to visit has
    // offset 0.
    rowsCopy_->resize(numToVisit);
    auto copy = rowsCopy_->data();
    // Subtract 'rowNumberBias_' from the rows to visit on this page.
    // 'copy' has a writable tail of SIMD width, so no special case for end of
    // loop.
    for (auto i = 0; i < numToVisit; i += xsimd::batch<int32_t>::size) {
      auto numbers = xsimd::batch<int32_t>::load_unaligned(
                         &visitorRows_[i + currentVisitorRow_]) -
          rowNumberBias_;
      numbers.store_unaligned(copy);
      copy += xsimd::batch<int32_t>::size;
    }
    nulls = readNulls(rowsCopy_->back() + 1, reader.nullsInReadRange());
    rows = folly::Range<const vector_size_t*>(
        rowsCopy_->data(), rowsCopy_->size());
  }
  reader.prepareNulls(rows, nulls != nullptr, currentVisitorRow_);
  currentVisitorRow_ += numToVisit;
  firstUnvisited_ = visitBase_ + visitorRows_[currentVisitorRow_ - 1] + 1;
  return true;
}

23.2 B. readWithVisitor：解码、结果追加和 null 拼接

源码摘录 · PageReader.h:596

template <typename Visitor>
void PageReader::readWithVisitor(Visitor& visitor) {
  constexpr bool hasFilter =
      !std::is_same_v<typename Visitor::FilterType, common::AlwaysTrue>;
  constexpr bool filterOnly =
      std::is_same_v<typename Visitor::Extract, dwio::common::DropValues>;
  constexpr bool hasHook = Visitor::kHasHook;
  static_assert(
      !(hasFilter && hasHook), "hasFilter and hasHook cannot both be true");

  bool mayProduceNulls = !filterOnly && visitor.allowNulls();
  auto rows = visitor.rows();
  auto numRows = visitor.numRows();
  auto& reader = visitor.reader();
  startVisit(folly::Range<const vector_size_t*>(rows, numRows));
  rowsCopy_ = &visitor.rowsCopy();
  folly::Range<const vector_size_t*> pageRows;
  int32_t numPageRowsRead = 0;
  const uint64_t* nulls = nullptr;
  bool isMultiPage = false;
  while (rowsForPage(reader, hasFilter, mayProduceNulls, pageRows, nulls)) {
    bool nullsFromFastPath = false;
    const int32_t numValuesBeforePage =
        numValuesRead<hasFilter, hasHook>(reader, numPageRowsRead);
    visitor.setNumValuesBias(numValuesBeforePage);
    visitor.setRows(pageRows);
    callDecoder(nulls, nullsFromFastPath, visitor);
    if (encoding_ == thrift::Encoding::DELTA_BINARY_PACKED &&
        deltaBpDecoder_->validValuesCount() == 0) {
      VELOX_DCHECK(
          deltaBpDecoder_->bufferStart() == pageData_ + encodedDataSize_,
          "Once all data in the delta binary packed decoder has been read, "
          "its buffer ptr should be moved to the end of the page.");
    }
    if (currentVisitorRow_ < numVisitorRows_ || isMultiPage) {
      if (mayProduceNulls) {
        if (!isMultiPage) {
          // Do not reuse nulls concatenation buffer if previous
          // results are hanging on to it.
          if (multiPageNulls_ && !multiPageNulls_->unique()) {
            multiPageNulls_ = nullptr;
          }
          nullConcatenation_.reset(multiPageNulls_);
        }
        if (!nulls) {
          nullConcatenation_.appendOnes(
              numValuesRead<hasFilter, hasHook>(reader, numPageRowsRead) -
              numValuesBeforePage);
        } else if (reader.returnReaderNulls()) {
          // Nulls from decoding go directly to result.
          nullConcatenation_.append(
              reader.nullsInReadRange()->template as<uint64_t>(),
              0,
              numValuesRead<hasFilter, hasHook>(reader, numPageRowsRead) -
                  numValuesBeforePage);
        } else {
          // Add the nulls produced from the decoder to the result.
          auto firstNullIndex = nullsFromFastPath ? 0 : numValuesBeforePage;
          nullConcatenation_.append(
              reader.mutableNulls(0),
              firstNullIndex,
              firstNullIndex +
                  numValuesRead<hasFilter, hasHook>(reader, numPageRowsRead) -
                  numValuesBeforePage);
        }
      }
      isMultiPage = true;
    }
    // The passing rows on non-first pages are relative to the start
    // of the page, adjust them to be relative to start of this
    // read. This can happen on the first processed page as well if
    // the first page of scan did not contain any of the rows to
    // visit.
    if (hasFilter && rowNumberBias_) {
      reader.offsetOutputRows(numValuesBeforePage, rowNumberBias_);
    }
    numPageRowsRead += pageRows.size();
  }
  if (isMultiPage) {
    reader.setNulls(mayProduceNulls ? nullConcatenation_.buffer() : nullptr);
  }
}

建议按这条链阅读：FileDataSource::next → ParquetRowReader::next → StructColumnReader::read → SelectiveStructColumnReaderBase::read → leaf reader / ColumnVisitor → PageReader；再回到 getValues → ExprSet → ColumnLoader，最后沿 DirectInputStream 查到 BufferedInput 和 CoalescedLoad。每次进入下一层，都先确认输入 rows 的坐标和对象生命周期。

24. 源码与参考资料

以下链接固定到本文核对的 commit。阅读入口从 HiveConnector / HiveDataSource 开始，而不是直接从 PageReader 开始。

• HiveDataSource.cpp:28
• HiveConnectorUtil.cpp:369
• HiveSplitReader.cpp:65
• TableScan.cpp:452
• AsyncSource.h:44
• ParquetTableScanTest.cpp:496
• ParquetReaderTest.cpp:1440

• velox/connectors/hive/FileDataSource.cpp
• velox/connectors/hive/FileSplitReader.cpp
• velox/dwio/parquet/reader/ParquetReader.cpp
• velox/dwio/parquet/reader/StructColumnReader.cpp
• velox/dwio/parquet/reader/ParquetColumnReader.cpp
• velox/dwio/common/ScanSpec.cpp
• velox/dwio/common/SelectiveIntegerColumnReader.h
• velox/dwio/common/SelectiveStructColumnReader.cpp
• velox/dwio/common/ColumnLoader.cpp
• velox/dwio/parquet/reader/PageReader.cpp
• velox/dwio/parquet/reader/PageReader.h
• velox/dwio/common/DirectBufferedInput.cpp
• velox/dwio/common/DirectInputStream.cpp
• velox/common/caching/AsyncDataCache.cpp
• velox/dwio/nimble/velox/selective/SelectiveNimbleReader.cpp

• 原分享：Velox 高性能 Parquet SelectiveReader（访问需原文权限）
• 向量化 - A Velox View
• Velox 社区早期 Parquet reader 设计讨论
• 原分享引用的 Spark LazyColumnVector 提案
• Photon: A Fast Query Engine for Lakehouse Systems
• Boudewijn Braams：native/vectorized reader 相关论文
• 原分享引用的列式文件格式研究
• Parquet Nested Encoding
• Spark SQL / Catalyst、Shark（原分享的历史背景引文）

原分享还链接了 Orri Erling、Michael Armbrust 的人物资料，以及 Spark 提案相关演讲。这些保留为背景资料，个人经历不作为实现或性能结论的证据。