be/src/format_v2/table_reader.h

Source
// 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 <bvar/status.h>

#include <algorithm>
#include <exception>
#include <map>
#include <memory>
#include <optional>
#include <span>
#include <string>
#include <string_view>
#include <unordered_map>
#include <utility>
#include <vector>

#include "common/cast_set.h"
#include "common/exception.h"
#include "common/logging.h"
#include "common/status.h"
#include "core/assert_cast.h"
#include "core/block/block.h"
#include "core/column/column_array.h"
#include "core/column/column_const.h"
#include "core/column/column_map.h"
#include "core/column/column_nullable.h"
#include "core/column/column_struct.h"
#include "core/column/column_vector.h"
#include "core/column/variant_v2/column_variant_v2.h"
#include "core/data_type/data_type.h"
#include "core/data_type/data_type_array.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/field.h"
#include "exec/common/stringop_substring.h"
#include "exprs/function/function_variant_element_v2.h"
#include "exprs/vexpr.h"
#include "exprs/vexpr_context.h"
#include "exprs/vexpr_fwd.h"
#include "exprs/vslot_ref.h"
#include "format/table/deletion_vector.h"
#include "format_v2/column_data.h"
#include "format_v2/column_mapper.h"
#include "format_v2/expr/cast.h"
#include "format_v2/expr/delete_predicate.h"
#include "format_v2/file_reader.h"
#include "format_v2/parquet/reader/column_reader.h"
#include "format_v2/schema_projection.h"
#include "gen_cpp/PlanNodes_types.h"
#include "io/io_common.h"
#include "runtime/descriptors.h"
#include "storage/segment/condition_cache.h"

namespace doris {
class Block;
struct DeleteFileDesc;
class RuntimeState;
} // namespace doris

namespace doris::format {

using DeleteRows = std::vector<int64_t>;

// Row-level predicates on table/global schema. They are rewritten to file-local expressions when
// possible, and remain the source of row-level filtering after localization.
struct TableFilter {
    VExprContextSPtr conjunct;
    std::vector<GlobalIndex> global_indices;
};

struct ScanTask {
    virtual ~ScanTask() = default;

    std::unique_ptr<io::FileDescription> data_file;
};

struct ProjectedColumnBuildContext {
    const TFileScanRangeParams* scan_params = nullptr;
    const TFileRangeDesc* range = nullptr;
    RuntimeState* runtime_state = nullptr;
    const SlotDescriptor* slot_desc = nullptr;
    std::optional<ColumnDefinition> schema_column = std::nullopt;
    size_t next_file_column_idx = 0;
};

struct ReadProfile {
    RuntimeProfile::Counter* total_timer = nullptr;
    RuntimeProfile::Counter* init_timer = nullptr;
    RuntimeProfile::Counter* num_delete_files = nullptr;
    RuntimeProfile::Counter* num_delete_rows = nullptr;
    RuntimeProfile::Counter* parse_delete_file_time = nullptr;
    RuntimeProfile::Counter* decoded_dv_cache_hit_count = nullptr;
    RuntimeProfile::Counter* decoded_dv_cache_miss_count = nullptr;
    RuntimeProfile::Counter* dv_file_cache_hit_count = nullptr;
    RuntimeProfile::Counter* dv_file_cache_miss_count = nullptr;
    RuntimeProfile::Counter* dv_file_cache_peer_read_count = nullptr;
    RuntimeProfile::Counter* exec_timer = nullptr;
    RuntimeProfile::Counter* prepare_split_timer = nullptr;
    RuntimeProfile::Counter* finalize_timer = nullptr;
    RuntimeProfile::Counter* create_reader_timer = nullptr;
    RuntimeProfile::Counter* pushdown_agg_timer = nullptr;
    RuntimeProfile::Counter* open_reader_timer = nullptr;
    RuntimeProfile::Counter* runtime_filter_partition_prune_timer = nullptr;
    RuntimeProfile::Counter* runtime_filter_partition_pruned_range_counter = nullptr;
    RuntimeProfile::Counter* close_timer = nullptr;
    RuntimeProfile::Counter* file_reader_total_timer = nullptr;
    RuntimeProfile::Counter* file_reader_init_timer = nullptr;
    RuntimeProfile::Counter* file_reader_schema_timer = nullptr;
    RuntimeProfile::Counter* file_reader_mapper_timer = nullptr;
    RuntimeProfile::Counter* file_reader_open_timer = nullptr;
    RuntimeProfile::Counter* file_reader_get_block_timer = nullptr;
    RuntimeProfile::Counter* file_reader_aggregate_timer = nullptr;
    RuntimeProfile::Counter* file_reader_close_timer = nullptr;
};

struct TableReadOptions {
    // Columns need to be read from file and output by table reader. They are all in table/global
    // schema semantics.
    const std::vector<ColumnDefinition> projected_columns;
    // All complex conjuncts from scan operator
    const VExprContextSPtrs conjuncts;
    // File format of the underlying data files, needed for reader initialization and reader-level
    // filter pushdown.
    const FileFormat format;
    TFileScanRangeParams* scan_params;
    std::shared_ptr<io::IOContext> io_ctx;
    RuntimeState* runtime_state;
    RuntimeProfile* scanner_profile;
    // File formats without complete self-describing metadata, such as CSV, Text, and JSON, need
    // the FE-planned physical file slots to build their file-local schema and deserialize values.
    const std::vector<SlotDescriptor*>* file_slot_descs = nullptr;
    // Push-down aggregate type.
    const TPushAggOp::type push_down_agg_type = TPushAggOp::type::NONE;
    // Table/global indices of explicit COUNT arguments. nullopt means an old FE did not send the
    // semantic argument field, while an explicit empty vector means COUNT(*)/COUNT(1). Keeping
    // those states separate prevents a rolling-upgrade plan from being reinterpreted by a new BE.
    const std::optional<std::vector<GlobalIndex>> push_down_count_columns = std::nullopt;
    // Initial digest of predicates available during scanner open. Scanner-driven splits override it
    // with SplitReadOptions::condition_cache_digest after collecting late-arrival runtime filters.
    // A zero digest disables condition cache.
    uint64_t condition_cache_digest = 0;
};

struct SplitReadOptions {
    // Split-level information for reader initialization, which may include file path, partition values, delete file info, etc. The content is table format specific and opaque to table reader base class; it's the responsibility of the concrete table reader implementation to parse necessary information for reader initialization and filter pushdown.
    std::map<std::string, Field> partition_values;
    // Latest scanner conjuncts rewritten to table/global column indices. Runtime filters may
    // arrive after TableReader::init(), so scanner-driven splits replace the initial snapshot.
    // nullopt preserves the initial snapshot for standalone TableReader callers.
    std::optional<VExprContextSPtrs> conjuncts = std::nullopt;
    // Independent clones used for partition pruning because evaluation prepares and opens them
    // against a synthetic partition block before the file reader opens its row-level conjuncts.
    VExprContextSPtrs partition_prune_conjuncts;
    // Table-level COUNT may emit one metadata-derived batch and resume on a later scheduler turn.
    // It is safe only after every runtime filter assigned to the scanner has arrived; otherwise a
    // filter could arrive after synthetic rows have already been returned and those rows cannot be
    // retracted. Standalone TableReader callers have no scanner runtime-filter lifecycle.
    bool all_runtime_filters_applied = true;
    // Digest for the exact scanner conjunct snapshot attached to this split. FileScannerV2 rebuilds
    // it after collecting late-arrival RFs, so different RF payloads cannot share a cache entry. A
    // zero value explicitly disables condition cache for this split.
    std::optional<uint64_t> condition_cache_digest;
    ShardedKVCache* cache = nullptr;
    TFileRangeDesc current_range;
    FileFormat current_split_format = FileFormat::PARQUET;
    std::optional<GlobalRowIdContext> global_rowid_context;
};

// Base class for table-level readers.
// This layer owns common table-level orchestration, such as split iteration, dynamic partition
// pruning, delete handling and conversion from file-local blocks to table-schema blocks. Concrete
// table-format readers only need to provide format-specific hooks for opening readers and parsing
// split metadata.
class TableReader {
public:
    virtual ~TableReader() = default;

    // Initialize common runtime options for the table reader. Subclasses may call this from their
    // own init(options); table-format schema and split metadata are provided later per split.
    virtual Status init(TableReadOptions&& options);

    // FileScannerV2 adjusts this before each get_block() using an adaptive bytes-per-row estimate.
    // Store it here as well as forwarding to the current reader so newly opened split readers start
    // with the latest predicted batch size.
    virtual void set_batch_size(size_t batch_size) {
        _batch_size = std::max<size_t>(1, batch_size);
        if (_data_reader.reader != nullptr) {
            _data_reader.reader->set_batch_size(_batch_size);
        }
    }

#ifdef BE_TEST
    size_t TEST_batch_size() const { return _batch_size; }
    void TEST_set_condition_cache_hit_count(int64_t hits) { _condition_cache_hit_count = hits; }
    bool TEST_current_data_file_is_immutable() const {
        DORIS_CHECK(_current_task != nullptr);
        DORIS_CHECK(_current_task->data_file != nullptr);
        DORIS_CHECK(_current_file_description.has_value());
        DORIS_CHECK(_current_task->data_file->is_immutable ==
                    _current_file_description->is_immutable);
        return _current_task->data_file->is_immutable;
    }
#endif

    // Prepare for reading a new split/task.
    // 1. Pass a new split/task to reader, which will be used in subsequent open_reader() to initialize the underlying file reader.
    // 2. Parse delete predicates from split/task information, which will be used for later dynamic filtering and delete handling.
    virtual Status prepare_split(const SplitReadOptions& options);

    virtual bool current_split_pruned() const { return _current_split_pruned; }
    virtual bool current_split_uses_metadata_count() const {
        return _current_split_uses_metadata_count;
    }

    // Largest logical input retained while materializing the last returned batch. FileScannerV2
    // combines this with the final output width when learning its adaptive batch size.
    size_t last_batch_materialization_input_bytes() const {
        return _last_batch_materialization_input_bytes;
    }

    // Discard the active split after the caller decides an error is ignorable, for example a
    // stale external-table file listing that returns NOT_FOUND. The next prepare_split() must start
    // with no concrete reader or split-local state left from the failed split.
    virtual Status abort_split() {
        // Ignored open failures still spend time closing partially initialized readers. Include
        // that recovery path in the common lifecycle profile so NOT_FOUND cannot become invisible.
        SCOPED_TIMER(_profile.total_timer);
        SCOPED_TIMER(_profile.close_timer);
        if (_data_reader.reader != nullptr) {
            RETURN_IF_ERROR(close_current_reader());
        } else {
            _current_task.reset();
            _current_file_description.reset();
        }
        _delete_rows = nullptr;
        _remaining_table_level_count = -1;
        _remaining_file_level_count = -1;
        _current_split_uses_metadata_count = false;
        _current_split_pruned = false;
        return Status::OK();
    }

