diff --git a/be/src/format_v2/AGENTS.md b/be/src/format_v2/AGENTS.md new file mode 100644 index 00000000000000..377a26ce530db8 --- /dev/null +++ b/be/src/format_v2/AGENTS.md @@ -0,0 +1,183 @@ +# Format V2 — Review Guide + +Use this guide when reviewing changes under `be/src/format_v2/`. Apply the repository-level +instructions as well; this file adds format-v2-specific review expectations. + +## Review Objective + +- Report actionable correctness, data-corruption, crash, resource-lifetime, and performance + regressions. Do not report style-only issues already enforced by the repository tooling. +- Trace the complete affected path instead of reviewing a changed function in isolation. The usual + path crosses `TableReader`, `TableColumnMapper`, schema projection/materialization, and a concrete + file or table reader. +- Verify claims against callers, implementations, and tests. Do not report a hypothetical failure + unless a reachable input or state demonstrates it. + +## Architecture and Interface Contracts + +- Use the [FileScannerV2 design document](../../../docs/file-scanner-v2-design.md) as the + architectural reference. Preserve the one-way responsibility chain: Scanner manages query + integration and Split progression, `TableReader` manages table semantics, and `FileReader` + interprets physical files. Layer boundaries take priority over incidental code reuse. +- `TableReader` owns table-level projection and column order, partition/default/virtual columns, + table predicates and delete semantics, per-Split state, reader orchestration, and final table-block + materialization. It may consume file schema and file-local blocks through stable contracts, but it + must not depend on a concrete format reader's metadata structures, decoding implementation, or + physical nested layout. +- `FileReader` owns physical schema discovery, file metadata, encoding and decoding, physical + pruning, lazy reads, and production of file-local blocks. It must not know query-global column + positions, table output order, partition/default/virtual-column construction, table-format + semantics, Scanner scheduling, or Split-source policy. +- `TableColumnMapper` is the only semantic bridge between table/global and file/local column + domains. It translates table projection and predicates plus file schema into `FileScanRequest`, + mapping/finalize metadata, constants, and localized expressions. It must not open or read files, + advance Splits, own reader lifecycle, or depend on concrete `TableReader`/`FileReader` + implementations. +- Coupling between these layers is allowed only through stable, format-neutral contracts such as + `ColumnDefinition`, `FileScanRequest`, mapper results, capability/status objects, and file-local + blocks. Flag new concrete-class includes, downcasts, reverse callbacks, shared mutable state, or + direct inspection of another layer's implementation details. +- Do not bypass `TableColumnMapper`: `TableReader` must not independently reproduce file-local + column matching or position logic, and `FileReader` must not independently resolve table schema, + defaults, partitions, virtual columns, or final table types. There must be one authoritative + mapping for projection, predicate localization, and final materialization. +- Keep identity namespaces explicit at every boundary. Query expressions and table output use + global identities; file requests and file blocks use local identities. A file-local ordinal, + field ID, physical child position, or format wrapper node must never leak upward as a table/global + identity. +- Localized predicates and delete information may be executed by a file reader only after the + mapper/table layer has converted them into a file-local contract. The file reader may optimize + execution but must not reinterpret or invent the table-level semantics. +- Format-specific capability or metadata needed by an upper layer should be exposed as the smallest + neutral capability/result contract. Do not add Parquet/ORC/JNI-specific conditionals to generic + table semantics when the decision belongs in a reader, factory, or capability interface. +- When reviewing an interface change, identify the owner layer, document input/output and lifecycle + invariants, inspect every caller and implementation, and verify that adding another file format or + table format would not require changes in unrelated layers. Require boundary-focused tests that + exercise mapping and materialization independently from physical decoding where possible. + +## Reader Lifecycle and Contracts + +- Preserve the reader lifecycle and state transitions across initialization, schema discovery, + opening, block production, EOF, split advancement, and close. +- Check that empty blocks, EOF, cancellation, early returns, and errors cannot skip required cleanup + or leave stale per-file/per-split state for the next reader. +- Keep `current_rows`, block row counts, selection vectors, row positions, and `eos` consistent on + every path, including fully filtered blocks and aggregate-pushdown paths. +- Check ownership and lifetime of file readers, column readers, blocks, columns, expression + contexts, callbacks, and objects referenced through raw pointers or views. + +## Schema Mapping and Materialization + +- Keep table/global identities and positions distinct from file/local identities and positions. + Review uses of `GlobalIndex`, `LocalColumnId`, `LocalIndex`, `ConstantIndex`, and nested child IDs + for accidental namespace or ordinal mixing. +- Verify mapping by field ID, name, and position against the intended table format. Missing columns, + partition columns, defaults, and virtual columns must be materialized with the correct type, + nullability, and row count. +- For schema evolution, check field additions, removals, renames, reordering, type changes, and + nullable/non-nullable transitions. +- For `STRUCT`, `ARRAY`, and `MAP`, verify recursive projection and reconstruction, child ordering, + file-local IDs, offsets, null maps, and empty collections. Remember that semantic Doris trees and + physical file-format trees may have different shapes. +- Check that casts and defaults preserve Doris semantics for overflow, precision/scale, timezone, + decimal, date/time, string, and nullable values. + +## Filtering, Deletes, and Pushdown + +- Predicate columns and lazily materialized non-predicate columns must refer to exactly the same + rows after filtering. Review selection-vector reuse, skipped row groups/pages, and row-position + accounting together. +- A pushed-down predicate, statistic, dictionary filter, bloom filter, or aggregate must be + semantically equivalent to evaluating it after materialization. Unsupported or unsafe cases must + follow the designed fallback or return an explicit error; they must not silently change results. +- Review equality deletes, position deletes, table-format predicates, and generated row-location + columns for ordering, null semantics, type conversion, file identity, and absolute row position. +- For Iceberg, Hive, Hudi, Paimon, Remote Doris, and JNI-backed readers, verify that the table-level + wrapper preserves the underlying file reader's schema, filtering, split, and EOF contracts. + +## Format-Specific Boundaries + +- Confirm file-format dispatch and capability checks match the actual implementation. New behavior + must not accidentally route unsupported formats or table modes into a reader that cannot handle + them. +- For Parquet and ORC, review physical-to-semantic schema conversion, nested levels/offsets, + statistics validity, page or stripe pruning, and corrupt/truncated input handling. +- For CSV, text, and JSON, review record boundaries, escaping/quoting, malformed rows, encoding, + column count, and partial-buffer behavior across reads. +- For JNI readers, review local/global reference lifetime, exception propagation, type conversion, + thread attachment assumptions, and cleanup on partial initialization. + +## Detailed FileReader Review Guides + +- Before reviewing any FileReader implementation, index, predicate path, cache, or virtual column, + read and apply the common checklist in + [FileScannerV2 Code Review Guide](../../../docs/file-scanner-v2-code-review-guide.md). +- For Parquet changes, also apply the guide's Parquet checklist and read + [FileScannerV2 Parquet Scan Design](../../../docs/file-scanner-v2-parquet-scan-design.md). +- For ORC changes, also apply the guide's ORC SARG and index checklist. +- These detailed guides are mandatory review instructions for their scope, not optional background + reading. Report any conflict between an implementation and the documented layer contract. + +## External Compatibility + +- Treat the external table-format specification and the behavior of supported external writers as + compatibility inputs. Do not assume Doris-generated fixtures or an existing Doris implementation + are authoritative when they conflict with the external contract. +- Do not require the external representation to behave like Doris internal storage. Verify the + complete translation from external semantics, through the format-v2 adapter, to the observable + Doris query result. Any intentional semantic difference must be documented and tested. +- Identify the compatibility matrix affected by a change: lake format and version, physical file + format and version, producing engine/writer, feature flags, encoding, compression codec, and + metadata version. Avoid fixes that only work for one writer's representation. +- Preserve backward compatibility with files and metadata produced by supported older versions. + For newer or unknown versions and features, follow the external specification's compatibility + rules; do not guess or silently reinterpret metadata. +- Review snapshot selection, time travel, manifest and partition evolution, schema and field IDs, + name and case matching, file identity, path normalization, and partition value decoding according + to the relevant lake-format semantics. +- Review writer-dependent physical representations, including Parquet logical annotations and + legacy encodings, ORC type attributes, timestamps and timezones, decimals, signedness, CHAR + padding, nested LIST/MAP layouts, null counts, NaN values, statistics, page/stripe indexes, and + optional or missing metadata. +- Capability detection and dispatch must happen before relying on a feature. Unsupported table + modes, metadata features, encodings, or semantic conversions must use the explicitly designed + fallback or return a clear error; they must never produce plausible but incorrect rows. +- Predicate, delete, statistics, and aggregate pushdown must return the same observable result as + reading and evaluating the external data without that optimization, including NULL, NaN, + timezone, collation/case, overflow, and precision edge cases. +- Check that a compatibility fix for one combination does not change existing behavior for other + lake formats, file formats, writers, or versions sharing the same abstraction. +- Require interoperability coverage using artifacts produced by representative external systems + such as Spark, Hive, Flink, or Trino when applicable. Prefer differential tests against a + non-pushdown path or the source system's expected result; do not rely only on files synthesized by + Doris test code. +- Each compatibility finding should state the affected external system or specification, versions + or writer variants, reachable input, Doris result, and expected result. + +## Performance and Observability + +- Treat per-row allocation, expression cloning, virtual dispatch, repeated schema work, unnecessary + column copies, and loss of lazy reads or pruning in hot paths as potential regressions. +- Check I/O ranges, caching, decompression, and batch sizing for accidental read amplification or + unbounded memory growth. +- Preserve profile counters and timers when control flow changes so filtered rows, bytes, reader + creation, and pushdown behavior remain diagnosable. + +## Tests + +- Require focused BE unit tests under `be/test/format_v2/`, following the source subdirectory when + possible. Add regression coverage when correctness depends on the FE-to-BE request or external + table integration. +- Include the relevant edge cases: empty input, all rows filtered, multiple blocks/splits/files, + EOF with and without output rows, nulls, missing/default columns, reordered or nested fields, and + malformed input. +- For bug fixes, require a test that fails for the original reachable path and validates the result, + row count, or explicit error after the fix. + +## Review Output + +- List findings first, ordered by severity. Each finding must identify the file and line, the + reachable execution path, and the concrete incorrect outcome. +- Distinguish verified defects from open questions. If no actionable defect is found, say so and + mention any important coverage or testing gap that remains. diff --git a/docs/doris-iceberg-parquet-api-design.md b/docs/doris-iceberg-parquet-api-design.md deleted file mode 100644 index 457550a932da67..00000000000000 --- a/docs/doris-iceberg-parquet-api-design.md +++ /dev/null @@ -1,511 +0,0 @@ -# Doris Iceberg + Parquet 新架构 API 设计 - -本文档用于描述 Doris 中 Iceberg + Parquet 新架构的 API 设计。本文档作为后续从 -`master` 新开重构分支时的起点，只定义 API 形状、职责边界、依赖方向和兼容原则， -不定义函数实现细节，不提供伪代码，不包含迁移 patch。 - -## 架构总览 - -目标架构包含 table 调度层、表格式语义层、schema 映射层、文件通用层和文件格式实现层： - -```text -FileScanner / split producer - -> -TableReader - -> -IcebergTableReader - -> -TableColumnMapper + FileReader - -> -ParquetReader -``` - -核心职责如下： - -- `TableReader` - 负责多文件、多 split 的上层调度，统一 scan 生命周期，对外输出 table block， - 并承接动态分区裁剪等 table-level 通用逻辑。 -- `IcebergTableReader` - 负责 Iceberg 表语义，包括 schema 绑定、scan task、delete file、虚拟列和 table - block finalize。 -- `TableColumnMapper` - 负责 table schema 到 file schema 的映射，负责 filter localization 和 schema - change 映射。 -- `FileReader` - 负责文件层通用读取接口，只理解 file-local schema 和 file-local scan request。 -- `ParquetReader` - 作为 `FileReader` 的 Parquet 实现，负责 Parquet 文件物理读取。 - -依赖方向必须保持单向： - -```text -TableReader - -> IcebergTableReader - -> TableColumnMapper - -> FileReader - -> ParquetReader -``` - -低层不反向理解高层语义，尤其 `ParquetReader` 不得反向理解 Iceberg/global schema。 - -## 核心 API 设计 - -### TableReader - -`TableReader` 是最上层读取接口，作为 `IcebergTableReader` 的基类，负责多 split / -多 file 调度，并承接 table-level 的通用裁剪逻辑，不下沉文件格式语义。 - -实际 API 文件： - -```text -be/src/format_v2/table_reader.h -``` - -实际命名空间： - -```cpp -namespace doris::format -``` - -建议职责： - -- 接收 split 列表或 scan task 列表； -- 控制当前 reader 的创建、切换和关闭； -- 管理 scan 生命周期； -- 承接动态分区裁剪等 table-level 通用过滤逻辑； -- 对外统一输出 table block。 -- `next` 是基类统一入口，内部负责 EOF 后切换 reader；具体表格式只提供打开和读取 - 当前 reader 的 hook。 - -建议接口形状： - -```cpp -namespace doris::format { - -class TableReader { -public: - virtual ~TableReader() = default; - - virtual Status init(const TableReadOptions& options); - virtual Status filter(const VExprContextSPtr& expr, bool* can_filter_all); - Status next(Block* table_block, size_t* rows, bool* eof); - virtual Status close(); - -protected: - Status next_reader(); - virtual Status open_next_reader(bool* has_reader); - virtual Status read_current(Block* table_block, size_t* rows, bool* eof); - virtual Status close_current_reader(); -}; - -} // namespace doris::format -``` - -接口约束： - -- `TableReader` 输出的是 table block，不输出 file-local block。 -- `TableReader` 负责多文件编排和 table-level 通用裁剪，不负责 schema mapping，不负责 - Parquet 物理解码。 -- `next_reader` 是 `TableReader` 自己的通用切换逻辑，不作为子类公开 override 接口。 -- 动态分区裁剪这类逻辑应下放到 `TableReader`，而不是散落在具体表格式 reader 中。 -- `TableReader` 不直接依赖旧 `vparquet` 表层语义。 - -### IcebergTableReader - -`IcebergTableReader` 是 Iceberg 表语义层，负责把单个 Iceberg data file 的读取组织成 -table 语义输出。 - -实际 API 文件： - -```text -be/src/format_v2/table/iceberg_reader.h -``` - -实际命名空间： - -```cpp -namespace doris::iceberg -``` - -建议职责： - -- 绑定 Iceberg 当前 table schema； -- 接收 `IcebergScanTask` 列表，并按 `TableReader` 的统一调度打开当前 task； -- 处理 position delete、equality delete、deletion vector； -- 物化 `_row_id`、`_last_updated_sequence_number` 等虚拟列； -- 将 `ParquetReader` 返回的 file-local block finalize 成 table block。 - -建议接口形状： - -```cpp -namespace doris::iceberg { - -class IcebergTableReader : public format::TableReader { -public: - virtual ~IcebergTableReader() = default; - - Status init(IcebergTableReadParams params); - Status close() override; - -protected: - Status open_next_reader(bool* has_reader) override; - Status read_current(Block* table_block, size_t* rows, bool* eof) override; - Status close_current_reader() override; -}; - -} // namespace doris::iceberg -``` - -接口约束： - -- `IcebergTableReader` 继承 `TableReader`，并通过组合使用 `FileReader`。 -- `IcebergTableReader` 不做 Parquet page/column 解码。 -- `IcebergTableReader` 负责 table-level finalize，不负责 file-local pruning 实现。 -- `IcebergTableReader` 的 schema、scan request、scan tasks 和底层 `FileReader` 应通过 - 一个初始化参数对象一次性传入；除非存在明确生命周期差异，不拆成 `bind` / - `init(TableScanRequest)` / `set_scan_tasks` 多阶段接口。 -- `IcebergTableReader` 不重新实现 reader 切换循环，只实现打开 Iceberg task、读取当前 - task 和关闭当前 reader 的 hook。 - -### TableColumnMapper - -`TableColumnMapper` 是 table schema 到 file schema 的通用映射层，不是 -Iceberg-only 组件。 - -实际 API 文件： - -```text -be/src/format_v2/table_reader.h -``` - -实际命名空间： - -```cpp -namespace doris::format -``` - -建议职责： - -- 输入 table schema、file schema、table scan request； -- 输出 `ColumnMapping` 和通用 `FileScanRequest`； -- 负责 filter localization； -- 负责 schema change 映射； -- 负责复杂列 child mapping； -- 负责缺失列、default、partition、generated 列的 finalize 语义描述。 - -建议接口形状： - -```cpp -namespace doris::format { - -class TableColumnMapper { -public: - explicit TableColumnMapper(TableColumnMapperOptions options = {}); - - virtual Status create_mapping(const std::vector& table_schema, - const std::vector& file_schema, - std::vector* mappings); - - virtual Status create_scan_request(const TableScanRequest& table_request, - const std::vector& mappings, - FileScanRequest* file_request); -}; - -} // namespace doris::format -``` - -接口约束： - -- `TableColumnMapper` 的输入是 table schema + file schema + table scan request。 -- `TableColumnMapper` 的输出是 `ColumnMapping` + `FileScanRequest`。 -- `TableColumnMapper` 必须是通用层，不做 Iceberg-only 命名。 -- Iceberg 场景默认按 field id 映射；按 name 映射不是本轮默认路径。 - -### FileReader - -`FileReader` 是文件物理读取层的通用接口，为后续 Parquet 之外的文件格式适配预留。 - -实际 API 文件： - -```text -be/src/format_v2/file_reader.h -``` - -实际命名空间： - -```cpp -namespace doris::format -``` - -建议职责： - -- 打开物理文件； -- 暴露 file-local schema； -- 接收 `FileScanRequest`； -- 输出 file-local block； -- 不理解 table/global schema。 - -建议接口形状： - -```cpp -namespace doris::format { - -class FileReader { -public: - virtual ~FileReader() = default; - - virtual Status open(io::FileReaderSPtr file, io::IOContext* io_ctx = nullptr); - virtual Status get_schema(std::vector* file_schema) const; - virtual Status init(const FileScanRequest& request); - virtual Status next(Block* file_block, size_t* rows, bool* eof); - virtual Status close(); -}; - -} // namespace doris::format -``` - -接口约束： - -- `FileReader` 输出的是 file-local block，不输出 table/global schema block。 -- `FileReader` 不处理 Iceberg schema evolution、default/generated/partition 列。 -- `IcebergTableReader` 组合 `FileReader`，不直接绑定具体文件格式 reader。 - -### ParquetReader - -`ParquetReader` 是 `FileReader` 的 Parquet 实现，只负责 Parquet file-local schema -和 Parquet file-local scan request。 - -实际 API 文件： - -```text -be/src/format/parquet/parquet_reader.h -``` - -实际命名空间： - -```cpp -namespace doris::parquet -``` - -建议职责： - -- 打开 Parquet 文件； -- 解析 footer 和 file schema； -- 接收 `ParquetScanRequest` 或通用 `FileScanRequest`； -- 执行 file-local projection 和 file-local filter； -- 输出 file-local block。 - -建议接口形状： - -```cpp -namespace doris::parquet { - -class ParquetReader : public format::FileReader { -public: - virtual ~ParquetReader() = default; - - virtual Status open(io::FileReaderSPtr file, io::IOContext* io_ctx = nullptr); - virtual Status get_schema(std::vector* file_schema) const; - virtual Status init(const ParquetScanRequest& request); - virtual Status next(Block* file_block, size_t* rows, bool* eof); - virtual Status close(); -}; - -} // namespace doris::parquet -``` - -接口约束： - -- `ParquetReader` 输出的是 file-local block，不输出 table/global schema block。 -- `ParquetReader` 不理解 Iceberg schema evolution。 -- `ParquetReader` 不负责 default/generated/partition 列。 -- 任何 table-level cast/default/generated/partition 语义都不能重新塞回 - `ParquetReader`。 - -## 关键类型 - -### SchemaField - -`SchemaField` 表示文件层 schema 中的列定义。 - -建议包含的信息： - -- file-local column id； -- 列名； -- 类型； -- child fields。 - -它服务于 `TableColumnMapper` 做 schema matching，不携带 table-level 语义。 - -### TableColumnDefinition - -`TableColumnDefinition` 表示 table/global schema 中的列定义。 - -建议包含的信息： - -- table column id； -- 列名； -- 类型； -- child columns。 - -Iceberg 场景下，column id 默认对应 field id。 - -### TableFilter - -`TableFilter` 表示 table 层过滤条件。 - -建议包含的信息： - -- `table_column_id` -- `conjunct` -- `predicates` - -职责约束： - -- `conjunct` 偏表达式过滤，适合表达 cast、复杂表达式、复杂列提取等语义； -- `predicates` 偏结构化单列下推，适合驱动 row group stats、page index、dictionary、 - bloom filter 等文件层优化。 - -### FileLocalFilter - -`FileLocalFilter` 表示已经 localize 到 file-local schema 的过滤条件。 - -建议包含的信息： - -- `file_column_id` -- `conjunct` -- `predicates` - -职责约束： - -- `conjunct` 用于 file-local 表达式过滤； -- `predicates` 用于 file-local 结构化下推； -- 其输入必须来自 `TableColumnMapper`，不能由具体文件 reader 自己推导 table 语义。 - -### ColumnMapping - -`ColumnMapping` 是 table schema 与 file schema 之间的核心边界对象。 - -建议包含的信息： - -- `table_column_id` -- `file_column_id` -- `file_type` -- `table_type` -- `finalize_expr` -- `reader_filter_expr` -- `child_mappings` - -职责约束： - -- `finalize_expr` 服务最终输出，把 file-local value 转成 table/global value； -- `reader_filter_expr` 服务读时 filter fallback； -- 二者语义不同，不能混用； -- `child_mappings` 用于复杂列 remap、复杂列裁剪和复杂列 schema change。 - -### TableScanRequest - -`TableScanRequest` 描述 table 层 scan 请求。 - -建议包含的信息： - -- projected table columns； -- table filters。 - -它由 `IcebergTableReader` 接收，再交给 `TableColumnMapper` 生成 file-local request。 - -### ParquetScanRequest - -`ParquetScanRequest` 继承 `FileScanRequest`，描述 Parquet file-local scan 请求。 - -### FileScanRequest - -`FileScanRequest` 描述通用 file-local scan 请求。 - -建议包含的信息： - -- projected file columns； -- local filters； -- reader expression map。 - -它是 `FileReader` 的唯一 scan 输入，不包含 table/global schema 语义。 - -### IcebergScanTask - -`IcebergScanTask` 表示一次 Iceberg data file 读取任务。 - -建议包含的信息： - -- data file 信息； -- position delete 文件； -- equality delete 文件； -- deletion vector 信息。 - -它是 `IcebergTableReader` 的输入，不应直接传给 `ParquetReader`。 - -### IcebergTableReadParams - -`IcebergTableReadParams` 表示一次 Iceberg table scan 的完整初始化输入。 - -建议包含的信息： - -- Iceberg read options； -- Iceberg table schema； -- table scan request； -- Iceberg scan task 列表； -- 底层 `FileReader`。 - -它用于避免 `IcebergTableReader` 暴露多个半初始化阶段。调用方应一次性构造完整 -参数并调用 `init`。 - -## 设计原则 - -### 边界原则 - -- `FileReader` 不理解 global schema，不直接处理 Iceberg schema evolution。 -- `ParquetReader` 是 `FileReader` 的 Parquet 实现。 -- `TableColumnMapper` 是 schema mapping 和 filter localization 的唯一入口。 -- `IcebergTableReader` 不做 Parquet 解码，只负责 table-level finalize、delete、 - virtual columns。 -- `TableReader` 只负责多文件编排和 table-level 通用裁剪，不下沉文件格式语义。 -- 任何 table-level cast/default/generated/partition 语义都不能重新塞回 - `ParquetReader`。 - -### 依赖原则 - -- 低层不能反向依赖高层语义。 -- `FileReader` 只依赖 file-local request。 -- `IcebergTableReader` 继承 `TableReader`，复用其多文件编排和通用裁剪能力。 -- `IcebergTableReader` 通过组合使用 `FileReader`。 -- `TableColumnMapper` 可以被 Iceberg 之外的其他表格式复用。 - -### 命名原则 - -- 表层抽象使用 `TableReader`、`IcebergTableReader`、`TableColumnMapper`、 - `FileReader`、`ParquetReader` 命名。 -- `TableColumnMapper` 不使用 Iceberg-only 命名。 -- file schema 类型使用 `SchemaField`，table schema 类型使用 `TableColumnDefinition`。 - -## 兼容原则 - -新架构重构期间，新旧代码允许并存，但必须遵守以下约束： - -- 旧 `vparquet` / Hive / Hudi / Paimon 路径在新架构稳定前允许保留。 -- 新架构实现不得继续向旧 `vparquet` 表层语义回灌依赖。 -- 先搭新框架 API，再逐步迁移调用点。 -- 不允许边改 API 边混入临时裸逻辑、实验性草稿或未收敛命名。 -- 兼容层可能需要存在，但本文档不定义兼容层的具体实现方案。 - -## 验收标准 - -该文档应满足以下目标： - -- 不引用错误实验代码作为既成事实； -- 不出现实现性草稿、裸伪代码、未收敛命名混用； -- 让另一个工程师从 `master` 新开分支时，可以直接按本文档搭 API 骨架； -- 读完文档后，不需要再讨论以下问题： - - 新架构分几层； - - 每层负责什么； - - 哪层理解 global schema； - - 哪层做 schema change / filter localization / finalize； - - 哪层允许依赖旧实现，哪层不允许。 diff --git a/docs/file-scanner-v2-code-review-guide.md b/docs/file-scanner-v2-code-review-guide.md new file mode 100644 index 00000000000000..b05ea798ee46b0 --- /dev/null +++ b/docs/file-scanner-v2-code-review-guide.md @@ -0,0 +1,173 @@ +# FileScannerV2 Code Review Guide + +This guide contains the detailed checklists referenced by +`be/src/format_v2/AGENTS.md`. Read the common checklist for every FileReader review, then apply the +format-specific checklist when reviewing Parquet or ORC. + +## Common FileReader: Indexes and Predicate Filtering + +- Inventory the reader's actual pruning capabilities before evaluating a change: metadata or + statistics, dictionary information, Bloom filters, page/stripe/row indexes, partition/Split + ranges, and format-specific encodings. Record the granularity, supported predicate/type set, + exactness, I/O cost, and conservative fallback for each capability. +- A FileReader consumes only predicates already localized by `TableColumnMapper` in + `FileScanRequest`. It may translate those predicates into format-native indexes or SDK filters, + but it must not reinterpret table-schema identity, defaults, partitions, or table-format + semantics. +- Every index may discard a candidate only when it proves that the candidate cannot match. Missing, + malformed, stale, truncated, unsupported, writer-incompatible, or unsafe metadata must retain the + candidate or return the format's explicit correctness-preserving error. +- Check logical-to-physical identity at every index boundary: file-local root and nested column IDs, + physical leaf IDs, row-group/stripe/page ordinals, byte ranges, file-global row offsets, and + selected row ranges. Index results for one column or unit must never be applied to another. +- Verify metadata semantics for NULL/all-NULL, empty units, NaN, signedness, truncated bounds, + decimal precision/scale, date/timestamp/timezone, string/binary ordering, CHAR padding, and + external-writer differences before trusting min/max or membership information. +- Preserve a cheap-to-expensive pruning order. Do not read or parse a finer index for a file, + row group, stripe, or page already eliminated by a cheaper layer. Measure index read/parse/build + cost as well as the I/O, decompression, decoding, and materialization it avoids. +- Trace each predicate through index pruning, exact format-native filtering, Doris residual VExpr, + delete predicates, and final materialization. A predicate not exactly covered by an earlier layer + must remain in the residual path. +- Preserve SQL three-valued logic and error behavior across AND/OR/NOT, comparisons, IN/NOT IN, + IS NULL, null-safe equality, casts, functions, stateful expressions, and exception-sensitive + operations. Splitting or reordering predicates requires proof of equivalence. +- Predicate columns and lazily read non-predicate columns must refer to the same original rows after + all skips and filters. Skipping must advance every physical reader consistently, including nested + definition/repetition state, offsets, row positions, and subsequent batches. +- Keep row-level deletes, equality deletes, position deletes, table filters, and query predicates in + their specified order. An index optimization must not bypass a delete or use post-filter row + numbering where file-global numbering is required. +- Readers without a native index or lazy-read capability must declare that boundary and preserve + correctness through residual evaluation. Do not add an imitation index in a generic layer merely + to make formats appear uniform. +- Require differential tests that compare exact results and errors with each index/filter + optimization enabled and disabled. Cover missing/invalid indexes, all/none/partially filtered + units, multiple files/Splits/batches, NULL and type boundaries, nested data, deletes, and + external-writer fixtures. + +## Common FileReader: Data and Condition Caches + +- Distinguish the cache layers and their value semantics: remote `FileCache` stores file bytes, + format metadata/page caches store format-specific serialized ranges or parsed metadata, + `ConditionCache` stores predicate survivor granules, and table-format caches may store deletion + vectors or decoded objects. Never reuse an entry as a different representation. +- A cache key must include every input that can change the value: filesystem and canonical path, + stable object/file version, size or mtime where reliable, byte/Split range, format/encoding + context, and predicate digest for filter results. Disable the cache when a stable identity cannot + be established; never trade stale rows for a hit. +- Validate hit, miss, partial coverage, overlapping/subrange reads, eviction, concurrent access, + cancellation, and error paths. A partial cache hit must read or conservatively retain uncovered + data rather than treating it as absent. +- `ConditionCache` can skip only file-global granules explicitly known to contain no surviving row. + Disable or expand the key when Runtime Filters, delete files/vectors, table snapshots, or other + changing semantics are not represented. Publish a miss result only after the physical reader + reaches EOF successfully so unvisited granules cannot become false negatives. +- Cache admission, prefetch, and range merging must follow pruning and lazy materialization. Do not + prefetch output columns or pruned units merely to improve hit rate, and account for read + amplification, request count, memory ownership, and cache pollution. +- Preserve resource accounting and source attribution across local, peer, and remote hits. Require + counters for hit/miss/write/eviction, bytes by source, wait/download time, requests, and avoided + reads so performance claims are diagnosable. +- Require warm/cold, enabled/disabled, overwrite/version-change, partial-range, concurrent, and + cancellation tests. Cached and uncached execution must return identical rows and errors. + +## Common FileReader: Virtual Columns + +- Keep file-coordinate virtual columns distinct from table-format virtual columns. FileReaders may + synthesize reserved file-local `ROW_POSITION` and `GLOBAL_ROWID`; `TableReader` and + `TableColumnMapper` own table semantics such as Iceberg `_row_id`, + `_last_updated_sequence_number`, and Doris Iceberg row locators. +- `ROW_POSITION` is the absolute zero-based physical row in the file, not an output, batch, + selected-row, row-group, stripe, or Split-local ordinal. It must advance across pruned units, + skipped pages/granules, rejected batches, lazy filters, and deletes without renumbering survivors. +- `GLOBAL_ROWID` must be stable and unique for its documented context. Review context version, + backend/file identity, serialization, physical row position, cross-file collisions, and retries; + filtering and batching must not change the generated ID for the same source row. +- Generate virtual values only when requested as output or needed by a predicate/delete. Support + virtual-only scans with no physical projected column, predicate-only virtual columns, selected-row + materialization, and EOF without forcing unrelated file I/O. +- Preserve declared type, nullability, nested shape, and `LocalColumnId`/`LocalIndex` mapping. Do not + let reserved negative IDs collide with invalid IDs, physical columns, table IDs, or block + positions. +- Require tests across multiple files, Splits, row groups/stripes/pages, batches, all rows filtered, + no rows filtered, index/cache skips, lazy materialization, deletes, and virtual-only projection. + Compare virtual values with the same scan when pruning, caching, and lazy reads are disabled. + +## Common FileReader: Performance and Observability + +- Keep index construction, predicate translation, cache lookup, and virtual-column setup out of + per-row and repeated batch paths unless the work is inherently row-local. Avoid repeated schema + traversal, expression cloning, metadata parsing, allocation, and conversion. +- Require format readers to populate the common `ReaderStatistics` accurately where applicable: + filtered/read row groups, Bloom and min/max pruning, filtered group/page/lazy rows, read rows and + bytes, metadata/footer/cache timing, page-index work, predicate time, dictionary rewrite, and + Bloom read time. +- Evaluate performance with representative format versions, writers, data ordering, predicate + selectivity, nested width, remote storage, batch sizes, and warm/cold caches. Report both the + optimization overhead and the avoided work; a low pruning ratio alone is not a defect. + +## Parquet Multi-Level Filtering + +- Use [FileScannerV2 Parquet Scan Design](file-scanner-v2-parquet-scan-design.md) as the detailed + architecture reference. Trace each affected predicate through localization, Row Group planning, + Page ranges, row-level residual evaluation, and final selected-column materialization. +- At Row Group level, check Split ownership and file-global row offsets, then verify Statistics, + Dictionary, and Bloom pruning independently. Dictionary pruning requires complete compatible + encoding. Bloom may prove absence only; a hit is never a matching row. +- Preserve the cost order from cheap to expensive. Footer Statistics should reduce candidates before + Dictionary/Bloom I/O, and ColumnIndex/OffsetIndex should be read only for surviving Row Groups. +- At Page level, require compatible ColumnIndex and OffsetIndex semantics. Check page-to-row mapping, + first/last row boundaries, empty or all-null pages, multi-column range intersection, and conversion + from logical `selected_ranges` to each leaf reader's physical `page_skip_plan`. +- Page skipping must keep every column reader aligned. Skipping values or pages must advance value, + definition, and repetition state consistently, especially for nested/repeated columns whose Page + boundaries do not align across leaves. +- At Row/Batch level, keep SelectionVector positions aligned with original Row Group rows across + dictionary-ID filters, incremental predicates, residual expressions, deletes, and output + materialization. Physical row positions must not be renumbered after pruning. +- Verify lazy materialization avoids reading and decoding non-predicate columns for rejected rows + while advancing all readers correctly. Predicate columns should be read/prefetched first; output + prefetch should wait for survivors when filtering is active. +- Register Parquet Page Cache ranges only for surviving projected Column Chunks, require a stable + file-version key, and assess FileCache, MergeRange, prefetch, requests, and read amplification + together. +- Require counters for Statistics/Dictionary/Bloom pruning, Page Index selected ranges and skipped + rows/pages, raw and filtered rows, dictionary-row filtering, lazy-read savings, cache sources, and + remote I/O. +- Differential tests must cover absent/invalid statistics, missing or partial Page Index, mixed + dictionary/plain encoding, Bloom false positives, NULL/NaN/type conversion, cross-Page batches, + nested/repeated columns, multiple Row Groups/Splits, and all/none filtered. + +## ORC SARG and Index Filtering + +- Trace every pushed predicate from localized `FileScanRequest` through + `build_orc_search_argument()`, ORC `SearchArgumentBuilder`, Stripe selection, SDK RowReader index + pruning, lazy callback filtering, and residual Doris VExpr. +- SARG conversion must be equivalent to the original Doris predicate for every value, including + NULL. Preserve AND/OR/NOT grouping, literal-on-left comparison direction, comparison/IN/NULL + semantics, and wrappers for Runtime Filter, direct-IN, and TopN predicates. +- Verify ORC predicate-domain and literal conversion for integer, floating-point, boolean, string, + binary, varchar, date, decimal, timestamp, and timestamp-instant, including overflow, non-finite + values, signed boundaries, precision/scale, CHAR/VARCHAR, timezone, and NULL. +- Treat schema-evolution casts as SARGable only when truth is preserved in the ORC domain. Review + numeric exactness, decimal widening, date-to-datetime boundary normalization, timestamp precision, + and string/binary casts. Lossy or timezone-changing casts must remain residual. +- For nested predicates, verify struct field name/ordinal traversal and the final ORC type ID. + Unsupported array/map/repeated/missing paths must not target another primitive child. +- Intersect the Split byte window with Stripe ownership before SARG selection, then let ORC RowReader + use row indexes and Bloom filters inside surviving Stripes. SARG must not reintroduce an + out-of-Split Stripe. +- Validate non-adjacent Stripe ranges, all-pruned/no-Stripe cases, file-global row positions, + deletes, and Condition Cache granules after every skipped Stripe or row group. +- Keep SDK filtering and Doris lazy materialization aligned: include the correct filter columns, + preserve selected-row indexes, and decode non-predicate columns only for survivors without + desynchronizing nested vectors or later batches. +- Review SARG cost for large IN lists, deep trees, many Runtime Filters, repeated literal conversion, + Stripe-statistics reads, and SDK index initialization. Build once per reader/Split setup and keep + expensive work out of batch loops. +- Require counters for evaluated/selected groups or Stripes, filtered rows/bytes, groups read, + lazy-filtered rows, I/O, decompression, and decoding. Explain pruning benefit and SARG/index cost. +- Differential tests must cover NULL truth tables, literal-on-left, nested AND/OR/NOT, IN/NOT IN + with NULL, casts, all literal domains, nested structs, unsupported arrays/maps, non-adjacent + Stripes, Split boundaries, row-index strides, Bloom present/absent, and all/none filtered. diff --git a/docs/file-scanner-v2-design.md b/docs/file-scanner-v2-design.md new file mode 100644 index 00000000000000..66b96de1204d14 --- /dev/null +++ b/docs/file-scanner-v2-design.md @@ -0,0 +1,325 @@ +# FileScannerV2 Scan Pipeline Design + +> **Core conclusion:** FileScannerV2 is not primarily about rewriting a file reader. Its purpose is +> to establish stable layer boundaries: Operator/Scheduler owns the control plane, Scanner owns +> query integration and the Split lifecycle, TableReader owns table semantics, and format readers +> own physical reads. All optimizations aim to eliminate unnecessary I/O as early as possible, +> control batch cost, preserve consistent semantics across formats, and make state reusable and +> observable. + +## 1. Design Goals and Boundaries + +FileScannerV2 targets external-data scans. It separates query execution, table-format semantics, +and file-format details into layers that can evolve independently. The design prioritizes durable +boundaries rather than isolated acceleration for a particular format. + +### Core goals + +- Unify the read pipeline for Parquet, ORC, text, JSON, JNI, and other formats. +- Complete pruning and short-circuit evaluation before file I/O whenever possible. +- Isolate table-level semantics from file-local schemas. +- Reuse heavyweight Scanner state across multiple Splits. +- Maintain consistent resource accounting and Profile conventions. + +### Non-goals + +- Do not reimplement every file format inside Scanner. +- Do not force every optimization onto every reader. +- Do not expose file-local column positions to the query layer. +- Do not sacrifice error semantics in order to continue execution. +- Full support for the Load path is currently out of scope. + +> **Design placement rule:** The correct layer for a capability depends on whether it manages the +> query, manages a Split, restores table semantics, or interprets a physical file. Layer boundaries +> take priority over code reuse. + +## 2. Overall Architecture + +V2 divides the scan pipeline into four layers. Upper layers depend only on stable contracts, while +lower layers may evolve independently for each format. + +```mermaid +flowchart TB + Q[Query Plan and Scan Operator] --> S[Scanner Scheduler and Split Source] + S --> F[FileScannerV2 Query Integration] + F --> T[TableReader Table-Semantics Orchestration] + T --> N[Native FileReader] + T --> J[JNI Reader] + N --> P[Parquet / ORC / CSV / Text / JSON] + J --> C[Paimon / Hudi / JDBC / Trino / MaxCompute] + F -. "Profile and Resource Accounting" .-> O[Query Observability] + N -. "FileCache and I/O Statistics" .-> O +``` + +| Layer | Primary responsibilities | Intentionally isolated concerns | +| --- | --- | --- | +| Operator / Scheduler | Select V1 or V2, control concurrency, distribute Splits, and apply late Runtime Filters | Does not understand file-schema mapping or interpret format metadata | +| FileScannerV2 | Maintain Scanner lifecycle, advance Splits, connect query context, predict batch size, handle errors, and collect statistics | Does not decode specific formats or implement table-format delete semantics | +| TableReader | Restore table-level column semantics and manage partition constants, predicates, deletes, Split state, and reader open order | Does not depend on Scanner scheduling | +| Format Reader | Interpret physical formats, metadata, encodings, pages, row groups, and JNI protocols | Does not control query-level concurrency or resource governance | + +> **Primary benefit:** Add a new format by extending a reader, add new table semantics by extending +> TableReader, and add query-level governance in Scanner/Operator. Each change remains in the layer +> that owns it. + +## 3. Core Scan Pipeline + +One Scanner consumes multiple Splits sequentially. The main pipeline advances through a loop rather +than reconstructing the entire scan object for every file. + +```mermaid +sequenceDiagram + participant O as FileScanOperator + participant S as ScannerScheduler + participant X as SplitSource + participant F as FileScannerV2 + participant T as TableReader + participant R as Format Reader + participant U as Upstream Operator + O->>S: Create and schedule Scanner + S->>F: prepare / open + F->>X: Fetch first or next Split + X-->>F: Split descriptor and partition values + S->>F: Refresh late Runtime Filters + F->>T: prepare split + alt Split is pruned early + T-->>F: pruned + F->>X: Continue with next Split + else Split must be read + F->>T: get block + T->>R: Lazily create and open concrete Reader + R-->>T: File-local Block + T-->>F: Table-level Block + F-->>U: Deliver upstream + loop Current Split is not finished + F->>T: get block + T->>R: read next batch + T-->>F: Table-level Block + F-->>U: Deliver upstream + end + F->>X: Advance to next Split + end +``` + +1. **Selection and scheduling:** Operator selects V2 from feature flags, the scan scenario, and the + complete format-support matrix. Multiple Scanners dynamically fetch work from one SplitSource. +2. **One-time initialization:** Expressions, projected columns, I/O Context, and TableReader are + reused throughout the Scanner lifecycle. +3. **Per-Split preparation:** Update only the current file, partition values, delete information, + and the latest available filters. +4. **Open on demand:** Construct the format reader only when data must actually be read, preserving + the opportunity for early pruning. +5. **Repeated delivery:** TableReader produces stable table-level Blocks. Scanner then applies the + common upstream filtering, projection, and statistics path. + +> **Core invariant:** Upstream operators always observe table-level column order and types. Split +> transitions, file-schema differences, cache sources, and concrete formats remain hidden below. + +## 4. Split Lifecycle and Early Pruning + +Split is the most important state-isolation unit in V2. Every transition clears the previous +Split's local state before deciding whether the current Split warrants reader construction. + +```mermaid +stateDiagram-v2 + [*] --> FetchSplit + FetchSplit --> PrepareSplit: Range acquired + PrepareSplit --> Pruned: Partition predicates reject all rows + PrepareSplit --> Ready: Read required + PrepareSplit --> Ignored: Ignorable NOT_FOUND + Ready --> Reading: First get block + Reading --> Reading: Return non-empty Block + Reading --> Finished: Current Split EOF + Reading --> Ignored: NOT_FOUND while reading + Pruned --> FetchSplit + Ignored --> FetchSplit + Finished --> FetchSplit + FetchSplit --> [*]: No more Splits or stopped +``` + +### Why pruning happens during prepare split + +```mermaid +sequenceDiagram + participant S as Scheduler + participant F as FileScannerV2 + participant T as TableReader + participant R as Concrete Reader + S->>F: Inject latest Runtime Filters + F->>T: Current Split and latest filter snapshot + T->>T: Build one-row semantics from partition constants + T->>T: Select only predicates fully answerable by partitions + alt All rows are filtered + T-->>F: Mark Split as pruned + Note over R: No construction, open, or file I/O + else Split may contain matches + T-->>F: Split ready + F->>T: get block + T->>R: Create and open Reader + end +``` + +### Benefits + +- Late Runtime Filters can affect subsequent Splits. +- Unnecessary object-storage requests and metadata reads are avoided. +- Delete-file parsing can also be skipped after pruning. +- All formats share the same Split-pruning semantics. + +### Required constraints + +- Make a pruning decision only when the current partition values fully determine the expression. +- Conservatively retain a Split when the result cannot be determined. +- Pruning, normal completion, and ignored errors must all advance the finished range consistently. +- Cleanup must cover native, JNI, and hybrid child readers. + +## 5. Block Reading and Table-Semantics Restoration + +A file reader returns a file-local Block, while query execution requires a table-level Block. V2 +models the conversion explicitly so schema evolution, partition columns, and virtual columns do not +leak into format readers. + +```mermaid +flowchart LR + A[Table Projection and Predicates] --> B[Global Column Semantics] + B --> C[Column Mapper] + D[File Schema and Local Column Positions] --> C + E[Partition Values / Defaults / Virtual Columns] --> C + C --> F[File Scan Request] + F --> G[Format Reader Reads Required Columns] + G --> H[File-local Block] + H --> I[Type Conversion and Nested-Column Reconstruction] + I --> J[Delete Semantics and Table-level Filtering] + J --> K[Stable Table-level Block] +``` + +| Design object | Problem addressed | Optimization enabled | +| --- | --- | --- | +| Global Index | Expressions use stable table-level positions independent of file-column order | Predicates can be relocated for different file schemas | +| Column Mapper | Handles names, positions, field IDs, missing columns, partition columns, and nested projection uniformly | Reads only required physical columns and enables nested-field pruning | +| File Scan Request | Translates table intent into a local request understood by a format reader | Predicate pushdown, lazy materialization, and dictionary/page/row-group pruning | +| Finalize | Restores file columns to the types, order, and virtual semantics required by the query | Upstream layers remain unaware of file-format differences | + +> **Tradeoff:** The mapping layer adds orchestration cost, but enables cross-format consistency, +> schema evolution, and fine-grained projection and filter optimizations. It is core V2 +> infrastructure. + +## 6. Key Optimizations + +V2 optimization is a continuous pipeline: eliminate work, control the cost of each remaining unit, +and reuse work already performed. + +```mermaid +flowchart TB + A[Eliminate Irrelevant Splits] --> A1[Runtime Filter Partition Pruning] + A --> A2[Constant-Predicate Short Circuit] + B[Eliminate Irrelevant Data] --> B1[Column and Subfield Projection] + B --> B2[Predicate Pushdown and Format-level Pruning] + B --> B3[Delete-Semantics Pushdown] + C[Control Per-batch Cost] --> C1[Small Probe Batch] + C --> C2[Adapt from Materialized Bytes per Row] + D[Reuse and Caching] --> D1[Scanner / TableReader Reuse Across Splits] + D --> D2[FileCache] + D --> D3[Condition Cache and Metadata Cache] + E[Avoid Materialization] --> E1[COUNT / MIN / MAX Aggregate Pushdown] +``` + +| Optimization | Design motivation | Key consideration | +| --- | --- | --- | +| Shared SplitSource with dynamic work assignment | Prevent a Scanner from binding to fixed files and reduce long-tail imbalance | Control concurrency by execution resources, not simply by file count | +| Lazy reader open | Allow pruning before remote I/O and format initialization | Define clear state contracts between prepare and read | +| Adaptive batches | A fixed row count cannot bound memory for wide or nested rows | Sample the final table-level Block's bytes per row; use a small probe without history | +| Projection and predicate localization | Translate table intent into the minimum physical read set | Pushdown must not change final query semantics | +| Layered caches | Reuse remote data and stabilize object-storage access cost | Attribute cache sources accurately to local, remote, and peer reads | +| Aggregate pushdown | Avoid data-page materialization when metadata can answer the query | Disable conservatively when filters or deletes may change the result | + +> **Optimization rule:** First prove that data need not be read, then decide what must be read, and +> finally optimize how much to read at once. Earlier optimizations usually provide greater benefit +> and require stricter correctness boundaries. + +## 7. Format Extension and Hybrid Readers + +V2 does not require every data source to use one physical execution mechanism. TableReader provides +uniform table semantics, while each Split can use native execution, JNI, or a hybrid reader that +dispatches between them. + +```mermaid +flowchart TB + S[Current Split] --> D{Table Format and Split Type} + D -->|Regular File| N[Native TableReader] + D -->|Java Connector| J[JNI TableReader] + D -->|Mixed Splits in One Table| H[Hybrid Reader] + H --> HN[Native Child] + H --> HJ[JNI Child] + N --> U[Uniform Table-level Block] + J --> U + HN --> U + HJ --> U + P[Pruning State / abort split / Profile] -. "Uniform Contract" .-> N + P -. "Uniform Contract" .-> J + P -. "Forward to Active Child" .-> H +``` + +### Adding a new file format + +- Implement schema discovery, reads, and format-level Profile reporting. +- Reuse TableReader mapping, deletes, constants, and finalize logic. +- Declare a capability matrix and select V2 only when every Split is supported. + +### Adding a new table format + +- Add field identity, historical schema, and delete semantics. +- Select native or JNI execution per Split. +- Ensure state queries and cleanup reach the actual child reader. + +> **Incremental migration:** V2 protects compatibility through a capability matrix instead of +> assuming that every format migrates at once. Coverage can expand gradually while retaining the V1 +> fallback path. + +## 8. Observability and Failure Semantics + +Scan optimization remains maintainable only when costs are visible, sources are distinguishable, +and failure semantics are explicit. V2 provides three complementary views: Query Profile, query +resource context, and global metrics. + +```mermaid +flowchart LR + R[FileReader and FileCache Raw Statistics] --> P[Query Profile] + R --> Q[Query Resource Context] + R --> M[Doris Metrics] + P --> P1[Per-query Layer Timings and Counts] + Q --> Q1[Resource Governance and Local/Remote I/O Attribution] + M --> M1[Long-term Node Trends] + C[Condition Cache / Pruning / NOT_FOUND] --> P + C --> Q +``` + +| Failure category | Default semantics | Design rationale | +| --- | --- | --- | +| Query cancellation / should stop | Stop Reader and Scanner loops promptly | Propagate the stop signal through I/O Context to avoid further remote-resource use | +| NOT_FOUND | Return an error by default; skip the current Split only when explicitly configured | Clean reader state and update counters before skipping; do not disguise another error as a missing file | +| Schema / decode / delete-semantics error | Fail immediately | These errors can affect result correctness and must not be swallowed defensively | +| Pruning | Complete the current Split normally | Pruning is an optimization result, not an error, and must be observed separately from Empty/NOT_FOUND | + +> **Observability rule:** Profile explains why one query is slow, ResourceContext explains what that +> query consumed, and DorisMetrics describes overall node health. Their measurements are related but +> not interchangeable. + +## 9. Summary of Design Tradeoffs + +| Design choice | Primary benefit | Cost and constraint | +| --- | --- | --- | +| Scanner, TableReader, and Format Reader layering | Stable responsibilities, extensible formats, and clear test boundaries | Adds translation and state contracts | +| One Scanner consumes multiple Splits | Reuses expressions, caches, and reader-orchestration state | Requires complete isolation of Split-local state | +| Separate table-global and file-local semantics | Supports schema evolution, field mapping, and complex-column pruning | Makes Column Mapper and finalize logic more complex | +| Prune before opening a reader | Maximizes avoided remote I/O and initialization | Can evaluate only predicates that are safe to decide early | +| Adapt batches from actual bytes | Controls memory peaks for wide and nested rows | Requires an initial probe and uses a dynamic estimate | +| Capability matrix with V1 fallback | Enables incremental migration without exposing incomplete format paths | Requires both paths to preserve equivalent semantics during migration | + +> **In one sentence:** FileScannerV2 separates whether to read, what to read, how to read, how to +> restore table semantics, and how to account for cost, allowing correctness, performance, and +> extensibility to evolve independently. + +## Further Reading + +- [FileScannerV2 profiling and pruning PR](https://github.com/apache/doris/pull/65449) diff --git a/docs/file-scanner-v2-parquet-scan-design.md b/docs/file-scanner-v2-parquet-scan-design.md new file mode 100644 index 00000000000000..63f63819670dd2 --- /dev/null +++ b/docs/file-scanner-v2-parquet-scan-design.md @@ -0,0 +1,503 @@ +# FileScannerV2 Parquet Scan Pipeline Design + +> **Reading goal:** Understand how the FileScannerV2 Parquet Reader progressively pushes +> table-level predicates down to Split, Row Group, Page, and Row granularity, then uses indexes, +> lazy materialization, and layered caches to reduce unnecessary I/O and decoding. + +## 1. Design Goals and Core Conclusions + +Parquet V2 is not simply a replacement decoder. It divides a file scan into a **planning phase** and +an **execution phase**: first eliminate impossible matches with lightweight metadata, then read only +predicate columns for surviving ranges, and finally defer output-column reads until matches exist. + +> **In one sentence:** Scan cost contracts through File and Split → Row Group → Page → Row → Column. +> The earlier a non-match is established, the more remote I/O, decompression, decoding, and +> materialization can be avoided. + +- **Uniform entry point:** TableReader maps table semantics to file semantics. ParquetReader handles + only localized columns and predicates. +- **Planning first:** After opening a file, read footer/schema and build `RowGroupReadPlan` objects + instead of making ad hoc decisions during reads. +- **Multi-level predicates:** The same table predicate may be reused at several granularities, but + each layer eliminates data only when it can do so safely. Uncertain cases remain candidates. +- **Predicate columns first:** Read filter columns first and maintain a SelectionVector. Read output + columns only for surviving rows. +- **Layered caches:** File-block cache, Parquet page cache, condition-result cache, and merged small + I/O solve different problems and are not interchangeable. + +**Scope:** This document focuses on the FileScannerV2 Parquet Reader design and core pipeline. It +does not cover Arrow decoder internals, complex-type reconstruction, or expression implementation. + +## 2. Overall Architecture + +Responsibilities are divided across scan orchestration, table-semantic adaptation, format planning, +Row Group execution, column decoding, and I/O. Upper layers own correctness semantics; lower layers +own format-aware pruning and reads. + +```mermaid +flowchart TB + A[FileScanOperator / ScannerScheduler
Schedule Scanners and Runtime Filters] --> B[FileScannerV2
Split Fetching, Batch Control, Profile Aggregation] + B --> C[TableReader
Schema Mapping, Partition/Default Values, Predicate Localization] + C --> D[ParquetReader
Footer/Schema and Row Group Scan Planning] + D --> E[ParquetScanScheduler
Row Group Lifecycle and Batch Reads] + E --> F[ParquetColumnReader
Page Skipping, Decompression, Decoding, Materialization] + F --> G[ParquetFileContext / Arrow RandomAccessFile
Page Cache, MergeRange, Prefetch] + G --> H[Doris FileReader / FileCache / Remote FS] +``` + +| Layer | Core responsibilities | Responsibilities intentionally excluded | +| --- | --- | --- | +| FileScannerV2 | Split lifecycle, reader reuse, dynamic batches, and unified Profile | Does not understand Parquet pages or encodings | +| TableReader | Map table columns, partition columns, missing columns, defaults, and conjuncts into file-local coordinates | Does not parse the Parquet footer directly | +| ParquetReader | Build file context, plan Row Groups, and aggregate format-level statistics | Does not implement table-level schema-evolution semantics | +| ParquetScanScheduler | Open planned Row Groups and order predicate/output column reads | Does not repeat global predicate analysis | +| ColumnReader | Locate and skip pages, decompress, decode, and materialize by Selection | Does not decide whether a Row Group is a candidate | +| FileContext / FileReader | Provide random reads, caches, merged reads, and remote access | Does not interpret SQL predicates | + +> **Design benefit:** Table format, file format, and storage medium remain decoupled. The Parquet +> layer can use footer, page index, dictionary, and other format knowledge while upper layers retain +> uniform scan semantics. + +## 3. From File Open to Scan Plan + +After a reader receives a Split, it opens the file and builds the scan plan. This phase determines +which Row Groups, row ranges, column chunks, and Page Skip Plans will be used later. + +```mermaid +sequenceDiagram + participant FS as FileScannerV2 + participant TR as TableReader + participant PR as ParquetReader + participant FC as ParquetFileContext + participant META as Footer/Metadata + participant PLAN as RowGroup Planner + participant SCH as ScanScheduler + FS->>TR: prepare/open split + TR->>TR: Map schema and localize predicates + TR->>PR: FileScanRequest + PR->>FC: Open FileReader + FC->>META: Read footer and schema + META-->>PR: Row Group / Column Chunk metadata + PR->>PLAN: Candidate Row Groups and local predicates + PLAN->>PLAN: Select by Split range + PLAN->>PLAN: Prune by Statistics/Dictionary/Bloom + PLAN->>PLAN: Prune pages by ColumnIndex+OffsetIndex + PLAN-->>PR: RowGroupReadPlan list + PR->>FC: Register Page Cache ranges for surviving chunks + PR->>SCH: Install plans and column-read request + SCH-->>FS: ready / EOF +``` + +### Key planning objects + +- **FileScanRequest:** Contains `predicate_columns`, `non_predicate_columns`, localized conjuncts, + delete conjuncts, and local column-position mappings. +- **RowGroupReadPlan:** Records the Row Group, its file-global starting row, `selected_ranges` + produced by page-index pruning, and the `page_skip_plan` for each leaf column. +- **ParquetFileContext:** Adapts Doris FileReader to Arrow RandomAccessFile and owns Page Cache, + FileCache prefetch, and MergeRange routing. + +> Planning intentionally proceeds from cheap to expensive. Split and metadata pruning reduce the +> candidate set before finer indexes are read for surviving Row Groups, avoiding index I/O for data +> that is already known to be irrelevant. + +## 4. Predicate Pushdown Design + +Predicate pushdown does not begin by passing table expressions directly to Parquet. TableReader and +ColumnMapper first translate a table expression into an expression understood by the current file. + +```mermaid +flowchart LR + A[Table Conjunct / Runtime Filter] --> B[Resolve Column References] + B --> C{Column Present in Current File?} + C -- "File Column" --> D[Map to LocalColumnId / Block Position] + C -- "Partition Column" --> E[Evaluate with Constant Value] + C -- "Missing Column" --> F[Apply Default or NULL Semantics] + D --> G[Separate Predicate and Output Columns] + E --> G + F --> G + G --> H[FileScanRequest] + H --> I[Reuse Localized Predicates at Parquet Index Levels] +``` + +### Design principles + +1. **Semantics before optimization:** Resolve partition constants, missing columns, defaults, and + type mappings before deciding whether pushdown is safe. +2. **Local coordinates:** Parquet sees only the current file's column IDs and block positions, so it + does not repeatedly interpret table-schema evolution. +3. **Capability checks:** ZoneMap, Dictionary, and Bloom use only expressions they can interpret + safely. All others remain row-level residual predicates. +4. **Prefer safe single-column predicates:** Single-column predicates can drive indexes and staged + filtering. Multi-column, stateful, or error-sensitive expressions retain whole-expression + evaluation. +5. **Runtime Filters can refresh:** ScannerScheduler refreshes late Runtime Filters before reading. + TableReader handles partition-range pruning during Split preparation, and passes file-pushable + parts as localized conjuncts. + +> Pushdown is not merely avoiding another expression evaluation. It projects deterministic facts +> from the expression onto cheaper data summaries. Any case that cannot prove a non-match must +> continue scanning. + +## 5. Predicate Evaluation at Different Granularities + +The same predicate may be attempted at several granularities. Each layer produces a smaller +candidate set that becomes the next layer's input. + +```mermaid +flowchart TB + A[Query / Runtime Filter] --> B[Split / Partition
Skip Entire File or Fragment] + B --> C[Row Group
Statistics / Dictionary / Bloom] + C --> D[Page
ColumnIndex + OffsetIndex] + D --> E[Batch / Row
Dictionary ID + VExpr + Delete Predicate] + E --> F[Column
Materialize Output Only for Surviving Rows] + style B fill:#e8f3ff + style C fill:#eaf7ea + style D fill:#fff5d6 + style E fill:#f4eaff + style F fill:#fce8e6 +``` + +| Granularity | Input information | Main cost avoided | Conservative fallback | +| --- | --- | --- | --- | +| Split / Partition | Partition values, Runtime Filter range, scan byte range | Opening and reading an entire file or fragment | Retain the Split when uncertain | +| Row Group | Footer statistics, dictionary, Bloom filter | I/O and decoding for all column chunks in the group | Retain the Row Group when an index is missing or incompatible | +| Page | ColumnIndex min/max/null data and OffsetIndex | Page I/O, decompression, and decoding | Read the affected range when page indexes are incomplete | +| Row / Batch | Actual column values, dictionary IDs, residual conjuncts | Later predicate-column and output-column materialization | Evaluate full VExpr semantics | +| Column | SelectionVector | Reads, decoding, and memory writes for non-predicate columns | Read all projected columns sequentially when no filtering applies | + +> **Key distinction:** Row Group and Page indexes generally eliminate impossible candidates; they +> do not produce final query results. Row-level predicates determine whether individual rows match. + +## 6. Row Group Planning and Index Coordination + +The Row Group Planner combines physical layout from the footer, the Split byte range, and predicate +index capabilities into an executable plan. The key property is a stable candidate-reduction order. + +```mermaid +sequenceDiagram + participant P as Planner + participant M as RowGroup Metadata + participant S as Statistics + participant D as Dictionary Page + participant B as Bloom Filter + participant I as Page Index + P->>M: Enumerate footer Row Groups + P->>M: Assign Split by Row Group midpoint + loop Each candidate Row Group + P->>S: Evaluate ZoneMap(min/max/null) + alt Proven non-match + S-->>P: Prune Row Group + else Match remains possible + P->>D: Read and evaluate available dictionary + alt Dictionary domain cannot match + D-->>P: Prune Row Group + else Match remains possible + P->>B: Probe Bloom Filter + B-->>P: Prune or retain + end + end + end + P->>I: Read ColumnIndex/OffsetIndex for survivors + I-->>P: selected_ranges + page_skip_plans +``` + +### Why this order is used + +- **Statistics:** Usually already in the footer, making them the lowest-cost option for range and + null semantics. +- **Dictionary:** Requires reading the dictionary page, but can prove a complete non-match for + low-cardinality string columns. +- **Bloom:** Requires Bloom data I/O and is useful for negative membership tests. A positive result + may be a false positive. +- **Page Index:** Builds page-level row ranges only for surviving Row Groups, avoiding index cost for + groups already eliminated. + +### How the plan drives physical skips + +ColumnIndex provides min/max/null semantics for each page. OffsetIndex maps pages to Row Group row +numbers and file offsets. Candidate ranges from multiple predicate columns are intersected into +`selected_ranges`; a `page_skip_plan` is then built for each leaf so its column reader can skip pages +that do not overlap surviving rows. + +> `selected_ranges` represents logical row ranges, while `page_skip_plan` represents physical page +> reads. Keeping them separate allows the scheduler to advance by row batch while each column skips +> according to its own page boundaries. + +## 7. Batch Reads, Dictionary Filtering, and Lazy Materialization + +Execution follows a filter-first, materialize-later strategy. The scheduler advances through +`selected_ranges`, asks column readers to skip gaps, and then reads the current batch. + +```mermaid +sequenceDiagram + participant S as ParquetScanScheduler + participant PC as Predicate Column Readers + participant SEL as SelectionVector + participant EX as Residual Expressions + participant OC as Output Column Readers + S->>S: Open next Row Group + S->>S: Skip row ranges rejected by Page Index + S->>PC: Read first predicate column set + PC->>SEL: Filter with dictionary IDs or actual values + loop Remaining safe single-column predicates + S->>PC: Read/materialize only surviving rows + PC->>SEL: Further reduce selection + end + S->>EX: Evaluate residual multi-column predicates and deletes + EX->>SEL: Produce final survivors + alt Rows survive + S->>OC: Prefetch and read non-predicate columns + OC->>OC: Materialize by Selection + S-->>S: Assemble output Block + else No rows survive + S->>S: Do not read deferred output columns + end +``` + +### Row-level dictionary filtering + +```mermaid +flowchart LR + A[Single-column Predicate] --> B{Column Fully Dictionary Encoded?} + B -- "No" --> F[Read Actual Values and Execute VExpr] + B -- "Yes" --> C[Read Dictionary Page] + C --> D[Evaluate Predicate on Dictionary Values
Build Dictionary-ID Bitmap] + D --> E[Decode Data-page Dictionary IDs
Update SelectionVector Directly] + E --> G[Materialize Only Survivors] +``` + +- Applies to non-repeated primitive, string-like BYTE_ARRAY / FIXED_LEN_BYTE_ARRAY columns whose + complete Column Chunk uses dictionary data encoding. +- Safe AND subexpressions may remove components exactly covered by dictionary evaluation. OR or + non-equivalent expressions are not rewritten aggressively. +- Stateful, potentially throwing, or whole-batch-sensitive expressions disable staged + single-column scheduling and fall back to reading required columns before whole-expression + evaluation. + +> **Optimization loop:** The earlier SelectionVector shrinks, the fewer values later predicate and +> output columns must decode and copy. This is the main benefit of lazy materialization in a +> columnar format. + +## 8. Supported Indexes and Their Boundaries + +V2 uses native Parquet metadata and encoding information. It does not construct Doris-internal +storage indexes for external Parquet files. + +| Capability | Granularity | Suitable predicates | Result property | Main limitations | +| --- | --- | --- | --- | --- | +| Footer Statistics / ZoneMap | Row Group | Ranges, comparisons, IS NULL/IS NOT NULL, and expressions safely convertible to ZoneMap | Can prove the entire group cannot match | Requires valid min/max/null_count and safe type conversion | +| Dictionary Pruning | Row Group | Single-column predicates exactly evaluable over the dictionary domain | Can prove the entire group cannot match | Low-cardinality string-like primitive with complete dictionary encoding | +| Parquet Bloom Filter | Row Group / Column Chunk | Equality and IN membership-negation predicates | Negative result can prune; positive result requires verification | Controlled by configuration; file must contain Bloom data; false positives are possible | +| ColumnIndex | Page | Predicates evaluable from min/max/null | Produces candidate pages and row ranges | Requires an index and decodable compatible types | +| OffsetIndex | Page → Row Range | Does not evaluate predicates directly | Maps page results to row numbers and physical skip plans | Normally used with ColumnIndex | +| Dictionary-ID Filter | Row / Batch | Safe single-column string-like predicates | Exact filtering of actual rows | Complete dictionary encoding and non-repeated primitive only | +| Condition Cache Bitmap | File-global granule | Stable cacheable conditions | Reuses previous filtering to reduce row ranges | Not a native Parquet index; uncovered ranges remain candidates | + +### Index-selection overview + +```mermaid +flowchart TD + A[Localized Predicate] --> B{ZoneMap Evaluatable?} + B -- "Yes" --> C[Row Group Statistics / Page ColumnIndex] + B -- "No" --> D{Single-column Dictionary Evaluatable?} + D -- "Yes" --> E[Row Group Dictionary + Row Dictionary-ID] + D -- "No" --> F{Bloom Negative-membership Test?} + F -- "Yes" --> G[Parquet Bloom Filter] + F -- "No" --> H[Retain as Row-level Residual VExpr] + C --> H + E --> H + G --> H +``` + +> Indexes are layered rather than mutually exclusive. An index may remove only ranges already +> proven impossible; residual predicates still guarantee final correctness. + +## 9. Cache and I/O Optimization + +Parquet V2 has four complementary cache and I/O paths: cache remote file blocks, cache serialized +Parquet ranges, cache predicate results, and merge small random reads. + +```mermaid +flowchart TB + A[Parquet Column Reader ReadAt] --> B{Parquet Page Cache Hit?} + B -- "Yes" --> C[Return Cached Serialized Range Bytes] + B -- "No" --> D{MergeRange Active?} + D -- "Yes" --> E[MergeRangeFileReader
Merge Adjacent Small I/O] + D -- "No" --> F[Base FileReader] + E --> G[CachedRemoteFileReader / FileCache] + F --> G + G --> H[Local Block / Peer / Remote Object Storage] + H --> I[Populate FileCache] + I --> J[Populate Page Cache for Registered Ranges] +``` + +| Mechanism | Cached or optimized object | Lifecycle and key | Problem addressed | +| --- | --- | --- | --- | +| FileCache | Remote file blocks | Related to filesystem/path and file version; may hit locally or through a peer | Avoid repeated object-storage access and support background prefetch | +| Parquet Page Cache | Serialized bytes within registered Column Chunk ranges | Stable file key depends on path, mtime/version, and file size; disabled when mtime is unreliable | Reduce repeated page reads and support exact/subrange coverage | +| Condition Cache | Condition-surviving granule bitmap | Managed by condition and file-range context | Reuse filtering results before reading columns | +| MergeRangeFileReader | Not a cache; merges small ranges into larger slices | Installed temporarily for projected chunks of the current Row Group | Reduce remote small-I/O count and request overhead | + +### Why Page Cache registers only surviving chunks + +The footer is read before Row Group planning and before Page Cache ranges are registered, so +footer/metadata bytes never enter the Parquet Page Cache. After planning, only projected Column +Chunks from surviving Row Groups are registered, limiting pollution and key count. + +### Relationship between prefetch and MergeRange + +- When the base reader is CachedRemoteFileReader, predicate/output ranges for the current Row Group + may be prefetched into FileCache. +- When average projected chunks are small and the reader is not in-memory, install + MergeRangeFileReader so subsequent Arrow `ReadAt` calls actually use merged reads. +- With row-level filters, prefetch predicate columns first. Prefetch non-predicate columns only after + at least one row survives, avoiding unnecessary bandwidth. + +## 10. Other Key Optimizations + +### 10.1 Condition Cache: Move Historical Filter Results Earlier + +```mermaid +sequenceDiagram + participant T as TableReader / Cache Context + participant S as ParquetScanScheduler + participant P as RowGroup Plans + participant R as Row Filter + alt Cache Hit + T->>S: Bitmap + base granule + S->>P: Intersect with selected_ranges + P-->>S: Smaller pending row ranges + else Cache Miss + T->>S: Empty bitmap context + S->>R: Execute normal row predicates + R-->>S: SelectionVector + S->>S: Mark granules containing survivors + S-->>T: Publish to Condition Cache later + end +``` + +On a hit, only granules explicitly proven unnecessary by the bitmap are removed. Rows outside +bitmap coverage remain candidates. On a miss, granules containing surviving rows are marked, +trading granularity for reuse and smaller cache entries. + +### 10.2 Adaptive Batches + +FileScannerV2 uses a small probe batch to measure bytes per row in the final table Block. It derives +later batch rows from a target Block size, bounded by the system batch-size limit. Wide rows use +smaller batches to reduce memory peaks; narrow rows use larger batches for throughput. + +```mermaid +flowchart LR + A[Small Probe Batch] --> B[Read and Complete Table-level Materialization] + B --> C[Estimate Bytes per Row] + C --> D[Target Block Bytes / Bytes per Row] + D --> E[Choose Next Row Cap] + E --> F[Bound by batch_size and Selected Range] +``` + +### 10.3 Aggregate Pushdown + +When TableReader proves that no filter or delete semantics can change the result, COUNT / MIN / MAX +may use Parquet metadata to compute all or part of an aggregate without scanning data pages. This is +a metadata aggregation optimization and is distinct from Row Group index pruning. + +### 10.4 Staged Prefetch + +Without row-level filtering, output columns may be warmed together. With filtering, warm predicate +columns first and defer non-predicate columns until at least one row survives, aligning network +bandwidth with lazy materialization. + +## 11. Correctness, Fallback, and Capability Boundaries + +V2 follows a prove-before-skip rule. Missing indexes, unsupported types, expressions that cannot be +split safely, or read anomalies must never change query semantics. + +> **Correctness baseline:** Index results only reduce candidate sets. Every expression not exactly +> covered remains a residual conjunct evaluated against actual data. + +| Scenario | V2 behavior | +| --- | --- | +| Missing Statistics or unsafe min/max conversion | Treat the column's ZoneMap as unavailable and retain the Row Group/Page | +| Bloom missing, disabled, or unreadable | Skip Bloom pruning and continue with later scan stages | +| Incomplete dictionary page, mixed non-dictionary encoding, complex/repeated column | Disable dictionary pruning and Dictionary-ID Filter; use actual values | +| Missing or inconsistent ColumnIndex/OffsetIndex | Disable fine-grained page pruning and read the full candidate range | +| Multi-column, OR, stateful, or error-order-sensitive expression | Preserve whole-expression evaluation to avoid changing SQL short-circuit or error semantics | +| No stable file-version identity for Page Cache | Disable Parquet Page Cache to prevent stale-byte reads | +| Incomplete Condition Cache coverage | Retain and recompute uncovered ranges | + +### Capability boundaries + +- Parquet Reader uses indexes and encoding metadata already present in the file; it does not build + new indexes for external files. +- Page boundaries and definition/repetition levels are more complex for nested/repeated columns, so + some dictionary and page-level optimizations conservatively fall back. +- Bloom is probabilistic and is safe only for proving absence. A positive Bloom result is not a row + match. +- Page Index benefit depends on whether the writer produced indexes, data ordering, and predicate + selectivity. + +## 12. Profile Observation and Troubleshooting + +Troubleshoot in this order: verify planning effectiveness, row filtering, lazy materialization, and +then I/O/cache health. Total ScanTime alone does not identify the cause. + +```mermaid +flowchart TD + A[Slow Scan] --> B{Many Row Groups Pruned?} + B -- "No" --> C[Check Statistics/Dictionary/Bloom Availability and Predicate Shape] + B -- "Yes" --> D{Page selected_ranges Shrink Significantly?} + D -- "No" --> E[Check ColumnIndex/OffsetIndex and Data Ordering] + D -- "Yes" --> F{Many Rows Filtered by Predicates?} + F -- "Yes" --> G[Check Deferred Prefetch and Selection-based Output Materialization] + F -- "No" --> H[Low Selectivity: Inspect Decode and I/O Throughput] + G --> I{Cache Hits and Small I/O Reasonable?} + H --> I + I -- "No" --> J[Check FileCache, Page Cache, MergeRange, and Remote Reads] + I -- "Yes" --> K[Check Type Conversion, Complex Columns, and Downstream Operators] +``` + +### Important metric families + +| Metric family | Question answered | +| --- | --- | +| Row Group pruning | How many total Row Groups were pruned by Statistics/Dictionary/Bloom, and how much time did each stage take? | +| Page index pruning | How many indexes were checked, pages/rows were pruned, ranges selected, and pages skipped? | +| Dictionary row filter | How often were predicates rewritten, dictionaries read, bitmaps built, and attempts successful or rejected? | +| Predicate / raw rows | How many rows were read and rejected, and was lazy materialization worthwhile? | +| Parquet Page Cache | What were hit/miss/write counts and compressed/decompressed hit shapes? | +| FileCache Profile | How many local/peer/remote bytes, waits, downloads, and hits occurred? | +| Merge / request I/O | Were small reads merged, and were request count and read amplification reasonable? | +| Condition Cache | How many rows were skipped early after a cache hit? | + +> Interpret pruning ratios in the context of write layout. Unsorted data produces wide min/max +> ranges, so Row Group/Page pruning may be ineffective even when the reader and indexes work +> correctly. + +## 13. Summary + +The FileScannerV2 Parquet scan pipeline has three primary threads: + +1. **Semantic thread:** TableReader maps table schema and predicates into stable file-local + semantics, preserving schema evolution, partition columns, and missing columns. +2. **Pruning thread:** Split → Row Group → Page → Row progressively applies Runtime Filters, + Statistics, Dictionary, Bloom, Page Index, and actual-value filters. +3. **I/O thread:** Predicate-first reads, SelectionVector, lazy materialization, adaptive batches, + FileCache/Page Cache/Condition Cache, and MergeRange reduce read amplification together. + +```mermaid +flowchart LR + A[Table-level Semantics] --> B[File-local Predicates] --> C[RowGroupReadPlan] --> D[selected_ranges] --> E[SelectionVector] --> F[Final Block] + G[Statistics / Dictionary / Bloom] --> C + H[ColumnIndex / OffsetIndex] --> D + I[FileCache / Page Cache / MergeRange] --> C + I --> D + I --> E +``` + +> **Final design criterion:** V2 turns format knowledge into an explicit scan plan and requires the +> executor to perform only the minimum necessary reads. Indexes safely reduce candidates, caches +> reuse cost, and lazy materialization avoids reading irrelevant columns for rejected rows. + +This document reflects the current code pipeline and is intended as a common reference for +architecture reviews, performance analysis, and Profile troubleshooting. diff --git a/docs/new-parquet-reader-column-index-refactor.md b/docs/new-parquet-reader-column-index-refactor.md deleted file mode 100644 index 56f8c7ca4a37d5..00000000000000 --- a/docs/new-parquet-reader-column-index-refactor.md +++ /dev/null @@ -1,404 +0,0 @@ -# New Reader 列标识实现说明 - -本文说明 Doris new table/file reader 栈中各种列标识的当前含义，以及它们在 -`FileScannerV2`、`TableReader`、`TableColumnMapper` 和 new Parquet reader 中的流转逻辑。 - -核心原则是把 **schema identity** 和 **执行期位置** 分开： - -- schema identity 用来判断 table column 和 file column 是否是同一列。 -- index/position 用来表示 block、projection tree、scan request 或 constant map 中的位置。 -- FE column unique id 只在 scanner 边界用于定位 slot，进入 table/file reader 后不再出现。 - -共享定义集中在 `be/src/format_v2/column_data.h`。file reader 通用请求定义在 -`be/src/format_v2/file_reader.h`。new Parquet reader 自己的 Parquet 内部 schema tree 定义在 -`be/src/format_v2/parquet/parquet_column_schema.h`。 - -## 层级边界 - -当前 reader 栈可以按语义分成三层。 - -### FileScannerV2：FE 标识到 reader 标识的边界 - -`FileScannerV2` 仍能看到 FE 下发的 `slot_id`、`col_unique_id`、`TFileScanSlotInfo` 和 -`TColumnAccessPath`。这些 FE 侧标识只在这里使用。 - -`FileScannerV2::_build_projected_columns()` 会把 `_params->required_slots` 转成 -`std::vector`： - -- vector 下标就是 `GlobalIndex`。 -- `_slot_id_to_global_index` 把 FE `slot_id` 转成 `GlobalIndex`，用于 row-level conjunct。 -- `_column_unique_id_to_global_index` 把 FE `col_unique_id` 转成 `GlobalIndex`，用于 column predicate。 -- `ColumnDefinition::identifier` 表示 table-side schema identity，默认是列名；如果外部 schema - 提供 field id，则改用 field id。 -- partition/default/generated 信息被挂到 `ColumnDefinition` 上，由 table reader 层处理。 - -从这一层往下，table/file reader 不再使用 FE column unique id。 - -### TableReader / TableColumnMapper：table schema 到 file schema - -`TableReader::open_reader()` 对每个 split 打开一个具体 `FileReader`，先通过 -`FileReader::get_schema()` 获取当前文件的 file-local schema，再用 `TableColumnMapper` 建立映射。 - -`TableColumnMapper` 的输入是： - -- table/global schema：`FileScannerV2` 构造的 `projected_columns`。 -- file-local schema：具体 file reader 返回的 `std::vector`。 -- per-split partition values。 -- table-level row filters 和 column predicates。 - -`TableColumnMapper` 的输出是： - -- `ColumnMapping`：构造阶段使用的 table column 到 file/constant/virtual source 的映射。 -- `FileScanRequest`：只含 file-local projection、file-local block layout 和 file-local filters。 -- `ColumnMapResult` / `ResultColumnMapping`：给 table reader finalize 阶段消费的最终映射。 -- `FilterEntry`：给 filter localization 使用的 `GlobalIndex -> LOCAL/CONSTANT/UNSET` target。 -- `ConstantMap`：partition/default/generated 常量列。 - -### FileReader / ParquetReader：只理解 file-local 请求 - -`FileReader` 只暴露两类 schema/request： - -- `get_schema(std::vector*)`：返回文件自身 schema。 -- `open(std::unique_ptr&)`：接收已经 localize 后的 file-local scan request。 - -具体 file reader 不理解 table/global schema、Iceberg default、partition column、FE slot id 或 -FE column unique id。 - -new Parquet reader 使用 `FileScanRequest` 中的 `LocalColumnIndex` 创建 column reader，并使用 -`local_positions` 决定 file-local block layout。 - -## ColumnDefinition - -定义位置：`be/src/format_v2/column_data.h` - -`ColumnDefinition` 是 table/global schema 和 file-local schema 共用的列定义。它表示列名、类型、 -nested children、默认表达式、partition 属性和 file-local column kind。 - -关键字段： - -- `identifier`：schema identity。用于 table column 和 file column 匹配。 -- `local_id`：file reader 返回的 schema node 在当前 parent 下的 reader-local id。 -- `name`：逻辑列名。BY_NAME 且没有显式 string identifier 时会回退到它。 -- `type`：当前 schema node 的 Doris 类型。 -- `children`：nested children。table/global schema 中是 table children；file schema 中是 - file-local children。 -- `default_expr`：missing/default/generated column 的物化表达式。 -- `is_partition_key`：partition column 标记。 -- `column_type`：file-local column kind，例如普通数据列或 row number virtual column。 - -`ColumnDefinition` 不保存 FE column unique id。它也不保存“应该按什么方式匹配”。匹配方式由 -`TableColumnMapperOptions::mode` 统一决定。 - -### identifier - -`identifier` 是一个 `Field`，语义接近 DuckDB `MultiFileColumnDefinition::identifier`： - -- `TYPE_NULL`：没有显式 identifier。BY_NAME 时使用 `name`。 -- `TYPE_INT`：在 BY_FIELD_ID 中表示 field id；在 BY_INDEX 中表示 file schema position。 -- `TYPE_STRING`：显式 name identifier。 - -访问 helper： - -- `has_identifier_field_id()` / `get_identifier_field_id()`：BY_FIELD_ID 使用。 -- `get_identifier_name()`：BY_NAME 使用；没有显式 string identifier 时返回 `name`。 -- `get_identifier_position()`：BY_INDEX 使用。 -- `file_local_id()`：file reader projection 使用；优先返回 `local_id`，否则回退到 int - identifier。这个回退只用于兼容某些 file schema 构造路径，不应重新引入 FE id 语义。 - -## 强类型位置 - -### GlobalIndex - -定义位置：`be/src/format_v2/column_data.h` - -`GlobalIndex` 表示 table/global output block 中的 top-level 列位置。当前等于 -`_params->required_slots` 的下标。 - -主要使用位置： - -- `ColumnMapping::global_index` -- `TableFilter::global_indices` -- `TableColumnPredicates` 的 key -- `ColumnMapResult` / `ResultColumnMapping` 的 key -- `FilterEntry` map 的 key - -`GlobalIndex` 不是 FE slot id，也不是 FE column unique id。 - -### LocalColumnId - -定义位置：`be/src/format_v2/column_data.h` - -`LocalColumnId` 表示当前物理文件 schema 的 top-level reader-local column id。 - -主要使用位置： - -- `FileScanRequest::local_positions` 的 key。 -- `LocalColumnIndex::top_level()`。 -- new Parquet reader 创建 top-level column reader。 -- page index、statistics、bloom filter 等 file-local pruning 的 root column key。 -- row position 这类 reader 内部 virtual column id。 - -`LocalColumnId` 不是 file-local block position。一个 top-level file column 在本次 scan request -输出 block 中的位置由 `LocalIndex` 表示。 - -### LocalIndex - -定义位置：`be/src/format_v2/column_data.h` - -`LocalIndex` 表示一次 `FileScanRequest` 内 file-local block 的列位置。 - -主要使用位置： - -- `FileScanRequest::local_positions` 的 value。 -- file-local rewritten `SlotRef` 的 input position。 -- `TableReader` 从 file block 取列。 -- `ParquetScanScheduler` 把 column reader 读出的数据写入 file block。 - -`LocalIndex` 是 request-local block layout，不是 file schema ordinal。 - -### ConstantIndex - -定义位置：`be/src/format_v2/column_data.h` - -`ConstantIndex` 表示 `ConstantMap` 中的 entry 位置。它用于 per-split/per-file 常量列： - -- partition column。 -- schema evolution default column。 -- generated/default expression column。 -- 将来可扩展到更多 virtual/constant source。 - -`FilterEntry` 可以指向 `ConstantIndex`。当一个 row-level conjunct 只引用 constant target 时， -`TableReader` 会在打开 file reader 前用 1 行常量 block 求值；如果结果为 false/NULL，当前 split -直接跳过。 - -### LocalColumnIndex - -定义位置：`be/src/format_v2/column_data.h` - -`LocalColumnIndex` 表示递归 file-local projection path： - -```cpp -struct LocalColumnIndex { - int32_t index = -1; - bool project_all_children = true; - std::vector children; -}; -``` - -语义： - -- root entry 的 `index` 是 `LocalColumnId`。 -- nested entry 的 `index` 是当前 parent 下的 file-local child id。 -- `project_all_children = true` 表示读取整个 subtree。 -- `project_all_children = false` 表示只读取 `children` 中列出的 child paths。 - -通用 helper： - -- `is_full_projection()` -- `is_partial_projection()` -- `find_child_projection()` -- `is_child_projected()` -- `merge_local_column_index()` - -new Parquet reader 的 STRUCT/LIST/MAP reader 都消费这套 projection helper： - -- STRUCT：只创建被投影 child 的 reader。 -- LIST：把 element projection 递归传给 element reader。 -- MAP：总是读取 key，把 value projection 递归传给 value reader。 - -## FileScanRequest - -定义位置：`be/src/format_v2/file_reader.h` - -`FileScanRequest` 是 table reader 交给 file reader 的唯一 scan 输入。它不包含 table/global schema。 - -关键字段： - -- `predicate_columns`：row-level conjunct/delete conjunct 需要先读取的 file-local projection。 -- `non_predicate_columns`：最终输出需要读取、且不需要先参与 row-level filter 的 file-local - projection。 -- `local_positions`：`LocalColumnId -> LocalIndex`，决定 file-local block layout。 -- `conjuncts` / `delete_conjuncts`：已经把 table/global slot 改写成 file-local slot 的表达式。 -- `column_predicate_filters`：file-layer pruning hints，只用于 min/max、page index、dictionary、 - bloom filter 等剪枝，不参与 batch row filtering。 - -`predicate_columns` 和 `non_predicate_columns` 都按 file-local schema 表达。file reader 只需要根据 -这两个列表创建 reader，并按 `local_positions` 写入 file block。 - -## TableColumnMapper 逻辑 - -定义位置： - -- `be/src/format_v2/column_mapper.h` -- `be/src/format_v2/column_mapper.cpp` - -### 匹配模式 - -`TableColumnMapperOptions::mode` 决定 `identifier` 的解释方式： - -- `BY_FIELD_ID`：`TYPE_INT` identifier 是 field id。 -- `BY_NAME`：`TYPE_STRING` identifier 或 `name` 是匹配名。 -- `BY_INDEX`：`TYPE_INT` identifier 是 file schema position。 - -`TableReader::open_reader()` 当前默认按 field id 映射；如果 file schema 首列没有 int identifier， -会 fallback 到 BY_NAME。Hive reader 可覆盖默认模式，Hive1 ORC 这类场景可使用 BY_INDEX。 - -### create_mapping() - -`create_mapping()` 为每个 `GlobalIndex` 生成一个 `ColumnMapping`： - -1. partition column 优先映射到 `ConstantMap`。 -2. BY_INDEX 时按 file position 取 file schema。 -3. 普通列通过 matcher 在 file schema 中找对应 file field。 -4. 缺失但带 default expr 的列映射到 `ConstantMap`。 -5. 特殊 virtual column 记录 virtual column type。 -6. 允许 missing column 时保留空 mapping，由 table finalize 阶段补 NULL/default。 - -`ColumnMapping::file_local_id` 是 table column 绑定到 file schema 后的 reader-local id： - -- root mapping 中可转成 `LocalColumnId`。 -- nested mapping 中表示 parent 下的 child id。 -- constant/missing/virtual mapping 没有 `file_local_id`。 - -schema identity field id 不保存在 `ColumnMapping` 中，只保存在 -`ColumnDefinition::identifier` 中，并由 mapper 的匹配模式解释。 - -### create_scan_request() - -`create_scan_request()` 把 table-level scan 信息转换成 file-local request： - -1. 先把不参与 row-level filter 的输出列加入 `non_predicate_columns`。 -2. 调用 `localize_filters()`，把 row-level conjunct 和 column predicates 定位到 file-local source。 -3. 为所有已读取 file column 重建 output projection，让 `ColumnMapping::projection` 指向正确的 - `LocalIndex`。 -4. 生成 `ColumnMapResult` 和 `ResultColumnMapping`，供 table reader finalize。 - -`local_positions` 在这个阶段确定。同一个 file column 如果同时被 filter 和 output 使用，只会有 -一个 `LocalIndex`。 - -### FilterEntry - -`FilterEntry` 是 `GlobalIndex` 到 filter target 的结果： - -- `LOCAL`：filter 可以在 file-local block 上求值，target 是 `LocalIndex`。 -- `CONSTANT`：filter 只依赖 `ConstantMap` entry。 -- `UNSET`：当前 split 无法下推到 file reader。 - -`TableColumnMapper::_build_filter_entries()` 在 `FileScanRequest::local_positions` 确定后生成 -`FilterEntry`。表达式改写时只把 `LOCAL` target 改写成 file-local slot；`CONSTANT` target 用于 -split-level constant filter evaluation。 - -### ColumnMapResult / ResultColumnMapping - -`ColumnMapResult` 记录一个 global result column 的递归映射结果： - -- `local_column_id`：root file column。 -- `column_index`：file-local projection tree。 -- `mapping`：root 指向 `LocalIndex`，nested child 通过 `IndexMapping::child_mapping` 递归映射。 - -`ResultColumnMapping` 是最终可消费的 `GlobalIndex -> ColumnMapEntry` map。`ColumnMapEntry` 包含： - -- `IndexMapping mapping` -- `local_type` -- `global_type` -- `filter_conversion` - -TableReader finalize 阶段用它把 file-local block 转成 table/global block。 - -### nested child mapping - -复杂列映射时，`IndexMapping::child_mapping` 的 key 是 table/global child ordinal，value 是对应 -file-local child mapping。这样 filter 中的 `STRUCT_EXTRACT` 可以按 table child ordinal 找到 -file child ordinal。 - -Doris 不再维护额外的 `NestedPredicateTargetInfo` / filter target path。nested filter localization -直接沿 `IndexMapping::child_mapping` 转换 selector path。 - -对于 `SELECT s.name WHERE s.id > 5` 这类 filter-only child： - -- `s.name` 进入 output projection。 -- `s.id` 会进入 predicate projection。 -- `original_file_children` 保留 projection 前的 file children，用于定位 filter-only child。 -- `child_mappings` 只描述输出 shape，避免 filter-only child 改变最终 STRUCT/LIST/MAP shape。 - -## Parquet 内部 schema 标识 - -定义位置：`be/src/format_v2/parquet/parquet_column_schema.h` - -`ParquetColumnSchema` 是 new Parquet reader 内部 schema tree。它描述 Parquet 逻辑字段和 primitive -leaf column 的关系，不暴露给 table reader。对外统一通过 `ParquetReader::get_schema()` 返回 -`std::vector`。 - -关键字段： - -- `local_id`：当前 parent 下的 reader-local id。top-level 是 root field ordinal，nested 是 child - ordinal。`LocalColumnIndex` 传给 `ParquetColumnReaderFactory` 的就是这个 id。 -- `parquet_field_id`：Parquet schema element 中可选的 field_id。Arrow 在不存在 field_id 时返回 - `-1`。它只作为 schema matching identifier，不用于读取 column chunk。 -- `name`：Parquet schema name。 -- `type`：转换后的 Doris 类型。 -- `leaf_column_id`：Parquet primitive leaf column ordinal。用于访问 `ColumnDescriptor`、 - row group column chunk、statistics、page index、bloom filter 等。复杂节点为 `-1`。 -- `type_descriptor`：primitive leaf 的 Parquet physical/logical type 信息。 -- `descriptor`：primitive leaf 的 Arrow Parquet `ColumnDescriptor`。 -- `max_definition_level` / `max_repetition_level`：该 node 下的最大 Dremel level。 -- `nullable_definition_level`：当前 node 自身为 NULL 时对应的 definition level。 -- `repeated_repetition_level`：当前或最近 repeated container 的 repetition level。 - -`ParquetReader::get_schema()` 会把 `ParquetColumnSchema` 转成 `ColumnDefinition`： - -- 如果 `parquet_field_id >= 0`，`ColumnDefinition::identifier` 是 `TYPE_INT` field id。 -- 否则 `identifier` 是 `TYPE_STRING` name。 -- `ColumnDefinition::local_id` 是 `ParquetColumnSchema::local_id`。 -- children 递归转换。 - -因此 table reader 可以按 field id 或 name 匹配，而 Parquet reader 自己仍只按 `local_id`、 -`leaf_column_id` 和 Dremel levels 读取数据。 - -## 端到端流转 - -一次 split 的列标识流转如下： - -1. `FileScannerV2::_build_projected_columns()`： - FE `slot_id` / `col_unique_id` 被翻译成 `GlobalIndex`，并生成 table-side - `ColumnDefinition`。 -2. `ParquetReader::init()`： - 解析 Arrow Parquet schema，构造内部 `ParquetColumnSchema`。 -3. `ParquetReader::get_schema()`： - 把 Parquet 内部 schema 暴露成 file-side `ColumnDefinition`。 -4. `TableReader::open_reader()`： - 根据 file schema 是否带 int identifier 选择 BY_FIELD_ID 或 BY_NAME，并调用 mapper。 -5. `TableColumnMapper::create_mapping()`： - 用 `ColumnDefinition::identifier` 匹配 table/global schema 和 file-local schema，生成 - `ColumnMapping`。 -6. `TableColumnMapper::create_scan_request()`： - 生成 `FileScanRequest`，其中所有 projection 和 block position 都是 file-local 的。 -7. `ParquetReader::open()`： - 校验 `LocalColumnId`，用 `LocalColumnIndex` 创建 column readers，并规划 row group pruning。 -8. `ParquetScanScheduler`： - 按 `local_positions` 把 predicate/non-predicate column 写入 file-local block。 -9. `TableReader` finalize： - 使用 `ResultColumnMapping`、`ConstantMap` 和 projection expression，把 file-local block 转成 - table/global output block。 - -## 使用约定 - -修改 new reader 代码时应遵守以下约定： - -- 不要在 table/file reader 层重新传递 FE column unique id。 -- 不要把 `ColumnDefinition::identifier` 当作 file reader 读取 id。 -- 不要把 `LocalColumnId` 当作 block position；block position 使用 `LocalIndex`。 -- 不要把 `LocalIndex` 当作 schema ordinal。 -- `LocalColumnIndex::index` 在 root 和 child 层含义不同，调用方必须知道当前 projection node - 所在层级。 -- file reader 只能消费 `FileScanRequest`，不能理解 partition/default/generated/table schema。 -- column predicate pruning 是 file-layer hint，不等价于 row-level filter。 -- constant filter 可以在 table reader 层提前求值，但不应下推到 file reader。 - -## 已知限制 - -TVF 查询 Parquet 且文件没有 field id 时，top-level BY_NAME 已经可以通过 name identifier 工作。 -但 nested access path 的 fallback 目前仍有一处 TODO：STRUCT child fallback 使用 struct ordinal -构造 int identifier。对于没有 field id 的 nested Parquet schema，BY_NAME 场景应保留 string -identifier，让 `TableColumnMapper` 从 Parquet file schema 中按 name 解析 file-local child id。 -该问题已在 `be/src/exec/scan/file_scanner_v2.cpp` 代码中记录，当前未修复。 diff --git a/docs/new-parquet-reader-ut-improvement-plan.md b/docs/new-parquet-reader-ut-improvement-plan.md deleted file mode 100644 index 4ece111d0d6323..00000000000000 --- a/docs/new-parquet-reader-ut-improvement-plan.md +++ /dev/null @@ -1,325 +0,0 @@ -# New Parquet Reader UT Improvement Plan - -本文档评估 Doris new parquet reader 当前 UT 覆盖方式，并给出更合理的测试分层、数据构造方法和落地优先级。 - -目标不是追求形式上的 100% 行覆盖率，而是让测试能够发现 new parquet reader 最容易出错的真实问题：schema 兼容、definition/repetition level 物化、投影/过滤交互、row group/page pruning、delete predicate 以及 schema evolution 组合。 - -## 当前覆盖方式评估 - -当前测试分层大体合理： - -| 层级 | 代表文件 | 当前价值 | -|---|---|---| -| Schema resolver UT | `be/test/format_v2/parquet/parquet_schema_test.cpp` | 直接构造 Parquet schema node，验证 `ParquetColumnSchema` 的 kind、type、level 和非法 schema 拒绝。速度快，适合覆盖 schema 分支。 | -| Type resolver UT | `be/test/format_v2/parquet/parquet_type_test.cpp` | 覆盖 physical/logical/converted type 到 Doris type 的映射。 | -| Leaf value UT | `be/test/format_v2/parquet/parquet_leaf_reader_test.cpp` | 覆盖 nullable spacing、binary/fixed/bool/float16 等 leaf append 细节。 | -| Column reader UT | `be/test/format_v2/parquet/parquet_column_reader_test.cpp` | 用 Arrow writer 生成真实 parquet 文件，覆盖 scalar/struct/list/map 的 read、skip、select、overflow。 | -| File reader UT | `be/test/format_v2/parquet/parquet_reader_test.cpp` | 覆盖 open/read、多 row group、predicate selection、statistics/dictionary/page index pruning、row position、delete predicate。 | -| Table reader UT | `be/test/format_v2/table_reader_test.cpp` | 覆盖 table schema 到 file schema mapping、aggregate pushdown、default value、Iceberg delete/virtual column 等跨层行为。 | - -这个方向是正确的，但目前有三个明显缺口： - -1. Schema 兼容测试和真实读取测试之间缺少桥接。`parquet_schema_test.cpp` 可以证明 legacy LIST/MAP schema 被解析成期望的 tree，但不能证明 `ListColumnReader`、`MapColumnReader` 可以正确消费对应 def/rep levels。 -2. 真实 parquet 文件主要由 Arrow writer 生成。Arrow 生成的文件通常符合标准 layout，不能充分代表 Hive、Spark、old parquet-mr、旧 Doris 或其它 legacy writer 的 schema 形态。 -3. 异常路径和组合路径覆盖不足。比如 optional map key 被 schema 接受后，真实数据中 key 为 null 必须在 materialize 阶段报错；key/value stream 不对齐、invalid repeated level、non-nullable complex column 读到 null 等 corruption 路径需要专门测试。 - -## 改进原则 - -1. 按风险分层测试，不用单一大 fixture 覆盖所有逻辑。 -2. Schema resolver 只验证 schema 归一化，不承担真实读取正确性的证明。 -3. Def/rep level materialization 要有直接单测，避免所有边界都依赖真实 parquet 文件构造。 -4. 对 legacy layout 使用 golden parquet corpus，而不是只用 Arrow writer 动态生成。 -5. Reader 集成测试覆盖跨模块行为，避免在 SQL regression 中验证过多 BE 内部细节。 -6. SQL regression 只保留用户可见和跨层最关键路径，避免回归测试过慢。 - -## 推荐测试分层 - -### L0: Schema Resolver Table-Driven UT - -位置：`be/test/format_v2/parquet/parquet_schema_test.cpp` - -职责：覆盖 `parquet_column_schema.cpp` 的 schema 归一化规则。建议把 LIST/MAP case 整理成 table-driven 形式，每个 case 明确： - -- 输入 schema layout -- 是否成功 -- top-level kind/type/nullability -- child kind/name/type/nullability -- definition/repetition level -- error message 关键字 - -必须覆盖的 schema 形态： - -| 类别 | Case | -|---|---| -| LIST 标准格式 | Standard 3-level list: `optional group a (LIST) { repeated group list { optional int32 element; } }` | -| LIST legacy | repeated primitive, repeated group named `array`, repeated group named `_tuple`, repeated group with multiple children | -| LIST wrapper 判定 | repeated group with logical annotation, repeated group whose only child is repeated, repeated group whose only child is optional scalar | -| Bare repeated | repeated primitive field, repeated group field inside struct | -| MAP 标准格式 | required/optional outer map, required/optional value | -| MAP 兼容格式 | optional key accepted at schema level, `MAP_KEY_VALUE` converted annotation | -| Invalid schema | LIST outer has zero/multiple children, non-repeated LIST child, MAP outer has zero/multiple children, primitive MAP entry, non-repeated MAP entry, entry child count not equal to 2, repeated outer LIST/MAP in normal mode | -| Unsupported type | UTC TIME rejection, unsupported physical/logical type | - -L0 的验收标准：schema branch 新增或修改时，必须有对应 table-driven case；但 L0 通过不代表 reader 行为充分。 - -### L1: Def/Rep Level Materializer UT - -位置建议： - -- `be/test/format_v2/parquet/parquet_nested_materializer_test.cpp` -- 或拆分为 `parquet_list_column_reader_test.cpp`、`parquet_map_column_reader_test.cpp` - -职责：用 fake child reader 直接喂 definition levels、repetition levels 和 leaf values，验证 `ListColumnReader` / `MapColumnReader` 的 offsets、nullmap、child values、cursor 和错误路径。 - -这种方式比构造真实 parquet 文件更适合覆盖边界，因为 def/rep level 是复杂类型 reader 的核心输入。 - -建议增加测试工具： - -```cpp -class FakeNestedColumnReader final : public ParquetColumnReader { -public: - Status load_nested_batch(int64_t rows) override; - Status build_nested_column(int64_t length_upper_bound, MutableColumnPtr& column, - int64_t* values_read) override; - const std::vector& nested_definition_levels() const override; - const std::vector& nested_repetition_levels() const override; - int64_t nested_levels_written() const override; -}; -``` - -必须覆盖的 materialize case： - -| 类别 | Case | -|---|---| -| LIST 正常路径 | null list, empty list, list with values, list with null element, consecutive repeated elements | -| LIST 操作 | read 分批、skip 后 read、select 非连续行、select 跨 overflow 边界 | -| LIST 异常 | first level has `rep_level == list.repetition_level`, non-nullable LIST 读到 null, child value count 不匹配 | -| MAP 正常路径 | null map, empty map, one entry, multiple entries, nullable value, complex value | -| MAP 操作 | read 分批、skip 后 read、select 非连续行、value scalar path 和 complex value path | -| MAP 异常 | null key, value stream ended before key stream, key/value repetition level 不对齐, key count 不匹配, value count 不匹配, non-nullable MAP 读到 null | - -L1 的验收标准：`ListColumnReader::build_nested_column()` 和 `MapColumnReader::build_nested_column()` 的主要分支必须有直接 UT；corruption path 不能只靠真实文件偶然触发。 - -### L2: Golden Parquet Corpus UT - -位置建议： - -- 数据文件：`be/test/exec/test_data/parquet_v2_compat/` -- 测试文件：`be/test/format_v2/parquet/parquet_compat_corpus_test.cpp` - -职责：保存小型真实 parquet 文件，覆盖非 Arrow 标准 writer 或难以用 Arrow writer 生成的 legacy layout。每个文件控制在几十行以内，配套记录 schema 来源和 expected output。 - -建议文件来源： - -| 来源 | 覆盖目标 | -|---|---| -| Arrow writer | 标准 LIST/MAP、page v2、dictionary/plain、不同 row group/page size | -| Spark | Spark nested list/map schema、nullable struct/list/map 混合 | -| Hive/parquet-mr | legacy two-level list、optional map key、`array` / `bag` / `key_value` 等命名兼容 | -| 手工生成 | malformed-but-parseable def/rep level edge case，或特殊 converted annotation | - -Golden 文件命名建议： - -```text -be/test/exec/test_data/parquet_v2_compat/ - list_two_level_repeated_primitive.parquet - list_tuple_struct_element.parquet - list_repeated_group_with_logical_map_element.parquet - map_optional_key_no_null.parquet - map_optional_key_with_null.parquet - map_value_list_nullable.parquet - nested_list_struct_map_list.parquet - README.md -``` - -每个 corpus case 至少验证： - -- `get_schema()` 输出是否符合预期 -- full read 输出是否符合预期 -- projection read 输出是否符合预期 -- skip/select 后输出是否符合预期 -- 预期失败文件是否返回明确错误 - -L2 的验收标准：每一个 schema compatibility rule 至少有一个真实 parquet 文件证明 reader 可以消费该 layout。 - -### L3: New Parquet Reader Integration UT - -位置：`be/test/format_v2/parquet/parquet_reader_test.cpp` - -职责：覆盖 file reader 层的组合行为，不重复 L1 的低层 def/rep 细节。 - -建议补充或保留以下组合： - -| 类别 | Case | -|---|---| -| Projection + predicate | `SELECT s.b WHERE s.a > x` 对应 file-local projection 与 predicate projection 合并 | -| Complex non-predicate select | predicate 过滤后，非谓词复杂列通过 selection vector 读取 | -| Row group/page pruning + complex projection | page index 缩小 row ranges 后，list/map/struct 输出行数和 offsets 正确 | -| Dictionary/statistics pruning | nested scalar leaf predicate 可 prune，但 repeated leaf 不做错误 aggregate/pruning | -| Delete predicate | delete predicate 和 query predicate 同时作用时 row position、selection、输出列一致 | -| Timestamp TZ | timestamp tz mapping 后 schema、read、min/max pushdown 一致 | -| Reopen split | 同一个 reader reopen 不残留 selection、cast、predicate projection、page skip state | - -L3 的验收标准：跨 reader state 的行为必须有 UT，尤其是 reopen、filter 后 selection、page skip 后 output column 不 double skip。 - -### L4: Table Reader And SQL Regression - -位置： - -- `be/test/format_v2/table_reader_test.cpp` -- `regression-test/suites/external_table_p*_parquet/` 或现有 parquet 外表相关目录 - -职责：覆盖用户可见行为和 FE/BE 接口组合，不在 regression 中验证 BE 内部 offset/nullmap 细节。 - -建议保留少量高价值 SQL regression： - -| 场景 | SQL 覆盖 | -|---|---| -| Legacy LIST/MAP 文件可读 | `SELECT *`, `SELECT nested_child`, `WHERE nested_child predicate` | -| Schema evolution | missing nested child with default, reordered/renamed nested field | -| Predicate pushdown 正确性 | row group/page pruning 开关开启时结果与关闭时一致 | -| Aggregate pushdown 正确性 | `count`, `min`, `max` 对 flat leaf 和 supported nested single leaf 正确；repeated leaf fallback | -| Iceberg/Paimon delete | delete vector / position delete / equality delete 与 parquet reader 组合结果正确 | - -L4 的验收标准：新增用户可见兼容能力时必须有 SQL regression；纯内部 refactor 不强制补 SQL regression，但需要 L0-L3 覆盖。 - -## 覆盖矩阵 - -下面的矩阵用于判断新改动应该补哪一层测试。 - -| 逻辑区域 | L0 Schema | L1 Def/Rep | L2 Corpus | L3 Reader | L4 SQL | -|---|---:|---:|---:|---:|---:| -| Parquet type mapping | 必须 | 不需要 | 可选 | 可选 | 可选 | -| LIST/MAP schema compatibility | 必须 | 可选 | 必须 | 可选 | 必须覆盖用户可见新增能力 | -| Bare repeated field | 必须 | 必须 | 必须 | 可选 | 可选 | -| List offsets/nullmap | 不足 | 必须 | 必须 | 必须 | 可选 | -| Map offsets/nullmap/key validation | 不足 | 必须 | 必须 | 必须 | 可选 | -| Projection pruning | 可选 | 可选 | 必须 | 必须 | 必须覆盖用户可见路径 | -| Predicate selection | 不需要 | 可选 | 可选 | 必须 | 必须覆盖关键路径 | -| Statistics/dictionary/page pruning | 不需要 | 不需要 | 可选 | 必须 | 结果一致性必须 | -| Aggregate pushdown | 不需要 | 不需要 | 可选 | 必须 | 必须 | -| Delete predicate / row position | 不需要 | 不需要 | 可选 | 必须 | Iceberg/Paimon 必须 | -| Error/corruption path | 必须覆盖 schema error | 必须覆盖 materialize error | 必须覆盖真实坏文件 | 可选 | 可选 | - -## 推荐优先级 - -### P0: 立即补齐的正确性保护 - -1. 为 legacy LIST schema 增加真实读取 corpus： - - repeated primitive list - - `_tuple` struct element - - repeated group with multiple children -2. 为 optional MAP key 增加两类真实读取： - - optional key 但所有 key 非 null，读取成功 - - optional key 且存在 null key，读取失败并包含 `contains null key` -3. 增加 fake def/rep level materializer UT： - - list null/empty/null element/multi element - - map null/empty/null value/multi entry/null key -4. 增加 skip/select 覆盖： - - legacy list corpus 上执行 skip/select - - map value list 或 list struct map list 上执行 select - -### P1: 组合路径保护 - -1. Projection + predicate 同时命中同一 nested struct 的不同 child。 -2. Page index pruning 后读取 complex output column，验证没有 double skip。 -3. Row group statistics/dictionary pruning 后从后续 row group 读取 nested column。 -4. Reopen split 后 predicate projection、selection vector、page skip plan 不残留。 - -### P2: 完整性和长期质量 - -1. 建立 `parquet_v2_compat` corpus README，记录文件生成方式、writer 版本、schema、预期行为。 -2. 对 changed files 定期跑 coverage，关注 branch coverage，不只看 line coverage。 -3. 对 schema resolver 增加 table-driven case，减少散落 assert。 -4. 对 materializer 增加 fuzz/property-style 小范围测试：随机生成合法 list/map rows，转换为 def/rep levels 后读回比较原始 logical rows。 - -## 测试数据构造建议 - -### 动态生成数据 - -适合： - -- Arrow 标准 schema -- row group/page size 控制 -- dictionary/plain/page index/statistics 行为 -- type mapping 常规 case - -优点是无需维护二进制文件，case 可读性高。 - -缺点是不能覆盖大量 legacy writer layout。 - -### Golden parquet 文件 - -适合： - -- Hive/Spark/parquet-mr legacy LIST/MAP schema -- Arrow writer 不容易生成的 converted annotation -- malformed-but-parseable 文件 -- 兼容性回归保护 - -要求： - -1. 文件尽量小，通常 3 到 20 行。 -2. 配套 README 说明生成命令、writer 版本、schema、逻辑数据。 -3. 不在 UT 中依赖外部网络或外部服务。 -4. 预期结果在 C++ UT 中直接断言，SQL regression 的 `.out` 仍由 regression 脚本生成。 - -### Fake reader 数据 - -适合： - -- def/rep level 边界 -- corruption path -- cursor/overflow 状态 -- non-nullable output 遇到 null - -要求： - -1. fake reader 只模拟 `ParquetColumnReader` 必需接口。 -2. 每个 case 明确输入 levels 和 expected logical rows。 -3. 错误 case 检查 `Status` 类型和关键错误文本。 - -## 验收标准 - -一个 new parquet reader 改动合入前，建议满足： - -1. 改动 schema resolver：至少补 L0；如果新增兼容能力，补 L2；如果用户可见，补 L4。 -2. 改动 list/map/struct reader：至少补 L1 和 L3；涉及 legacy layout 时补 L2。 -3. 改动 pruning/predicate/aggregate：至少补 L3；用户可见 SQL 语义补 L4。 -4. 改动 table reader mapping/schema evolution：至少补 `table_reader_test.cpp`，必要时补 L4。 -5. 新增 error handling：必须有负向 UT，不能只依赖代码审查。 - -推荐执行命令： - -```bash -./run-be-ut.sh --run '--filter=ParquetSchemaTest.*' -./run-be-ut.sh --run '--filter=ParquetColumnReaderTest.*:NewParquetReaderTest.*:ParquetScanTest.*' -./run-be-ut.sh --run '--filter=TableReaderTest.*' -``` - -对重要重构或发布前验证，建议执行： - -```bash -./run-be-ut.sh --run '--filter=Parquet*:*TableReaderTest*' --coverage -``` - -如果本地工具链无法执行 UT，需要在提交说明或 PR 中明确说明失败原因，并在 CI 或可用环境补跑。 - -## 不建议的方式 - -1. 不建议用更多 schema-only case 替代真实读取 case。schema 正确不等于 reader 正确。 -2. 不建议只用 Arrow writer 动态生成文件证明 compatibility。兼容性问题通常来自非 Arrow writer。 -3. 不建议把所有复杂类型组合塞进一个巨大 fixture 后只断言少量输出。失败定位困难，覆盖意图不清晰。 -4. 不建议把内部 def/rep level 边界全部放到 SQL regression。执行慢、定位差、难覆盖异常路径。 -5. 不建议用 100% line coverage 作为合入门槛。更合理的是 changed branch coverage + 风险矩阵覆盖。 - -## 最小落地计划 - -第一阶段只需要完成 P0： - -1. 新增 `parquet_nested_materializer_test.cpp`，覆盖 list/map def/rep 核心正常和异常路径。 -2. 新增 `be/test/exec/test_data/parquet_v2_compat/README.md` 和 4 到 6 个小型 golden parquet 文件。 -3. 新增 `parquet_compat_corpus_test.cpp`，对 golden 文件做 schema/full read/projection/skip/select 断言。 -4. 将现有 `parquet_schema_test.cpp` 中 LIST/MAP schema case 整理为 table-driven 或至少按类别分组。 - -完成第一阶段后，才能较有信心地说 new parquet reader 的关键逻辑有有效测试保护；否则当前 UT 只能证明主路径和部分 schema 分支，不能充分发现 legacy compatibility 和 complex materialization 的问题。 diff --git a/docs/parquet-list-map-compat-design.md b/docs/parquet-list-map-compat-design.md deleted file mode 100644 index a02ca6e822aaf0..00000000000000 --- a/docs/parquet-list-map-compat-design.md +++ /dev/null @@ -1,664 +0,0 @@ -# Parquet LIST/MAP Compatibility Design - -本文描述如何参考 Arrow Parquet 的 LIST/MAP 兼容策略，在 Doris new parquet reader 中支持更多 Parquet 标准和 legacy 复杂类型 schema。 - -目标不是改变 `ListColumnReader` / `MapColumnReader` 的读取模型，而是在 schema 构建阶段把不同物理 schema 归一化成 Doris 当前 reader 可以消费的统一 `ParquetColumnSchema` tree。 - -## 背景 - -Parquet 的复杂类型是通过 group schema、logical/converted annotation、definition levels 和 repetition levels 共同表达的。 - -标准 LIST/MAP schema 比较明确，但历史 writer 产生过多种 legacy 形态。例如 LIST 可能缺少标准 `list.element` wrapper，MAP entry group 可能叫 `key_value`、`entries` 或其它名字。 - -Arrow C++ 的处理思路是： - -1. 在 Parquet schema conversion 阶段识别标准和 legacy schema。 -2. 将这些 schema 归一化为 Arrow `ListType` / `MapType` / `StructType`。 -3. 后续 reader 只消费归一化后的 nested field tree，不在读取阶段继续判断 legacy schema 名字。 - -Doris new parquet reader 应采用相同边界： - -1. `parquet_column_schema.cpp` 负责兼容不同 LIST/MAP physical schema。 -2. `ParquetColumnSchema` 输出统一的 LIST/MAP child tree。 -3. `ListColumnReader` / `MapColumnReader` / `ParquetLeafReader` 不感知 legacy schema 形态。 - -## 当前 Doris 限制 - -当前 `build_node_schema()` 的 LIST 分支只支持标准 3-level LIST： - -```text -optional group a (LIST) { - repeated group list { - optional int32 element; - } -} -``` - -当前限制： - -- outer LIST group 必须只有一个 child。 -- repeated child 必须是 group。 -- repeated group 必须只有一个 child。 -- 不支持 repeated primitive list。 -- 不支持 repeated group 多字段 struct element。 -- 不支持 `array` / `_tuple` 这类 legacy structural name。 - -当前 MAP 分支支持标准 MAP 结构： - -```text -optional group m (MAP) { - repeated group key_value { - required binary key; - optional int32 value; - } -} -``` - -当前限制： - -- outer MAP group 必须只有一个 child。 -- entry child 必须 repeated group。 -- entry group 必须正好两个 children。 -- key 必须 required。 -- 不支持 key-only map。 -- 不支持没有 repeated entry layer 的非标准 MAP。 - -## 设计原则 - -1. 兼容逻辑只放在 schema 构建阶段。 -2. reader 层继续消费统一 schema tree。 -3. 不支持会改变 reader model 的格式，例如没有 repeated entry layer 的 MAP。 -4. 第一阶段不支持 key-only map，因为 Doris `ColumnMap` 需要 values column。 -5. 对容易误判的 schema 保持严格，避免把普通 struct 错解析成 LIST/MAP。 -6. 支持范围对齐 Arrow 的稳定 legacy compatibility 规则，而不是无限放宽。 - -MAP projection 语义也保持收敛： - -- partial MAP projection 只表示 value subtree pruning，例如 `MAP>` 投影 `value.b` 后输出 `MAP>`。 -- key 不作为可裁剪 projection 子树。reader 始终读取完整 key stream，因为 key stream 决定 entry existence、offsets，并且 key 本身承载 MAP 的 key equality 语义。 -- schema projection 重建 `DataTypeMap` 时保留原始 key type，只根据 projected value child 重建 value type。 - -## LIST 兼容规则 - -对于 outer group annotated as `LIST`： - -```text -optional group a (LIST) { - repeated ... repeated_child; -} -``` - -先要求： - -- outer LIST group 必须只有一个 child。 -- child 必须是 repeated。 - -然后根据 repeated child 形态判断 element schema node。 - -### 1. 标准 3-level LIST - -```text -optional group a (LIST) { - repeated group list { - optional int32 element; - } -} -``` - -解析： - -- repeated child 是 wrapper。 -- element 是 wrapper 的唯一 child：`list.element`。 -- `ParquetColumnSchema(LIST).children[0]` 指向 element schema。 - -### 2. Repeated primitive legacy LIST - -```text -optional group a (LIST) { - repeated int32 element; -} -``` - -解析： - -- repeated primitive 本身是 element。 -- element 本身不 nullable，因为 repeated primitive 不提供额外 optional element level。 -- array 自身 nullable 仍由 outer LIST group 决定。 - -### 3. Repeated group as struct element - -```text -optional group a (LIST) { - repeated group element { - optional int32 x; - optional binary y; - } -} -``` - -解析： - -- repeated group 有多个 children。 -- repeated group 本身是 element。 -- element type 是 `STRUCT`。 - -### 4. Legacy structural name - -Arrow 会将某些名字视作 structural element，而不是标准 wrapper。 - -```text -optional group a (LIST) { - repeated group array { - optional int32 item; - } -} -``` - -```text -optional group a (LIST) { - repeated group a_tuple { - optional int32 item; - } -} -``` - -解析： - -- repeated group 名为 `array`，或名为 `_tuple`。 -- repeated group 本身是 element。 -- 即使它只有一个 child，也不要剥掉这一层。 - -### 5. One-child repeated group wrapper - -```text -optional group a (LIST) { - repeated group list { - optional int32 element; - } -} -``` - -如果 repeated group 只有一个 child，且不是 legacy structural name，则按 wrapper 处理： - -- element 是 repeated group 的唯一 child。 - -但这里不能只按 child 数量判断。需要额外保持 Arrow / parquet-format 的 backward compatibility 规则： - -- 如果 repeated group 自身带 `LIST` 或 `MAP` annotation，则 repeated group 本身是 element，不剥 wrapper。 -- 如果 repeated group 的唯一 child 也是 repeated，则 repeated group 本身是 element，不剥 wrapper。 -- 只有当 repeated group 无 logical annotation、唯一 child 非 repeated、且不是 legacy structural name 时，才把它当作标准 wrapper 剥掉。 - -这样可以避免把 two-level `List>`、two-level `List>` 或单字段 repeated struct element 错解析成少一层的结构。 - -## LIST schema resolver - -建议在 `parquet_column_schema.cpp` 中新增 helper： - -```cpp -struct ListElementResolution { - const parquet::schema::Node* repeated_node = nullptr; - const parquet::schema::Node* element_node = nullptr; - SchemaBuildContext repeated_context; - SchemaBuildContext element_context; - bool element_is_repeated_node = false; -}; - -Status resolve_list_element_node( - const parquet::SchemaDescriptor& schema, - const parquet::schema::GroupNode& list_group, - const SchemaBuildContext& list_context, - ListElementResolution* result); -``` - -Resolver 逻辑： - -```text -if list_group.field_count != 1: - reject - -repeated_node = list_group.field(0) -if !repeated_node.is_repeated: - reject - -repeated_context = child_context(list_context, repeated_node, 0) - -if repeated_node.is_primitive: - element_node = repeated_node - element_context = repeated_context - element_is_repeated_node = true - return - -repeated_group = as_group(repeated_node) -if repeated_group.field_count == 0: - reject - -if repeated_group.field_count > 1: - element_node = repeated_node - element_context = repeated_context - element_is_repeated_node = true - return - -if has_structural_list_name(list_group.name, repeated_group.name): - element_node = repeated_node - element_context = repeated_context - element_is_repeated_node = true - return - -if repeated_group has LIST or MAP annotation: - element_node = repeated_node - element_context = repeated_context - element_is_repeated_node = true - return - -only_child = repeated_group.field(0) -if only_child.is_repeated: - element_node = repeated_node - element_context = repeated_context - element_is_repeated_node = true - return - -element_node = only_child -element_context = child_context(repeated_context, only_child, 0) -element_is_repeated_node = false -``` - -`has_structural_list_name()` 对齐 Arrow 的 legacy rule： - -```text -name == "array" || name == list_name + "_tuple" -``` - -## LIST schema build - -`build_node_schema()` 的 LIST 分支改为： - -```text -resolve_list_element_node(...) - -column_schema.kind = LIST -column_schema.definition_level = repeated_context.definition_level -column_schema.repetition_level = repeated_context.repetition_level -column_schema.repeated_repetition_level = repeated_context.repeated_repetition_level - -build child schema from resolved element_node and element_context -column_schema.type = nullable_if_needed(DataTypeArray(child.type), list_node) -column_schema.children = [child] -propagate_child_levels(column_schema) -``` - -### repeated group itself as element - -当 element 是 repeated group 本身时，需要注意不要把这个 repeated group 再解释成一层 LIST。 - -预期效果： - -```text -optional group a (LIST) { - repeated group element { - optional int32 x; - optional binary y; - } -} -``` - -应构造成： - -```text -LIST - child: STRUCT -``` - -而不是： - -```text -LIST - child: LIST or extra repeated container -``` - -实现上可以新增一个 internal build mode： - -```cpp -enum class SchemaBuildMode { - NORMAL, - REPEATED_GROUP_AS_LIST_ELEMENT, -}; -``` - -当 mode 是 `REPEATED_GROUP_AS_LIST_ELEMENT`： - -- 当前 repeated group 作为 element 本身构造成 STRUCT 或 annotated logical type。 -- 它的 repeated level 已经由 list entry 层消费，不再把 repeated 当作额外 array 层。 -- 如果当前 repeated group 是普通 group，则构造成 `STRUCT` element。 -- 如果当前 repeated group 带 `LIST` annotation，则继续按 LIST 解析它的 child repeated layer，构造成 nested list element。 -- 如果当前 repeated group 带 `MAP` 或 `MAP_KEY_VALUE` annotation，则继续按 MAP 解析它的 child repeated entry layer，构造成 map element。 -- 构造当前 element schema 时，不得再次因为“当前节点本身是 repeated”引入隐式 list；只有它内部的 child repeated layer 才能产生下一层 list/map repetition 语义。 - -如果希望保持改动更小，也可以新增专用函数： - -```cpp -Status build_repeated_group_as_list_element_schema(...); -``` - -该函数至少需要处理 repeated group 作为普通 struct element 的场景；如果选择不用通用 build mode，则还需要显式覆盖 repeated group annotated as LIST/MAP 的场景。 - -## MAP 兼容规则 - -对于 outer group annotated as `MAP` 或 legacy `MAP_KEY_VALUE`： - -```text -optional group m (MAP) { - repeated group entries { - required binary key; - optional int32 value; - } -} -``` - -支持： - -- 只有 outer group 带 `MAP` / `MAP_KEY_VALUE` annotation 时，才进入 MAP 兼容解析。 -- entry group 名字可以是 `key_value`、`entries` 或其它。 -- key/value 字段名不强制必须叫 `key` / `value`。 -- 第一个 child 是 key。 -- 第二个 child 是 value。 -- key 必须 required。 -- value 可以 required 或 optional。 - -不支持： - -- outer MAP group 多个 children。 -- entry child 非 repeated。 -- entry child 是 primitive。 -- entry group 没有 value，即 key-only map。 -- 没有 repeated entry layer 的 MAP。 -- nullable key。 - -## MAP schema resolver - -建议新增 helper： - -```cpp -struct MapEntryResolution { - const parquet::schema::GroupNode* entry_group = nullptr; - SchemaBuildContext entry_context; -}; - -Status resolve_map_entry_group( - const parquet::schema::GroupNode& map_group, - const SchemaBuildContext& map_context, - MapEntryResolution* result); -``` - -Resolver 逻辑： - -```text -if map_group.field_count != 1: - reject - -entry_node = map_group.field(0) -if !entry_node.is_repeated: - reject -if entry_node.is_primitive: - reject - -entry_group = as_group(entry_node) -if entry_group.field_count != 2: - reject - -key_node = entry_group.field(0) -value_node = entry_group.field(1) -if key_node.repetition != REQUIRED: - reject - -entry_context = child_context(map_context, entry_node, 0) -return -``` - -## MAP schema build - -`build_node_schema()` 的 MAP 分支应和 LIST 一样在 schema 构建阶段折叠物理 wrapper。 -`key_value` / `entries` / 任意合法 entry group 只用于解析 repeated entry level，不出现在 -最终 `ParquetColumnSchema.children` 中： - -```text -MAP - child[0]: key - child[1]: value -``` - -构造流程： - -```text -resolve_map_entry_group(...) - -column_schema.kind = MAP -column_schema.definition_level = entry_context.definition_level -column_schema.repetition_level = entry_context.repetition_level -column_schema.repeated_repetition_level = entry_context.repeated_repetition_level - -build key child from entry_group.field(0) -build value child from entry_group.field(1) - -column_schema.type = nullable_if_needed(DataTypeMap(nullable(key.type), nullable(value.type)), map_node) -column_schema.children = [key_schema, value_schema] -propagate_child_levels(column_schema) -``` - -这里保持 `MapColumnReader` 的直接 key/value 假设： - -- `column_schema.children[0]` 是 key。 -- `column_schema.children[1]` 是 value。 -- MAP node 自身保存 entry repeated group 的 `definition_level` / `repetition_level` / - `repeated_repetition_level`，用于 materialize offsets、null map 和 empty map。 - -注意：`DataTypeMap` 中把 key type 包成 nullable 是 Doris nested column materialization 的内部类型约定，不代表 Parquet nullable key 被支持。Schema resolver 仍必须在 `key_node.repetition != REQUIRED` 时 reject。 - -## 不支持 key-only map 的原因 - -Key-only map 可能长这样： - -```text -optional group m (MAP) { - repeated group entries { - required binary key; - } -} -``` - -理论上可以解释为 set-like map 或 `MAP`，但 Doris `ColumnMap` 需要 keys column 和 values column。 - -若要支持，需要额外设计： - -- synthetic null value schema。 -- constant-null value reader。 -- `MapColumnReader` value stream 缺失时的特殊路径。 - -这会改变 reader tree，不属于本次 schema compatibility 的最小范围。因此第一阶段明确 reject。 - -## 不支持 no-entry MAP 的原因 - -No-entry MAP 可能长这样： - -```text -optional group m (MAP) { - required binary key; - optional int32 value; -} -``` - -它缺少 repeated entry layer，因此没有 repetition level 可以表达多个 map entries，也无法生成 Doris `ColumnMap` offsets。 - -这不是标准 MAP，也不是 Arrow 主要兼容的 legacy 形态。第一阶段应 reject。 - -## 对 reader 层的影响 - -预期不修改 reader 层核心逻辑。 - -保持： - -- `ListColumnReader` 只读取 `column_schema.children[0]` 作为 element reader。 -- `MapColumnReader` 读取 `column_schema.children[0/1]` 作为 key/value reader。 -- `MapColumnReader` 对 partial MAP projection 只接受 value child projection，显式 key child projection 应 reject；即使只裁剪 value，reader 也必须完整读取 key stream。 -- `ParquetLeafReader` 只负责 leaf records/levels/values 读取和 batch materialization。 -- `nested_column_materializer.*` 只负责 Doris nested Column 构造 helper。 - -风险点在 LIST repeated group as element： - -- 如果该 repeated group 是 struct element，需要确保 schema builder 不把 repeated group 再解释成一个额外 repeated container。 -- 这个风险应通过专用 build mode 或专用 helper 解决。 - -## 错误处理策略 - -错误信息应明确指出具体 unsupported schema 原因： - -- LIST outer group child count invalid。 -- LIST child is not repeated。 -- LIST repeated group has no child。 -- MAP outer group child count invalid。 -- MAP entry is not repeated group。 -- MAP entry child count is not 2。 -- MAP key is nullable。 - -不要用过于笼统的 `Unsupported parquet LIST encoding` 覆盖所有错误，否则后续排查文件兼容性问题会困难。 - -## 测试计划 - -### LIST 正例 - -1. 标准 3-level LIST： - -```text -optional group a (LIST) { - repeated group list { - optional int32 element; - } -} -``` - -2. Repeated primitive legacy LIST： - -```text -optional group a (LIST) { - repeated int32 element; -} -``` - -3. Repeated group struct element： - -```text -optional group a (LIST) { - repeated group element { - optional int32 x; - optional binary y; - } -} -``` - -4. Legacy `array` name： - -```text -optional group a (LIST) { - repeated group array { - optional int32 item; - } -} -``` - -5. Legacy `_tuple` name： - -```text -optional group a (LIST) { - repeated group a_tuple { - optional int32 item; - } -} -``` - -6. Repeated group annotated as nested LIST： - -```text -optional group a (LIST) { - repeated group array (LIST) { - repeated int32 array; - } -} -``` - -预期解析为 `ARRAY>`，不要剥掉 `array (LIST)` 这一层。 - -7. Repeated group annotated as MAP： - -```text -optional group a (LIST) { - repeated group array (MAP) { - repeated group key_value { - required binary key; - optional int32 value; - } - } -} -``` - -预期解析为 `ARRAY>`，不要剥掉 `array (MAP)` 这一层。 - -8. One-child repeated group whose child is repeated： - -```text -optional group a (LIST) { - repeated group element { - repeated int32 items; - } -} -``` - -预期 repeated group 本身是 struct element，解析为 `ARRAY>>`，不要把 `items` 提升成 list element。 - -### LIST 反例 - -1. outer LIST group 多 child。 -2. outer LIST child 非 repeated。 -3. repeated group 无 child。 -4. repeated LIST-annotated outer group，除非它作为 another two-level LIST 的 element 被专门支持。 - -### MAP 正例 - -1. 标准 `key_value` entry group。 -2. `entries` entry group name。 -3. entry group 任意名字，但结构为 repeated group with required key and value。 -4. `MAP_KEY_VALUE` legacy converted type。 -5. key/value 字段名非 `key`/`value`，但位置正确。 - -### MAP 反例 - -1. nullable key。 -2. outer MAP group 多 child。 -3. entry child 非 repeated。 -4. entry child 是 primitive。 -5. key-only map。 -6. no-entry MAP。 - -## 实施步骤 - -1. 在 `parquet_column_schema.cpp` 增加 LIST helper： - - `has_structural_list_name()` - - `resolve_list_element_node()` - - 必要时增加 repeated group as element 的 build helper。 -2. 改造 LIST 分支，输出统一 `ParquetColumnSchemaKind::LIST` schema tree。 -3. 增加 LIST schema/unit/regression 测试。 - - 覆盖 repeated primitive、multi-field struct element、`array` / `_tuple` structural name。 - - 覆盖 two-level `List>`、two-level `List>`、单 child repeated group 且 child repeated 的 struct element。 - - read 测试至少覆盖 null list、empty list、单元素、多元素，验证 def/rep materialization。 -4. 增加 MAP helper： - - `resolve_map_entry_group()` -5. 改造 MAP 分支，放宽 entry group 名字限制，但保持 key/value 结构严格，并在 schema build 阶段折叠 entry wrapper，输出 `MAP -> key,value`。 -6. 增加 MAP schema/unit/regression 测试。 - - 覆盖 entry group 名字兼容。 - - 覆盖 `ParquetColumnSchema(MAP).children == [key, value]`。 - - 覆盖 partial MAP projection 只允许 value child，key child projection reject。 -7. 如后续确有需求，再单独设计 key-only map 或 key subtree projection 支持。 - -## 预期收益 - -- 支持更多由 Arrow、Spark、Hive、旧 Parquet writer 产生的 LIST/MAP schema。 -- 兼容逻辑集中在 schema builder，reader 层保持稳定。 -- 为后续 complex parquet reader 的兼容性测试建立清晰边界。