从一到无穷大 #35 Velox Parquet Reader 能力边界

本作品采用知识共享署名-非商业性使用-相同方式共享 4.0 国际许可协议进行许可。

本作品 (李兆龙 博文, 由 李兆龙 创作)，由 李兆龙 确认，转载请注明版权。

引言

InfluxDB IOX这样完全不使用索引，只是基于执行引擎与Arrow-rs Parquet Reader的极致工程化道路无疑是相对极端的，这样的做法在大多数低时间线场景性能基本不可能高于VictoriaMetrics和Influxdb v1.x这样的全倒排索引实现，尤其是在筛选条件较多时。

权衡之下，像GreptimeDB这样对Parquet构建稀疏倒排索引[7]的方案就成了非多引擎下的合理选择，依托于Datafusion和Arrow-rs活跃的社区和完备的功能，事实上可以相对低成本，低风险的在各方面指标达到良好的表现，在初创公司的角度来看这当然是一条合理，有效，且高效的道路。

但是世界并不是只由rust构成，也不是所有团队都有资源愿意像InfluxData一样对DataFusion，Arrow-rs这样基础库做大量的投入，并稳操社区的控制权，最后才反哺自己的产品，当然回馈了不少上层软件产品，同时孕育了一众开源的时序数据库产品。

对大多数团队来说如何在现有资源下低风险，高人效的拿成果就成了需要思考的问题。

在21世纪20年代来看，自研数据库计算引擎是一件极高投入，较低回报的事情，算子扩展，并行化，性能提升，稳定性等无不需要大量的精力投入，到最后性能，功能也不及世界顶尖的执行引擎产品，这事实上是基本可预料的，一个数据库产品团队的内核研发能有多少人力去做专用计算引擎呢？这也是

这条路Meta已经走过了[1]，其设计了Velox用来替换Presto，Spark，XStream等系统的执行引擎，基础语言为CPP。

我们的系统语言为Cpp，在经过技术调研后除去从DuckDB，Clickhouse等知名项目中抠执行引擎外，可行的技术选择只剩下了Arrow Acero和Velox。Arrow Acero虽然依托于Arrow-cpp社区，且愿景宏大，但是整体还处于实验阶段，且没有值得信赖的项目背书。相对之下Velox确实就成了唯一的选择。

对Velox研究了一段时间后，认为Velox满足了90%以上的功能需求，但是部分性能关键点存在缺失，有比较大的修改空间，本篇文章聚集在Velox Parquet Reader，探究其功能缺失点。

源码分析

Velox的Velox Parquet Reader并不是原生的Arrow-cpp的实现，而是100%自主实现的，官方给出的解释[5]是：

Velox implements a visitor pattern suitable for most columnar file formats. This visitor pattern implements file-format agnostic features like predicate pruning, filter evaluation, etc… There are also some IO optimizations like prefetching and I/O coalescing.
The format-specific implementation (for Parquet, ORC, and DWRF) involves extending these visitor APIs. Wrapping these implementations around apache arrow parquet would involve a lot more work.
Velox implements a visitor pattern suitable for most columnar file formats. This visitor pattern implements file-format agnostic features like predicate pruning, filter evaluation, etc… There are also some IO optimizations like prefetching and I/O coalescing.

Velox代码中Parquet Reader（velox/dwio/parquet/reader）的部分只涉及了文件格式的编解码，连Decoder部分也是公有的，而类似filter pushdown这样的功能则是通过Decoder，放在visitor pattern中去实现。

这样来看，确实实现更多的优化功能会十分麻烦，因为这部分代码需要新增在common部分，且需要新增大量新的公有接口。

先给出一个我自己写的Filter pushdown的读取Parquet文件样例，example中对这部分描述比较少，而且都是很底层的接口，不研究下代码还是比较难写出来的，我认为对初学者有比较大的学习价值。

void processParquetFile(const std::string& filePath, std::vector<RowVectorPtr>& rowBatches, memory::MemoryPool* pool) {
    dwio::common::ReaderOptions readerOpts{pool};
    readerOpts.setFileFormat(FileFormat::PARQUET);
    auto reader = getReaderFactory(FileFormat::PARQUET)
                      ->createReader(
                          std::make_unique<BufferedInput>(
                              std::make_shared<LocalReadFile>(filePath),
                              readerOpts.memoryPool()),
                          readerOpts);
    if (!reader) {
        std::cerr << "Failed to create reader for file: " << filePath << std::endl;
        return;
    }

    RowReaderOptions rowReaderOptions;
    auto rowType = ROW({"measurement", "timestamp", "arch", "datacenter", "usage_user"},
                       {VARCHAR(), TIMESTAMP(), VARCHAR(), VARCHAR(), DOUBLE()});
    rowReaderOptions.select(
      std::make_shared<facebook::velox::dwio::common::ColumnSelector>(
        rowType, rowType->names(), nullptr, false));
    auto scanSpec = std::make_shared<facebook::velox::common::ScanSpec>("");
  
    auto untyped = parse::parseExpr("usage_steal >= 3.0", parse::ParseOptions());
    auto filterExpr = core::Expressions::inferTypes(untyped, rowType, pool);
    std::shared_ptr<core::QueryCtx> queryCtx{core::QueryCtx::create()};
    exec::SimpleExpressionEvaluator evaluator{queryCtx.get(), pool};
    auto [subfield, filter] = exec::toSubfieldFilter(filterExpr, &evaluator);
    auto fieldSpec = scanSpec->getOrCreateChild(subfield);
    fieldSpec->addFilter(*filter);

    scanSpec->addAllChildFields(*rowType);
    rowReaderOptions.setScanSpec(scanSpec);
    auto rowReader = reader->createRowReader(rowReaderOptions);
    std::cout << "The type of rowReader is: " << typeid(*rowReader).name() << std::endl;

    auto rowBatch = BaseVector::create(rowType, 50000, pool);

    while (rowReader->next(50000, rowBatch)) {
        auto rowVector = std::dynamic_pointer_cast<RowVector>(rowBatch);
        if (rowVector) {
            std::lock_guard<std::mutex> lock(batchMutex);
            rowBatches.push_back(rowVector);
        } else {
            std::cerr << "Error: Batch is not a RowVector for file: " << filePath << std::endl;
        }
    }
}