    // Public entry point for reading a table-schema block. The base class opens the current reader,
    // advances across EOF, and closes exhausted readers. Subclasses provide protected hooks for
    // table-format-specific behavior.
    virtual Status get_block(Block* block, bool* eos) {
        SCOPED_TIMER(_profile.total_timer);
        SCOPED_TIMER(_profile.exec_timer);
        DORIS_CHECK(block->columns() == _projected_columns.size());
        block->clear_column_data(_projected_columns.size());
        _last_batch_materialization_input_bytes = 0;

        while (true) {
            if (*eos) {
                return Status::OK();
            }
            if (_io_ctx != nullptr && _io_ctx->should_stop) {
                *eos = true;
                return Status::OK();
            }
            if (!_data_reader.reader) {
                if (_is_table_level_count_active()) {
                    RETURN_IF_ERROR(_read_table_level_count(block, eos));
                    return Status::OK();
                }
                if (_is_file_level_count_active()) {
                    RETURN_IF_ERROR(_read_file_level_count(block, eos));
                    return Status::OK();
                }
                RETURN_IF_ERROR(create_next_reader(eos));
                if (!_data_reader.reader) {
                    DCHECK(*eos);
                    return Status::OK();
                }
            }

            // Materialize a reduced row set for upper aggregate operators when aggregate
            // pushdown can be applied. This is not the final aggregate result: COUNT emits
            // `count` default rows for the upper COUNT(*), and MIN/MAX emits two rows containing
            // file-level min/max values for the upper MIN/MAX.
            if (!_aggregate_pushdown_tried) {
                SCOPED_TIMER(_profile.pushdown_agg_timer);
                bool pushed_down = false;
                const auto status = _try_materialize_aggregate_pushdown_rows(block, &pushed_down);
                if (!status.ok()) {
                    if (_io_ctx != nullptr && _io_ctx->should_stop &&
                        status.is<ErrorCode::END_OF_FILE>()) {
                        *eos = true;
                        return Status::OK();
                    }
                    return status;
                }
                if (pushed_down) {
                    return Status::OK();
                }
            }

            bool current_eof = false;
            _data_reader.block_template.clear_column_data(
                    cast_set<int64_t>(_data_reader.file_block_layout.size()));
            size_t current_rows = 0;
            {
                SCOPED_TIMER(_profile.file_reader_total_timer);
                SCOPED_TIMER(_profile.file_reader_get_block_timer);
                RETURN_IF_ERROR(_data_reader.reader->get_block(&_data_reader.block_template,
                                                               &current_rows, &current_eof));
            }
            const bool stopped_during_read = _io_ctx != nullptr && _io_ctx->should_stop;
            if (current_rows == 0) {
                if (current_eof) {
                    _current_reader_reached_eof = !stopped_during_read;
                    RETURN_IF_ERROR(close_current_reader());
                }
                continue;
            }
            DCHECK_EQ(_data_reader.block_template.columns(), _data_reader.file_block_layout.size())
                    << _data_reader.block_template.dump_structure();
#ifndef NDEBUG
            RETURN_IF_ERROR(_check_file_block_columns("after file reader get_block", current_rows));
#endif
            DORIS_CHECK(block->columns() == _data_reader.column_mapper->mappings().size());
            RETURN_IF_ERROR(finalize_chunk(block, current_rows));
#ifndef NDEBUG
            RETURN_IF_ERROR(
                    _check_table_block_columns("after finalize_chunk", block, current_rows));
#endif
            if (current_eof) {
                _current_reader_reached_eof = !stopped_during_read;
                RETURN_IF_ERROR(close_current_reader());
            }
            return Status::OK();
        }
    }

    // Close the table reader and the currently active file reader. Subclasses that hold additional
    // table-format resources should override this and call TableReader::close() first.
    virtual Status close() {
        SCOPED_TIMER(_profile.total_timer);
        SCOPED_TIMER(_profile.close_timer);
        if (_data_reader.reader) {
            RETURN_IF_ERROR(close_current_reader());
        }
        _current_task.reset();
        _current_file_description.reset();
        _remaining_table_level_count = -1;
        _remaining_file_level_count = -1;
        _current_split_uses_metadata_count = false;
        return Status::OK();
    }

    virtual int64_t condition_cache_hit_count() const { return _condition_cache_hit_count; }

    virtual std::string debug_string() const;

    virtual Status annotate_projected_column(const TFileScanSlotInfo& slot_info,
                                             ProjectedColumnBuildContext* context,
                                             ColumnDefinition* column) const;

    virtual Status validate_projected_columns(const ProjectedColumnBuildContext& context) const {
        (void)context;
        return Status::OK();
    }

protected:
    // TableReader keeps the active file description both in the scan task and separately for
    // creating the physical reader. Table-format readers must update both copies when their
    // snapshot protocol guarantees that a file path is never overwritten with different bytes.
    // This guarantee lets readers safely build cache keys without mtime; it must not be used for
    // ordinary Hive/TVF files whose paths may be overwritten in place.
    void mark_current_data_file_immutable() {
        DORIS_CHECK(_current_task != nullptr);
        DORIS_CHECK(_current_task->data_file != nullptr);
        DORIS_CHECK(_current_file_description.has_value());
        _current_task->data_file->is_immutable = true;
        _current_file_description->is_immutable = true;
    }

    std::optional<ColumnDefinition> _find_current_table_column_by_field_id(int32_t field_id,
                                                                           DataTypePtr type) const;

    // Parse deletion vector information from table format specific file description.
    virtual Status _parse_deletion_vector_file(const TTableFormatFileDesc& t_desc,
                                               DeleteFileDesc* desc, bool* has_delete_file) {
        *has_delete_file = false;
        return Status::OK();
    }

    // Advance to the next reader. This closes the current reader first and then opens the next
    // concrete reader. Subclasses should not duplicate this loop.
    Status create_next_reader(bool* eos);
    virtual Status create_file_reader(std::unique_ptr<FileReader>* reader);
    virtual TableColumnMappingMode mapping_mode() const { return TableColumnMappingMode::BY_NAME; }
    virtual void configure_mapper_options(TableColumnMapperOptions*) const {}
    virtual Status annotate_file_schema(std::vector<ColumnDefinition>* file_schema) {
        DORIS_CHECK(file_schema != nullptr);
        return Status::OK();
    }

    // Open the concrete reader for the current split/task and build the file-local scan request.
    virtual Status open_reader() {
        SCOPED_TIMER(_profile.open_reader_timer);
        // 1. Get file schema and create column mapping.
        std::vector<ColumnDefinition> file_schema;
        {
            SCOPED_TIMER(_profile.file_reader_total_timer);
            SCOPED_TIMER(_profile.file_reader_schema_timer);
            RETURN_IF_ERROR(_data_reader.reader->get_schema(&file_schema));
        }
        // For Paimon/Hudi, FE can provide field ids through `history_schema_info`. Annotate the
        // file schema before column mapping when the table format maps columns by field id.
        RETURN_IF_ERROR(annotate_file_schema(&file_schema));
        _data_reader.file_schema = file_schema;
        _mapper_options.mode = mapping_mode();
        configure_mapper_options(&_mapper_options);

        {
            SCOPED_TIMER(_profile.file_reader_total_timer);
            SCOPED_TIMER(_profile.file_reader_mapper_timer);
            _data_reader.column_mapper = _data_reader.reader->create_column_mapper(_mapper_options);
        }
        DORIS_CHECK(_data_reader.column_mapper != nullptr);
        RETURN_IF_ERROR(_data_reader.column_mapper->create_mapping(_projected_columns,
                                                                   _partition_values, file_schema));
        DORIS_CHECK(_data_reader.column_mapper->mappings().size() == _projected_columns.size());

        // 2. Build table filters based on conjuncts and column predicates.
        RETURN_IF_ERROR(_build_table_filters_from_conjuncts());

        // 3. Create file scan request based on column mapping and table filters, then open file
        // reader with the request. File scan request carries row-level expression filters and
        // file-level pruning hints. Only expression filters decide returned rows.
        auto file_request = std::make_shared<FileScanRequest>();
        RETURN_IF_ERROR(_data_reader.column_mapper->create_scan_request(
                _table_filters, _projected_columns, file_request.get(), _runtime_state));
        bool constant_filter_pruned_split = false;
        RETURN_IF_ERROR(_evaluate_constant_filters(&constant_filter_pruned_split));
        if (constant_filter_pruned_split) {
            RETURN_IF_ERROR(close_current_reader());
            return Status::OK();
        }
        // COUNT(*) has no semantic column argument, but Nereids retains a minimum-width scan slot
        // so the scan node still has an output tuple. Record only the current non-predicate file
        // columns before table-format hooks add row-position or equality-delete dependencies. This
        // marker is independent of aggregate eligibility: with position deletes, for example,
        // metadata COUNT must fall back to reading rows, but an arbitrary unsupported TIME_MILLIS
        // placeholder still must not be validated or decoded merely to carry the surviving count.
        if (_push_down_agg_type == TPushAggOp::type::COUNT &&
            _push_down_count_columns.has_value() && _push_down_count_columns->empty()) {
            file_request->count_star_placeholder_columns.reserve(
                    file_request->non_predicate_columns.size());
            for (const auto& column : file_request->non_predicate_columns) {
                file_request->count_star_placeholder_columns.push_back(column.column_id());
            }
        }
        RETURN_IF_ERROR(customize_file_scan_request(file_request.get()));
        RETURN_IF_ERROR(_open_local_filter_exprs(*file_request));
        _data_reader.file_block_layout.clear();
        _data_reader.block_template.clear();
        _data_reader.file_block_layout.resize(file_request->local_positions.size());

        // 4. Build file block layout from file schema and column mapping. The layout describes
        // the block returned by file reader before table-column materialization.
        for (const auto& [file_column_id, block_position] : file_request->local_positions) {
            DORIS_CHECK(block_position.value() < _data_reader.file_block_layout.size());
            const auto* field = _find_column_definition(_data_reader.file_schema, file_column_id);
            DORIS_CHECK(field != nullptr);

            ColumnDefinition projected_field;
            {
                auto it = std::find_if(
                        file_request->non_predicate_columns.begin(),
                        file_request->non_predicate_columns.end(),
                        [&](const LocalColumnIndex& p) { return p.column_id() == file_column_id; });
                if (it != file_request->non_predicate_columns.end()) {
                    RETURN_IF_ERROR(project_column_definition(*field, *it, &projected_field));
                }
            }
            {
                auto it = std::find_if(
                        file_request->predicate_columns.begin(),
                        file_request->predicate_columns.end(),
                        [&](const LocalColumnIndex& p) { return p.column_id() == file_column_id; });
                if (it != file_request->predicate_columns.end()) {
                    RETURN_IF_ERROR(project_column_definition(*field, *it, &projected_field));
                }
            }
            _data_reader.file_block_layout[block_position.value()] = {
                    .file_column_id = file_column_id,
                    .name = projected_field.name,
                    .type = projected_field.type,
            };
            DORIS_CHECK(_data_reader.file_block_layout[block_position.value()].type != nullptr);
        }

        // 5. Prepare block template from file block layout. The block template stores the block
        // returned by file reader before table-column materialization.
        _data_reader.block_template.reserve(_data_reader.file_block_layout.size());
        for (const auto& column : _data_reader.file_block_layout) {
            _data_reader.block_template.insert(
                    {column.type->create_column(), column.type, column.name});
        }
        if (VLOG_DEBUG_IS_ON) {
            VLOG_DEBUG << "TableReader debug: " << debug_string();
        }
        RETURN_IF_ERROR(_open_mapping_exprs());
        {
            SCOPED_TIMER(_profile.file_reader_total_timer);
            SCOPED_TIMER(_profile.file_reader_open_timer);
            RETURN_IF_ERROR(_data_reader.reader->open(file_request));
        }
        RETURN_IF_ERROR(_init_reader_condition_cache(*file_request));
        return Status::OK();
    }

