xylaaaaa commented on code in PR #63862:
URL: https://github.com/apache/doris/pull/63862#discussion_r3487963720
##########
be/src/format_v2/table/iceberg_reader.cpp:
##########
@@ -448,7 +449,10 @@ Status
IcebergTableReader::_append_equality_delete_predicates(format::FileScanRe
if (field_it->type->equals(*filter.key_types[idx])) {
delete_predicate->add_child(std::move(slot));
} else {
- auto cast_expr = Cast::create_shared(filter.key_types[idx]);
+ const auto cast_type = field_it->type->is_nullable()
Review Comment:
这里保留是为了 equality delete 的正确性：delete key 可能不在查询投影里，但 delete_conjuncts 需要在
file-local block 上先执行删除过滤，所以必须把 equality delete key 加进 predicate_columns，让 file
reader 能在 finalize 前过滤掉被删行。
##########
be/src/format_v2/new_orc/orc_reader.h:
##########
@@ -0,0 +1,270 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#pragma once
+
+#include <memory>
+#include <orc/Reader.hh>
+#include <set>
+#include <string>
+#include <vector>
+
+#include "common/status.h"
+#include "core/column/column.h"
+#include "format_v2/file_reader.h"
+#include "runtime/runtime_profile.h"
+
+namespace doris::new_orc {
+
+struct OrcReaderScanState;
+
+struct OrcScanRequest : public format::FileScanRequest {};
+
+//
=============================================================================
+// new_orc::OrcReader —— 新框架 ORC FileReader
+//
=============================================================================
+//
+// 架构层次：
+// TableReader (上层：表语义、schema evolution、partition、delete)
+// ↓
+// format::FileReader 接口
+// ↓
+// new_orc::OrcReader (本类：把 ORC C++ 库包成 file-local reader)
+// ↓
+// ORC C++ 库 (Reader / RowReader / SearchArgument)
+//
+// 设计原则（详见 file_reader.h 注释）：
+// - 文件层只看 file-local concept，不假设表语义
+// - 输出列永远 nullable（让 table 层判断 NOT NULL 约束）
+// - 用 LocalColumnId 而非 SQL slot id；Block 位置通过 local_positions 映射
+//
+// 生命周期（6 个 override 方法）：
+// new OrcReader → init → get_schema → open(request)
+// → get_block 循环（或 get_aggregate_result，二选一）
+// → close
+//
+// lazy materialization：
+// ORC lazy : ORC 库内部 LEADERS/FOLLOWERS 模式 + filter callback
+// 适用于无嵌套投影、filter 不引用虚拟列、有 filter 且 filter 引用列已投影的场景
+// 收益：ORC 库内部 column reader decode 跳行；
+// get_block 只对 ORC callback 选中的行 materialize non-predicate
列
+// 普通路径 : 先 decode 完请求列，再由 Doris row-level filter 裁 block
+// 当前不在 reader 层单独做 predicate-first lazy materialization
+//
+// 4 类 filter 串行 AND（_build_keep_filter）：
+// 1. column_predicate_filters : 单列 ColumnPredicate（zonemap-friendly，可转
SARG）
+// 2. reader_expression_map : column_mapper 改写不了的 file-local 表达式
+// 3. conjuncts : 标准 SQL VExpr 表达式（主通道）
+// 4. delete_conjuncts : Iceberg/Hudi delete 标记（不转 SARG）
+//
+// 三层 pruning（粒度从粗到细）：
+// 1. Stripe-level : SARG (ORC 库内) + Doris zonemap (本 reader) 双闸取交集
+// 2. Row group-level: ORC 库内部用 SARG + row group statistics（不可见）
+// 3. Row-level : 4 类 filter 行级精确（_build_keep_filter）
+//
+// 虚拟列（OrcReader 自己生成的列）：
+// __orc_row_position : 当前行在文件里的物理行号（BIGINT）
+// 用于 Iceberg position delete
+// column_id = -10002（负数避免和真实列冲突）
+class OrcReader final : public format::FileReader {
+public:
+ static constexpr int ROW_POSITION_COLUMN_ID = -10002;
+ static constexpr const char* ROW_POSITION_COLUMN_NAME =
"__orc_row_position";
+
+ OrcReader(std::shared_ptr<io::FileSystemProperties>& system_properties,
+ std::unique_ptr<io::FileDescription>& file_description,
+ std::shared_ptr<io::IOContext> io_ctx, RuntimeProfile* profile);
+ ~OrcReader() override;
+
+ static format::ColumnDefinition row_position_column_definition();
+
+ // 6 个生命周期方法（实现 format::FileReader 接口）
+ Status init(RuntimeState* state) override; // 打开 ORC 文件，建 ORC Reader 对象
+ Status get_schema(std::vector<format::ColumnDefinition>* const
file_schema) const override;
+ Status open(std::shared_ptr<format::FileScanRequest> request) override; //
配置投影/SARG/pruning
+ Status get_block(Block* file_block, size_t* rows, bool* eof) override; //
读一批
+ Status get_aggregate_result(const format::FileAggregateRequest& request,
+ format::FileAggregateResult* result) override;
+ Status close() override;
+
+private:
+ // RuntimeProfile counters (跟 Parquet 命名对齐)
+ // _init_profile 在 init() 时由基类调用注册；_collect_profile 在 close() 时统一推数据
+ struct OrcProfile {
+ RuntimeProfile::Counter* filtered_row_groups = nullptr; //
RowGroupsFiltered
+ RuntimeProfile::Counter* filtered_row_groups_by_min_max = nullptr;
+ RuntimeProfile::Counter* filtered_group_rows = nullptr; //
FilteredRowsByGroup
+ RuntimeProfile::Counter* filtered_bytes = nullptr;
+ RuntimeProfile::Counter* read_row_groups = nullptr; // RowGroupsReadNum
+ RuntimeProfile::Counter* lazy_read_filtered_rows = nullptr;
+ RuntimeProfile::Counter* orc_lazy_read_filtered_rows = nullptr;
+ RuntimeProfile::Counter* open_file_num = nullptr;
+ };
+
+ // ORC 库 filter callback 的实现类（cpp 中定义）。
+ // ORC 库要的是 ::orc::ORCFilter 接口，本类用 PIMPL 风格转发到
OrcReader::_filter_orc_batch。
+ class OrcFilterImpl;
+
+ void _init_profile() override;
+ void _collect_profile() const;
+
+ // ============ Schema 转换：ORC type → Doris ColumnDefinition ============
+ // _convert_to_doris_type 单个 ORC type 转 Doris DataType（永远 nullable）
+ // _fill_schema_field 递归填一个 ColumnDefinition（含 Iceberg.id 透传）
+ // _fill_*_schema_children STRUCT / LIST / MAP 的 children 递归
+ DataTypePtr _convert_to_doris_type(const ::orc::Type& type) const;
+ DataTypePtr _convert_list_to_doris_type(const ::orc::Type& type) const;
+ DataTypePtr _convert_map_to_doris_type(const ::orc::Type& type) const;
+ DataTypePtr _convert_struct_to_doris_type(const ::orc::Type& type) const;
+ Status _fill_schema_field(const ::orc::Type& type, int32_t field_id,
+ const std::string& field_name,
+ format::ColumnDefinition* const field) const;
+ Status _fill_struct_schema_children(const ::orc::Type& type,
+ format::ColumnDefinition* const field)
const;
+ Status _fill_list_schema_children(const ::orc::Type& type,
+ format::ColumnDefinition* const field)
const;
+ Status _fill_map_schema_children(const ::orc::Type& type,
+ format::ColumnDefinition* const field)
const;
+
+ // ============ open() 阶段配置 ============
+ // _configure_row_reader_projection 简单 include vs 复杂 includeTypes
+ // _can_enable_orc_lazy_read 决定是否启用 ORC 库 lazy callback
+ // _init_search_argument_from_conjuncts 4 类 filter → ORC SARG
+ // _select_stripe_ranges_by_statistics 双闸 stripe pruning（SARG + Doris
zonemap）
+ // _apply_current_stripe_range / _advance_to_next_stripe_range
+ // stripe pruning 后剩下的可能是不连续区间，要分段读
+ // _create_row_reader 实际创建 ORC RowReader（含 lazy callback 注入）
+ Status _configure_row_reader_projection();
+ bool _can_enable_orc_lazy_read() const;
Review Comment:
复杂列 lazy 已经用 filterTypes() 支持。这里保留的是 correctness guard，不是外部开关：只有当所有
row-level filters 都能在 ORC callback 已解码的 predicate columns 上执行时才启用 ORC lazy；例如引用
global rowid 或尚未 decode 的 non-predicate column 时必须回退普通路径，避免 callback 阶段误过滤。
##########
be/src/format_v2/new_orc/orc_reader.cpp:
##########
@@ -0,0 +1,4393 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "format_v2/new_orc/orc_reader.h"
+
+#include <cctz/time_zone.h>
+#include <gen_cpp/Types_types.h>
+
+#include <algorithm>
+#include <charconv>
+#include <cmath>
+#include <cstdint>
+#include <cstring>
+#include <limits>
+#include <list>
+#include <map>
+#include <memory>
+#include <optional>
+#include <orc/OrcFile.hh>
+#include <orc/Vector.hh>
+#include <orc/sargs/Literal.hh>
+#include <orc/sargs/SearchArgument.hh>
+#include <set>
+#include <string>
+#include <system_error>
+#include <utility>
+#include <vector>
+
+#include "common/cast_set.h"
+#include "common/consts.h"
+#include "common/exception.h"
+#include "core/block/block.h"
+#include "core/column/column_array.h"
+#include "core/column/column_decimal.h"
+#include "core/column/column_map.h"
+#include "core/column/column_nullable.h"
+#include "core/column/column_string.h"
+#include "core/column/column_struct.h"
+#include "core/column/column_vector.h"
+#include "core/column/predicate_column.h"
+#include "core/data_type/data_type_array.h"
+#include "core/data_type/data_type_date_or_datetime_v2.h"
+#include "core/data_type/data_type_date_time.h"
+#include "core/data_type/data_type_decimal.h"
+#include "core/data_type/data_type_map.h"
+#include "core/data_type/data_type_nullable.h"
+#include "core/data_type/data_type_number.h"
+#include "core/data_type/data_type_string.h"
+#include "core/data_type/data_type_struct.h"
+#include "core/types.h"
+#include "core/value/vdatetime_value.h"
+#include "exprs/vexpr_context.h"
+#include "exprs/vliteral.h"
+#include "exprs/vslot_ref.h"
+#include "io/fs/file_reader.h"
+#include "runtime/exec_env.h"
+#include "runtime/runtime_profile.h"
+#include "storage/index/zone_map/zone_map_index.h"
+#include "storage/predicate/column_predicate.h"
+#include "storage/predicate/predicate_literal_provider.h"
+#include "util/slice.h"
+
+namespace doris::new_orc {
+namespace {
+
+constexpr uint64_t DEFAULT_ORC_READ_BATCH_SIZE = 4096;
+constexpr uint64_t DEFAULT_ORC_NATURAL_READ_SIZE = 128 * 1024;
+constexpr int DECIMAL_PRECISION_FOR_HIVE11 =
BeConsts::MAX_DECIMAL128_PRECISION;
+constexpr int DECIMAL_SCALE_FOR_HIVE11 = 10;
+constexpr const char* ORC_LIST_ELEMENT_NAME = "element";
+constexpr const char* ORC_MAP_ENTRY_NAME = "key_value";
+constexpr const char* ORC_MAP_KEY_NAME = "key";
+constexpr const char* ORC_MAP_VALUE_NAME = "value";
+constexpr const char* ORC_ICEBERG_ID_ATTRIBUTE = "iceberg.id";
+
+// Thin adapter from Doris FileReader to ORC's InputStream API. Keep IO policy,
+// tracing, and retry behavior in the underlying FileReader.
+class DorisOrcInputStream final : public ::orc::InputStream {
+public:
+ DorisOrcInputStream(std::string file_name, io::FileReaderSPtr file_reader,
+ io::IOContext* io_ctx)
+ : _file_name(std::move(file_name)),
+ _file_reader(std::move(file_reader)),
+ _io_ctx(io_ctx) {}
+
+ uint64_t getLength() const override { return _file_reader->size(); }
+
+ uint64_t getNaturalReadSize() const override { return
DEFAULT_ORC_NATURAL_READ_SIZE; }
+
+ void read(void* buf, uint64_t length, uint64_t offset) override {
+ uint64_t bytes_read = 0;
+ auto* out = static_cast<uint8_t*>(buf);
+ while (bytes_read < length) {
+ size_t loop_read = 0;
+ Status st = _file_reader->read_at(
+ static_cast<size_t>(offset + bytes_read),
+ Slice(out + bytes_read, static_cast<size_t>(length -
bytes_read)), &loop_read,
+ _io_ctx);
+ if (!st.ok()) {
+ throw ::orc::ParseError("Failed to read " + _file_name + ": " +
+ st.to_string_no_stack());
+ }
+ if (loop_read == 0) {
+ break;
+ }
+ bytes_read += loop_read;
+ }
+ if (bytes_read != length) {
+ throw ::orc::ParseError("Short read from " + _file_name);
+ }
+ }
+
+ const std::string& getName() const override { return _file_name; }
+
+private:
+ std::string _file_name;
+ io::FileReaderSPtr _file_reader;
+ io::IOContext* _io_ctx = nullptr;
+};
+
+// selected_rows is a source-row remap produced by ORC lazy callback:
+// predicate columns are decoded first, then surviving row ids drive follower
decodes.
+bool is_null_at(const ::orc::ColumnVectorBatch& batch, size_t row) {
+ return batch.hasNulls && !batch.notNull[row];
+}
+
+size_t decode_row_count(size_t rows, const std::vector<size_t>* selected_rows)
{
+ if (selected_rows == nullptr) {
+ return rows;
+ }
+ return selected_rows->size();
+}
+
+size_t source_row_at(size_t row, const std::vector<size_t>* selected_rows) {
+ if (selected_rows == nullptr) {
+ return row;
+ }
+ return (*selected_rows)[row];
+}
+
+// ColumnPredicate evaluates over PredicateColumn implementations. These
helpers
+// adapt decoded Doris columns into that shape for row-level predicate
fallback.
+template <PrimitiveType Primitive>
+Status build_predicate_nested_column(const IColumn& src_column, size_t rows,
+ MutableColumnPtr* const predicate_column)
{
+ DORIS_CHECK(predicate_column != nullptr);
+ using EvaluatePrimitive =
+ std::integral_constant<PrimitiveType,
PredicateEvaluateType<Primitive>>;
+ using PredicateColumn = PredicateColumnType<EvaluatePrimitive::value>;
+ using SourceColumn = typename
PrimitiveTypeTraits<EvaluatePrimitive::value>::ColumnType;
+
+ const auto* source_column =
check_and_get_column<SourceColumn>(&src_column);
+ if (source_column == nullptr) {
+ return Status::NotSupported("ORC column predicate does not support
column {} as {}",
+ src_column.get_name(),
+ type_to_string(EvaluatePrimitive::value));
+ }
+
+ auto result = PredicateColumn::create();
+ result->reserve(rows);
+ for (size_t row = 0; row < rows; ++row) {
+ result->insert_data(source_column->get_data_at(row).data,
+ source_column->get_data_at(row).size);
+ }
+ *predicate_column = std::move(result);
+ return Status::OK();
+}
+
+template <PrimitiveType Primitive>
+Status build_fixed_length_predicate_nested_column(const IColumn& src_column,
size_t rows,
+ MutableColumnPtr* const
predicate_column) {
+ DORIS_CHECK(predicate_column != nullptr);
+ using EvaluatePrimitive =
+ std::integral_constant<PrimitiveType,
PredicateEvaluateType<Primitive>>;
+ using PredicateColumn = PredicateColumnType<EvaluatePrimitive::value>;
+ using SourceColumn = typename
PrimitiveTypeTraits<EvaluatePrimitive::value>::ColumnType;
+
+ const auto* source_column =
check_and_get_column<SourceColumn>(&src_column);
+ if (source_column == nullptr) {
+ return Status::NotSupported("ORC column predicate does not support
column {} as {}",
+ src_column.get_name(),
+ type_to_string(EvaluatePrimitive::value));
+ }
+ DORIS_CHECK(rows <= source_column->size());
+
+ auto result = PredicateColumn::create();
+ result->reserve(rows);
+ result->insert_many_fix_len_data(source_column->get_raw_data().data, rows);
+ *predicate_column = std::move(result);
+ return Status::OK();
+}
+
+Status build_predicate_column(const IColumn& src_column, const IDataType&
src_type, size_t rows,
+ MutableColumnPtr* const predicate_column) {
+ DORIS_CHECK(predicate_column != nullptr);
+ const auto src_nested_column =
+ src_column.is_nullable()
+ ? assert_cast<const
ColumnNullable&>(src_column).get_nested_column_ptr()
+ : src_column.get_ptr();
+ DORIS_CHECK(src_nested_column.get() != nullptr);
+
+ MutableColumnPtr predicate_nested_column;
+ switch (src_type.get_primitive_type()) {
+ case TYPE_BOOLEAN:
+
RETURN_IF_ERROR(build_fixed_length_predicate_nested_column<TYPE_BOOLEAN>(
+ *src_nested_column, rows, &predicate_nested_column));
+ break;
+ case TYPE_TINYINT:
+
RETURN_IF_ERROR(build_fixed_length_predicate_nested_column<TYPE_TINYINT>(
+ *src_nested_column, rows, &predicate_nested_column));
+ break;
+ case TYPE_SMALLINT:
+
RETURN_IF_ERROR(build_fixed_length_predicate_nested_column<TYPE_SMALLINT>(
+ *src_nested_column, rows, &predicate_nested_column));
+ break;
+ case TYPE_INT:
+ RETURN_IF_ERROR(build_fixed_length_predicate_nested_column<TYPE_INT>(
+ *src_nested_column, rows, &predicate_nested_column));
+ break;
+ case TYPE_BIGINT:
+
RETURN_IF_ERROR(build_fixed_length_predicate_nested_column<TYPE_BIGINT>(
+ *src_nested_column, rows, &predicate_nested_column));
+ break;
+ case TYPE_LARGEINT:
+
RETURN_IF_ERROR(build_fixed_length_predicate_nested_column<TYPE_LARGEINT>(
+ *src_nested_column, rows, &predicate_nested_column));
+ break;
+ case TYPE_FLOAT:
+ RETURN_IF_ERROR(build_fixed_length_predicate_nested_column<TYPE_FLOAT>(
+ *src_nested_column, rows, &predicate_nested_column));
+ break;
+ case TYPE_DOUBLE:
+
RETURN_IF_ERROR(build_fixed_length_predicate_nested_column<TYPE_DOUBLE>(
+ *src_nested_column, rows, &predicate_nested_column));
+ break;
+ case TYPE_DATE:
+
RETURN_IF_ERROR(build_predicate_nested_column<TYPE_DATE>(*src_nested_column,
rows,
+
&predicate_nested_column));
+ break;
+ case TYPE_DATETIME:
+
RETURN_IF_ERROR(build_predicate_nested_column<TYPE_DATETIME>(*src_nested_column,
rows,
+
&predicate_nested_column));
+ break;
+ case TYPE_DATEV2:
+
RETURN_IF_ERROR(build_predicate_nested_column<TYPE_DATEV2>(*src_nested_column,
rows,
+
&predicate_nested_column));
+ break;
+ case TYPE_DATETIMEV2:
+
RETURN_IF_ERROR(build_fixed_length_predicate_nested_column<TYPE_DATETIMEV2>(
+ *src_nested_column, rows, &predicate_nested_column));
+ break;
+ case TYPE_DECIMALV2:
+
RETURN_IF_ERROR(build_predicate_nested_column<TYPE_DECIMALV2>(*src_nested_column,
rows,
+
&predicate_nested_column));
+ break;
+ case TYPE_DECIMAL32:
+
RETURN_IF_ERROR(build_fixed_length_predicate_nested_column<TYPE_DECIMAL32>(
+ *src_nested_column, rows, &predicate_nested_column));
+ break;
+ case TYPE_DECIMAL64:
+
RETURN_IF_ERROR(build_fixed_length_predicate_nested_column<TYPE_DECIMAL64>(
+ *src_nested_column, rows, &predicate_nested_column));
+ break;
+ case TYPE_DECIMAL128I:
+
RETURN_IF_ERROR(build_fixed_length_predicate_nested_column<TYPE_DECIMAL128I>(
+ *src_nested_column, rows, &predicate_nested_column));
+ break;
+ case TYPE_DECIMAL256:
+
RETURN_IF_ERROR(build_fixed_length_predicate_nested_column<TYPE_DECIMAL256>(
+ *src_nested_column, rows, &predicate_nested_column));
+ break;
+ case TYPE_CHAR:
+
RETURN_IF_ERROR(build_predicate_nested_column<TYPE_CHAR>(*src_nested_column,
rows,
+
&predicate_nested_column));
+ break;
+ case TYPE_VARCHAR:
+
RETURN_IF_ERROR(build_predicate_nested_column<TYPE_VARCHAR>(*src_nested_column,
rows,
+
&predicate_nested_column));
+ break;
+ case TYPE_STRING:
+
RETURN_IF_ERROR(build_predicate_nested_column<TYPE_STRING>(*src_nested_column,
rows,
+
&predicate_nested_column));
+ break;
+ case TYPE_IPV4:
+ RETURN_IF_ERROR(build_fixed_length_predicate_nested_column<TYPE_IPV4>(
+ *src_nested_column, rows, &predicate_nested_column));
+ break;
+ case TYPE_IPV6:
+ RETURN_IF_ERROR(build_fixed_length_predicate_nested_column<TYPE_IPV6>(
+ *src_nested_column, rows, &predicate_nested_column));
+ break;
+ default:
+ return Status::NotSupported("ORC column predicate does not support
column type {}",
+ src_type.get_name());
+ }
+
+ if (!src_column.is_nullable()) {
+ *predicate_column = std::move(predicate_nested_column);
+ return Status::OK();
+ }
+
+ const auto& nullable_column = assert_cast<const
ColumnNullable&>(src_column);
+ auto null_map = nullable_column.get_null_map_column().clone_resized(rows);
+ *predicate_column =
+ ColumnNullable::create(std::move(predicate_nested_column),
std::move(null_map));
+ return Status::OK();
+}
+
+// Scalar decoders append into already-created mutable Doris columns. Null
values
+// write default payloads here; the nullable wrapper is filled by
_decode_column().
+template <typename ColumnType, typename ValueType, typename OrcBatchType>
+Status append_numeric_values(MutableColumnPtr& column, const
::orc::ColumnVectorBatch& batch,
+ size_t rows, const std::vector<size_t>*
selected_rows) {
+ const auto* orc_batch = dynamic_cast<const OrcBatchType*>(&batch);
+ if (orc_batch == nullptr) {
+ return Status::InternalError("Unexpected ORC numeric batch type {}",
batch.toString());
+ }
+ auto& data = assert_cast<ColumnType&>(*column).get_data();
+ const size_t old_size = data.size();
+ const auto output_rows = decode_row_count(rows, selected_rows);
+ data.resize(old_size + output_rows);
+ for (size_t row = 0; row < output_rows; ++row) {
+ const auto source_row = source_row_at(row, selected_rows);
+ if (is_null_at(batch, source_row)) {
+ data[old_size + row] = ValueType {};
+ continue;
+ }
+ data[old_size + row] =
static_cast<ValueType>(orc_batch->data[source_row]);
+ }
+ return Status::OK();
+}
+
+template <typename ColumnType, typename ValueType>
+Status append_floating_values(MutableColumnPtr& column, const
::orc::ColumnVectorBatch& batch,
+ size_t rows, const std::vector<size_t>*
selected_rows) {
+ const auto* orc_batch = dynamic_cast<const
::orc::DoubleVectorBatch*>(&batch);
+ if (orc_batch == nullptr) {
+ return Status::InternalError("Unexpected ORC floating batch type {}",
batch.toString());
+ }
+ auto& data = assert_cast<ColumnType&>(*column).get_data();
+ const size_t old_size = data.size();
+ const auto output_rows = decode_row_count(rows, selected_rows);
+ data.resize(old_size + output_rows);
+ for (size_t row = 0; row < output_rows; ++row) {
+ const auto source_row = source_row_at(row, selected_rows);
+ if (is_null_at(batch, source_row)) {
+ data[old_size + row] = ValueType {};
+ continue;
+ }
+ data[old_size + row] =
static_cast<ValueType>(orc_batch->data[source_row]);
+ }
+ return Status::OK();
+}
+
+size_t trim_right_spaces(const char* value, size_t length) {
+ while (length > 0 && value[length - 1] == ' ') {
+ --length;
+ }
+ return length;
+}
+
+Int128 to_int128(::orc::Int128 value) {
+ const auto high_bits =
static_cast<__uint128_t>(static_cast<uint64_t>(value.getHighBits()));
+ const auto low_bits = static_cast<__uint128_t>(value.getLowBits());
+ return static_cast<Int128>((high_bits << 64) | low_bits);
+}
+
+// ORC nested projection is type-id based. These helpers translate Doris'
+// LocalColumnIndex tree into the ORC type ids expected by includeTypes().
+Status get_projection_child_index(const format::LocalColumnIndex& child,
int32_t child_count,
+ const std::string& column_name, int32_t*
child_idx) {
+ DORIS_CHECK(child_idx != nullptr);
+ *child_idx = child.local_id();
+ if (*child_idx < 0 || *child_idx >= child_count) {
+ return Status::InvalidArgument("Invalid ORC projection child index {}
for column {}",
+ *child_idx, column_name);
+ }
+ return Status::OK();
+}
+
+void collect_type_and_descendant_ids(const ::orc::Type& type,
std::set<uint64_t>* const type_ids) {
+ DORIS_CHECK(type_ids != nullptr);
+ type_ids->insert(type.getColumnId());
+ for (uint64_t child_idx = 0; child_idx < type.getSubtypeCount();
++child_idx) {
+ const auto* child_type = type.getSubtype(child_idx);
+ DORIS_CHECK(child_type != nullptr);
+ collect_type_and_descendant_ids(*child_type, type_ids);
+ }
+}
+
+Status collect_projected_type_ids(const ::orc::Type& type,
+ const format::LocalColumnIndex& projection,
+ std::set<uint64_t>* const type_ids);
+
+Status collect_projected_map_type_ids(const ::orc::Type& type,
+ const format::LocalColumnIndex&
projection,
+ std::set<uint64_t>* const type_ids);
+
+Status collect_projected_map_entry_type_ids(const ::orc::Type& type,
+ const format::LocalColumnIndex&
entry_projection,
+ std::set<uint64_t>* const type_ids,
+ bool* const selected_key, bool*
const selected_value) {
+ DORIS_CHECK(type.getKind() == ::orc::TypeKind::MAP);
+ DORIS_CHECK(type.getSubtypeCount() == 2);
+ DORIS_CHECK(selected_key != nullptr);
+ DORIS_CHECK(selected_value != nullptr);
+
+ int32_t entry_idx = 0;
+ RETURN_IF_ERROR(
+ get_projection_child_index(entry_projection, 1,
ORC_MAP_ENTRY_NAME, &entry_idx));
+ DORIS_CHECK(entry_idx == 0);
+ if (entry_projection.project_all_children) {
+ collect_type_and_descendant_ids(*type.getSubtype(0), type_ids);
+ collect_type_and_descendant_ids(*type.getSubtype(1), type_ids);
+ *selected_key = true;
+ *selected_value = true;
+ return Status::OK();
+ }
+ if (entry_projection.children.empty()) {
+ return Status::NotSupported("ORC MAP entry projection contains no
children");
+ }
+ for (const auto& key_value_projection : entry_projection.children) {
+ int32_t key_value_idx = 0;
+ RETURN_IF_ERROR(get_projection_child_index(key_value_projection, 2,
ORC_MAP_ENTRY_NAME,
+ &key_value_idx));
+ const auto* child_type =
type.getSubtype(static_cast<uint64_t>(key_value_idx));
+ DORIS_CHECK(child_type != nullptr);
+ RETURN_IF_ERROR(collect_projected_type_ids(*child_type,
key_value_projection, type_ids));
+ *selected_key = *selected_key || key_value_idx == 0;
+ *selected_value = *selected_value || key_value_idx == 1;
+ }
+ return Status::OK();
+}
+
+Status collect_projected_map_type_ids(const ::orc::Type& type,
+ const format::LocalColumnIndex&
projection,
+ std::set<uint64_t>* const type_ids) {
+ DORIS_CHECK(type.getKind() == ::orc::TypeKind::MAP);
+ DORIS_CHECK(type.getSubtypeCount() == 2);
+ type_ids->insert(type.getColumnId());
+ if (projection.project_all_children) {
+ collect_type_and_descendant_ids(type, type_ids);
+ return Status::OK();
+ }
+ if (projection.children.empty()) {
+ return Status::NotSupported("ORC MAP projection for column {} contains
no children",
+ projection.local_id());
+ }
+
+ bool selected_key = false;
+ bool selected_value = false;
+ for (const auto& entry_projection : projection.children) {
+ RETURN_IF_ERROR(collect_projected_map_entry_type_ids(type,
entry_projection, type_ids,
+ &selected_key,
&selected_value));
+ }
+ if (!selected_key || !selected_value) {
+ return Status::NotSupported("ORC MAP projection must include both key
and value");
+ }
+ return Status::OK();
+}
+
+Status collect_projected_type_ids(const ::orc::Type& type,
+ const format::LocalColumnIndex& projection,
+ std::set<uint64_t>* const type_ids) {
+ DORIS_CHECK(type_ids != nullptr);
+ type_ids->insert(type.getColumnId());
+ if (projection.project_all_children) {
+ collect_type_and_descendant_ids(type, type_ids);
+ return Status::OK();
+ }
+ if (projection.children.empty()) {
+ return Status::NotSupported("ORC projection contains no children");
+ }
+ if (type.getKind() == ::orc::TypeKind::MAP) {
+ return collect_projected_map_type_ids(type, projection, type_ids);
+ }
+ if (type.getKind() != ::orc::TypeKind::STRUCT && type.getKind() !=
::orc::TypeKind::LIST) {
+ return Status::InvalidArgument("Cannot project children from
non-complex ORC type {}",
+ static_cast<int>(type.getKind()));
+ }
+
+ const auto child_count = static_cast<int32_t>(type.getSubtypeCount());
+ for (const auto& child_projection : projection.children) {
+ int32_t child_idx = 0;
+ RETURN_IF_ERROR(get_projection_child_index(child_projection,
child_count, "orc_complex",
+ &child_idx));
+ const auto* child_type =
type.getSubtype(static_cast<uint64_t>(child_idx));
+ DORIS_CHECK(child_type != nullptr);
+ RETURN_IF_ERROR(collect_projected_type_ids(*child_type,
child_projection, type_ids));
+ }
+ return Status::OK();
+}
+
+// For arrays/maps, selected parent rows expand into selected element rows. The
+// returned offsets are compacted so downstream child decoding can append
densely.
+Status append_orc_offsets(ColumnArray::Offsets64& doris_offsets,
+ const ::orc::DataBuffer<int64_t>& orc_offsets,
size_t rows,
+ size_t* element_size, const std::vector<size_t>*
selected_rows = nullptr,
+ std::vector<size_t>* element_selection = nullptr) {
+ if (selected_rows != nullptr) {
+ DORIS_CHECK(element_selection != nullptr);
+ const auto prev_offset = doris_offsets.empty() ? 0 :
doris_offsets.back();
+ ColumnArray::Offset64 current_offset = prev_offset;
+ element_selection->clear();
+ for (size_t row = 0; row < selected_rows->size(); ++row) {
+ const auto source_row = (*selected_rows)[row];
+ DORIS_CHECK(source_row < rows);
+ const auto begin_offset = orc_offsets[source_row];
+ const auto end_offset = orc_offsets[source_row + 1];
+ if (end_offset < begin_offset) {
+ return Status::Corruption("Invalid ORC offsets");
+ }
+ const auto delta = static_cast<size_t>(end_offset - begin_offset);
+ for (size_t element_idx = 0; element_idx < delta; ++element_idx) {
+ element_selection->push_back(static_cast<size_t>(begin_offset)
+ element_idx);
+ }
+ current_offset += static_cast<ColumnArray::Offset64>(delta);
+ doris_offsets.push_back(current_offset);
+ }
+ *element_size = element_selection->size();
+ return Status::OK();
+ }
+
+ const auto prev_offset = doris_offsets.empty() ? 0 : doris_offsets.back();
+ const auto base_offset = orc_offsets[0];
+ for (size_t idx = 1; idx <= rows; ++idx) {
+ const auto delta = orc_offsets[idx] - base_offset;
+ if (delta < 0) {
+ return Status::Corruption("Invalid ORC offsets");
+ }
+ doris_offsets.push_back(prev_offset +
static_cast<ColumnArray::Offset64>(delta));
+ }
+ const auto total_delta = orc_offsets[rows] - base_offset;
+ if (total_delta < 0) {
+ return Status::Corruption("Invalid ORC offsets");
+ }
+ *element_size = static_cast<size_t>(total_delta);
+ return Status::OK();
+}
+
+int64_t find_struct_child_index(const ::orc::Type& type, const std::string&
field_name) {
+ DORIS_CHECK(type.getKind() == ::orc::TypeKind::STRUCT);
+ for (uint64_t child_idx = 0; child_idx < type.getSubtypeCount();
++child_idx) {
+ if (type.getFieldName(child_idx) == field_name) {
+ return static_cast<int64_t>(child_idx);
+ }
+ }
+ return -1;
+}
+
+bool is_row_position_column(format::LocalColumnId file_column_id) {
+ return file_column_id ==
format::LocalColumnId(OrcReader::ROW_POSITION_COLUMN_ID);
+}
+
+const format::LocalColumnIndex* find_projection(
+ const std::vector<format::LocalColumnIndex>& projections,
+ format::LocalColumnId file_column_id) {
+ const auto it = std::find_if(projections.begin(), projections.end(),
+ [&](const format::LocalColumnIndex&
projection) {
+ return projection.column_id() ==
file_column_id;
+ });
+ return it == projections.end() ? nullptr : &*it;
+}
+
+const format::LocalColumnIndex* find_request_projection(const
format::FileScanRequest& request,
+ format::LocalColumnId
file_column_id) {
+ if (const auto* projection = find_projection(request.predicate_columns,
file_column_id);
+ projection != nullptr) {
+ return projection;
+ }
+ return find_projection(request.non_predicate_columns, file_column_id);
+}
+
+bool has_pruned_projection(const format::LocalColumnIndex& projection) {
+ return !projection.project_all_children;
+}
+
+// Stripe pruning maps ORC stripe statistics into Doris ZoneMap semantics.
Missing
+// or unsupported statistics are treated conservatively and never prune.
+bool set_integer_zone_map(const ::orc::Type& type, const
::orc::ColumnStatistics& statistics,
+ segment_v2::ZoneMap* zone_map) {
+ const auto* integer_statistics =
+ dynamic_cast<const ::orc::IntegerColumnStatistics*>(&statistics);
+ if (integer_statistics == nullptr || !integer_statistics->hasMinimum() ||
+ !integer_statistics->hasMaximum()) {
+ return false;
+ }
+ switch (type.getKind()) {
+ case ::orc::TypeKind::BYTE:
+ zone_map->min_value =
+
Field::create_field<TYPE_TINYINT>(cast_set<Int8>(integer_statistics->getMinimum()));
+ zone_map->max_value =
+
Field::create_field<TYPE_TINYINT>(cast_set<Int8>(integer_statistics->getMaximum()));
+ return true;
+ case ::orc::TypeKind::SHORT:
+ zone_map->min_value = Field::create_field<TYPE_SMALLINT>(
+ cast_set<Int16>(integer_statistics->getMinimum()));
+ zone_map->max_value = Field::create_field<TYPE_SMALLINT>(
+ cast_set<Int16>(integer_statistics->getMaximum()));
+ return true;
+ case ::orc::TypeKind::INT:
+ zone_map->min_value =
+
Field::create_field<TYPE_INT>(cast_set<Int32>(integer_statistics->getMinimum()));
+ zone_map->max_value =
+
Field::create_field<TYPE_INT>(cast_set<Int32>(integer_statistics->getMaximum()));
+ return true;
+ case ::orc::TypeKind::LONG:
+ zone_map->min_value =
Field::create_field<TYPE_BIGINT>(integer_statistics->getMinimum());
+ zone_map->max_value =
Field::create_field<TYPE_BIGINT>(integer_statistics->getMaximum());
+ return true;
+ default:
+ return false;
+ }
+}
+
+bool set_boolean_zone_map(const ::orc::ColumnStatistics& statistics,
+ segment_v2::ZoneMap* zone_map) {
+ const auto* boolean_statistics =
+ dynamic_cast<const ::orc::BooleanColumnStatistics*>(&statistics);
+ if (boolean_statistics == nullptr || !boolean_statistics->hasCount()) {
+ return false;
+ }
+ const bool has_false = boolean_statistics->getFalseCount() > 0;
+ const bool has_true = boolean_statistics->getTrueCount() > 0;
+ if (!has_false && !has_true) {
+ return false;
+ }
+ zone_map->min_value =
Field::create_field<TYPE_BOOLEAN>(static_cast<UInt8>(has_false ? 0 : 1));
+ zone_map->max_value =
Field::create_field<TYPE_BOOLEAN>(static_cast<UInt8>(has_true ? 1 : 0));
+ return true;
+}
+
+bool set_floating_zone_map(const ::orc::Type& type, const
::orc::ColumnStatistics& statistics,
+ segment_v2::ZoneMap* zone_map) {
+ const auto* double_statistics = dynamic_cast<const
::orc::DoubleColumnStatistics*>(&statistics);
+ if (double_statistics == nullptr || !double_statistics->hasMinimum() ||
+ !double_statistics->hasMaximum()) {
+ return false;
+ }
+ if (type.getKind() == ::orc::TypeKind::FLOAT) {
+ zone_map->min_value = Field::create_field<TYPE_FLOAT>(
+ static_cast<Float32>(double_statistics->getMinimum()));
+ zone_map->max_value = Field::create_field<TYPE_FLOAT>(
+ static_cast<Float32>(double_statistics->getMaximum()));
+ return true;
+ }
+ if (type.getKind() == ::orc::TypeKind::DOUBLE) {
+ zone_map->min_value =
Field::create_field<TYPE_DOUBLE>(double_statistics->getMinimum());
+ zone_map->max_value =
Field::create_field<TYPE_DOUBLE>(double_statistics->getMaximum());
+ return true;
+ }
+ return false;
+}
+
+bool set_string_zone_map(const ::orc::ColumnStatistics& statistics,
segment_v2::ZoneMap* zone_map) {
+ const auto* string_statistics = dynamic_cast<const
::orc::StringColumnStatistics*>(&statistics);
+ if (string_statistics == nullptr || !string_statistics->hasMinimum() ||
+ !string_statistics->hasMaximum()) {
+ return false;
+ }
+ zone_map->min_value =
Field::create_field<TYPE_STRING>(string_statistics->getMinimum());
+ zone_map->max_value =
Field::create_field<TYPE_STRING>(string_statistics->getMaximum());
+ return true;
+}
+
+bool set_date_zone_map(const ::orc::ColumnStatistics& statistics,
segment_v2::ZoneMap* zone_map) {
+ const auto* date_statistics = dynamic_cast<const
::orc::DateColumnStatistics*>(&statistics);
+ if (date_statistics == nullptr || !date_statistics->hasMinimum() ||
+ !date_statistics->hasMaximum()) {
+ return false;
+ }
+ auto& date_dict = date_day_offset_dict::get();
+ zone_map->min_value =
+
Field::create_field<TYPE_DATEV2>(date_dict[date_statistics->getMinimum()]);
+ zone_map->max_value =
+
Field::create_field<TYPE_DATEV2>(date_dict[date_statistics->getMaximum()]);
+ return true;
+}
+
+DateV2Value<DateTimeV2ValueType> datetime_v2_from_orc_millis(int64_t millis,
int32_t nanos_tail) {
+ int64_t seconds = millis / 1000;
+ int64_t millis_remainder = millis % 1000;
+ if (millis_remainder < 0) {
+ --seconds;
+ millis_remainder += 1000;
+ }
+ const auto extra_nanos = std::max<int32_t>(nanos_tail, 0);
+ const auto microseconds = cast_set<uint64_t>(millis_remainder * 1000 +
extra_nanos / 1000);
+ DateV2Value<DateTimeV2ValueType> value;
+ value.from_unixtime(seconds, cctz::utc_time_zone());
+ value.set_microsecond(microseconds);
+ return value;
+}
+
+bool set_timestamp_zone_map(const ::orc::ColumnStatistics& statistics,
+ segment_v2::ZoneMap* zone_map) {
+ const auto* timestamp_statistics =
+ dynamic_cast<const ::orc::TimestampColumnStatistics*>(&statistics);
+ if (timestamp_statistics == nullptr || !timestamp_statistics->hasMinimum()
||
+ !timestamp_statistics->hasMaximum()) {
+ return false;
+ }
+ zone_map->min_value =
Field::create_field<TYPE_DATETIMEV2>(datetime_v2_from_orc_millis(
+ timestamp_statistics->getMinimum(),
timestamp_statistics->getMinimumNanos()));
+ zone_map->max_value =
Field::create_field<TYPE_DATETIMEV2>(datetime_v2_from_orc_millis(
+ timestamp_statistics->getMaximum(),
timestamp_statistics->getMaximumNanos()));
+ return true;
+}
+
+int32_t decimal_scale_for_orc_type(const ::orc::Type& type) {
+ return type.getPrecision() == 0 ? DECIMAL_SCALE_FOR_HIVE11 :
cast_set<int32_t>(type.getScale());
+}
+
+std::optional<Decimal128V3> decimal_value_at_scale(const ::orc::Decimal&
decimal,
+ int32_t target_scale) {
+ if (decimal.scale == target_scale) {
+ return Decimal128V3(to_int128(decimal.value));
+ }
+ if (decimal.scale < target_scale) {
+ bool overflow = false;
+ const auto scaled = ::orc::scaleUpInt128ByPowerOfTen(
+ decimal.value, target_scale - decimal.scale, overflow);
+ if (overflow) {
+ return std::nullopt;
+ }
+ return Decimal128V3(to_int128(scaled));
+ }
+
+ const auto scale_diff = decimal.scale - target_scale;
+ const auto scaled = ::orc::scaleDownInt128ByPowerOfTen(decimal.value,
scale_diff);
+ bool overflow = false;
+ const auto restored = ::orc::scaleUpInt128ByPowerOfTen(scaled, scale_diff,
overflow);
+ if (overflow || restored != decimal.value) {
+ return std::nullopt;
+ }
+ return Decimal128V3(to_int128(scaled));
+}
+
+bool set_decimal_zone_map(const ::orc::Type& type, const
::orc::ColumnStatistics& statistics,
+ segment_v2::ZoneMap* zone_map) {
+ const auto* decimal_statistics =
+ dynamic_cast<const ::orc::DecimalColumnStatistics*>(&statistics);
+ if (decimal_statistics == nullptr || !decimal_statistics->hasMinimum() ||
+ !decimal_statistics->hasMaximum()) {
+ return false;
+ }
+ const auto min = decimal_statistics->getMinimum();
+ const auto max = decimal_statistics->getMaximum();
+ const auto expected_scale = decimal_scale_for_orc_type(type);
+ const auto min_value = decimal_value_at_scale(min, expected_scale);
+ const auto max_value = decimal_value_at_scale(max, expected_scale);
+ if (!min_value.has_value() || !max_value.has_value()) {
+ return false;
+ }
+ zone_map->min_value = Field::create_field<TYPE_DECIMAL128I>(*min_value);
+ zone_map->max_value = Field::create_field<TYPE_DECIMAL128I>(*max_value);
+ return true;
+}
+
+bool build_zone_map_from_orc_statistics(const ::orc::Type& type,
+ const ::orc::ColumnStatistics&
statistics,
+ segment_v2::ZoneMap* zone_map) {
+ DORIS_CHECK(zone_map != nullptr);
+ zone_map->has_null = statistics.hasNull();
+ zone_map->has_not_null = statistics.getNumberOfValues() > 0;
+ if (!zone_map->has_not_null) {
+ return true;
+ }
+ switch (type.getKind()) {
+ case ::orc::TypeKind::BOOLEAN:
+ return set_boolean_zone_map(statistics, zone_map);
+ case ::orc::TypeKind::BYTE:
+ case ::orc::TypeKind::SHORT:
+ case ::orc::TypeKind::INT:
+ case ::orc::TypeKind::LONG:
+ return set_integer_zone_map(type, statistics, zone_map);
+ case ::orc::TypeKind::FLOAT:
+ case ::orc::TypeKind::DOUBLE:
+ return set_floating_zone_map(type, statistics, zone_map);
+ case ::orc::TypeKind::STRING:
+ case ::orc::TypeKind::VARCHAR:
+ case ::orc::TypeKind::CHAR:
+ return set_string_zone_map(statistics, zone_map);
+ case ::orc::TypeKind::DATE:
+ return set_date_zone_map(statistics, zone_map);
+ case ::orc::TypeKind::TIMESTAMP:
+ case ::orc::TypeKind::TIMESTAMP_INSTANT:
+ return set_timestamp_zone_map(statistics, zone_map);
+ case ::orc::TypeKind::DECIMAL:
+ return set_decimal_zone_map(type, statistics, zone_map);
+ default:
+ return false;
+ }
+}
+
+Status find_projected_minmax_leaf_in_type(const ::orc::Type& type,
+ const format::LocalColumnIndex&
projection,
+ const ::orc::Type** leaf_type) {
+ DORIS_CHECK(leaf_type != nullptr);
+ if (projection.project_all_children || projection.children.empty()) {
+ if (type.getSubtypeCount() > 0) {
+ return Status::NotSupported(
+ "ORC aggregate pushdown only supports primitive column
kind {}",
+ static_cast<int>(type.getKind()));
+ }
+ *leaf_type = &type;
+ return Status::OK();
+ }
+ if (projection.children.size() != 1) {
+ return Status::NotSupported(
+ "ORC aggregate pushdown only supports a single nested leaf
under column kind {}",
+ static_cast<int>(type.getKind()));
+ }
+ if (type.getKind() != ::orc::TypeKind::STRUCT) {
+ return Status::NotSupported(
+ "ORC aggregate pushdown only supports struct nested leaf
projection, got kind {}",
+ static_cast<int>(type.getKind()));
+ }
+ const auto& child_projection = projection.children[0];
+ if (child_projection.local_id() < 0 ||
+ child_projection.local_id() >=
static_cast<int32_t>(type.getSubtypeCount())) {
+ return Status::InvalidArgument("Invalid ORC aggregate child local id
{} for kind {}",
+ child_projection.local_id(),
+ static_cast<int>(type.getKind()));
+ }
+ const auto* child_type =
type.getSubtype(static_cast<uint64_t>(child_projection.local_id()));
+ DORIS_CHECK(child_type != nullptr);
+ return find_projected_minmax_leaf_in_type(*child_type, child_projection,
leaf_type);
+}
+
+Status find_projected_minmax_leaf(const ::orc::Type& root_type,
+ const format::LocalColumnIndex& projection,
+ const ::orc::Type** leaf_type) {
+ DORIS_CHECK(leaf_type != nullptr);
+ if (root_type.getKind() != ::orc::TypeKind::STRUCT) {
+ return Status::NotSupported("ORC aggregate pushdown requires top-level
struct schema");
+ }
+ const auto file_column_id = projection.column_id();
+ if (!file_column_id.is_valid() ||
+ file_column_id.value() >=
static_cast<int32_t>(root_type.getSubtypeCount())) {
+ return Status::InvalidArgument("Invalid ORC aggregate column id {}",
+ file_column_id.value());
+ }
+ const auto* column_type =
root_type.getSubtype(static_cast<uint64_t>(file_column_id.value()));
+ DORIS_CHECK(column_type != nullptr);
+ return find_projected_minmax_leaf_in_type(*column_type, projection,
leaf_type);
+}
+
+bool predicate_can_evaluate_orc_zone_map(const ::orc::Type& type,
+ const ColumnPredicate& predicate) {
+ if (type.getKind() != ::orc::TypeKind::DECIMAL) {
+ return true;
+ }
+ if (predicate.type() == PredicateType::IS_NULL ||
+ predicate.type() == PredicateType::IS_NOT_NULL) {
+ return true;
+ }
+ const auto* literal_provider = dynamic_cast<const
ColumnPredicateLiteralProvider*>(&predicate);
+ if (literal_provider == nullptr) {
+ return true;
+ }
+ const auto literal_type_info =
literal_provider->predicate_literal_type_info();
+ return literal_type_info.has_value() &&
+ literal_type_info->scale ==
cast_set<uint32_t>(decimal_scale_for_orc_type(type));
+}
+
+const ::orc::Type* resolve_predicate_type(const ::orc::Type& root_type,
+ const
format::FileColumnPredicateFilter& column_filter) {
+ const auto file_column_id = column_filter.effective_file_column_id();
+ if (file_column_id.value() < 0 ||
+ file_column_id.value() >=
static_cast<int32_t>(root_type.getSubtypeCount())) {
+ return nullptr;
+ }
+ const auto* type =
root_type.getSubtype(static_cast<uint64_t>(file_column_id.value()));
+ for (const auto child_id : column_filter.effective_file_child_id_path()) {
+ if (type == nullptr || child_id < 0 ||
+ child_id >= static_cast<int32_t>(type->getSubtypeCount())) {
+ return nullptr;
+ }
+ type = type->getSubtype(static_cast<uint64_t>(child_id));
+ }
+ return type;
+}
+
+bool stripe_excludes_filter(const ::orc::Type& root_type, const
::orc::StripeStatistics& statistics,
+ const format::FileColumnPredicateFilter&
column_filter) {
+ const auto* predicate_type = resolve_predicate_type(root_type,
column_filter);
+ if (predicate_type == nullptr) {
+ return false;
+ }
+ const auto* column_statistics =
+
statistics.getColumnStatistics(cast_set<uint32_t>(predicate_type->getColumnId()));
+ if (column_statistics == nullptr) {
+ return false;
+ }
+
+ segment_v2::ZoneMap zone_map;
+ if (!build_zone_map_from_orc_statistics(*predicate_type,
*column_statistics, &zone_map)) {
+ return false;
+ }
+ for (const auto& predicate : column_filter.predicates) {
+ if (predicate == nullptr || !predicate->support_zonemap()) {
+ continue;
+ }
+ if (!predicate_can_evaluate_orc_zone_map(*predicate_type, *predicate))
{
+ continue;
+ }
+ if (!predicate->evaluate_and(zone_map)) {
+ return true;
+ }
+ }
+ return false;
+}
+
+// SARG conversion is intentionally conservative: only direct slots,
+// struct_element chains, and whitelisted schema-evolution casts become ORC
+// SearchArgument predicates. Everything else stays as row-level filtering.
+// ORC TypeKind → PredicateDataType
+//
+// SARG 评估器只认 7 种 PredicateDataType（这是 ORC 标准）：
+// LONG / FLOAT / STRING / DATE / DECIMAL / TIMESTAMP / BOOLEAN
+//
+// 注意"合并"：BYTE / SHORT / INT / LONG 都映射到 LONG（int64_t 内部表示）；
+// FLOAT / DOUBLE 都映射到 FLOAT（double 内部表示）；
+// STRING / BINARY / VARCHAR 都映射到 STRING。
+// 这样 SARG 评估器内部只需要按"类型大类"实现一套比较逻辑。
+//
+// 复杂类型（STRUCT / LIST / MAP / UNION）→ nullopt → SARG 不支持 → row-level fallback。
+// 注意：SARG 不支持的是"用复杂类型整体做谓词"。
+// "用 STRUCT 内部 primitive 子字段做谓词"（addr.zip > 100）是支持的，
+// 通过 struct_element 链解析到子字段的 ORC type 后再调本函数。
+std::optional<::orc::PredicateDataType> predicate_type_for_orc_type(const
::orc::Type& type) {
+ switch (type.getKind()) {
+ case ::orc::TypeKind::BYTE:
+ case ::orc::TypeKind::SHORT:
+ case ::orc::TypeKind::INT:
+ case ::orc::TypeKind::LONG:
+ return ::orc::PredicateDataType::LONG;
+ case ::orc::TypeKind::FLOAT:
+ case ::orc::TypeKind::DOUBLE:
+ return ::orc::PredicateDataType::FLOAT;
+ case ::orc::TypeKind::STRING:
+ case ::orc::TypeKind::BINARY:
+ case ::orc::TypeKind::VARCHAR:
+ return ::orc::PredicateDataType::STRING;
+ case ::orc::TypeKind::DATE:
+ return ::orc::PredicateDataType::DATE;
+ case ::orc::TypeKind::DECIMAL:
+ return ::orc::PredicateDataType::DECIMAL;
+ case ::orc::TypeKind::TIMESTAMP:
+ case ::orc::TypeKind::TIMESTAMP_INSTANT:
+ return ::orc::PredicateDataType::TIMESTAMP;
+ case ::orc::TypeKind::BOOLEAN:
+ return ::orc::PredicateDataType::BOOLEAN;
+ default:
+ return std::nullopt;
+ }
+}
+
+// SARG 把"列引用"抽象成 OrcSargColumn：
+// column_id ORC type id（含嵌套，比如 addr.zip 用的是 zip 子字段的 type id）
+// predicate_type SARG 评估器用的类型（按上面的合并规则）
+// orc_type 底层 ORC type 指针（构造 literal 时用）
+//
+// predicate_type 默认 LONG 只是个 placeholder（C++ 聚合 struct 默认值要求），
+// 实际使用时一定会被覆盖成真实值。
+struct OrcSargColumn {
+ uint64_t column_id = 0;
+ ::orc::PredicateDataType predicate_type = ::orc::PredicateDataType::LONG;
+ const ::orc::Type* orc_type = nullptr;
+};
+
+// 普通比较的 SARG 中间结果（非 cast 反解）：
+// column 左操作数指向的列
+// literal 右操作数（已转成 ORC Literal）
+// normalized_op 规范化后的 op（操作数顺序已挪到 "slot op literal" 形式）
+struct OrcSargComparison {
+ OrcSargColumn column;
+ ::orc::Literal literal;
+ TExprOpcode::type normalized_op = TExprOpcode::INVALID_OPCODE;
+};
+
+std::optional<format::LocalColumnId> file_column_id_for_slot_position(
+ const format::FileScanRequest& request, int slot_column_id) {
+ if (slot_column_id < 0) {
+ return std::nullopt;
+ }
+ // Localized VSlotRef::column_id() points to the file block position, not
the file schema id.
+ const auto slot_position = cast_set<size_t>(slot_column_id);
+ for (const auto& [file_column_id, local_position] :
request.local_positions) {
+ if (local_position.value() == slot_position) {
+ return file_column_id;
+ }
+ }
+ return std::nullopt;
+}
+
+const ::orc::Type* orc_type_for_slot(const format::FileScanRequest& request,
+ const ::orc::Type& root_type, const
VExprSPtr& slot_expr) {
+ if (slot_expr == nullptr || !slot_expr->is_slot_ref()) {
+ return nullptr;
+ }
+ const auto* slot_ref = dynamic_cast<const VSlotRef*>(slot_expr.get());
+ if (slot_ref == nullptr || slot_ref->column_id() < 0) {
+ return nullptr;
+ }
+ const auto file_column_id = file_column_id_for_slot_position(request,
slot_ref->column_id());
+ if (!file_column_id.has_value() || !file_column_id->is_valid()) {
+ return nullptr;
+ }
+ const auto file_column_position =
cast_set<uint64_t>(file_column_id->value());
+ if (file_column_position >= root_type.getSubtypeCount()) {
+ return nullptr;
+ }
+ return root_type.getSubtype(file_column_position);
+}
+
+std::optional<OrcSargColumn> sarg_column_for_orc_type(const ::orc::Type*
orc_type) {
+ if (orc_type == nullptr) {
+ return std::nullopt;
+ }
+ const auto predicate_type = predicate_type_for_orc_type(*orc_type);
+ if (!predicate_type.has_value()) {
+ return std::nullopt;
+ }
+ return OrcSargColumn {
+ .column_id = orc_type->getColumnId(),
+ .predicate_type = *predicate_type,
+ .orc_type = orc_type,
+ };
+}
+
+const ::orc::Type* orc_type_for_sarg_source_expr(const
format::FileScanRequest& request,
+ const ::orc::Type& root_type,
+ const VExprSPtr& expr);
+
+std::optional<Field> literal_field_for_sarg(const VExprSPtr& literal_expr) {
+ if (literal_expr == nullptr || !literal_expr->is_literal()) {
+ return std::nullopt;
+ }
+ const auto* literal = dynamic_cast<const VLiteral*>(literal_expr.get());
+ if (literal == nullptr || literal->get_column_ptr().get() == nullptr ||
+ literal->get_column_ptr()->is_null_at(0)) {
+ return std::nullopt;
+ }
+ Field field;
+ literal->get_column_ptr()->get(0, field);
+ return field;
+}
+
+// 识别 VExpr 是不是 struct_element(struct_value, 'field_name') 函数调用
+//
+// SQL 里访问 struct 子字段（addr.zip）的语法糖，被 SQL 编译器降级成这个函数调用：
+// addr.region.code → struct_element(struct_element(slot_ref(addr),
'region'), 'code')
+//
+// 用途：SARG 解析左操作数时识别这种链，递归剥到最深处的 primitive 子字段，
+// 拿到 ORC type id 后构造 SARG。这就是 SARG 支持嵌套 struct 子字段的关键。
+bool is_struct_element_expr(const VExprSPtr& expr) {
+ return expr != nullptr && expr->children().size() == 2 &&
+ expr->fn().name.function_name == "struct_element";
+}
+
+// 在 ORC struct type 里按字段名查找子字段索引
+// struct<id, addr, name>: struct_child_index(t, 'addr') = 1
+// 也支持 INT 字段索引：struct_element(s, 1) → 直接拿 index 1
+std::optional<uint64_t> struct_child_index(const ::orc::Type& struct_type,
+ const VExprSPtr& selector_expr) {
+ if (struct_type.getKind() != ::orc::TypeKind::STRUCT) {
+ return std::nullopt;
+ }
+ const auto field = literal_field_for_sarg(selector_expr);
+ if (!field.has_value()) {
+ return std::nullopt;
+ }
+ switch (field->get_type()) {
+ case TYPE_STRING:
+ case TYPE_CHAR:
+ case TYPE_VARCHAR: {
+ const auto child_name = std::string(field->as_string_view());
+ for (uint64_t child_idx = 0; child_idx <
struct_type.getSubtypeCount(); ++child_idx) {
+ if (struct_type.getFieldName(child_idx) == child_name) {
+ return child_idx;
+ }
+ }
+ return std::nullopt;
+ }
+ case TYPE_TINYINT:
+ if (field->get<TYPE_TINYINT>() <= 0) {
+ return std::nullopt;
+ }
+ return cast_set<uint64_t>(field->get<TYPE_TINYINT>() - 1);
+ case TYPE_SMALLINT:
+ if (field->get<TYPE_SMALLINT>() <= 0) {
+ return std::nullopt;
+ }
+ return cast_set<uint64_t>(field->get<TYPE_SMALLINT>() - 1);
+ case TYPE_INT:
+ if (field->get<TYPE_INT>() <= 0) {
+ return std::nullopt;
+ }
+ return cast_set<uint64_t>(field->get<TYPE_INT>() - 1);
+ case TYPE_BIGINT:
+ if (field->get<TYPE_BIGINT>() <= 0) {
+ return std::nullopt;
+ }
+ return cast_set<uint64_t>(field->get<TYPE_BIGINT>() - 1);
+ default:
+ return std::nullopt;
+ }
+}
+
+const ::orc::Type* orc_type_for_struct_element(const format::FileScanRequest&
request,
+ const ::orc::Type& root_type,
+ const VExprSPtr& expr) {
+ if (!is_struct_element_expr(expr)) {
+ return nullptr;
+ }
+ const auto* parent_type =
+ orc_type_for_sarg_source_expr(request, root_type,
expr->children()[0]);
+ if (parent_type == nullptr || parent_type->getKind() !=
::orc::TypeKind::STRUCT) {
+ return nullptr;
+ }
+ const auto child_idx = struct_child_index(*parent_type,
expr->children()[1]);
+ if (!child_idx.has_value() || *child_idx >=
parent_type->getSubtypeCount()) {
+ return nullptr;
+ }
+ return parent_type->getSubtype(*child_idx);
+}
+
+const ::orc::Type* orc_type_for_sarg_source_expr(const
format::FileScanRequest& request,
+ const ::orc::Type& root_type,
+ const VExprSPtr& expr) {
+ if (is_struct_element_expr(expr)) {
+ return orc_type_for_struct_element(request, root_type, expr);
+ }
+ return orc_type_for_slot(request, root_type, expr);
+}
+
+std::optional<OrcSargColumn> sarg_column_for_source_expr(const
format::FileScanRequest& request,
+ const ::orc::Type&
root_type,
+ const VExprSPtr&
expr) {
+ return sarg_column_for_orc_type(orc_type_for_sarg_source_expr(request,
root_type, expr));
+}
+
+// Safe cast checks protect pruning correctness. A cast is SARGable only when
the
+// comparison in ORC's domain is equivalent to the original Doris expression.
+std::optional<uint8_t> orc_integer_width(const ::orc::Type& type) {
+ switch (type.getKind()) {
+ case ::orc::TypeKind::BYTE:
+ return 8;
+ case ::orc::TypeKind::SHORT:
+ return 16;
+ case ::orc::TypeKind::INT:
+ return 32;
+ case ::orc::TypeKind::LONG:
+ return 64;
+ default:
+ return std::nullopt;
+ }
+}
+
+std::optional<uint8_t> signed_integer_width(PrimitiveType type) {
+ switch (type) {
+ case TYPE_TINYINT:
+ return 8;
+ case TYPE_SMALLINT:
+ return 16;
+ case TYPE_INT:
+ return 32;
+ case TYPE_BIGINT:
+ return 64;
+ default:
+ return std::nullopt;
+ }
+}
+
+std::optional<uint8_t> orc_floating_width(const ::orc::Type& type) {
+ switch (type.getKind()) {
+ case ::orc::TypeKind::FLOAT:
+ return 32;
+ case ::orc::TypeKind::DOUBLE:
+ return 64;
+ default:
+ return std::nullopt;
+ }
+}
+
+std::optional<uint8_t> floating_width(PrimitiveType type) {
+ switch (type) {
+ case TYPE_FLOAT:
+ return 32;
+ case TYPE_DOUBLE:
+ return 64;
+ default:
+ return std::nullopt;
+ }
+}
+
+std::optional<uint8_t> floating_exact_integer_width(PrimitiveType type) {
+ switch (type) {
+ case TYPE_FLOAT:
+ return 24;
+ case TYPE_DOUBLE:
+ return 53;
+ default:
+ return std::nullopt;
+ }
+}
+
+struct DecimalPrecisionScale {
+ UInt32 precision = 0;
+ UInt32 scale = 0;
+};
+
+std::optional<DecimalPrecisionScale> decimal_precision_scale(const
DataTypePtr& type) {
+ if (type == nullptr || !is_decimal(type->get_primitive_type())) {
+ return std::nullopt;
+ }
+ const auto precision = type->get_precision();
+ const auto scale = type->get_scale();
+ if (precision == 0 || scale > precision) {
+ return std::nullopt;
+ }
+ return DecimalPrecisionScale {.precision = precision, .scale = scale};
+}
+
+bool is_safe_widening_cast(const OrcSargColumn& column, const VExprSPtr&
cast_expr) {
+ if (column.orc_type == nullptr || cast_expr == nullptr ||
cast_expr->data_type() == nullptr) {
+ return false;
+ }
+ const auto target_type = remove_nullable(cast_expr->data_type());
+ if (target_type == nullptr) {
+ return false;
+ }
+ const auto target_primitive_type = target_type->get_primitive_type();
+ if (const auto source_width = orc_integer_width(*column.orc_type)) {
+ const auto target_width = signed_integer_width(target_primitive_type);
+ return target_width.has_value() && *target_width >= *source_width;
+ }
+ if (const auto source_width = orc_floating_width(*column.orc_type)) {
+ const auto target_width = floating_width(target_primitive_type);
+ return target_width.has_value() && *target_width >= *source_width;
+ }
+ return false;
+}
+
+bool is_safe_date_schema_evolution_cast(const OrcSargColumn& column, const
VExprSPtr& cast_expr) {
+ if (column.orc_type == nullptr || cast_expr == nullptr ||
cast_expr->data_type() == nullptr ||
+ column.orc_type->getKind() != ::orc::TypeKind::DATE) {
+ return false;
+ }
+ const auto target_type = remove_nullable(cast_expr->data_type());
+ if (target_type == nullptr) {
+ return false;
+ }
+ const auto target_primitive_type = target_type->get_primitive_type();
+ return target_primitive_type == TYPE_DATE || target_primitive_type ==
TYPE_DATEV2 ||
+ target_primitive_type == TYPE_DATETIME || target_primitive_type ==
TYPE_DATETIMEV2;
+}
+
+bool is_safe_string_schema_evolution_cast(const OrcSargColumn& column, const
VExprSPtr& cast_expr) {
+ if (column.orc_type == nullptr || cast_expr == nullptr ||
cast_expr->data_type() == nullptr) {
+ return false;
+ }
+ const auto orc_kind = column.orc_type->getKind();
+ switch (column.orc_type->getKind()) {
+ case ::orc::TypeKind::STRING:
+ case ::orc::TypeKind::BINARY:
+ case ::orc::TypeKind::VARCHAR:
+ break;
+ default:
+ return false;
+ }
+ if (cast_expr->children().size() != 1 || cast_expr->children()[0] ==
nullptr ||
+ cast_expr->children()[0]->data_type() == nullptr) {
+ return false;
+ }
+ const auto source_type =
remove_nullable(cast_expr->children()[0]->data_type());
+ if (source_type == nullptr) {
+ return false;
+ }
+ const auto source_primitive_type = source_type->get_primitive_type();
+ const bool source_is_string_like =
+ source_primitive_type == TYPE_STRING || source_primitive_type ==
TYPE_VARCHAR ||
+ (orc_kind == ::orc::TypeKind::BINARY && source_primitive_type ==
TYPE_VARBINARY);
+ if (!source_is_string_like) {
+ return false;
+ }
+ const auto target_type = remove_nullable(cast_expr->data_type());
+ if (target_type == nullptr) {
+ return false;
+ }
+ return target_type->get_primitive_type() == TYPE_STRING;
+}
+
+bool is_safe_decimal_schema_evolution_cast(const OrcSargColumn& column,
+ const VExprSPtr& cast_expr) {
+ if (column.orc_type == nullptr || column.orc_type->getKind() !=
::orc::TypeKind::DECIMAL ||
+ cast_expr == nullptr || cast_expr->data_type() == nullptr ||
+ cast_expr->children().size() != 1 || cast_expr->children()[0] ==
nullptr ||
+ cast_expr->children()[0]->data_type() == nullptr) {
+ return false;
+ }
+
+ const auto source_type =
remove_nullable(cast_expr->children()[0]->data_type());
+ const auto target_type = remove_nullable(cast_expr->data_type());
+ const auto source_decimal = decimal_precision_scale(source_type);
+ const auto target_decimal = decimal_precision_scale(target_type);
+ if (!source_decimal.has_value() || !target_decimal.has_value()) {
+ return false;
+ }
+ if (source_decimal->precision !=
cast_set<UInt32>(column.orc_type->getPrecision()) ||
+ source_decimal->scale !=
cast_set<UInt32>(column.orc_type->getScale())) {
+ return false;
+ }
+
+ const auto source_integer_digits = source_decimal->precision -
source_decimal->scale;
+ const auto target_integer_digits = target_decimal->precision -
target_decimal->scale;
+ return target_decimal->scale >= source_decimal->scale &&
+ target_integer_digits >= source_integer_digits;
+}
+
+bool is_safe_integer_to_floating_cast(const OrcSargColumn& column, const
VExprSPtr& cast_expr) {
+ if (column.orc_type == nullptr || cast_expr == nullptr ||
cast_expr->data_type() == nullptr) {
+ return false;
+ }
+ const auto source_width = orc_integer_width(*column.orc_type);
+ if (!source_width.has_value()) {
+ return false;
+ }
+ const auto target_type = remove_nullable(cast_expr->data_type());
+ if (target_type == nullptr) {
+ return false;
+ }
+ const auto target_width =
floating_exact_integer_width(target_type->get_primitive_type());
+ return target_width.has_value() && *source_width <= *target_width;
+}
+
+// 安全 cast 白名单总入口
+//
+// 原则：只把"绝对不会让 SARG 错裁"的 cast 形态放进 SARG。
+// 不安全的 cast 跳过 SARG（保守），row-level filter 兜底。
+//
+// 5 种安全 cast：
+// widening_cast 整数 widening (INT → BIGINT)、float widening
(FLOAT → DOUBLE)
+// 和 DECIMAL 精确 widening
+// date_schema_evolution_cast DATE → DATE/DATEV2/DATETIME/DATETIMEV2 安全子集
+// string_schema_evolution_cast STRING/VARCHAR → STRING（注意 CHAR 不安全）
+// decimal_schema_evolution_cast DECIMAL widening (scale 不降低、整数位不减少)
+// integer_to_floating_cast BYTE/SHORT → FLOAT、BYTE/SHORT/INT → DOUBLE
的精确表示
+//
+// 不安全的反例：
+// narrowing cast (BIGINT → INT)：可能溢出，stripe 统计的 BIGINT 范围跟 cast 后 INT 不一致
+// 非整数 INT → FLOAT 等精度丢失场景：cast 后值跟 stripe 统计对不齐
+bool is_safe_cast_for_sarg(const OrcSargColumn& column, const VExprSPtr&
cast_expr) {
+ return is_safe_widening_cast(column, cast_expr) ||
+ is_safe_date_schema_evolution_cast(column, cast_expr) ||
+ is_safe_string_schema_evolution_cast(column, cast_expr) ||
+ is_safe_decimal_schema_evolution_cast(column, cast_expr) ||
+ is_safe_integer_to_floating_cast(column, cast_expr);
+}
+
+// 解析 SARG 的左操作数：直接 slot 或 安全 cast 包裹的 slot
+//
+// SQL 例子：
+// id > 100 expr = slot_ref(id) 直接是 slot
+// cast(id AS BIGINT)>100 expr = cast(slot_ref(id), BIGINT) cast 包裹 slot
+//
+// 处理顺序：
+// 1. 先试 sarg_column_for_source_expr：识别 slot / struct_element 链
+// 2. 如果是 cast 节点，递归剥 cast，验证是 safe cast，再返回剥到的 slot 信息
+// 3. 不安全 / 不是 slot / 不是 cast → nullopt → 整体跳过 SARG
+//
+// 注意：返回的 OrcSargColumn 是**底层 file 列**的信息，不是 cast 目标类型。
+// SARG 评估用的是文件 stripe 统计，而 stripe 统计是 file 列原始类型的。
+std::optional<OrcSargColumn> sarg_column_for_slot_or_safe_cast(
+ const format::FileScanRequest& request, const ::orc::Type& root_type,
+ const VExprSPtr& expr) {
+ auto column = sarg_column_for_source_expr(request, root_type, expr);
+ if (column.has_value()) {
+ return column; // 直接 slot / struct_element 链
+ }
+ if (expr == nullptr || expr->node_type() != TExprNodeType::CAST_EXPR ||
+ expr->children().size() != 1) {
+ return std::nullopt;
+ }
+ // 是 cast 节点，剥皮验证
+ column = sarg_column_for_source_expr(request, root_type,
expr->children()[0]);
+ if (!column.has_value() || !is_safe_cast_for_sarg(*column, expr)) {
+ return std::nullopt;
+ }
+ return column;
+}
+
+std::optional<OrcSargColumn> sarg_column_for_column_filter(
+ const ::orc::Type& root_type, const format::FileColumnPredicateFilter&
column_filter) {
+ if (!column_filter.effective_file_column_id().is_valid()) {
+ return std::nullopt;
+ }
+ const auto* orc_type = resolve_predicate_type(root_type, column_filter);
+ if (orc_type == nullptr) {
+ return std::nullopt;
+ }
+ const auto predicate_type = predicate_type_for_orc_type(*orc_type);
+ if (!predicate_type.has_value()) {
+ return std::nullopt;
+ }
+ return OrcSargColumn {
+ .column_id = orc_type->getColumnId(),
+ .predicate_type = *predicate_type,
+ .orc_type = orc_type,
+ };
+}
+
+// Literal conversion preserves ORC PredicateDataType rules. Unsupported
+// literal/type pairs simply make the surrounding predicate non-SARGable.
+std::optional<::orc::Literal> make_long_literal(const Field& field) {
+ switch (field.get_type()) {
+ case TYPE_TINYINT:
+ return ::orc::Literal(static_cast<int64_t>(field.get<TYPE_TINYINT>()));
+ case TYPE_SMALLINT:
+ return
::orc::Literal(static_cast<int64_t>(field.get<TYPE_SMALLINT>()));
+ case TYPE_INT:
+ return ::orc::Literal(static_cast<int64_t>(field.get<TYPE_INT>()));
+ case TYPE_BIGINT:
+ return ::orc::Literal(static_cast<int64_t>(field.get<TYPE_BIGINT>()));
+ default:
+ return std::nullopt;
+ }
+}
+
+std::optional<::orc::Literal> make_float_literal(const Field& field) {
+ switch (field.get_type()) {
+ case TYPE_FLOAT:
+ return ::orc::Literal(static_cast<double>(field.get<TYPE_FLOAT>()));
+ case TYPE_DOUBLE:
+ return ::orc::Literal(field.get<TYPE_DOUBLE>());
+ default:
+ return std::nullopt;
+ }
+}
+
+std::optional<::orc::Literal> make_string_literal(const Field& field) {
+ if (!is_string_type(field.get_type()) && !is_varbinary(field.get_type())) {
+ return std::nullopt;
+ }
+ const auto value = field.as_string_view();
+ return ::orc::Literal(value.data(), value.size());
+}
+
+std::optional<::orc::Literal> make_bool_literal(const Field& field) {
+ if (field.get_type() != TYPE_BOOLEAN) {
+ return std::nullopt;
+ }
+ return ::orc::Literal(field.get<TYPE_BOOLEAN>() != 0);
+}
+
+std::optional<::orc::Literal> make_date_literal(const Field& field) {
+ static const cctz::time_zone utc0 = cctz::utc_time_zone();
+ switch (field.get_type()) {
+ case TYPE_DATE: {
+ const auto& date = field.get<TYPE_DATE>();
+ const cctz::civil_day civil_date(date.year(), date.month(),
date.day());
+ const auto day_offset =
+ cctz::convert(civil_date, utc0).time_since_epoch().count() /
(24 * 60 * 60);
+ return ::orc::Literal(::orc::PredicateDataType::DATE, day_offset);
+ }
+ case TYPE_DATEV2: {
+ const auto& date = field.get<TYPE_DATEV2>();
+ const cctz::civil_day civil_date(date.year(), date.month(),
date.day());
+ const auto day_offset =
+ cctz::convert(civil_date, utc0).time_since_epoch().count() /
(24 * 60 * 60);
+ return ::orc::Literal(::orc::PredicateDataType::DATE, day_offset);
+ }
+ case TYPE_DATETIME: {
+ const auto& datetime = field.get<TYPE_DATETIME>();
+ if (datetime.hour() != 0 || datetime.minute() != 0 ||
datetime.second() != 0) {
+ return std::nullopt;
+ }
+ const cctz::civil_day civil_date(datetime.year(), datetime.month(),
datetime.day());
+ const auto day_offset =
+ cctz::convert(civil_date, utc0).time_since_epoch().count() /
(24 * 60 * 60);
+ return ::orc::Literal(::orc::PredicateDataType::DATE, day_offset);
+ }
+ case TYPE_DATETIMEV2: {
+ const auto& datetime = field.get<TYPE_DATETIMEV2>();
+ if (datetime.hour() != 0 || datetime.minute() != 0 ||
datetime.second() != 0 ||
+ datetime.microsecond() != 0) {
+ return std::nullopt;
+ }
+ const cctz::civil_day civil_date(datetime.year(), datetime.month(),
datetime.day());
+ const auto day_offset =
+ cctz::convert(civil_date, utc0).time_since_epoch().count() /
(24 * 60 * 60);
+ return ::orc::Literal(::orc::PredicateDataType::DATE, day_offset);
+ }
+ default:
+ return std::nullopt;
+ }
+}
+
+std::optional<::orc::Literal> make_date_literal(int year, int month, int day) {
+ static const cctz::time_zone utc0 = cctz::utc_time_zone();
+ const cctz::civil_day civil_date(year, month, day);
+ const auto day_offset =
+ cctz::convert(civil_date, utc0).time_since_epoch().count() / (24 *
60 * 60);
+ return ::orc::Literal(::orc::PredicateDataType::DATE, day_offset);
+}
+
+struct DateTimeLiteralParts {
+ int year = 0;
+ int month = 0;
+ int day = 0;
+ bool has_time = false;
+};
+
+std::optional<DateTimeLiteralParts> date_time_literal_parts(const Field&
field) {
+ switch (field.get_type()) {
+ case TYPE_DATETIME: {
+ const auto& datetime = field.get<TYPE_DATETIME>();
+ return DateTimeLiteralParts {
+ .year = datetime.year(),
+ .month = datetime.month(),
+ .day = datetime.day(),
+ .has_time =
+ datetime.hour() != 0 || datetime.minute() != 0 ||
datetime.second() != 0,
+ };
+ }
+ case TYPE_DATETIMEV2: {
+ const auto& datetime = field.get<TYPE_DATETIMEV2>();
+ return DateTimeLiteralParts {
+ .year = datetime.year(),
+ .month = datetime.month(),
+ .day = datetime.day(),
+ .has_time = datetime.hour() != 0 || datetime.minute() != 0 ||
+ datetime.second() != 0 || datetime.microsecond()
!= 0,
+ };
+ }
+ default:
+ return std::nullopt;
+ }
+}
+
+std::optional<::orc::Literal> make_timestamp_literal(const Field& field) {
+ static const cctz::time_zone utc0 = cctz::utc_time_zone();
+ switch (field.get_type()) {
+ case TYPE_DATETIME: {
+ const auto& datetime = field.get<TYPE_DATETIME>();
+ const cctz::civil_second civil_seconds(datetime.year(),
datetime.month(), datetime.day(),
+ datetime.hour(),
datetime.minute(),
+ datetime.second());
+ return ::orc::Literal(cctz::convert(civil_seconds,
utc0).time_since_epoch().count(), 0);
+ }
+ case TYPE_DATETIMEV2: {
+ const auto& datetime = field.get<TYPE_DATETIMEV2>();
+ const cctz::civil_second civil_seconds(datetime.year(),
datetime.month(), datetime.day(),
+ datetime.hour(),
datetime.minute(),
+ datetime.second());
+ const auto seconds = cctz::convert(civil_seconds,
utc0).time_since_epoch().count();
+ const auto nanos = cast_set<int32_t>(datetime.microsecond() * 1000);
+ return ::orc::Literal(seconds, nanos);
+ }
+ default:
+ return std::nullopt;
+ }
+}
+
+std::optional<::orc::Literal> make_decimal_literal(const ::orc::Type& orc_type,
+ const VLiteral& literal,
const Field& field) {
+ if (orc_type.getKind() != ::orc::TypeKind::DECIMAL) {
+ return std::nullopt;
+ }
+ const auto& literal_type = literal.get_data_type();
+ if (literal_type == nullptr) {
+ return std::nullopt;
+ }
+
+ Int128 decimal_value = 0;
+ switch (field.get_type()) {
+ case TYPE_DECIMALV2:
+ decimal_value = binary_cast<DecimalV2Value,
Int128>(field.get<TYPE_DECIMALV2>());
+ break;
+ case TYPE_DECIMAL32:
+ decimal_value = static_cast<Int128>(field.get<TYPE_DECIMAL32>().value);
+ break;
+ case TYPE_DECIMAL64:
+ decimal_value = static_cast<Int128>(field.get<TYPE_DECIMAL64>().value);
+ break;
+ case TYPE_DECIMAL128I:
+ decimal_value = field.get<TYPE_DECIMAL128I>().value;
+ break;
+ default:
+ return std::nullopt;
+ }
+
+ const auto precision = literal_type->get_precision() == 0
+ ? cast_set<UInt32>(orc_type.getPrecision())
+ : literal_type->get_precision();
+ const auto scale = literal_type->get_scale();
+ return ::orc::Literal(::orc::Int128(static_cast<uint64_t>(decimal_value >>
64),
+ static_cast<uint64_t>(decimal_value)),
+ cast_set<int>(precision), cast_set<int>(scale));
+}
+
+std::optional<::orc::Literal> make_decimal_literal(
+ const ::orc::Type& orc_type, const ColumnPredicateLiteralTypeInfo&
literal_type_info,
+ const Field& field) {
+ if (orc_type.getKind() != ::orc::TypeKind::DECIMAL) {
+ return std::nullopt;
+ }
+
+ Int128 decimal_value = 0;
+ switch (field.get_type()) {
+ case TYPE_DECIMALV2:
+ decimal_value = binary_cast<DecimalV2Value,
Int128>(field.get<TYPE_DECIMALV2>());
+ break;
+ case TYPE_DECIMAL32:
+ decimal_value = static_cast<Int128>(field.get<TYPE_DECIMAL32>().value);
+ break;
+ case TYPE_DECIMAL64:
+ decimal_value = static_cast<Int128>(field.get<TYPE_DECIMAL64>().value);
+ break;
+ case TYPE_DECIMAL128I:
+ decimal_value = field.get<TYPE_DECIMAL128I>().value;
+ break;
+ default:
+ return std::nullopt;
+ }
+
+ const auto precision = literal_type_info.precision == 0
+ ?
cast_set<uint32_t>(orc_type.getPrecision())
+ : literal_type_info.precision;
+ return ::orc::Literal(::orc::Int128(static_cast<uint64_t>(decimal_value >>
64),
+ static_cast<uint64_t>(decimal_value)),
+ cast_set<int>(precision),
cast_set<int>(literal_type_info.scale));
+}
+
+bool column_predicate_can_execute_on_decoded_column(
+ const format::FileColumnPredicateFilter& column_filter, const
DataTypePtr& decoded_type) {
+ if (!column_filter.effective_file_child_id_path().empty()) {
+ return false;
+ }
+ const auto nested_type = remove_nullable(decoded_type);
+ if (nested_type == nullptr ||
!is_decimal(nested_type->get_primitive_type())) {
+ return true;
+ }
+ for (const auto& predicate : column_filter.predicates) {
+ if (predicate == nullptr) {
+ return false;
+ }
+ if (predicate->type() == PredicateType::IS_NULL ||
+ predicate->type() == PredicateType::IS_NOT_NULL) {
+ continue;
+ }
+ const auto* literal_provider =
+ dynamic_cast<const
ColumnPredicateLiteralProvider*>(predicate.get());
+ if (literal_provider == nullptr) {
+ return false;
+ }
+ const auto literal_type_info =
literal_provider->predicate_literal_type_info();
+ if (!literal_type_info.has_value() ||
+ literal_type_info->scale !=
cast_set<uint32_t>(nested_type->get_scale())) {
+ return false;
+ }
+ }
+ return true;
+}
+
+std::optional<::orc::Literal> make_orc_literal(const OrcSargColumn&
sarg_column,
+ const Field& field) {
+ switch (sarg_column.predicate_type) {
+ case ::orc::PredicateDataType::LONG:
+ return make_long_literal(field);
+ case ::orc::PredicateDataType::FLOAT:
+ return make_float_literal(field);
+ case ::orc::PredicateDataType::STRING:
+ return make_string_literal(field);
+ case ::orc::PredicateDataType::BOOLEAN:
+ return make_bool_literal(field);
+ case ::orc::PredicateDataType::DATE:
+ return make_date_literal(field);
+ case ::orc::PredicateDataType::TIMESTAMP:
+ return make_timestamp_literal(field);
+ case ::orc::PredicateDataType::DECIMAL:
+ // ColumnPredicate does not currently carry decimal scale/precision.
+ return std::nullopt;
+ }
+ return std::nullopt;
+}
+
+std::optional<::orc::Literal> make_orc_literal(
+ const OrcSargColumn& sarg_column, const Field& field,
+ const std::optional<ColumnPredicateLiteralTypeInfo>&
literal_type_info) {
+ if (sarg_column.predicate_type != ::orc::PredicateDataType::DECIMAL) {
+ return make_orc_literal(sarg_column, field);
+ }
+ if (sarg_column.orc_type == nullptr || !literal_type_info.has_value()) {
+ return std::nullopt;
+ }
+ return make_decimal_literal(*sarg_column.orc_type, *literal_type_info,
field);
+}
+
+std::optional<::orc::Literal> make_orc_literal(const OrcSargColumn&
sarg_column,
+ const VExprSPtr& literal_expr) {
+ if (literal_expr == nullptr || !literal_expr->is_literal()) {
+ return std::nullopt;
+ }
+ const auto* literal = dynamic_cast<const VLiteral*>(literal_expr.get());
+ if (literal == nullptr || literal->get_column_ptr().get() == nullptr ||
+ literal->get_column_ptr()->is_null_at(0)) {
+ return std::nullopt;
+ }
+
+ Field field;
+ literal->get_column_ptr()->get(0, field);
+ if (sarg_column.predicate_type == ::orc::PredicateDataType::DECIMAL) {
+ if (sarg_column.orc_type == nullptr) {
+ return std::nullopt;
+ }
+ return make_decimal_literal(*sarg_column.orc_type, *literal, field);
+ }
+ return make_orc_literal(sarg_column, field);
+}
+
+bool is_null_literal(const VExprSPtr& literal_expr) {
+ if (literal_expr == nullptr || !literal_expr->is_literal()) {
+ return false;
+ }
+ const auto* literal = dynamic_cast<const VLiteral*>(literal_expr.get());
+ return literal != nullptr && literal->get_column_ptr().get() != nullptr &&
+ literal->get_column_ptr()->is_null_at(0);
+}
+
+// Buildability is checked before emission so the builder path can assume the
+// expression shape was validated and use DORIS_CHECK for impossible branches.
+bool can_build_search_argument(const format::FileScanRequest& request, const
::orc::Type& root_type,
+ const VExprSPtr& expr);
+
+std::optional<TExprOpcode::type> reverse_comparison_op(TExprOpcode::type op) {
+ switch (op) {
+ case TExprOpcode::GE:
+ return TExprOpcode::LE;
+ case TExprOpcode::GT:
+ return TExprOpcode::LT;
+ case TExprOpcode::LE:
+ return TExprOpcode::GE;
+ case TExprOpcode::LT:
+ return TExprOpcode::GT;
+ case TExprOpcode::EQ:
+ case TExprOpcode::NE:
+ return op;
+ default:
+ return std::nullopt;
+ }
+}
+
+bool is_date_to_datetime_cast_for_sarg(const OrcSargColumn& column, const
VExprSPtr& expr) {
+ if (column.orc_type == nullptr || column.orc_type->getKind() !=
::orc::TypeKind::DATE ||
+ expr == nullptr || expr->node_type() != TExprNodeType::CAST_EXPR ||
+ expr->data_type() == nullptr) {
+ return false;
+ }
+ const auto target_type = remove_nullable(expr->data_type());
+ if (target_type == nullptr) {
+ return false;
+ }
+ const auto target_type_id = target_type->get_primitive_type();
+ return target_type_id == TYPE_DATETIME || target_type_id ==
TYPE_DATETIMEV2;
+}
+
+bool is_integer_to_floating_cast_for_sarg(const OrcSargColumn& column, const
VExprSPtr& expr) {
+ return expr != nullptr && expr->node_type() == TExprNodeType::CAST_EXPR &&
+ is_safe_integer_to_floating_cast(column, expr);
+}
+
+std::optional<TExprOpcode::type>
normalize_date_to_datetime_comparison_op(TExprOpcode::type op,
+ bool
literal_has_time) {
+ if (!literal_has_time) {
+ return op;
+ }
+ switch (op) {
+ case TExprOpcode::GT:
+ case TExprOpcode::GE:
+ return TExprOpcode::GT;
+ case TExprOpcode::LT:
+ case TExprOpcode::LE:
+ return TExprOpcode::LE;
+ default:
+ return std::nullopt;
+ }
+}
+
+std::optional<double> floating_literal_value_for_sarg(const VExprSPtr&
literal_expr) {
+ const auto field = literal_field_for_sarg(literal_expr);
+ if (!field.has_value()) {
+ return std::nullopt;
+ }
+ switch (field->get_type()) {
+ case TYPE_FLOAT:
+ return static_cast<double>(field->get<TYPE_FLOAT>());
+ case TYPE_DOUBLE:
+ return field->get<TYPE_DOUBLE>();
+ default:
+ return std::nullopt;
+ }
+}
+
+std::optional<int64_t> double_to_int64_boundary(double value, double
(*round_func)(double)) {
+ if (!std::isfinite(value)) {
+ return std::nullopt;
+ }
+ const auto rounded = round_func(value);
+ const auto int64_min =
static_cast<double>(std::numeric_limits<int64_t>::min());
+ // INT64_MAX rounds to 2^63 as double, so keep the upper bound exclusive.
+ const auto int64_max_exclusive = std::ldexp(1.0, 63);
+ if (rounded < int64_min || rounded >= int64_max_exclusive) {
+ return std::nullopt;
+ }
+ return static_cast<int64_t>(rounded);
+}
+
+std::optional<int64_t> double_to_integral_int64(double value) {
+ if (!std::isfinite(value) || std::trunc(value) != value) {
+ return std::nullopt;
+ }
+ return double_to_int64_boundary(value, std::trunc);
+}
+
+// cast 反解的中间结果
+//
+// 跟 OrcSargComparison 区别：用在 cast 反解的子函数里。
+// 反解是把 cast(slot AS X) op literal 等价改写成 slot op' literal'，
+// 可能会改变 op（比如 cast(int as DOUBLE) > 500.5 反解成 int >= 501），
+// 所以多了 normalized_op 字段，由外层根据反解结果重新选 SARG API。
+//
+// 不带 column：因为外层已经知道左操作数指向哪个列了，反解只关心右操作数和新 op。
+struct OrcSargComparisonLiteral {
+ ::orc::Literal literal;
+ TExprOpcode::type normalized_op = TExprOpcode::INVALID_OPCODE;
+};
+
+std::optional<OrcSargComparisonLiteral>
make_integer_to_floating_comparison_literal(
+ TExprOpcode::type normalized_op, double literal_value) {
+ std::optional<int64_t> integer_literal;
+ TExprOpcode::type integer_op = normalized_op;
+ switch (normalized_op) {
+ case TExprOpcode::GT:
+ integer_literal = double_to_int64_boundary(literal_value, std::floor);
+ break;
+ case TExprOpcode::GE:
+ integer_literal = double_to_int64_boundary(literal_value, std::ceil);
+ break;
+ case TExprOpcode::LT:
+ integer_literal = double_to_int64_boundary(literal_value, std::ceil);
+ break;
+ case TExprOpcode::LE:
+ integer_literal = double_to_int64_boundary(literal_value, std::floor);
+ break;
+ case TExprOpcode::EQ:
+ case TExprOpcode::NE:
+ integer_literal = double_to_integral_int64(literal_value);
+ break;
+ default:
+ return std::nullopt;
+ }
+ if (!integer_literal.has_value()) {
+ return std::nullopt;
+ }
+ return OrcSargComparisonLiteral {
+ .literal = ::orc::Literal(*integer_literal),
+ .normalized_op = integer_op,
+ };
+}
+
+std::optional<OrcSargComparisonLiteral> make_comparison_literal_for_sarg(
+ const OrcSargColumn& column, const VExprSPtr& source_expr, const
VExprSPtr& literal_expr,
+ TExprOpcode::type normalized_op) {
+ if (is_date_to_datetime_cast_for_sarg(column, source_expr)) {
+ const auto field = literal_field_for_sarg(literal_expr);
+ if (!field.has_value()) {
+ return std::nullopt;
+ }
+ const auto parts = date_time_literal_parts(*field);
+ if (!parts.has_value()) {
+ return std::nullopt;
+ }
+ const auto adjusted_op =
+ normalize_date_to_datetime_comparison_op(normalized_op,
parts->has_time);
+ if (!adjusted_op.has_value()) {
+ return std::nullopt;
+ }
+ const auto literal = make_date_literal(parts->year, parts->month,
parts->day);
+ if (!literal.has_value()) {
+ return std::nullopt;
+ }
+ return OrcSargComparisonLiteral {
+ .literal = *literal,
+ .normalized_op = *adjusted_op,
+ };
+ }
+ if (is_integer_to_floating_cast_for_sarg(column, source_expr)) {
+ const auto literal_value =
floating_literal_value_for_sarg(literal_expr);
+ if (!literal_value.has_value()) {
+ return std::nullopt;
+ }
+ return make_integer_to_floating_comparison_literal(normalized_op,
*literal_value);
+ }
+
+ const auto literal = make_orc_literal(column, literal_expr);
+ if (!literal.has_value()) {
+ return std::nullopt;
+ }
+ return OrcSargComparisonLiteral {
+ .literal = *literal,
+ .normalized_op = normalized_op,
+ };
+}
+
+std::optional<std::vector<::orc::Literal>> make_in_literals_for_sarg(
+ const OrcSargColumn& column, const VExprSPtr& source_expr,
+ const std::vector<VExprSPtr>& children) {
+ if (children.size() < 2) {
+ return std::nullopt;
+ }
+
+ std::vector<::orc::Literal> literals;
+ literals.reserve(children.size() - 1);
+ if (is_date_to_datetime_cast_for_sarg(column, source_expr)) {
+ for (auto child_it = children.begin() + 1; child_it != children.end();
++child_it) {
+ if (is_null_literal(*child_it)) {
+ continue;
+ }
+ const auto field = literal_field_for_sarg(*child_it);
+ if (!field.has_value()) {
+ return std::nullopt;
+ }
+ const auto parts = date_time_literal_parts(*field);
+ if (!parts.has_value()) {
+ return std::nullopt;
+ }
+ if (parts->has_time) {
+ continue;
+ }
+ const auto literal = make_date_literal(parts->year, parts->month,
parts->day);
+ if (!literal.has_value()) {
+ return std::nullopt;
+ }
+ literals.push_back(*literal);
+ }
+ if (literals.empty()) {
+ return std::nullopt;
+ }
+ return literals;
+ }
+ if (is_integer_to_floating_cast_for_sarg(column, source_expr)) {
+ for (auto child_it = children.begin() + 1; child_it != children.end();
++child_it) {
+ if (is_null_literal(*child_it)) {
+ continue;
+ }
+ const auto literal_value =
floating_literal_value_for_sarg(*child_it);
+ if (!literal_value.has_value()) {
+ return std::nullopt;
+ }
+ const auto integer_literal =
double_to_integral_int64(*literal_value);
+ if (!integer_literal.has_value()) {
+ continue;
+ }
+ literals.emplace_back(*integer_literal);
+ }
+ if (literals.empty()) {
+ return std::nullopt;
+ }
+ return literals;
+ }
+
+ for (auto child_it = children.begin() + 1; child_it != children.end();
++child_it) {
+ if (is_null_literal(*child_it)) {
+ continue;
+ }
+ auto literal = make_orc_literal(column, *child_it);
+ if (!literal.has_value()) {
+ return std::nullopt;
+ }
+ literals.push_back(*literal);
+ }
+ if (literals.empty()) {
+ return std::nullopt;
+ }
+ return literals;
+}
+
+std::optional<OrcSargComparison> sarg_comparison_for_expr(const
format::FileScanRequest& request,
+ const ::orc::Type&
root_type,
+ const VExprSPtr&
expr) {
+ if (expr == nullptr || expr->children().size() != 2) {
+ return std::nullopt;
+ }
+
+ const VExprSPtr* slot_expr = nullptr;
+ const VExprSPtr* literal_expr = nullptr;
+ auto normalized_op = expr->op();
+ if (sarg_column_for_slot_or_safe_cast(request, root_type,
expr->children()[0]).has_value() &&
+ expr->children()[1]->is_literal()) {
+ slot_expr = &expr->children()[0];
+ literal_expr = &expr->children()[1];
+ } else if (expr->children()[0]->is_literal() &&
+ sarg_column_for_slot_or_safe_cast(request, root_type,
expr->children()[1])
+ .has_value()) {
+ const auto reversed_op = reverse_comparison_op(expr->op());
+ if (!reversed_op.has_value()) {
+ return std::nullopt;
+ }
+ slot_expr = &expr->children()[1];
+ literal_expr = &expr->children()[0];
+ normalized_op = *reversed_op;
+ } else {
+ return std::nullopt;
+ }
+
+ auto sarg_column = sarg_column_for_slot_or_safe_cast(request, root_type,
*slot_expr);
+ if (!sarg_column.has_value()) {
+ return std::nullopt;
+ }
+ const auto comparison_literal =
make_comparison_literal_for_sarg(*sarg_column, *slot_expr,
+
*literal_expr, normalized_op);
+ if (!comparison_literal.has_value()) {
+ return std::nullopt;
+ }
+ return OrcSargComparison {
+ .column = *sarg_column,
+ .literal = comparison_literal->literal,
+ .normalized_op = comparison_literal->normalized_op,
+ };
+}
+
+bool can_build_slot_literal_predicate(const format::FileScanRequest& request,
+ const ::orc::Type& root_type, const
VExprSPtr& expr) {
+ if (expr == nullptr || expr->children().size() != 2) {
+ return false;
+ }
+ return sarg_comparison_for_expr(request, root_type, expr).has_value();
+}
+
+bool can_build_in_predicate(const format::FileScanRequest& request, const
::orc::Type& root_type,
+ const VExprSPtr& expr) {
+ if (expr == nullptr || expr->children().size() < 2) {
+ return false;
+ }
+ const auto sarg_column =
+ sarg_column_for_slot_or_safe_cast(request, root_type,
expr->children()[0]);
+ if (!sarg_column.has_value()) {
+ return false;
+ }
+ return make_in_literals_for_sarg(*sarg_column, expr->children()[0],
expr->children())
+ .has_value();
+}
+
+bool can_build_is_null_predicate(const format::FileScanRequest& request,
+ const ::orc::Type& root_type, const
VExprSPtr& expr) {
+ return expr != nullptr && expr->children().size() == 1 &&
+ sarg_column_for_slot_or_safe_cast(request, root_type,
expr->children()[0]).has_value();
+}
+
+std::optional<OrcSargColumn> sarg_column_for_null_safe_equal_null(
+ const format::FileScanRequest& request, const ::orc::Type& root_type,
+ const VExprSPtr& expr) {
+ if (expr == nullptr || expr->node_type() !=
TExprNodeType::NULL_AWARE_BINARY_PRED ||
+ expr->op() != TExprOpcode::EQ_FOR_NULL || expr->children().size() !=
2) {
+ return std::nullopt;
+ }
+
+ auto column_if_other_child_is_null_literal =
+ [&](size_t slot_idx, size_t literal_idx) ->
std::optional<OrcSargColumn> {
+ if (!is_null_literal(expr->children()[literal_idx])) {
+ return std::nullopt;
+ }
+ return sarg_column_for_slot_or_safe_cast(request, root_type,
expr->children()[slot_idx]);
+ };
+ auto column = column_if_other_child_is_null_literal(0, 1);
+ if (column.has_value()) {
+ return column;
+ }
+ return column_if_other_child_is_null_literal(1, 0);
+}
+
+std::optional<OrcSargComparison> sarg_comparison_for_null_safe_equal_literal(
+ const format::FileScanRequest& request, const ::orc::Type& root_type,
+ const VExprSPtr& expr) {
+ if (expr == nullptr || expr->node_type() !=
TExprNodeType::NULL_AWARE_BINARY_PRED ||
+ expr->op() != TExprOpcode::EQ_FOR_NULL || expr->children().size() !=
2) {
+ return std::nullopt;
+ }
+
+ auto comparison_if_other_child_is_non_null_literal =
+ [&](size_t slot_idx, size_t literal_idx) ->
std::optional<OrcSargComparison> {
+ const auto& literal_expr = expr->children()[literal_idx];
+ if (literal_expr == nullptr || !literal_expr->is_literal() ||
+ is_null_literal(literal_expr)) {
+ return std::nullopt;
+ }
+ const auto& source_expr = expr->children()[slot_idx];
+ auto column = sarg_column_for_slot_or_safe_cast(request, root_type,
source_expr);
+ if (!column.has_value()) {
+ return std::nullopt;
+ }
+ const auto comparison_literal = make_comparison_literal_for_sarg(
+ *column, source_expr, literal_expr, TExprOpcode::EQ);
+ if (!comparison_literal.has_value() ||
+ comparison_literal->normalized_op != TExprOpcode::EQ) {
+ return std::nullopt;
+ }
+ return OrcSargComparison {
+ .column = *column,
+ .literal = comparison_literal->literal,
+ .normalized_op = TExprOpcode::EQ,
+ };
+ };
+ auto comparison = comparison_if_other_child_is_non_null_literal(0, 1);
+ if (comparison.has_value()) {
+ return comparison;
+ }
+ return comparison_if_other_child_is_non_null_literal(1, 0);
+}
+
+bool can_build_null_safe_equal_predicate(const format::FileScanRequest&
request,
+ const ::orc::Type& root_type, const
VExprSPtr& expr) {
+ return sarg_column_for_null_safe_equal_null(request, root_type,
expr).has_value() ||
+ sarg_comparison_for_null_safe_equal_literal(request, root_type,
expr).has_value();
+}
+
+bool contains_null_safe_equal(const VExprSPtr& expr) {
+ if (expr == nullptr) {
+ return false;
+ }
+ if (expr->op() == TExprOpcode::EQ_FOR_NULL) {
+ return true;
+ }
+ return std::ranges::any_of(expr->children(), contains_null_safe_equal);
+}
+
+bool can_build_search_argument(const format::FileScanRequest& request, const
::orc::Type& root_type,
+ const VExprSPtr& expr) {
+ if (expr == nullptr) {
+ return false;
+ }
+
+ switch (expr->op()) {
+ case TExprOpcode::COMPOUND_AND:
+ return std::ranges::any_of(expr->children(), [&](const auto& child) {
+ return can_build_search_argument(request, root_type, child);
+ });
+ case TExprOpcode::COMPOUND_OR:
+ if (contains_null_safe_equal(expr)) {
+ return false;
+ }
+ return !expr->children().empty() &&
+ std::ranges::all_of(expr->children(), [&](const auto& child) {
+ return can_build_search_argument(request, root_type, child);
+ });
+ case TExprOpcode::COMPOUND_NOT:
+ if (contains_null_safe_equal(expr)) {
+ return false;
+ }
+ return expr->children().size() == 1 &&
+ can_build_search_argument(request, root_type,
expr->children()[0]);
+ case TExprOpcode::GE:
+ case TExprOpcode::GT:
+ case TExprOpcode::LE:
+ case TExprOpcode::LT:
+ case TExprOpcode::EQ:
+ case TExprOpcode::NE:
+ return expr->node_type() != TExprNodeType::NULL_AWARE_BINARY_PRED &&
+ can_build_slot_literal_predicate(request, root_type, expr);
+ case TExprOpcode::EQ_FOR_NULL:
+ return can_build_null_safe_equal_predicate(request, root_type, expr);
+ case TExprOpcode::FILTER_IN:
+ case TExprOpcode::FILTER_NOT_IN:
+ return expr->node_type() != TExprNodeType::NULL_AWARE_IN_PRED &&
+ can_build_in_predicate(request, root_type, expr);
+ case TExprOpcode::INVALID_OPCODE:
+ return expr->node_type() == TExprNodeType::FUNCTION_CALL &&
+ (expr->fn().name.function_name == "is_null_pred" ||
+ expr->fn().name.function_name == "is_not_null_pred") &&
+ can_build_is_null_predicate(request, root_type, expr);
+ default:
+ return false;
+ }
+}
+
+void build_less_than(const OrcSargColumn& column, const ::orc::Literal&
literal,
+ std::unique_ptr<::orc::SearchArgumentBuilder>& builder) {
+ builder->lessThan(column.column_id, column.predicate_type, literal);
+}
+
+void build_less_than(const OrcSargComparison& comparison,
+ std::unique_ptr<::orc::SearchArgumentBuilder>& builder) {
+ build_less_than(comparison.column, comparison.literal, builder);
+}
+
+void build_less_than_equals(const OrcSargColumn& column, const ::orc::Literal&
literal,
+ std::unique_ptr<::orc::SearchArgumentBuilder>&
builder) {
+ builder->lessThanEquals(column.column_id, column.predicate_type, literal);
+}
+
+void build_less_than_equals(const OrcSargComparison& comparison,
+ std::unique_ptr<::orc::SearchArgumentBuilder>&
builder) {
+ build_less_than_equals(comparison.column, comparison.literal, builder);
+}
+
+void build_equals(const OrcSargColumn& column, const ::orc::Literal& literal,
+ std::unique_ptr<::orc::SearchArgumentBuilder>& builder) {
+ builder->equals(column.column_id, column.predicate_type, literal);
+}
+
+void build_equals(const OrcSargComparison& comparison,
+ std::unique_ptr<::orc::SearchArgumentBuilder>& builder) {
+ build_equals(comparison.column, comparison.literal, builder);
+}
+
+void build_comparison_predicate(const format::FileScanRequest& request,
+ const ::orc::Type& root_type, const VExprSPtr&
expr,
+ std::unique_ptr<::orc::SearchArgumentBuilder>&
builder) {
+ const auto comparison = *sarg_comparison_for_expr(request, root_type,
expr);
+ switch (comparison.normalized_op) {
+ case TExprOpcode::GE:
+ builder->startNot();
+ build_less_than(comparison, builder);
+ builder->end();
+ return;
+ case TExprOpcode::GT:
+ builder->startNot();
+ build_less_than_equals(comparison, builder);
+ builder->end();
+ return;
+ case TExprOpcode::LE:
+ build_less_than_equals(comparison, builder);
+ return;
+ case TExprOpcode::LT:
+ build_less_than(comparison, builder);
+ return;
+ case TExprOpcode::EQ:
+ build_equals(comparison, builder);
+ return;
+ case TExprOpcode::NE:
+ builder->startNot();
+ build_equals(comparison, builder);
+ builder->end();
+ return;
+ default:
+ DORIS_CHECK(false) << "Unsupported normalized ORC SARG comparison op "
+ << comparison.normalized_op;
+ }
+}
+
+bool build_column_comparison_predicate(const OrcSargColumn& column,
+ const ColumnPredicate& predicate,
+
std::unique_ptr<::orc::SearchArgumentBuilder>& builder) {
+ if (!PredicateTypeTraits::is_comparison(predicate.type())) {
+ return false;
+ }
+ const auto* literal_provider = dynamic_cast<const
ColumnPredicateLiteralProvider*>(&predicate);
+ if (literal_provider == nullptr) {
+ return false;
+ }
+ const auto field = literal_provider->predicate_value();
+ if (!field.has_value()) {
+ return false;
+ }
+ const auto literal =
+ make_orc_literal(column, *field,
literal_provider->predicate_literal_type_info());
+ if (!literal.has_value()) {
+ return false;
+ }
+
+ switch (predicate.type()) {
+ case PredicateType::GE:
+ builder->startNot();
+ build_less_than(column, *literal, builder);
+ builder->end();
+ return true;
+ case PredicateType::GT:
+ builder->startNot();
+ build_less_than_equals(column, *literal, builder);
+ builder->end();
+ return true;
+ case PredicateType::LE:
+ build_less_than_equals(column, *literal, builder);
+ return true;
+ case PredicateType::LT:
+ build_less_than(column, *literal, builder);
+ return true;
+ case PredicateType::EQ:
+ build_equals(column, *literal, builder);
+ return true;
+ case PredicateType::NE:
+ builder->startNot();
+ build_equals(column, *literal, builder);
+ builder->end();
+ return true;
+ default:
+ return false;
+ }
+}
+
+void build_in_predicate(const format::FileScanRequest& request, const
::orc::Type& root_type,
+ const VExprSPtr& expr,
+ std::unique_ptr<::orc::SearchArgumentBuilder>&
builder) {
+ const auto sarg_column =
+ *sarg_column_for_slot_or_safe_cast(request, root_type,
expr->children()[0]);
+ auto literals = *make_in_literals_for_sarg(sarg_column,
expr->children()[0], expr->children());
+ DORIS_CHECK(!literals.empty());
+ if (literals.size() == 1) {
+ builder->equals(sarg_column.column_id, sarg_column.predicate_type,
literals.front());
+ return;
+ }
+ builder->in(sarg_column.column_id, sarg_column.predicate_type, literals);
+}
+
+bool build_column_in_predicate(const OrcSargColumn& column, const
ColumnPredicate& predicate,
+ std::unique_ptr<::orc::SearchArgumentBuilder>&
builder) {
+ if (!PredicateTypeTraits::is_list(predicate.type())) {
+ return false;
+ }
+ const auto* literal_provider = dynamic_cast<const
ColumnPredicateLiteralProvider*>(&predicate);
+ if (literal_provider == nullptr) {
+ return false;
+ }
+
+ std::vector<Field> fields;
+ if (!literal_provider->predicate_values(&fields)) {
+ return false;
+ }
+ std::vector<::orc::Literal> literals;
+ literals.reserve(fields.size());
+ for (const auto& field : fields) {
+ auto literal =
+ make_orc_literal(column, field,
literal_provider->predicate_literal_type_info());
+ if (!literal.has_value()) {
+ return false;
+ }
+ literals.push_back(*literal);
+ }
+ if (literals.empty()) {
+ return false;
+ }
+
+ if (predicate.type() == PredicateType::NOT_IN_LIST) {
+ builder->startNot();
+ }
+ if (literals.size() == 1) {
+ build_equals(column, literals.front(), builder);
+ } else {
+ builder->in(column.column_id, column.predicate_type, literals);
+ }
+ if (predicate.type() == PredicateType::NOT_IN_LIST) {
+ builder->end();
+ }
+ return true;
+}
+
+void build_is_null(const format::FileScanRequest& request, const ::orc::Type&
root_type,
+ const VExprSPtr& expr,
std::unique_ptr<::orc::SearchArgumentBuilder>& builder) {
+ const auto sarg_column =
+ *sarg_column_for_slot_or_safe_cast(request, root_type,
expr->children()[0]);
+ builder->isNull(sarg_column.column_id, sarg_column.predicate_type);
+}
+
+void build_null_safe_equal(const format::FileScanRequest& request, const
::orc::Type& root_type,
+ const VExprSPtr& expr,
+ std::unique_ptr<::orc::SearchArgumentBuilder>&
builder) {
+ const auto sarg_column = sarg_column_for_null_safe_equal_null(request,
root_type, expr);
+ if (sarg_column.has_value()) {
+ builder->isNull(sarg_column->column_id, sarg_column->predicate_type);
+ return;
+ }
+ const auto comparison =
*sarg_comparison_for_null_safe_equal_literal(request, root_type, expr);
+ build_equals(comparison, builder);
+}
+
+bool build_column_null_predicate(const OrcSargColumn& column, const
ColumnPredicate& predicate,
+
std::unique_ptr<::orc::SearchArgumentBuilder>& builder) {
+ switch (predicate.type()) {
+ case PredicateType::IS_NULL:
+ builder->isNull(column.column_id, column.predicate_type);
+ return true;
+ case PredicateType::IS_NOT_NULL:
+ builder->startNot();
+ builder->isNull(column.column_id, column.predicate_type);
+ builder->end();
+ return true;
+ default:
+ return false;
+ }
+}
+
+bool build_column_predicate_search_argument(
+ const OrcSargColumn& column, const ColumnPredicate& predicate,
+ std::unique_ptr<::orc::SearchArgumentBuilder>& builder) {
+ if (predicate.opposite()) {
+ return false;
+ }
+ if (build_column_comparison_predicate(column, predicate, builder)) {
+ return true;
+ }
+ if (build_column_in_predicate(column, predicate, builder)) {
+ return true;
+ }
+ return build_column_null_predicate(column, predicate, builder);
+}
+
+bool build_search_argument(const ::orc::Type& root_type,
+ const format::FileColumnPredicateFilter&
column_filter,
+ std::unique_ptr<::orc::SearchArgumentBuilder>&
builder) {
+ const auto column = sarg_column_for_column_filter(root_type,
column_filter);
+ if (!column.has_value()) {
+ return false;
+ }
+
+ bool has_pushdown = false;
+ for (const auto& predicate : column_filter.predicates) {
+ if (predicate == nullptr) {
+ continue;
+ }
+ has_pushdown = build_column_predicate_search_argument(*column,
*predicate, builder) ||
+ has_pushdown;
+ }
+ return has_pushdown;
+}
+
+bool build_search_argument(const format::FileScanRequest& request, const
::orc::Type& root_type,
+ const VExprSPtr& expr,
+ std::unique_ptr<::orc::SearchArgumentBuilder>&
builder) {
+ if (!can_build_search_argument(request, root_type, expr)) {
+ return false;
+ }
+
+ switch (expr->op()) {
+ case TExprOpcode::COMPOUND_AND:
+ builder->startAnd();
+ for (const auto& child : expr->children()) {
+ static_cast<void>(build_search_argument(request, root_type, child,
builder));
+ }
+ builder->end();
+ return true;
+ case TExprOpcode::COMPOUND_OR:
+ builder->startOr();
+ for (const auto& child : expr->children()) {
+ const auto built = build_search_argument(request, root_type,
child, builder);
+ DORIS_CHECK(built);
+ }
+ builder->end();
+ return true;
+ case TExprOpcode::COMPOUND_NOT:
+ builder->startNot();
+ DORIS_CHECK(build_search_argument(request, root_type,
expr->children()[0], builder));
+ builder->end();
+ return true;
+ case TExprOpcode::GE:
+ case TExprOpcode::GT:
+ case TExprOpcode::LE:
+ case TExprOpcode::LT:
+ case TExprOpcode::EQ:
+ case TExprOpcode::NE:
+ build_comparison_predicate(request, root_type, expr, builder);
+ return true;
+ case TExprOpcode::EQ_FOR_NULL:
+ build_null_safe_equal(request, root_type, expr, builder);
+ return true;
+ case TExprOpcode::FILTER_IN:
+ build_in_predicate(request, root_type, expr, builder);
+ return true;
+ case TExprOpcode::FILTER_NOT_IN:
+ builder->startNot();
+ build_in_predicate(request, root_type, expr, builder);
+ builder->end();
+ return true;
+ case TExprOpcode::INVALID_OPCODE:
+ if (expr->fn().name.function_name == "is_null_pred") {
+ build_is_null(request, root_type, expr, builder);
+ return true;
+ }
+ if (expr->fn().name.function_name == "is_not_null_pred") {
+ builder->startNot();
+ build_is_null(request, root_type, expr, builder);
+ builder->end();
+ return true;
+ }
+ return false;
+ default:
+ return false;
+ }
+}
+
+} // namespace
+
+class OrcReader::OrcFilterImpl final : public ::orc::ORCFilter {
+public:
+ explicit OrcFilterImpl(OrcReader* reader) : _reader(reader) {}
+
+ void filter(::orc::ColumnVectorBatch& data, uint16_t* sel, uint16_t size,
+ void* arg) const override {
+ THROW_IF_ERROR(_reader->_filter_orc_batch(data, sel, size, arg));
+ }
+
+private:
+ OrcReader* _reader = nullptr;
+};
+
+// Per-open mutable ORC state. close() publishes counters first, then resets
this
+// object so the reader can be opened again without carrying stale scan state.
+//
=============================================================================
+// OrcReaderScanState —— 所有运行时可变状态都装这一个 struct 里。
+//
=============================================================================
+// 设计原则：close() 时直接 _state = std::make_unique<OrcReaderScanState>()
+// 就能把状态彻底重置，避免逐个字段 reset 漏掉。
+struct OrcReaderScanState {
+ // pruning 后剩下的 stripe 可能不连续（[1, 3, 4]），按连续段合并成多个 range
+ // 每个 range 对应一次 RowReader 重建（ORC 库不支持单个 RowReader 内 seek 跳过 stripe）
+ struct StripeRange {
+ uint64_t first_stripe = 0;
+ uint64_t last_stripe = 0;
+ uint64_t offset = 0; // 本 range 起始字节
+ uint64_t length = 0; // 本 range 字节长度
+ };
+
+ // ===== ORC 库对象 =====
+ std::unique_ptr<::orc::Reader> reader; // 文件级 reader (init 时建)
+ const ::orc::Type* root_type = nullptr; // 文件 schema 根 (root_type 必须是
STRUCT)
+ ::orc::ReaderMetrics reader_metrics;
+ ::orc::RowReaderOptions row_reader_options; // projection + filter +
SARG + stripe range
+ std::unique_ptr<::orc::RowReader> row_reader; // 行级 reader (open / 切
range 时建)
+ const ::orc::Type* selected_type = nullptr; // projection 后的 selected
schema
+ std::unique_ptr<::orc::ColumnVectorBatch> batch; // 批次缓冲（next() 时复用）
+
+ // ===== 投影信息 =====
+ std::vector<format::LocalColumnId> read_columns; // 要读的列（去重排序后）
+ // file_column_id → 在 ORC selected_type 里的 child index（projection 后顺序可能变）
+ std::map<format::LocalColumnId, size_t> column_to_selected_batch_index;
+
+ // ===== 当前批次状态 =====
+ uint64_t current_batch_first_row = 0; // 本批起始物理行号（虚拟列要用）
+
+ // ===== ORC lazy 路径状态（filter callback 写、get_block 收尾用）=====
+ std::vector<size_t> orc_lazy_selected_rows; // 命中行的 batch 内偏移
+ size_t orc_lazy_input_rows = 0; // callback 拿到的原始批次大小
+ bool orc_lazy_read_enabled = false; // 是否走 ORC lazy 路径（open 时定）
+ bool orc_lazy_selection_valid = false; // selection 是否有效（防止状态错位）
+
+ // ===== Stripe pruning 状态 =====
+ std::vector<StripeRange> selected_stripe_ranges; // pruning 后剩下的连续段
+ size_t current_stripe_range = 0; // 当前在读哪一段
+ bool stripe_pruning_applied = false; // pruning 是否实际生效（决定能否切段）
+
+ bool row_reader_created = false; // row_reader 已创建（open 后或切段后）
+};
+
+OrcReader::OrcReader(std::shared_ptr<io::FileSystemProperties>&
system_properties,
+ std::unique_ptr<io::FileDescription>& file_description,
+ std::shared_ptr<io::IOContext> io_ctx, RuntimeProfile*
profile)
+ : FileReader(system_properties, file_description, io_ctx, profile) {}
+
+OrcReader::~OrcReader() = default;
+
+// Expose ORC pruning and lazy-read statistics in RuntimeProfile. These
counters
+// are the quickest way to confirm whether SARG/stripe pruning actually fired.
+void OrcReader::_init_profile() {
+ if (_profile == nullptr) {
+ return;
+ }
+
+ static const char* orc_profile = "OrcReader";
+ ADD_TIMER_WITH_LEVEL(_profile, orc_profile, 1);
+ _orc_profile.filtered_row_groups = ADD_CHILD_COUNTER_WITH_LEVEL(_profile,
"RowGroupsFiltered",
+
TUnit::UNIT, orc_profile, 1);
+ _orc_profile.filtered_row_groups_by_min_max = ADD_CHILD_COUNTER_WITH_LEVEL(
+ _profile, "RowGroupsFilteredByMinMax", TUnit::UNIT, orc_profile,
1);
+ _orc_profile.read_row_groups =
+ ADD_CHILD_COUNTER_WITH_LEVEL(_profile, "RowGroupsReadNum",
TUnit::UNIT, orc_profile, 1);
+ _orc_profile.filtered_group_rows = ADD_CHILD_COUNTER_WITH_LEVEL(_profile,
"FilteredRowsByGroup",
+
TUnit::UNIT, orc_profile, 1);
+ _orc_profile.lazy_read_filtered_rows = ADD_CHILD_COUNTER_WITH_LEVEL(
+ _profile, "FilteredRowsByLazyRead", TUnit::UNIT, orc_profile, 1);
+ _orc_profile.orc_lazy_read_filtered_rows = ADD_CHILD_COUNTER_WITH_LEVEL(
+ _profile, "FilteredRowsByOrcLazyRead", TUnit::UNIT, orc_profile,
1);
+ _orc_profile.filtered_bytes =
+ ADD_CHILD_COUNTER_WITH_LEVEL(_profile, "FilteredBytes",
TUnit::BYTES, orc_profile, 1);
+ _orc_profile.open_file_num =
+ ADD_CHILD_COUNTER_WITH_LEVEL(_profile, "FileNum", TUnit::UNIT,
orc_profile, 1);
+}
+
+void OrcReader::_collect_profile() const {
+ if (_profile == nullptr) {
+ return;
+ }
+
+ COUNTER_UPDATE(_orc_profile.filtered_row_groups,
_reader_statistics.filtered_row_groups);
+ COUNTER_UPDATE(_orc_profile.filtered_row_groups_by_min_max,
+ _reader_statistics.filtered_row_groups_by_min_max);
+ COUNTER_UPDATE(_orc_profile.read_row_groups,
_reader_statistics.read_row_groups);
+ COUNTER_UPDATE(_orc_profile.filtered_group_rows,
_reader_statistics.filtered_group_rows);
+ COUNTER_UPDATE(_orc_profile.lazy_read_filtered_rows,
+ _reader_statistics.lazy_read_filtered_rows);
+ COUNTER_UPDATE(_orc_profile.orc_lazy_read_filtered_rows,
+ _reader_statistics.orc_lazy_read_filtered_rows);
+ COUNTER_UPDATE(_orc_profile.filtered_bytes,
_reader_statistics.filtered_bytes);
+ COUNTER_UPDATE(_orc_profile.open_file_num,
_reader_statistics.open_file_num);
+}
+
+format::ColumnDefinition OrcReader::row_position_column_definition() {
+ format::ColumnDefinition field;
+ field.identifier = Field::create_field<TYPE_INT>(ROW_POSITION_COLUMN_ID);
+ field.local_id = ROW_POSITION_COLUMN_ID;
+ field.name = ROW_POSITION_COLUMN_NAME;
+ field.type = std::make_shared<DataTypeInt64>();
+ field.column_type = format::ColumnType::ROW_NUMBER;
+ return field;
+}
+
+// 阶段 1：init —— 打开 ORC 文件，建 ORC Reader 对象
+//
+// 关键点：
+// 1. 用 ExecEnv 全局 memory pool —— ORC 库的内存分配走 Doris 内存追踪
+// 2. DorisOrcInputStream 是 ORC IO 适配层，把 Doris io::FileReader 包成
+// ORC 库要的 ::orc::InputStream 接口
+// 3. 这一步只读 file footer（schema + stripe statistics），还没读数据
+Status OrcReader::init(RuntimeState* state) {
+ RETURN_IF_ERROR(format::FileReader::init(state)); // 基类会调 _init_profile()
+ _state = std::make_unique<OrcReaderScanState>();
+
+ ::orc::ReaderOptions options;
+ options.setMemoryPool(*ExecEnv::GetInstance()->orc_memory_pool());
+ options.setReaderMetrics(&_state->reader_metrics);
+
+ auto input_stream =
std::make_unique<DorisOrcInputStream>(_file_description->path,
+
_tracing_file_reader, _io_ctx.get());
+ try {
+ _state->reader = ::orc::createReader(std::move(input_stream), options);
+ _state->root_type = &_state->reader->getType();
+ } catch (const std::exception& e) {
+ return Status::InternalError("Failed to open ORC file {}: {}",
_file_description->path,
+ e.what());
+ }
+ return Status::OK();
+}
+
+DataTypePtr OrcReader::_convert_to_doris_type(const ::orc::Type& type) const {
+ DataTypePtr data_type;
+ switch (type.getKind()) {
+ case ::orc::TypeKind::BOOLEAN:
+ data_type = std::make_shared<DataTypeUInt8>();
+ break;
+ case ::orc::TypeKind::BYTE:
+ data_type = std::make_shared<DataTypeInt8>();
+ break;
+ case ::orc::TypeKind::SHORT:
+ data_type = std::make_shared<DataTypeInt16>();
+ break;
+ case ::orc::TypeKind::INT:
+ data_type = std::make_shared<DataTypeInt32>();
+ break;
+ case ::orc::TypeKind::LONG:
+ data_type = std::make_shared<DataTypeInt64>();
+ break;
+ case ::orc::TypeKind::FLOAT:
+ data_type = std::make_shared<DataTypeFloat32>();
+ break;
+ case ::orc::TypeKind::DOUBLE:
+ data_type = std::make_shared<DataTypeFloat64>();
+ break;
+ case ::orc::TypeKind::STRING:
+ case ::orc::TypeKind::BINARY:
+ data_type = std::make_shared<DataTypeString>();
+ break;
+ case ::orc::TypeKind::VARCHAR:
+ data_type =
std::make_shared<DataTypeString>(cast_set<int>(type.getMaximumLength()),
+
PrimitiveType::TYPE_VARCHAR);
+ break;
+ case ::orc::TypeKind::CHAR:
+ data_type =
std::make_shared<DataTypeString>(cast_set<int>(type.getMaximumLength()),
+ PrimitiveType::TYPE_CHAR);
+ break;
+ case ::orc::TypeKind::DATE:
+ data_type = std::make_shared<DataTypeDateV2>();
+ break;
+ case ::orc::TypeKind::TIMESTAMP:
+ case ::orc::TypeKind::TIMESTAMP_INSTANT:
+ data_type = std::make_shared<DataTypeDateTimeV2>(6);
+ break;
+ case ::orc::TypeKind::DECIMAL:
+ data_type = std::make_shared<DataTypeDecimal<TYPE_DECIMAL128I>>(
+ type.getPrecision() == 0 ? DECIMAL_PRECISION_FOR_HIVE11
+ : cast_set<int>(type.getPrecision()),
+ type.getPrecision() == 0 ? DECIMAL_SCALE_FOR_HIVE11
+ : cast_set<int>(type.getScale()));
+ break;
+ case ::orc::TypeKind::LIST:
+ data_type = _convert_list_to_doris_type(type);
+ break;
+ case ::orc::TypeKind::MAP:
+ data_type = _convert_map_to_doris_type(type);
+ break;
+ case ::orc::TypeKind::STRUCT:
+ data_type = _convert_struct_to_doris_type(type);
+ break;
+ default:
+ throw doris::Exception(
+ Status::NotSupported("ORC type {} is not supported by new ORC
reader",
+ static_cast<int>(type.getKind())));
+ }
+ return make_nullable(data_type);
+}
+
+DataTypePtr OrcReader::_convert_list_to_doris_type(const ::orc::Type& type)
const {
+ DORIS_CHECK(type.getKind() == ::orc::TypeKind::LIST);
+ DORIS_CHECK(type.getSubtypeCount() == 1);
+ const auto* element_type = type.getSubtype(0);
+ DORIS_CHECK(element_type != nullptr);
+ return
std::make_shared<DataTypeArray>(_convert_to_doris_type(*element_type));
+}
+
+DataTypePtr OrcReader::_convert_map_to_doris_type(const ::orc::Type& type)
const {
+ DORIS_CHECK(type.getKind() == ::orc::TypeKind::MAP);
+ DORIS_CHECK(type.getSubtypeCount() == 2);
+ const auto* key_type = type.getSubtype(0);
+ const auto* value_type = type.getSubtype(1);
+ DORIS_CHECK(key_type != nullptr);
+ DORIS_CHECK(value_type != nullptr);
+ return std::make_shared<DataTypeMap>(_convert_to_doris_type(*key_type),
+ _convert_to_doris_type(*value_type));
+}
+
+DataTypePtr OrcReader::_convert_struct_to_doris_type(const ::orc::Type& type)
const {
+ DORIS_CHECK(type.getKind() == ::orc::TypeKind::STRUCT);
+ DataTypes child_types;
+ Strings child_names;
+ child_types.reserve(type.getSubtypeCount());
+ child_names.reserve(type.getSubtypeCount());
+ for (uint64_t child_idx = 0; child_idx < type.getSubtypeCount();
++child_idx) {
+ const auto* child_type = type.getSubtype(child_idx);
+ DORIS_CHECK(child_type != nullptr);
+ child_types.push_back(_convert_to_doris_type(*child_type));
+ child_names.push_back(type.getFieldName(child_idx));
+ }
+ return std::make_shared<DataTypeStruct>(child_types, child_names);
+}
+
+Status OrcReader::_fill_schema_field(const ::orc::Type& type, int32_t field_id,
+ const std::string& field_name,
+ format::ColumnDefinition* const field)
const {
+ if (field == nullptr) {
+ return Status::InvalidArgument("schema field is null");
+ }
+ field->local_id = field_id;
+ field->name = field_name;
+ field->column_type = format::ColumnType::DATA_COLUMN;
+ if (type.hasAttributeKey(ORC_ICEBERG_ID_ATTRIBUTE)) {
+ const auto iceberg_id =
type.getAttributeValue(ORC_ICEBERG_ID_ATTRIBUTE);
+ int32_t parsed_id = 0;
+ const auto* begin = iceberg_id.data();
+ const auto* end = begin + iceberg_id.size();
+ const auto [ptr, ec] = std::from_chars(begin, end, parsed_id);
+ if (ec != std::errc() || ptr != end) {
+ return Status::InvalidArgument("Invalid ORC Iceberg field id '{}'
for column {}",
+ iceberg_id, field_name);
+ }
+ field->identifier = Field::create_field<TYPE_INT>(parsed_id);
+ }
+ try {
+ field->type = _convert_to_doris_type(type);
+ } catch (const doris::Exception& e) {
+ return e.to_status();
+ }
+ field->children.clear();
+ switch (type.getKind()) {
+ case ::orc::TypeKind::STRUCT:
+ return _fill_struct_schema_children(type, field);
+ case ::orc::TypeKind::LIST:
+ return _fill_list_schema_children(type, field);
+ case ::orc::TypeKind::MAP:
+ return _fill_map_schema_children(type, field);
+ default:
+ break;
+ }
+ return Status::OK();
+}
+
+Status OrcReader::_fill_struct_schema_children(const ::orc::Type& type,
+ format::ColumnDefinition* const
field) const {
+ DORIS_CHECK(type.getKind() == ::orc::TypeKind::STRUCT);
+ field->children.reserve(type.getSubtypeCount());
+ for (uint64_t child_idx = 0; child_idx < type.getSubtypeCount();
++child_idx) {
+ const auto* child_type = type.getSubtype(child_idx);
+ DORIS_CHECK(child_type != nullptr);
+ const auto child_name = type.getFieldName(child_idx);
+ format::ColumnDefinition child_field;
+ RETURN_IF_ERROR(_fill_schema_field(*child_type,
static_cast<int32_t>(child_idx), child_name,
+ &child_field));
+ field->children.push_back(std::move(child_field));
+ }
+ return Status::OK();
+}
+
+Status OrcReader::_fill_list_schema_children(const ::orc::Type& type,
+ format::ColumnDefinition* const
field) const {
+ DORIS_CHECK(type.getKind() == ::orc::TypeKind::LIST);
+ DORIS_CHECK(type.getSubtypeCount() == 1);
+ const auto* element_type = type.getSubtype(0);
+ DORIS_CHECK(element_type != nullptr);
+
+ format::ColumnDefinition element_field;
+ RETURN_IF_ERROR(_fill_schema_field(*element_type, 0,
ORC_LIST_ELEMENT_NAME, &element_field));
+ field->children.push_back(std::move(element_field));
+ return Status::OK();
+}
+
+Status OrcReader::_fill_map_schema_children(const ::orc::Type& type,
+ format::ColumnDefinition* const
field) const {
+ DORIS_CHECK(type.getKind() == ::orc::TypeKind::MAP);
+ DORIS_CHECK(type.getSubtypeCount() == 2);
+ const auto* key_type = type.getSubtype(0);
+ const auto* value_type = type.getSubtype(1);
+ DORIS_CHECK(key_type != nullptr);
+ DORIS_CHECK(value_type != nullptr);
+
+ format::ColumnDefinition key_field;
+ RETURN_IF_ERROR(_fill_schema_field(*key_type, 0, ORC_MAP_KEY_NAME,
&key_field));
+ format::ColumnDefinition value_field;
+ RETURN_IF_ERROR(_fill_schema_field(*value_type, 1, ORC_MAP_VALUE_NAME,
&value_field));
+
+ format::ColumnDefinition entry_field;
+ entry_field.local_id = 0;
+ entry_field.name = ORC_MAP_ENTRY_NAME;
+ entry_field.column_type = format::ColumnType::DATA_COLUMN;
+ entry_field.children.push_back(std::move(key_field));
+ entry_field.children.push_back(std::move(value_field));
+
+ DataTypes entry_child_types;
+ Strings entry_child_names;
+ entry_child_types.reserve(entry_field.children.size());
+ entry_child_names.reserve(entry_field.children.size());
+ for (const auto& child : entry_field.children) {
+ entry_child_types.push_back(child.type);
+ entry_child_names.push_back(child.name);
+ }
+ entry_field.type = std::make_shared<DataTypeStruct>(entry_child_types,
entry_child_names);
+ field->children.push_back(std::move(entry_field));
+ return Status::OK();
+}
+
+// 阶段 2：get_schema —— ORC schema → Doris ColumnDefinition
+//
+// 注意：
+// - ORC root 必须是 STRUCT（这是 ORC 标准约定）
+// - 每个子字段递归走 _fill_schema_field，处理 STRUCT/LIST/MAP 嵌套
+// - 返回的 type 永远是 nullable（文件层不假设 NOT NULL，table 层判断）
+// - Iceberg 的 field id 通过 ORC type attribute "iceberg.id" 透传到
+// ColumnDefinition::identifier，给 equality delete 按 field id 匹配列用
+Status OrcReader::get_schema(std::vector<format::ColumnDefinition>* const
file_schema) const {
+ if (file_schema == nullptr) {
+ return Status::InvalidArgument("file_schema is null");
+ }
+ if (_state == nullptr || _state->root_type == nullptr) {
+ return Status::Uninitialized("OrcReader is not open");
+ }
+ if (_state->root_type->getKind() != ::orc::TypeKind::STRUCT) {
+ return Status::NotSupported("ORC reader only supports top-level struct
schema");
+ }
+ file_schema->clear();
+ file_schema->reserve(_state->root_type->getSubtypeCount());
+ for (uint64_t child_idx = 0; child_idx <
_state->root_type->getSubtypeCount(); ++child_idx) {
+ const auto* child_type = _state->root_type->getSubtype(child_idx);
+ DORIS_CHECK(child_type != nullptr);
+ const auto child_name = _state->root_type->getFieldName(child_idx);
+ format::ColumnDefinition field;
+ RETURN_IF_ERROR(_fill_schema_field(*child_type,
static_cast<int32_t>(child_idx), child_name,
+ &field));
+ file_schema->push_back(std::move(field));
+ }
+ return Status::OK();
+}
+
+// 阶段 3：open(FileScanRequest) —— 最复杂的一步
+//
+// 8 步流程：
+// 1. fallback 计算 local_positions（如果上层没传）
+// 2. 收集 read_columns，去重排序
+// 3. _validate_*：护栏检查（早期发现非法 request）
+// 4. _can_enable_orc_lazy_read：决定是否启用 ORC 库 lazy callback
+// 5. _configure_row_reader_projection：simple include vs nested includeTypes
+// 6. _init_search_argument_from_conjuncts：4 类 filter → ORC SARG
+// 7. _select_stripe_ranges_by_statistics：双闸 stripe pruning
+// 8. _create_row_reader：实际创建 ORC RowReader（含 lazy callback 注入）
+Status OrcReader::open(std::shared_ptr<format::FileScanRequest> request) {
+ if (_state == nullptr || _state->reader == nullptr || _state->root_type ==
nullptr) {
+ return Status::Uninitialized("OrcReader is not open");
+ }
+ RETURN_IF_ERROR(format::FileReader::open(std::move(request)));
+
+ // 步骤 1：上层没传 local_positions 就按 predicate + non_predicate 顺序自动派号
+ // 上层 TableReader 通常会主动填，这是 fallback 兜底
+ if (_request->local_positions.empty()) {
+ size_t next_position = 0;
+ for (const auto& projection : _request->predicate_columns) {
+ if (_request->local_positions
+ .emplace(projection.column_id(),
format::LocalIndex(next_position))
+ .second) {
+ ++next_position;
+ }
+ }
+ for (const auto& projection : _request->non_predicate_columns) {
+ if (_request->local_positions
+ .emplace(projection.column_id(),
format::LocalIndex(next_position))
+ .second) {
+ ++next_position;
+ }
+ }
+ }
+
+ // 步骤 2：收集要读的列（predicate + non_predicate 合并去重）
+ _state->read_columns.clear();
+ _state->read_columns.reserve(_request->predicate_columns.size() +
+ _request->non_predicate_columns.size());
+ for (const auto& projection : _request->predicate_columns) {
+ _state->read_columns.push_back(projection.column_id());
+ }
+ for (const auto& projection : _request->non_predicate_columns) {
+ _state->read_columns.push_back(projection.column_id());
+ }
+ std::sort(_state->read_columns.begin(), _state->read_columns.end());
+ _state->read_columns.erase(
+ std::unique(_state->read_columns.begin(),
_state->read_columns.end()),
+ _state->read_columns.end());
+
+ // 步骤 3：护栏（防御性检查，不合法 request 早期失败）
+ RETURN_IF_ERROR(_validate_column_predicate_filters());
+ RETURN_IF_ERROR(_validate_reader_expression_map());
+
+ // 步骤 4：是否启用 ORC lazy（ORC 库内部 column reader 跳行）
+ _state->orc_lazy_read_enabled = _can_enable_orc_lazy_read();
+
+ // 步骤 5：投影配置
+ RETURN_IF_ERROR(_configure_row_reader_projection());
+ _state->row_reader_options.setTimezoneName("UTC"); // 统一 UTC 解读
TIMESTAMP_INSTANT
+
_state->row_reader_options.setEnableLazyDecoding(_state->orc_lazy_read_enabled);
+ _state->row_reader_options.setUseTightNumericVector(false);
+
+ // 步骤 6：SARG 构造（conjuncts / reader_expression_map /
column_predicate_filters → ORC SARG）
+ RETURN_IF_ERROR(_init_search_argument_from_conjuncts());
Review Comment:
这里已收敛边界：TableColumnMapper 仍负责 table schema 到 file-local FileScanRequest 的
localize；ORC SearchArgument 构造依赖 ORC C++ Type/type id，所以没有放进通用 mapper，而是拆到了
format_v2/orc/orc_search_argument.{h,cpp}，OrcReader 只调用这个 ORC 专属 lowering
helper。
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