以一个Filter pushdown的流程来入门，整体代码流程如下：

velox/dwio/parquet/reader/ParquetColumnReader.cpp:build
velox/dwio/parquet/reader/StructColumnReader.h:StructColumnReader 在构造函数中构造每一列的column reader
velox/dwio/common/SelectiveStructColumnReader.cpp:next
velox/dwio/common/SelectiveStructColumnReader.cpp:read 从此处处获取每一列的数据
velox/dwio/common/SelectiveStructColumnReader.cpp:getValues 分别调用child的getValues
velox/dwio/parquet/reader/FloatingPointColumnReader.h:read 
    velox/dwio/common/SelectiveFloatingPointColumnReader.h:readCommon
    velox/dwio/common/SelectiveFloatingPointColumnReader.h:processFilter
    velox/dwio/common/SelectiveFloatingPointColumnReader.h:readHelper
    回调 velox/dwio/parquet/reader/FloatingPointColumnReader.h:readWithVisitor filter作为参数构造ColumnVisitor
    velox/dwio/parquet/reader/ParquetData.h:readWithVisitor(Visitor visitor)  visitor中包含filter
    velox/dwio/parquet/reader/PageReader.h:readWithVisitor
    velox/dwio/parquet/reader/PageReader.h:callDecoder
        velox/dwio/parquet/reader/PageReader.cpp:seekToPage 
        velox/dwio/parquet/reader/PageReader.cpp:prepareDataPageV2
        velox/dwio/parquet/reader/PageReader.cpp:makeDecoder 构造decode
    velox/dwio/dwrf/common/ByteRLE.h:readWithVisitor 各种code被赋值给decode
    velox/dwio/common/ColumnVisitors.h:process 最重要的逻辑，判断是否应该过滤列
    velox/type/Filter.h:applyFilter
    velox/dwio/common/ColumnVisitors.h:filterPassed
    velox/dwio/common/ColumnVisitors.h:addResult 然后把下标加入addOutputRow
    velox/dwio/common/ColumnVisitors.h:addOutputRow 
    回调 velox/dwio/common/SelectiveColumnReader.h:addOutputRow 把row加入outputRows_

功能描述

Velox Parquet Reader基本能力可以表述如下：

总体来看的评价是：

1. 基本功能完善
2. 性能可想的不够优异
3. 架构追求美感，包袱重，很难复用Arrow-cpp
4. 不支持倒排索引

其次从部分测试结果看，Velox filter pushdown存在一些性能问题，还没有filter Operator快，这个具体的原因后续仔细分析以下，暂时倒不是特别急切。

功能展望

我们对于Velox Parquet Reader的核心诉求是支持Parquet级别的倒排索引，这样可以无缝适配到TableScan算子的异步Push模式中去。执行引擎本身当然也需要改一些地方，但都可以逐步迭代，并不急切。

我改了一版的Velox代码，以支持Parquet Reader中Row级别的过滤，只需要在AddSplit时添加文件对应的RowFilter信息即可，接口部分模仿DataFusion+Arrow-rs，但是还有部分测试和代码优化工作需要细化；

肉眼可见的，Velox社区还有大量的核心特性的贡献机会，虽然Meta的开源社区维护一直被人诟病，但是有Presto，Spark背书近五年到不必担心项目爆雷。

参考：

1. 从一到无穷大 #26 Velox：Meta用cpp实现的大一统模块化执行引擎
2. https://blog.mwish.me/
3. Velox: Meta’s Unified Execution Engine vldb2022
4. Querying Parquet with Millisecond Latency
5. [Design] Native Parquet Reader
6. PARQUET-1820: [C++] pre-buffer specified columns of row group #6744
7. GrepTimeDB index
8. advanced_parquet_index.rs
9. parquet/src/arrow/arrow_reader/selection.rs
10. Faster C++ Apache Parquet performance on dictionary-encoded string data coming in Apache Arrow 0.15
11. Parquet SBBF
12. Split block Bloom filters