    Status _build_table_filters_from_conjuncts();
    Status _evaluate_partition_prune_conjuncts(const VExprContextSPtrs& conjuncts,
                                               bool* can_filter_all);
    static bool _is_safe_to_pre_execute(const VExprContextSPtr& conjunct);
    Status _build_partition_prune_block(Block* block) const;
    Status _open_local_filter_exprs(const FileScanRequest& file_request);
    Status _init_reader_condition_cache(const FileScanRequest& file_request);
    void _finalize_reader_condition_cache();
    bool _should_enable_condition_cache(const FileScanRequest& file_request) const;

    Status _evaluate_constant_filters(bool* can_filter_all) {
        DORIS_CHECK(can_filter_all != nullptr);
        DORIS_CHECK_LE(_constant_pruning_safe_filter_count, _table_filters.size());
        *can_filter_all = false;
        // The bound was derived from the original `_conjuncts` order, which includes slotless
        // expressions omitted from `_table_filters`. Iterating only this prefix therefore cannot
        // skip an unsafe row-level predicate and pre-execute a later constant predicate.
        for (size_t i = 0; i < _constant_pruning_safe_filter_count; ++i) {
            const auto& table_filter = _table_filters[i];
            if (table_filter.conjunct == nullptr) {
                continue;
            }
            DORIS_CHECK(_is_safe_to_pre_execute(table_filter.conjunct));
            // RuntimeFilterExpr does not implement execute_column_impl(); it is evaluated by the
            // row-level filter path through execute_filter(). Constant split pruning uses
            // VExprContext::execute() on a one-row synthetic block, so runtime filters must not be
            // pre-executed here even when their referenced slot maps to a constant value.
            if (table_filter.conjunct->root()->is_rf_wrapper() ||
                !_table_filter_has_only_constant_entries(table_filter)) {
                continue;
            }
            Block eval_block;
            RETURN_IF_ERROR(_build_constant_filter_block(table_filter, &eval_block));
            RowDescriptor row_desc;
            RETURN_IF_ERROR(table_filter.conjunct->prepare(_runtime_state, row_desc));
            RETURN_IF_ERROR(table_filter.conjunct->open(_runtime_state));
            int result_column_id = -1;
            RETURN_IF_ERROR(table_filter.conjunct->execute(&eval_block, &result_column_id));
            DORIS_CHECK(result_column_id >= 0);
            if (_filter_result_filters_all(eval_block.get_by_position(result_column_id).column)) {
                *can_filter_all = true;
                return Status::OK();
            }
        }
        return Status::OK();
    }

    bool _table_filter_has_only_constant_entries(const TableFilter& table_filter) const {
        const auto& filter_entries = _data_reader.column_mapper->filter_entries();
        for (const auto global_index : table_filter.global_indices) {
            const auto entry_it = filter_entries.find(global_index);
            if (entry_it == filter_entries.end() || !entry_it->second.is_constant()) {
                return false;
            }
        }
        return !table_filter.global_indices.empty();
    }

    Status _build_constant_filter_block(const TableFilter& table_filter, Block* eval_block) {
        DORIS_CHECK(eval_block != nullptr);
        eval_block->clear();
        const auto& mappings = _data_reader.column_mapper->mappings();
        const auto& filter_entries = _data_reader.column_mapper->filter_entries();
        DORIS_CHECK(mappings.size() == _projected_columns.size());
        for (size_t column_idx = 0; column_idx < mappings.size(); ++column_idx) {
            const auto global_index = GlobalIndex(column_idx);
            const auto& mapping = mappings[column_idx];
            const auto entry_it = filter_entries.find(global_index);
            const bool referenced_by_filter =
                    std::find(table_filter.global_indices.begin(),
                              table_filter.global_indices.end(),
                              global_index) != table_filter.global_indices.end();
            if (referenced_by_filter && entry_it != filter_entries.end() &&
                entry_it->second.is_constant()) {
                ColumnPtr constant_column;
                RETURN_IF_ERROR(_materialize_constant_filter_column(
                        entry_it->second.constant_index(), &constant_column));
                eval_block->insert({std::move(constant_column), mapping.table_type,
                                    mapping.table_column_name});
            } else {
                eval_block->insert({mapping.table_type->create_column_const_with_default_value(1),
                                    mapping.table_type, mapping.table_column_name});
            }
        }
        return Status::OK();
    }

    Status _materialize_constant_filter_column(ConstantIndex constant_index, ColumnPtr* column) {
        DORIS_CHECK(column != nullptr);
        const auto& constant_entry = _data_reader.column_mapper->constant_map().get(constant_index);
        DORIS_CHECK(constant_entry.expr != nullptr);
        DORIS_CHECK(constant_entry.type != nullptr);
        RowDescriptor row_desc;
        RETURN_IF_ERROR(constant_entry.expr->prepare(_runtime_state, row_desc));
        RETURN_IF_ERROR(constant_entry.expr->open(_runtime_state));
        Block eval_block;
        eval_block.insert({constant_entry.type->create_column_const_with_default_value(1),
                           constant_entry.type, "__table_reader_constant_filter"});
        int result_column_id = -1;
        RETURN_IF_ERROR(constant_entry.expr->execute(&eval_block, &result_column_id));
        DORIS_CHECK(result_column_id >= 0);
        *column = eval_block.get_by_position(result_column_id).column;
        DORIS_CHECK((*column)->size() == 1);
        return Status::OK();
    }

    static bool _filter_result_filters_all(const ColumnPtr& filter_column) {
        DORIS_CHECK(filter_column.get() != nullptr);
        DORIS_CHECK(filter_column->size() == 1);
        return !filter_column->get_bool(0);
    }

    virtual Status customize_file_scan_request(FileScanRequest* file_request) {
        return _append_delete_predicate(file_request);
    }

    bool _is_table_level_count_active() const { return _remaining_table_level_count >= 0; }

    bool _is_file_level_count_active() const { return _remaining_file_level_count >= 0; }

    Status _materialize_count_rows(size_t rows, Block* block) const {
        DORIS_CHECK(block != nullptr);
        DORIS_CHECK(block->columns() > 0 || rows == 0);
        for (size_t column_idx = 0; column_idx < block->columns(); ++column_idx) {
            auto column = block->get_by_position(column_idx).type->create_column();
            if (auto* nullable = check_and_get_column<ColumnNullable>(*column)) {
                // Metadata COUNT emits synthetic input rows for the unchanged upper aggregate.
                // They must be non-NULL for COUNT(nullable_col), and constructing them explicitly
                // also keeps every nullable null map boolean-valid in debug/ASAN block checks.
                nullable->get_nested_column().insert_many_defaults(rows);
                nullable->get_null_map_data().resize_fill(rows, 0);
            } else {
                column->insert_many_defaults(rows);
            }
            block->replace_by_position(column_idx, std::move(column));
        }
        return Status::OK();
    }

    Status _materialize_next_count_batch(int64_t* remaining_rows, Block* block) const {
        DORIS_CHECK(remaining_rows != nullptr);
        DORIS_CHECK(*remaining_rows > 0);
        const int64_t batch_size = _runtime_state == nullptr
                                           ? *remaining_rows
                                           : static_cast<int64_t>(_runtime_state->batch_size());
        const auto rows = std::min(*remaining_rows, batch_size);
        RETURN_IF_ERROR(_materialize_count_rows(cast_set<size_t>(rows), block));
        *remaining_rows -= rows;
        return Status::OK();
    }

    Status _read_count_batch(int64_t* remaining_rows, Block* block, bool* eos) {
        DORIS_CHECK(block != nullptr);
        DORIS_CHECK(eos != nullptr);
        DORIS_CHECK(_push_down_agg_type == TPushAggOp::type::COUNT);
        DORIS_CHECK(remaining_rows != nullptr);
        DORIS_CHECK(*remaining_rows >= 0);
        if (*remaining_rows == 0) {
            *remaining_rows = -1;
            _current_task.reset();
            *eos = true;
            return Status::OK();
        }
        RETURN_IF_ERROR(_materialize_next_count_batch(remaining_rows, block));
        *eos = false;
        return Status::OK();
    }

    Status _read_table_level_count(Block* block, bool* eos) {
        return _read_count_batch(&_remaining_table_level_count, block, eos);
    }

    Status _read_file_level_count(Block* block, bool* eos) {
        return _read_count_batch(&_remaining_file_level_count, block, eos);
    }

    void _append_file_scan_column(FileScanRequest* request, LocalColumnId column_id,
                                  std::vector<LocalColumnIndex>* scan_columns) {
        DORIS_CHECK(request != nullptr);
        DORIS_CHECK(scan_columns != nullptr);
        FileScanRequestBuilder builder(request);
        Status status;
        if (scan_columns == &request->predicate_columns) {
            status = builder.add_predicate_column(column_id);
        } else {
            DORIS_CHECK(scan_columns == &request->non_predicate_columns);
            status = builder.add_non_predicate_column(column_id);
        }
        DORIS_CHECK(status.ok()) << status.to_string();
        if (column_id == LocalColumnId(ROW_POSITION_COLUMN_ID) &&
            _find_column_definition(_data_reader.file_schema, column_id) == nullptr) {
            _data_reader.file_schema.push_back(row_position_column_definition());
        }
    }

    // Append DeletePredicate to file scan request if there are deletes. The predicate will be evaluated in file reader level and filter out deleted rows before returning data to table reader.
    Status _append_delete_predicate(FileScanRequest* request) {
        DORIS_CHECK(request != nullptr);
        if ((_delete_rows == nullptr || _delete_rows->empty()) &&
            (_deletion_vector == nullptr || _deletion_vector->isEmpty())) {
            return Status::OK();
        }
        const auto row_position_column_id = LocalColumnId(ROW_POSITION_COLUMN_ID);
        _append_file_scan_column(request, row_position_column_id, &request->predicate_columns);

        const auto block_position = request->local_positions.at(row_position_column_id);
        auto append_predicate = [&](auto& deleted_rows) {
            auto delete_predicate = std::make_shared<DeletePredicate>(deleted_rows);
            delete_predicate->add_child(VSlotRef::create_shared(
                    cast_set<int>(block_position.value()), cast_set<int>(block_position.value()),
                    -1, std::make_shared<DataTypeInt64>(), ROW_POSITION_COLUMN_NAME));
            request->delete_conjuncts.push_back(
                    VExprContext::create_shared(std::move(delete_predicate)));
        };
        if (_delete_rows != nullptr && !_delete_rows->empty()) {
            append_predicate(*_delete_rows);
        }
        if (_deletion_vector != nullptr && !_deletion_vector->isEmpty()) {
            append_predicate(*_deletion_vector);
        }
        return Status::OK();
    }

    // Close the current concrete reader. This hook is called by both create_next_reader() and
    // close(), so it should remain idempotent.
    virtual Status close_current_reader() {
        _finalize_reader_condition_cache();
        {
            SCOPED_TIMER(_profile.file_reader_total_timer);
            SCOPED_TIMER(_profile.file_reader_close_timer);
            RETURN_IF_ERROR(_data_reader.reader->close());
        }
        _data_reader.reader.reset();
        if (_data_reader.column_mapper != nullptr) {
            _data_reader.column_mapper->clear();
            _data_reader.column_mapper.reset();
        }
        _table_filters.clear();
        _constant_pruning_safe_filter_count = 0;
        _data_reader.file_schema.clear();
        _data_reader.file_block_layout.clear();
        _data_reader.block_template.clear();
        _current_task.reset();
        _current_file_description.reset();
        _current_reader_reached_eof = false;
        return Status::OK();
    }

    void _record_scan_rows(size_t rows) {
        if (_io_ctx != nullptr && _io_ctx->file_reader_stats != nullptr) {
            _io_ctx->file_reader_stats->read_rows += rows;
        }
    }

    // Finalize file-local block to table/global schema block.
    Status finalize_chunk(Block* block, const size_t rows) {
        SCOPED_TIMER(_profile.finalize_timer);
        const auto& mappings = _data_reader.column_mapper->mappings();
        const size_t file_column_count = _data_reader.block_template.columns();
        const bool has_variant_path = std::ranges::any_of(
                mappings,
                [](const ColumnMapping& mapping) { return !mapping.column_paths.empty(); });
        // Record before any projection transfers a carrier out of the file-local block. Current
        // Variant-aware Parquet readers may return a full root even when only a tiny Path Slot is
        // requested; future physical pruning automatically lowers this sample.
        _last_batch_materialization_input_bytes =
                has_variant_path ? _data_reader.block_template.bytes() : 0;
        auto materialize_mapping = [&](size_t mapping_index,
                                       bool take_projection_result) -> Status {
            ColumnPtr column;
            RETURN_IF_ERROR(_materialize_mapping_column(mappings[mapping_index],
                                                        &_data_reader.block_template, rows, &column,
                                                        take_projection_result));
            // Materialization order is internal only. Always write the original global index
            // position so the table/global output schema remains unchanged.
            block->replace_by_position(mapping_index, IColumn::mutate(std::move(column)));
            return Status::OK();
        };

        if (!has_variant_path) {
            // Keep the common path identical to the original linear finalization. Building a
            // carrier ownership plan is only necessary when a Path Slot shares its root source.
            for (size_t mapping_index = 0; mapping_index < mappings.size(); ++mapping_index) {
                RETURN_IF_ERROR(
                        materialize_mapping(mapping_index, mapping_index + 1 == mappings.size()));
            }
        } else {
            // Record the last consumer in execution order once per chunk. This avoids scanning all
            // mappings for every output column when a Path Slot is present.
            std::unordered_map<int32_t, size_t> last_file_column_consumers;
            last_file_column_consumers.reserve(mappings.size());
            auto record_phase = [&](bool record_paths) {
                for (size_t mapping_index = 0; mapping_index < mappings.size(); ++mapping_index) {
                    const auto& mapping = mappings[mapping_index];
                    if ((!mapping.column_paths.empty()) != record_paths ||
                        !mapping.file_local_id.has_value() || mapping.projection == nullptr) {
                        continue;
                    }
                    last_file_column_consumers[*mapping.file_local_id] = mapping_index;
                }
            };
            // Every Path Slot reads its shared carrier before any root/normal mapping can transfer
            // that carrier out of the file block.
            record_phase(true);
            record_phase(false);

            auto materialize_phase = [&](bool materialize_paths) -> Status {
                for (size_t mapping_index = 0; mapping_index < mappings.size(); ++mapping_index) {
                    const auto& mapping = mappings[mapping_index];
                    if ((!mapping.column_paths.empty()) != materialize_paths) {
                        continue;
                    }
                    bool take_projection_result = false;
                    if (mapping.file_local_id.has_value() && mapping.projection != nullptr) {
                        const auto last_consumer_it =
                                last_file_column_consumers.find(*mapping.file_local_id);
                        DORIS_CHECK(last_consumer_it != last_file_column_consumers.end());
                        take_projection_result = last_consumer_it->second == mapping_index;
                    }
                    RETURN_IF_ERROR(materialize_mapping(mapping_index, take_projection_result));
                }
                return Status::OK();
            };
            RETURN_IF_ERROR(materialize_phase(true));
            RETURN_IF_ERROR(materialize_phase(false));
        }
        // Expressions can append temporary result columns, while carrier transfer leaves same-row
        // constant placeholders so later expressions retain Block::rows(). Remove the temporary
        // results and restore only transferred file columns before the next physical batch. Other
        // file columns keep their allocated buffers for clear_column_data() to reuse.
        _data_reader.block_template.erase_tail(file_column_count);
        for (size_t idx = 0; idx < file_column_count; ++idx) {
            auto& file_column = _data_reader.block_template.get_by_position(idx);
            if (is_column_const(*file_column.column)) {
                DORIS_CHECK(file_column.type != nullptr);
                file_column.column = file_column.type->create_column();
            }
        }
        RETURN_IF_ERROR(materialize_virtual_columns(block));
        // Enforce CHAR/VARCHAR length declared by the table schema after all file-to-table
        // materialization has finished.
        RETURN_IF_ERROR(_truncate_char_or_varchar_columns(block));
        return Status::OK();
    }

    // Materialize virtual columns in the table block, such as Iceberg _row_id and
    // _last_updated_sequence_number. This runs after normal column materialization so finalize
    // expressions can reference those virtual columns.
    virtual Status materialize_virtual_columns(Block* table_block) { return Status::OK(); }

#ifndef NDEBUG
    Status _check_file_block_columns(std::string_view stage, size_t rows) {
        DORIS_CHECK(_data_reader.block_template.columns() == _data_reader.file_block_layout.size());
        for (size_t idx = 0; idx < _data_reader.block_template.columns(); ++idx) {
            const auto& file_block_column = _data_reader.file_block_layout[idx];
            const auto& column_with_type = _data_reader.block_template.get_by_position(idx);
            const auto* column = column_with_type.column.get();
            try {
                if (column == nullptr) {
                    auto st = Status::InternalError(
                            "Invalid file block column {} at {}: file_column_id={}, name='{}', "
                            "type={}, column=null, expected_rows={}, reader={}",
                            idx, stage, file_block_column.file_column_id.value(),
                            file_block_column.name,
                            file_block_column.type == nullptr ? "null"
                                                              : file_block_column.type->get_name(),
                            rows, debug_string());
                    LOG(WARNING) << st;
                    return st;
                }
                column->sanity_check();
                auto st = column_with_type.check_type_and_column_match();
                if (!st.ok()) {
                    auto contextual_status = Status::InternalError(
                            "Invalid file block column {} at {}: file_column_id={}, name='{}', "
                            "type={}, column={}, column_size={}, expected_rows={}, error={}, "
                            "reader={}",
                            idx, stage, file_block_column.file_column_id.value(),
                            file_block_column.name,
                            file_block_column.type == nullptr ? "null"
                                                              : file_block_column.type->get_name(),
                            column->get_name(), column->size(), rows, st.to_string(),
                            debug_string());
                    LOG(WARNING) << contextual_status;
                    return contextual_status;
                }
            } catch (const Exception& e) {
                auto st = Status::InternalError(
                        "Invalid file block column {} at {}: file_column_id={}, name='{}', "
                        "type={}, column={}, column_size={}, expected_rows={}, error={}, "
                        "reader={}",
                        idx, stage, file_block_column.file_column_id.value(),
                        file_block_column.name,
                        file_block_column.type == nullptr ? "null"
                                                          : file_block_column.type->get_name(),
                        column == nullptr ? "null" : column->get_name(),
                        column == nullptr ? 0 : column->size(), rows, e.to_string(),
                        debug_string());
                LOG(WARNING) << st;
                return st;
            } catch (const std::exception& e) {
                auto st = Status::InternalError(
                        "Invalid file block column {} at {}: file_column_id={}, name='{}', "
                        "type={}, column={}, column_size={}, expected_rows={}, error={}, "
                        "reader={}",
                        idx, stage, file_block_column.file_column_id.value(),
                        file_block_column.name,
                        file_block_column.type == nullptr ? "null"
                                                          : file_block_column.type->get_name(),
                        column == nullptr ? "null" : column->get_name(),
                        column == nullptr ? 0 : column->size(), rows, e.what(), debug_string());
                LOG(WARNING) << st;
                return st;
            }
        }
        return Status::OK();
    }

    Status _check_table_block_columns(std::string_view stage, const Block* block, size_t rows) {
        DORIS_CHECK(block != nullptr);
        DORIS_CHECK(block->columns() == _data_reader.column_mapper->mappings().size());
        for (size_t idx = 0; idx < block->columns(); ++idx) {
            const auto& mapping = _data_reader.column_mapper->mappings()[idx];
            const auto& column_with_type = block->get_by_position(idx);
            const auto* column = column_with_type.column.get();
            try {
                if (column == nullptr) {
                    auto st = Status::InternalError(
                            "Invalid table block column {} at {}: table_column='{}', "
                            "global_index={}, type={}, column=null, expected_rows={}, mapping={}",
                            idx, stage, mapping.table_column_name, mapping.global_index.value(),
                            mapping.table_type == nullptr ? "null" : mapping.table_type->get_name(),
                            rows, mapping.debug_string());
                    LOG(WARNING) << st;
                    return st;
                }
                column->sanity_check();
                auto st = column_with_type.check_type_and_column_match();
                if (!st.ok()) {
                    auto contextual_status = Status::InternalError(
                            "Invalid table block column {} at {}: table_column='{}', "
                            "global_index={}, type={}, column={}, column_size={}, "
                            "expected_rows={}, error={}, mapping={}",
                            idx, stage, mapping.table_column_name, mapping.global_index.value(),
                            mapping.table_type == nullptr ? "null" : mapping.table_type->get_name(),
                            column->get_name(), column->size(), rows, st.to_string(),
                            mapping.debug_string());
                    LOG(WARNING) << contextual_status;
                    return contextual_status;
                }
            } catch (const Exception& e) {
                auto st = Status::InternalError(
                        "Invalid table block column {} at {}: table_column='{}', global_index={}, "
                        "type={}, column={}, column_size={}, expected_rows={}, error={}, "
                        "mapping={}",
                        idx, stage, mapping.table_column_name, mapping.global_index.value(),
                        mapping.table_type == nullptr ? "null" : mapping.table_type->get_name(),
                        column == nullptr ? "null" : column->get_name(),
                        column == nullptr ? 0 : column->size(), rows, e.to_string(),
                        mapping.debug_string());
                LOG(WARNING) << st;
                return st;
            } catch (const std::exception& e) {
                auto st = Status::InternalError(
                        "Invalid table block column {} at {}: table_column='{}', global_index={}, "
                        "type={}, column={}, column_size={}, expected_rows={}, error={}, "
                        "mapping={}",
                        idx, stage, mapping.table_column_name, mapping.global_index.value(),
                        mapping.table_type == nullptr ? "null" : mapping.table_type->get_name(),
                        column == nullptr ? "null" : column->get_name(),
                        column == nullptr ? 0 : column->size(), rows, e.what(),
                        mapping.debug_string());
                LOG(WARNING) << st;
                return st;
            }
        }
        return Status::OK();
    }
#endif

    Status _truncate_char_or_varchar_columns(Block* block) {
        DORIS_CHECK(block != nullptr);
        if (_runtime_state == nullptr ||
            !_runtime_state->query_options().truncate_char_or_varchar_columns) {
            return Status::OK();
        }
        DORIS_CHECK(block->columns() == _data_reader.column_mapper->mappings().size());
        for (size_t idx = 0; idx < _data_reader.column_mapper->mappings().size(); ++idx) {
            const auto& mapping = _data_reader.column_mapper->mappings()[idx];
            if (!_should_truncate_char_or_varchar_column(mapping)) {
                continue;
            }
            const auto target_len =
                    assert_cast<const DataTypeString*>(remove_nullable(mapping.table_type).get())
                            ->len();
            _truncate_char_or_varchar_column(block, idx, target_len);
        }
        return Status::OK();
    }

    // Return true when the table schema has a bounded CHAR/VARCHAR length that is stricter than
    // the file-side type. Examples:
    // - table VARCHAR(10), file VARCHAR(20): truncate to 10;
    // - table VARCHAR(10), file STRING: truncate to 10 because STRING has no declared bound;
    // - table STRING, any file type: no truncation because the target has no bound.
    static bool _should_truncate_char_or_varchar_column(const ColumnMapping& mapping) {
        if (mapping.table_type == nullptr) {
            return false;
        }
        const auto table_type = remove_nullable(mapping.table_type);
        const auto primitive_type = table_type->get_primitive_type();
        if (primitive_type != TYPE_VARCHAR && primitive_type != TYPE_CHAR) {
            return false;
        }
        const auto target_len = assert_cast<const DataTypeString*>(table_type.get())->len();
        if (target_len <= 0) {
            return false;
        }
        if (mapping.file_type == nullptr) {
            return true;
        }
        const auto file_type = remove_nullable(mapping.file_type);
        DORIS_CHECK(file_type != nullptr);
        int file_len = -1;
        if (file_type->get_primitive_type() == TYPE_VARCHAR ||
            file_type->get_primitive_type() == TYPE_CHAR ||
            file_type->get_primitive_type() == TYPE_STRING) {
            file_len = assert_cast<const DataTypeString*>(file_type.get())->len();
        }

        return file_len < 0 || target_len < file_len;
    }

    // Truncate a materialized CHAR/VARCHAR column in place by reusing the vectorized substring
    // implementation: substring(column, 1, len). Nullable columns are unwrapped before substring
    // execution and wrapped back with the original null map afterward, because substring operates
    // on the nested string payload only.
    static void _truncate_char_or_varchar_column(Block* block, size_t idx, int len) {
        DORIS_CHECK(block != nullptr);
        auto int_type = std::make_shared<DataTypeInt32>();
        const auto num_columns_without_result = cast_set<uint32_t>(block->columns());
        auto& target = block->get_by_position(idx);
        const bool is_nullable = target.type->is_nullable();
        ColumnPtr input_column = target.column;
        ColumnPtr null_map_column;
        if (is_nullable) {
            const auto* nullable_column = assert_cast<const ColumnNullable*>(target.column.get());
            input_column = nullable_column->get_nested_column_ptr();
            null_map_column = nullable_column->get_null_map_column_ptr();
        }
        block->replace_by_position(idx, std::move(input_column));
        block->insert({int_type->create_column_const(block->rows(), to_field<TYPE_INT>(1)),
                       int_type, "const 1"});
        block->insert({int_type->create_column_const(block->rows(), to_field<TYPE_INT>(len)),
                       int_type, "const len"});
        block->insert({nullptr, std::make_shared<DataTypeString>(), "result"});

        ColumnNumbers temp_arguments(3);
        temp_arguments[0] = cast_set<uint32_t>(idx);
        temp_arguments[1] = num_columns_without_result;
        temp_arguments[2] = num_columns_without_result + 1;
        const uint32_t result_column_id = num_columns_without_result + 2;
        SubstringUtil::substring_execute(*block, temp_arguments, result_column_id, block->rows());

        ColumnPtr result_column = block->get_by_position(result_column_id).column;
        if (is_nullable) {
            result_column = ColumnNullable::create(std::move(result_column), null_map_column);
        }
        block->replace_by_position(idx, std::move(result_column));
        block->erase_tail(num_columns_without_result);
    }

    Status _try_materialize_aggregate_pushdown_rows(Block* block, bool* pushed_down) {
        DORIS_CHECK(block != nullptr);
        DORIS_CHECK(pushed_down != nullptr);
        *pushed_down = false;
        block->clear_column_data(_projected_columns.size());
        _aggregate_pushdown_tried = true;
        if (!_supports_aggregate_pushdown(_push_down_agg_type)) {
            return Status::OK();
        }

        FileAggregateRequest file_request;
        RETURN_IF_ERROR(_build_file_aggregate_request(_push_down_agg_type, &file_request));
        FileAggregateResult file_result;
        Status status;
        {
            SCOPED_TIMER(_profile.file_reader_total_timer);
            SCOPED_TIMER(_profile.file_reader_aggregate_timer);
            status = _data_reader.reader->get_aggregate_result(file_request, &file_result);
        }
        if (status.is<ErrorCode::NOT_IMPLEMENTED_ERROR>()) {
            return Status::OK();
        }
        RETURN_IF_ERROR(status);
        if (_push_down_agg_type == TPushAggOp::type::COUNT) {
            DORIS_CHECK(file_result.count >= 0);
            // The upper aggregate consumes synthetic input rows, but emitting the whole metadata
            // count in one block bypasses the runtime batch contract and can allocate by file size.
            // Keep the remaining cardinality as split state and expose at most one batch per call.
            _remaining_file_level_count = file_result.count;
            _current_split_uses_metadata_count = true;
            if (_remaining_file_level_count > 0) {
                RETURN_IF_ERROR(_materialize_next_count_batch(&_remaining_file_level_count, block));
            }
        } else {
            RETURN_IF_ERROR(
                    _materialize_aggregate_pushdown_rows(_push_down_agg_type, file_result, block));
        }
        *pushed_down = true;
        RETURN_IF_ERROR(close_current_reader());
        return Status::OK();
    }

    virtual bool _supports_aggregate_pushdown(TPushAggOp::type agg_type) const {
        // Only COUNT and MIN/MAX can be push down.
        if (agg_type != TPushAggOp::type::COUNT && agg_type != TPushAggOp::type::MINMAX) {
            return false;
        }
        // Aggregate pushdown returns reduced synthetic rows and may close the physical reader
        // before the next scheduler turn. If a runtime filter is still pending, those rows could
        // escape before the filter arrives and cannot later be reconstructed from real file rows.
        // This is the same irreversibility constraint as table-level metadata COUNT, and applies
        // to COUNT and MIN/MAX for Parquet/ORC as well as COUNT for text readers.
        if (!_all_runtime_filters_applied_for_split) {
            return false;
        }
        // Scanner owns the original conjunct list and evaluates it after TableReader finalizes
        // rows. Even a slotless conjunct that cannot become a TableFilter must see every source
        // row before an aggregate reduces the stream to synthetic COUNT/MINMAX rows.
        if (!_conjuncts.empty()) {
            return false;
        }
        // Only support aggregate pushdown when there is no delete or filter, so
        // the reduced rows consumed by the upper aggregate remain semantically equivalent to a
        // normal scan.
        if ((_delete_rows != nullptr && !_delete_rows->empty()) ||
            (_deletion_vector != nullptr && !_deletion_vector->isEmpty())) {
            return false;
        }
        if (!_table_filters.empty()) {
            return false;
        }
        if (agg_type == TPushAggOp::type::COUNT) {
            // Old FEs do not serialize push_down_count_slot_ids. During the supported BE-first
            // rolling upgrade, nullopt therefore means "COUNT semantics are unknown", not
            // COUNT(*). Fall back to reading rows until the FE explicitly sends either an empty
            // list for COUNT(*) or one slot for COUNT(col).
            if (!_push_down_count_columns.has_value()) {
                return false;
            }
            // COUNT(*) needs no column metadata. COUNT(col) currently supports one direct file
            // column; multiple COUNT arguments fall back to the normal scan so every upper
            // aggregate receives the original rows.
            if (_push_down_count_columns->empty()) {
                return true;
            }
            if (_push_down_count_columns->size() != 1) {
                return false;
            }
            const auto& mapping = _push_down_count_mapping();
            // Metadata COUNT skips TableReader's normal materialization path. Only a trivial
            // mapping is safe: for example, a nullable Parquet INT mapped to a NOT NULL table
            // BIGINT normally needs both an INT->BIGINT cast and nullability validation. Counting
            // footer values directly would bypass both operations and could hide invalid data.
            return mapping.file_local_id.has_value() && mapping.file_type != nullptr &&
                   mapping.table_type != nullptr && mapping.is_trivial &&
                   mapping.virtual_column_type == TableVirtualColumnType::INVALID &&
                   mapping.default_expr == nullptr;
        }
        // For MIN/MAX, only support direct file-to-table column mappings. The two emitted rows
        // must be enough for the upper MIN/MAX aggregate without evaluating default expressions or
        // virtual columns.
        for (const auto& mapping : _data_reader.column_mapper->mappings()) {
            if (!mapping.file_local_id.has_value() ||
                mapping.virtual_column_type != TableVirtualColumnType::INVALID ||
                mapping.default_expr != nullptr || mapping.file_type == nullptr ||
                mapping.table_type == nullptr) {
                return false;
            }
            if (!_can_push_down_minmax_for_mapping(mapping)) {
                return false;
            }
        }
        return true;
    }

    static ColumnPtr _detach_column(ColumnPtr column) {
        DORIS_CHECK(column.get() != nullptr);
        return IColumn::mutate(std::move(column));
    }

    static ColumnPtr _take_and_detach_block_column(Block* block, int position) {
        DORIS_CHECK(block != nullptr);
        DORIS_CHECK(position >= 0 && position < static_cast<int>(block->columns()));
        auto& source = block->get_by_position(position);
        ColumnPtr column = source.column;
        // The final mapping no longer needs the file block. Release its COW owner before mutate(),
        // otherwise nested MAP/STRING columns are deep-copied and a multi-GB payload can OOM.
        // Keep a constant placeholder with the original row count: Block::rows() is derived from
        // its first column, and later mapping/default expressions must still see the full batch
        // after an earlier Path Slot transfers the carrier at position zero.
        block->replace_by_position(
                position, source.type->create_column_const_with_default_value(column->size()));
        return _detach_column(std::move(column));
    }

    static Status _align_column_nullability(ColumnPtr* column, const DataTypePtr& table_type) {
        DORIS_CHECK(column != nullptr);
        DORIS_CHECK(column->get() != nullptr);
        DORIS_CHECK(table_type != nullptr);
        // Must return non-const column
        *column = (*column)->convert_to_full_column_if_const();
        if (table_type->is_nullable()) {
            const auto& nested_type =
                    assert_cast<const DataTypeNullable&>(*table_type).get_nested_type();
            if (!(*column)->is_nullable()) {
                RETURN_IF_ERROR(_align_column_nullability(column, nested_type));
                *column = make_nullable(*column);
                return Status::OK();
            }
            const auto& nullable_column = assert_cast<const ColumnNullable&>(**column);
            ColumnPtr nested_column = nullable_column.get_nested_column_ptr();
            RETURN_IF_ERROR(_align_column_nullability(&nested_column, nested_type));
            *column = ColumnNullable::create(nested_column,
                                             nullable_column.get_null_map_column_ptr());
            return Status::OK();
        }
        if ((*column)->is_nullable()) {
            const auto& nullable_column = assert_cast<const ColumnNullable&>(**column);
            if (nullable_column.has_null()) {
                return Status::InternalError(
                        "Default expression produced NULL for non-nullable table column");
            }
            ColumnPtr nested_column = nullable_column.get_nested_column_ptr();
            RETURN_IF_ERROR(_align_column_nullability(&nested_column, table_type));
            *column = nested_column;
            return Status::OK();
        }
        if (const auto* array_type = typeid_cast<const DataTypeArray*>(table_type.get())) {
            const auto& array_column = assert_cast<const ColumnArray&>(**column);
            ColumnPtr nested_column = array_column.get_data_ptr();
            RETURN_IF_ERROR(
                    _align_column_nullability(&nested_column, array_type->get_nested_type()));
            *column = ColumnArray::create(nested_column, array_column.get_offsets_ptr());
            return Status::OK();
        }
        if (const auto* map_type = typeid_cast<const DataTypeMap*>(table_type.get())) {
            const auto& map_column = assert_cast<const ColumnMap&>(**column);
            ColumnPtr key_column = map_column.get_keys_ptr();
            ColumnPtr value_column = map_column.get_values_ptr();
            RETURN_IF_ERROR(_align_column_nullability(&key_column, map_type->get_key_type()));
            RETURN_IF_ERROR(_align_column_nullability(&value_column, map_type->get_value_type()));
            *column = ColumnMap::create(key_column, value_column, map_column.get_offsets_ptr());
            return Status::OK();
        }
        if (const auto* struct_type = typeid_cast<const DataTypeStruct*>(table_type.get())) {
            const auto& struct_column = assert_cast<const ColumnStruct&>(**column);
            Columns columns = struct_column.get_columns_copy();
            DORIS_CHECK(columns.size() == struct_type->get_elements().size());
            for (size_t i = 0; i < columns.size(); ++i) {
                RETURN_IF_ERROR(
                        _align_column_nullability(&columns[i], struct_type->get_element(i)));
            }
            *column = ColumnStruct::create(columns);
            return Status::OK();
        }
        return Status::OK();
    }

    static Status _execute_default_expr_without_root_type_check(
            const VExprContextSPtr& default_expr, const Block* block,
            ColumnWithTypeAndName* result_data) {
        DORIS_CHECK(default_expr != nullptr);
        DORIS_CHECK(block != nullptr);
        DORIS_CHECK(result_data != nullptr);
        ColumnPtr result_column;
        Status st;
        RETURN_IF_CATCH_EXCEPTION({
            st = default_expr->root()->execute_column_impl(default_expr.get(), block, nullptr,
                                                           block->rows(), result_column);
        });
        RETURN_IF_ERROR(st);
        DORIS_CHECK(result_column.get() != nullptr);
        if (result_column->size() != block->rows()) {
            return Status::InternalError(
                    "Default expr {} return column size {} not equal to expected size {}",
                    default_expr->expr_name(), result_column->size(), block->rows());
        }
        result_data->column = result_column;
        result_data->type = default_expr->execute_type(block);
        result_data->name = default_expr->expr_name();
        return Status::OK();
    }

    Status _cast_column_to_type(ColumnPtr* column, const DataTypePtr& file_type,
                                const DataTypePtr& table_type,
                                const std::string& column_name) const {
        DORIS_CHECK(column != nullptr);
        DORIS_CHECK(column->get() != nullptr);
        DORIS_CHECK(file_type != nullptr);
        DORIS_CHECK(table_type != nullptr);
        if (file_type->equals(*table_type)) {
            return Status::OK();
        }

        DataTypePtr input_type = file_type;
        // Cast wrappers unwrap nullable inputs according to the declared input type, so keep the
        // root nullability of the declared type aligned with the actual column shape.
        if ((*column)->is_nullable() && !input_type->is_nullable()) {
            input_type = make_nullable(input_type);
        } else if (!(*column)->is_nullable() && input_type->is_nullable()) {
            input_type = remove_nullable(input_type);
        }
        Block cast_block;
        cast_block.insert({*column, input_type, column_name});
        auto slot_ref = VSlotRef::create_shared(0, 0, -1, input_type, column_name);
        auto cast_expr = Cast::create_shared(table_type);
        cast_expr->add_child(std::move(slot_ref));
        auto cast_ctx = VExprContext::create_shared(std::move(cast_expr));
        RowDescriptor row_desc;
        RETURN_IF_ERROR(cast_ctx->prepare(_runtime_state, row_desc));
        RETURN_IF_ERROR(cast_ctx->open(_runtime_state));
        ColumnPtr cast_column;
        RETURN_IF_ERROR(cast_ctx->execute(&cast_block, cast_column));
        *column = std::move(cast_column);
        return Status::OK();
    }

    Status _materialize_present_child_mapping_column(const ColumnMapping& mapping,
                                                     const ColumnPtr& file_column,
                                                     const size_t rows, ColumnPtr* column) {
        DORIS_CHECK(column != nullptr);
        DORIS_CHECK(mapping.file_type != nullptr);
        DORIS_CHECK(mapping.table_type != nullptr);
        *column = file_column;
        if (!mapping.is_trivial) {
            if (!mapping.child_mappings.empty()) {
                RETURN_IF_ERROR(
                        _materialize_complex_mapping_column(mapping, *column, rows, column));
            } else {
                RETURN_IF_ERROR(_cast_column_to_type(column, mapping.file_type, mapping.table_type,
                                                     mapping.file_column_name));
            }
        }
        RETURN_IF_ERROR(_align_column_nullability(column, mapping.table_type));
        return Status::OK();
    }

    Status _materialize_mapping_column(const ColumnMapping& mapping, Block* current_block,
                                       const size_t rows, ColumnPtr* column,
                                       bool take_projection_result = false) {
        if (!mapping.column_paths.empty()) {
            return _materialize_variant_path_mapping_column(mapping, current_block, rows, column,
                                                            take_projection_result);
        }
        if (!mapping.is_trivial && mapping.file_local_id.has_value() &&
            !mapping.child_mappings.empty()) {
            DCHECK(mapping.projection != nullptr);
            int res_id;
            auto st = mapping.projection->execute(current_block, &res_id);
            if (!st.ok()) {
                return Status::InternalError(
                        "Failed to execute complex mapping projection for table column '{}' "
                        "(global_index={}, file_local_id={}, rows={}): {}, mapping={}",
                        mapping.table_column_name, mapping.global_index.value(),
                        *mapping.file_local_id, rows, st.to_string(), mapping.debug_string());
            }
            ColumnPtr result_column = take_projection_result
                                              ? _take_and_detach_block_column(current_block, res_id)
                                              : current_block->get_by_position(res_id).column;
            RETURN_IF_ERROR(
                    _materialize_complex_mapping_column(mapping, result_column, rows, column));
            return Status::OK();
        }
        if (mapping.projection != nullptr) {
            int res_id;
            auto st = mapping.projection->execute(current_block, &res_id);
            if (!st.ok()) {
                std::string file_local_id = "null";
                if (mapping.file_local_id.has_value()) {
                    file_local_id = std::to_string(*mapping.file_local_id);
                }
                return Status::InternalError(
                        "Failed to execute mapping projection for table column '{}' "
                        "(global_index={}, file_local_id={}, rows={}): {}, mapping={}",
                        mapping.table_column_name, mapping.global_index.value(), file_local_id,
                        rows, st.to_string(), mapping.debug_string());
            }
            if (take_projection_result) {
                *column = _take_and_detach_block_column(current_block, res_id);
            } else {
                ColumnPtr result_column = current_block->get_by_position(res_id).column;
                *column = _detach_column(std::move(result_column));
            }
            return Status::OK();
        }
        if (mapping.default_expr != nullptr) {
            return _materialize_default_mapping_column(mapping, current_block, rows, column);
        }
        ColumnPtr result_column = mapping.table_type->create_column_const_with_default_value(rows);
        *column = _detach_column(std::move(result_column));
        return Status::OK();
    }

    Status _materialize_default_mapping_column(const ColumnMapping& mapping, Block* current_block,
                                               size_t rows, ColumnPtr* column) {
        DORIS_CHECK(mapping.default_expr != nullptr);
        DORIS_CHECK(mapping.table_type != nullptr);
        DORIS_CHECK(current_block != nullptr);
        DORIS_CHECK(column != nullptr);
        if (current_block->rows() == rows) {
            ColumnWithTypeAndName result;
            RETURN_IF_ERROR(_execute_default_expr_without_root_type_check(mapping.default_expr,
                                                                          current_block, &result));
            ColumnPtr result_column = result.column;
            RETURN_IF_ERROR(_align_column_nullability(&result_column, mapping.table_type));
            *column = _detach_column(std::move(result_column));
            return Status::OK();
        }

        DORIS_CHECK(mapping.constant_index.has_value());
        Block eval_block;
        eval_block.insert({mapping.table_type->create_column_const_with_default_value(rows),
                           mapping.table_type, "__table_reader_const_rows"});
        ColumnWithTypeAndName result;
        RETURN_IF_ERROR(_execute_default_expr_without_root_type_check(mapping.default_expr,
                                                                      &eval_block, &result));
        ColumnPtr result_column = result.column;
        RETURN_IF_ERROR(_align_column_nullability(&result_column, mapping.table_type));
        *column = _detach_column(std::move(result_column));
        return Status::OK();
    }

    Status _materialize_variant_path_mapping_column(const ColumnMapping& mapping,
                                                    Block* current_block, size_t rows,
                                                    ColumnPtr* column,
                                                    bool take_projection_result) {
        DORIS_CHECK(!mapping.column_paths.empty());
        DORIS_CHECK(mapping.resolved_variant_path != nullptr);
        DORIS_CHECK(mapping.table_type != nullptr);
        DORIS_CHECK(current_block != nullptr);
        DORIS_CHECK(column != nullptr);

        ColumnPtr root_column;
        if (mapping.projection != nullptr) {
            int result_id;
            auto status = mapping.projection->execute(current_block, &result_id);
            if (!status.ok()) {
                return Status::InternalError(
                        "Failed to read Variant root carrier for Path Slot '{}' "
                        "(global_index={}, rows={}): {}, mapping={}",
                        mapping.table_column_name, mapping.global_index.value(), rows,
                        status.to_string(), mapping.debug_string());
            }
            root_column = take_projection_result
                                  ? _take_and_detach_block_column(current_block, result_id)
                                  : current_block->get_by_position(result_id).column;
        } else if (mapping.default_expr != nullptr) {
            // Schema-evolution defaults describe the missing root column. Apply the same path
            // extraction after evaluating that root value instead of exposing the whole default.
            RETURN_IF_ERROR(_materialize_default_mapping_column(mapping, current_block, rows,
                                                                &root_column));
        } else {
            // A missing root without a default makes every requested subpath SQL NULL.
            *column = _detach_column(
                    mapping.table_type->create_column_const_with_default_value(rows));
            return Status::OK();
        }

        const ColumnPtr materialized = root_column->convert_to_full_column_if_const();
        _last_batch_materialization_input_bytes =
                std::max(_last_batch_materialization_input_bytes, materialized->byte_size());
        const IColumn* physical = materialized.get();
        std::span<const uint8_t> outer_nulls;
        if (const auto* nullable = check_and_get_column<ColumnNullable>(physical)) {
            outer_nulls = nullable->get_null_map_data();
            physical = &nullable->get_nested_column();
        }
        const auto* variant = check_and_get_column<ColumnVariantV2>(physical);
        if (variant == nullptr) {
            return Status::NotSupported(
                    "Variant Path Slot '{}' requires a ColumnVariantV2 root carrier, got {}. "
                    "Legacy ColumnVariant path extraction is not equivalent because it also needs "
                    "sparse/document fallback semantics",
                    mapping.table_column_name, materialized->get_name());
        }
        if (variant->size() != rows) {
            return Status::InternalError(
                    "Variant root carrier for Path Slot '{}' has {} rows, expected {}",
                    mapping.table_column_name, variant->size(), rows);
        }

        ColumnPtr result;
        RETURN_IF_ERROR(extract_variant_element_v2(*variant, *mapping.resolved_variant_path,
                                                   outer_nulls, &result));
        RETURN_IF_ERROR(_align_column_nullability(&result, mapping.table_type));
        *column = std::move(result);
        return Status::OK();
    }

    Status _materialize_complex_mapping_column(const ColumnMapping& mapping,
                                               const ColumnPtr& file_column, const size_t rows,
                                               ColumnPtr* column) {
        DORIS_CHECK(mapping.table_type != nullptr);
        DORIS_CHECK(file_column.get() != nullptr);
        const auto table_type = remove_nullable(mapping.table_type);
        switch (table_type->get_primitive_type()) {
        case TYPE_STRUCT:
            RETURN_IF_ERROR(_materialize_struct_mapping_column(mapping, file_column, rows, column));
            break;
        case TYPE_ARRAY:
            RETURN_IF_ERROR(_materialize_array_mapping_column(mapping, file_column, rows, column));
            break;
        case TYPE_MAP:
            RETURN_IF_ERROR(_materialize_map_mapping_column(mapping, file_column, rows, column));
            break;
        default:
            *column = _detach_column(file_column);
            break;
        }
        return Status::OK();
    }

    static std::vector<const ColumnMapping*> _present_child_mappings_in_file_order(
            const std::vector<ColumnMapping>& child_mappings) {
        std::vector<const ColumnMapping*> result;
        result.reserve(child_mappings.size());
        for (const auto& child_mapping : child_mappings) {
            if (child_mapping.file_local_id.has_value()) {
                result.push_back(&child_mapping);
            }
        }
        std::ranges::sort(result, [](const ColumnMapping* lhs, const ColumnMapping* rhs) {
            DORIS_CHECK(lhs->file_local_id.has_value());
            DORIS_CHECK(rhs->file_local_id.has_value());
            return *lhs->file_local_id < *rhs->file_local_id;
        });
        return result;
    }

    static size_t _file_child_ordinal_for_mapping(
            const ColumnMapping& mapping, const ColumnMapping& child_mapping,
            const std::vector<const ColumnMapping*>& file_ordered_children) {
        DORIS_CHECK(child_mapping.file_local_id.has_value());
        if (!mapping.projected_file_children.empty()) {
            const auto child_it = std::ranges::find_if(
                    mapping.projected_file_children, [&](const ColumnDefinition& file_child) {
                        return file_child.file_local_id() == *child_mapping.file_local_id;
                    });
            DORIS_CHECK(child_it != mapping.projected_file_children.end());
            return static_cast<size_t>(
                    std::distance(mapping.projected_file_children.begin(), child_it));
        }
        const auto child_it = std::ranges::find(file_ordered_children, &child_mapping);
        DORIS_CHECK(child_it != file_ordered_children.end());
        return static_cast<size_t>(std::distance(file_ordered_children.begin(), child_it));
    }

    static std::vector<const ColumnMapping*> _child_mappings_in_table_type_order(
            const ColumnMapping& mapping, const DataTypeStruct& table_type) {
        std::vector<const ColumnMapping*> result;
        result.reserve(mapping.child_mappings.size());
        for (size_t child_idx = 0; child_idx < table_type.get_elements().size(); ++child_idx) {
            const auto& child_name = table_type.get_element_name(child_idx);
            const auto child_it = std::ranges::find_if(
                    mapping.child_mappings, [&](const ColumnMapping& child_mapping) {
                        return child_mapping.table_column_name == child_name;
                    });
            DORIS_CHECK(child_it != mapping.child_mappings.end())
                    << mapping.debug_string() << ", table_child_name=" << child_name;
            result.push_back(&*child_it);
        }
        return result;
    }

    static const IColumn* _nested_column_if_nullable(const ColumnPtr& column,
                                                     const NullMap** null_map) {
        DORIS_CHECK(column.get() != nullptr);
        if (const auto* nullable_column = check_and_get_column<ColumnNullable>(*column)) {
            if (null_map != nullptr) {
                *null_map = &nullable_column->get_null_map_data();
            }
            return &nullable_column->get_nested_column();
        }
        return column.get();
    }

    Status _materialize_struct_mapping_column(const ColumnMapping& mapping,
                                              const ColumnPtr& file_column, const size_t rows,
                                              ColumnPtr* column) {
        DORIS_CHECK(mapping.table_type != nullptr);
        const auto* table_type =
                assert_cast<const DataTypeStruct*>(remove_nullable(mapping.table_type).get());
        const auto full_file_column = file_column->convert_to_full_column_if_const();
        const NullMap* parent_null_map = nullptr;
        const auto* nested_file_column =
                _nested_column_if_nullable(full_file_column, &parent_null_map);
        const auto* file_struct = assert_cast<const ColumnStruct*>(nested_file_column);
        DORIS_CHECK(table_type->get_elements().size() == mapping.child_mappings.size());

        Columns child_columns;
        child_columns.reserve(mapping.child_mappings.size());
        const auto file_ordered_children =
                _present_child_mappings_in_file_order(mapping.child_mappings);
        const auto table_ordered_children =
                _child_mappings_in_table_type_order(mapping, *table_type);
        for (const auto* child_mapping : table_ordered_children) {
            DORIS_CHECK(child_mapping != nullptr);
            if (!child_mapping->file_local_id.has_value()) {
                child_columns.push_back(
                        (child_mapping->initial_default_column
                                 ? child_mapping->initial_default_column->clone_resized(rows)
                                 : child_mapping->table_type
                                           ->create_column_const_with_default_value(rows))
                                ->convert_to_full_column_if_const());
                continue;
            }
            const auto file_child_idx =
                    _file_child_ordinal_for_mapping(mapping, *child_mapping, file_ordered_children);
            DORIS_CHECK(file_child_idx < file_struct->get_columns().size());
            ColumnPtr child_column = file_struct->get_column_ptr(file_child_idx);
            RETURN_IF_ERROR(_materialize_present_child_mapping_column(*child_mapping, child_column,
                                                                      rows, &child_column));
            child_columns.push_back(std::move(child_column));
        }
        MutableColumns mutable_child_columns;
        mutable_child_columns.reserve(child_columns.size());
        for (auto& child_column : child_columns) {
            mutable_child_columns.push_back(IColumn::mutate(std::move(child_column)));
        }
        auto result = ColumnStruct::create(std::move(mutable_child_columns));
        if (mapping.table_type->is_nullable()) {
            auto null_map = ColumnUInt8::create();
            auto& null_map_data = null_map->get_data();
            null_map_data.resize(rows);
            if (parent_null_map != nullptr) {
                DORIS_CHECK(parent_null_map->size() == rows);
                null_map_data.assign(parent_null_map->begin(), parent_null_map->end());
            } else {
                std::fill(null_map_data.begin(), null_map_data.end(), 0);
            }
            *column = ColumnNullable::create(std::move(result), std::move(null_map));
        } else {
            *column = std::move(result);
        }
        return Status::OK();
    }

    Status _materialize_array_mapping_column(const ColumnMapping& mapping,
                                             const ColumnPtr& file_column, const size_t rows,
                                             ColumnPtr* column) {
        DORIS_CHECK(mapping.child_mappings.size() == 1);
        const auto full_file_column = file_column->convert_to_full_column_if_const();
        const NullMap* parent_null_map = nullptr;
        const auto* nested_file_column =
                _nested_column_if_nullable(full_file_column, &parent_null_map);
        const auto* file_array = assert_cast<const ColumnArray*>(nested_file_column);
        ColumnPtr nested_column = file_array->get_data_ptr();
        const auto& element_mapping = mapping.child_mappings[0];
        RETURN_IF_ERROR(_materialize_present_child_mapping_column(
                element_mapping, nested_column, nested_column->size(), &nested_column));
        auto offsets_column = file_array->get_offsets_ptr()->convert_to_full_column_if_const();
        auto result = ColumnArray::create(IColumn::mutate(std::move(nested_column)),
                                          IColumn::mutate(std::move(offsets_column)));
        if (mapping.table_type->is_nullable()) {
            auto null_map = ColumnUInt8::create();
            auto& null_map_data = null_map->get_data();
            null_map_data.resize(rows);
            if (parent_null_map != nullptr) {
                DORIS_CHECK(parent_null_map->size() == rows);
                null_map_data.assign(parent_null_map->begin(), parent_null_map->end());
            } else {
                std::fill(null_map_data.begin(), null_map_data.end(), 0);
            }
            *column = ColumnNullable::create(std::move(result), std::move(null_map));
        } else {
            *column = std::move(result);
        }
        return Status::OK();
    }

    Status _materialize_map_mapping_column(const ColumnMapping& mapping,
                                           const ColumnPtr& file_column, const size_t rows,
                                           ColumnPtr* column) {
        const auto full_file_column = file_column->convert_to_full_column_if_const();
        const NullMap* parent_null_map = nullptr;
        const auto* nested_file_column =
                _nested_column_if_nullable(full_file_column, &parent_null_map);
        const auto* file_map = assert_cast<const ColumnMap*>(nested_file_column);
        ColumnPtr key_column = file_map->get_keys_ptr();
        ColumnPtr value_column = file_map->get_values_ptr();

        const ColumnMapping* key_mapping = nullptr;
        const ColumnMapping* value_mapping = nullptr;
        for (const auto& child_mapping : mapping.child_mappings) {
            if (!child_mapping.file_local_id.has_value()) {
                continue;
            }
            if (*child_mapping.file_local_id == 0) {
                key_mapping = &child_mapping;
            } else if (*child_mapping.file_local_id == 1) {
                value_mapping = &child_mapping;
            }
        }

        if (key_mapping != nullptr) {
            RETURN_IF_ERROR(_materialize_present_child_mapping_column(
                    *key_mapping, key_column, key_column->size(), &key_column));
        }
        if (value_mapping != nullptr) {
            RETURN_IF_ERROR(_materialize_present_child_mapping_column(
                    *value_mapping, value_column, value_column->size(), &value_column));
        }
        auto offsets_column = file_map->get_offsets_ptr()->convert_to_full_column_if_const();
        auto result = ColumnMap::create(IColumn::mutate(std::move(key_column)),
                                        IColumn::mutate(std::move(value_column)),
                                        IColumn::mutate(std::move(offsets_column)));
        if (mapping.table_type->is_nullable()) {
            auto null_map = ColumnUInt8::create();
            auto& null_map_data = null_map->get_data();
            null_map_data.resize(rows);
            if (parent_null_map != nullptr) {
                DORIS_CHECK(parent_null_map->size() == rows);
                null_map_data.assign(parent_null_map->begin(), parent_null_map->end());
            } else {
                std::fill(null_map_data.begin(), null_map_data.end(), 0);
            }
            *column = ColumnNullable::create(std::move(result), std::move(null_map));
        } else {
            *column = std::move(result);
        }
        return Status::OK();
    }

    Status _open_mapping_exprs() {
        RowDescriptor row_desc;
        for (const auto& mapping : _data_reader.column_mapper->mappings()) {
            if (mapping.projection != nullptr) {
                RETURN_IF_ERROR(mapping.projection->prepare(_runtime_state, row_desc));
                RETURN_IF_ERROR(mapping.projection->open(_runtime_state));
            }
            if (mapping.default_expr != nullptr) {
                RETURN_IF_ERROR(mapping.default_expr->prepare(_runtime_state, row_desc));
                RETURN_IF_ERROR(mapping.default_expr->open(_runtime_state));
            }
        }
        return Status::OK();
    }

    Status _build_file_aggregate_request(TPushAggOp::type agg_type,
                                         FileAggregateRequest* request) const {
        DORIS_CHECK(request != nullptr);
        DORIS_CHECK(_supports_aggregate_pushdown(agg_type));
        request->agg_type = agg_type;
        request->columns.clear();
        if (agg_type == TPushAggOp::type::COUNT) {
            DORIS_CHECK(_push_down_count_columns.has_value());
            // An empty explicit list is the semantic signal for COUNT(*). Do not inspect the
            // mapping count: `SELECT COUNT(*) FROM t` may still project one nullable column because
            // the planner keeps a placeholder slot. In a 10,000-row file where that arbitrary slot
            // has 9,015 non-null values, passing the slot would ask Parquet/ORC metadata for
            // COUNT(slot)=9,015 instead of the required row count 10,000.
            if (!_push_down_count_columns->empty()) {
                const auto& mapping = _push_down_count_mapping();
                DORIS_CHECK(mapping.file_local_id.has_value());
                FileAggregateRequest::Column column;
                column.projection =
                        LocalColumnIndex::top_level(LocalColumnId(*mapping.file_local_id));
                request->columns.push_back(std::move(column));
            }
            return Status::OK();
        }
        request->columns.reserve(_data_reader.column_mapper->mappings().size());
        for (const auto& mapping : _data_reader.column_mapper->mappings()) {
            DORIS_CHECK(mapping.file_local_id.has_value());
            FileAggregateRequest::Column column;
            column.projection = LocalColumnIndex::top_level(LocalColumnId(*mapping.file_local_id));
            if (!mapping.child_mappings.empty()) {
                RETURN_IF_ERROR(build_aggregate_projection(mapping, &column.projection));
            }
            request->columns.push_back(std::move(column));
        }
        return Status::OK();
    }

    const ColumnMapping& _push_down_count_mapping() const {
        DORIS_CHECK(_push_down_count_columns.has_value());
        DORIS_CHECK(_push_down_count_columns->size() == 1);
        const auto mapping_it =
                std::ranges::find(_data_reader.column_mapper->mappings(),
                                  _push_down_count_columns->front(), &ColumnMapping::global_index);
        // FileScannerV2 translates FE SlotIds through the same projected-column list used to build
        // the mapper, so a missing mapping is an FE/BE contract violation rather than a fallback.
        DORIS_CHECK(mapping_it != _data_reader.column_mapper->mappings().end());
        return *mapping_it;
    }

    Status _materialize_aggregate_pushdown_rows(TPushAggOp::type agg_type,
                                                const FileAggregateResult& file_result,
                                                Block* block) {
        DORIS_CHECK(agg_type == TPushAggOp::type::MINMAX);
        // MIN/MAX pushdown emits two rows, min first and max second, for each projected column.
        // The upper MIN/MAX aggregate consumes those two rows to produce the final aggregate value.
        DORIS_CHECK(file_result.columns.size() == _data_reader.column_mapper->mappings().size());
        DORIS_CHECK(block->columns() == _data_reader.column_mapper->mappings().size());
        Block file_block;
        file_block.reserve(_data_reader.file_block_layout.size());
        for (const auto& column : _data_reader.file_block_layout) {
            file_block.insert({column.type->create_column(), column.type, column.name});
        }
        for (size_t column_idx = 0; column_idx < file_result.columns.size(); ++column_idx) {
            const auto& result_column = file_result.columns[column_idx];
            if (!result_column.has_min || !result_column.has_max) {
                return Status::NotSupported("Missing min/max aggregate result for column {}",
                                            _projected_columns[column_idx].name);
            }
            bool found_file_column = false;
            for (size_t block_position = 0; block_position < _data_reader.file_block_layout.size();
                 ++block_position) {
                if (_data_reader.file_block_layout[block_position].file_column_id ==
                    file_result.columns[column_idx].projection.column_id()) {
                    found_file_column = true;
                    auto column = file_block.get_by_position(block_position)
                                          .type->create_column()
                                          ->assert_mutable();
                    RETURN_IF_ERROR(_insert_aggregate_projection_value(
                            file_result.columns[column_idx].projection, result_column.min_value,
                            column.get()));
                    RETURN_IF_ERROR(_insert_aggregate_projection_value(
                            file_result.columns[column_idx].projection, result_column.max_value,
                            column.get()));
                    file_block.replace_by_position(block_position, std::move(column));
                    break;
                }
            }
            DORIS_CHECK(found_file_column);
        }
        for (size_t column_idx = 0; column_idx < _data_reader.column_mapper->mappings().size();
             ++column_idx) {
            ColumnPtr table_column;
            RETURN_IF_ERROR(_materialize_mapping_column(
                    _data_reader.column_mapper->mappings()[column_idx], &file_block, 2,
                    &table_column,
                    column_idx + 1 == _data_reader.column_mapper->mappings().size()));
            block->replace_by_position(column_idx, std::move(table_column));
        }
        return Status::OK();
    }

    struct FileBlockColumn {
        LocalColumnId file_column_id = LocalColumnId::invalid();
        std::string name;
        DataTypePtr type;
    };

    struct DataReader {
        std::unique_ptr<FileReader> reader;
        std::unique_ptr<TableColumnMapper> column_mapper;
        // Schema of the data file, also including virtual column (row position).
        std::vector<ColumnDefinition> file_schema;
        // Layout of the block returned by file reader, determined by column mapping and file
        // schema. It is used for file reader to materialize columns into correct type and position.
        std::vector<FileBlockColumn> file_block_layout;
        Block block_template;
    };
    DataReader _data_reader;
    std::vector<ColumnDefinition> _projected_columns;
    std::unique_ptr<ScanTask> _current_task;
    std::optional<io::FileDescription> _current_file_description;
    // Range-level compression has higher priority than scan-param compression. TVF/load can keep
    // the logical format as CSV/TEXT while carrying the concrete compression such as GZ or LZO on
    // each TFileRangeDesc, matching the old FileScanner reader contract.
    TFileCompressType::type _current_range_compress_type = TFileCompressType::UNKNOWN;
    std::optional<TUniqueId> _current_range_load_id;
    TFileRangeDesc _current_file_range_desc;
    std::shared_ptr<io::FileSystemProperties> _system_properties;
    // partition key -> value
    std::map<std::string, Field> _partition_values;
    // Predicates built from scan conjuncts before file-level localization.
    std::vector<TableFilter> _table_filters;
    // Number of localized filters before the first unsafe conjunct in the original row-level
    // order. This differs from scanning `_table_filters` for safety because slotless predicates are
    // intentionally absent from that vector but must still act as ordering barriers.
    size_t _constant_pruning_safe_filter_count = 0;
    VExprContextSPtrs _conjuncts;
    ReadProfile _profile;
    // Parsed from row-position based delete files, including position delete and deletion vector.
    DeleteRows* _delete_rows = nullptr;
    DeletionVector* _deletion_vector = nullptr;
    TFileScanRangeParams* _scan_params;
    std::shared_ptr<io::IOContext> _io_ctx;
    RuntimeState* _runtime_state;
    RuntimeProfile* _scanner_profile;
    const std::vector<SlotDescriptor*>* _file_slot_descs = nullptr;
    FileFormat _format;
    TPushAggOp::type _push_down_agg_type = TPushAggOp::type::NONE;
    std::optional<std::vector<GlobalIndex>> _push_down_count_columns;
    size_t _batch_size = 0;
    size_t _last_batch_materialization_input_bytes = 0;
    uint64_t _initial_condition_cache_digest = 0;
    uint64_t _condition_cache_digest = 0;
    // True only when prepare_split() received a digest for the exact conjunct snapshot used by
    // this split. Standalone callers that only supplied TableReadOptions::condition_cache_digest
    // keep the conservative runtime-filter guard.
    bool _condition_cache_digest_covers_current_split = false;
    segment_v2::ConditionCache::ExternalCacheKey _condition_cache_key;
    std::shared_ptr<std::vector<bool>> _condition_cache;
    std::shared_ptr<ConditionCacheContext> _condition_cache_ctx;
    int64_t _condition_cache_hit_count = 0;
    bool _current_reader_reached_eof = false;
    int64_t _remaining_table_level_count = -1;
    int64_t _remaining_file_level_count = -1;
    // True only after the active split selects a table-level row-count shortcut or successfully
    // materializes COUNT rows from file metadata. FileScannerV2 uses this result, rather than the
    // raw aggregate opcode, to keep adaptive batching enabled for normal row-scan fallbacks.
    bool _current_split_uses_metadata_count = false;
    // Snapshot supplied by FileScannerV2 for the active split. It gates every shortcut that emits
    // irreversible aggregate rows, not only the table-level row-count shortcut in prepare_split().
    bool _all_runtime_filters_applied_for_split = true;
    std::optional<GlobalRowIdContext> _global_rowid_context;
    bool _aggregate_pushdown_tried = false;
    bool _current_split_pruned = false;
    TableColumnMapperOptions _mapper_options;

private:
    static const ColumnDefinition* _find_column_definition(
            const std::vector<ColumnDefinition>& schema, LocalColumnId column_id) {
        for (const auto& field : schema) {
            if (field.file_local_id() == column_id.value()) {
                return &field;
            }
        }
        return nullptr;
    }

    static bool _can_push_down_minmax_for_mapping(const ColumnMapping& mapping) {
        if (mapping.child_mappings.empty()) {
            // Direct mappings use a slot-ref projection to materialize the file column. The
            // projection does not transform ordering; casts and other conversions are already
            // represented by a non-trivial mapping and must fall back to row scanning.
            return mapping.is_trivial;
        }
        const auto primitive_type = remove_nullable(mapping.file_type)->get_primitive_type();
        if (primitive_type != TYPE_STRUCT) {
            return false;
        }
        size_t mapped_children = 0;
        const ColumnMapping* mapped_child = nullptr;
        for (const auto& child_mapping : mapping.child_mappings) {
            if (!child_mapping.file_local_id.has_value()) {
                continue;
            }
            ++mapped_children;
            mapped_child = &child_mapping;
        }
        return mapped_children == 1 && mapped_child != nullptr &&
               _can_push_down_minmax_for_mapping(*mapped_child);
    }

    static Status build_aggregate_projection(const ColumnMapping& mapping,
                                             LocalColumnIndex* projection) {
        DORIS_CHECK(projection != nullptr);
        DORIS_CHECK(mapping.file_local_id.has_value());
        *projection = LocalColumnIndex::local(*mapping.file_local_id);
        projection->children.clear();
        projection->project_all_children = true;
        if (mapping.child_mappings.empty()) {
            return Status::OK();
        }
        projection->project_all_children = false;
        for (const auto& child_mapping : mapping.child_mappings) {
            if (!child_mapping.file_local_id.has_value()) {
                continue;
            }
            LocalColumnIndex child_projection;
            RETURN_IF_ERROR(build_aggregate_projection(child_mapping, &child_projection));
            projection->children.push_back(std::move(child_projection));
        }
        DORIS_CHECK(projection->children.size() == 1);
        return Status::OK();
    }

    static Status _insert_aggregate_projection_value(const LocalColumnIndex& projection,
                                                     const Field& value, IColumn* column) {
        DORIS_CHECK(column != nullptr);
        if (auto* nullable_column = check_and_get_column<ColumnNullable>(*column)) {
            RETURN_IF_ERROR(_insert_aggregate_projection_value(
                    projection, value, &nullable_column->get_nested_column()));
            nullable_column->get_null_map_data().push_back(0);
            return Status::OK();
        }
        if (projection.project_all_children || projection.children.empty()) {
            column->insert(value);
            return Status::OK();
        }
        auto* struct_column = assert_cast<ColumnStruct*>(column);
        DORIS_CHECK(projection.children.size() == 1);
        const auto& child_projection = projection.children[0];
        DORIS_CHECK(struct_column->get_columns().size() == 1);
        RETURN_IF_ERROR(_insert_aggregate_projection_value(child_projection, value,
                                                           &struct_column->get_column(0)));
        return Status::OK();
    }

    // Parse a DV into its compressed bitmap. Position delete files continue to use _delete_rows.
    Status _parse_delete_predicates(const SplitReadOptions& options);
};

} // namespace doris::format

Coverage Report

Created: 2026-07-31 00:52