// 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 <string>

#include <string_view>

#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/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/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;

    bool metadata_pruning_safe = true;

};

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* refresh_conjuncts_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_refresh_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);

    // Refresh row-level predicates for an already prepared split. Physical readers that support

    // this operation decide the safe boundary at which the new immutable request becomes active.

    virtual Status refresh_conjuncts(VExprContextSPtrs conjuncts);

    virtual bool current_split_pruned() const { return _current_split_pruned; }

    virtual bool current_split_uses_metadata_count() const {

        return _current_split_uses_metadata_count;

    }

    // 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());

        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_table_column_by_field_id(

            int32_t field_id, DataTypePtr type, bool include_historical_schemas) const;

    std::optional<std::vector<ColumnDefinition>> _find_table_column_path_by_field_id(

            int32_t field_id, DataTypePtr leaf_type, bool include_historical_schemas) const;

    std::optional<std::vector<ColumnDefinition>> _find_table_column_identity_path_by_field_id(

            int32_t field_id, bool include_historical_schemas) 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();

    }

    virtual Status validate_file_mapping(const TableColumnMapper&) const { 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));

        _constant_pruning_safe_filter_count =

                std::min(_constant_pruning_safe_filter_count,

                         file_request->constant_pruning_safe_table_filter_count);

        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();

        }

        RETURN_IF_ERROR(validate_file_mapping(*_data_reader.column_mapper));

        // 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.

        // Pending runtime filters may later target this retained slot, so placeholder values are

        // safe only after every filter for the split has arrived.

        if (_push_down_agg_type == TPushAggOp::type::COUNT &&

            _push_down_count_columns.has_value() && _push_down_count_columns->empty() &&

            _all_runtime_filters_applied_for_split) {

            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();

        _file_scan_request.reset();

        _data_reader.file_block_layout.resize(file_request->block_column_count());

        // 4. Build file block layout from file schema and column mapping. The layout describes

        // the block returned by file reader before table-column materialization.

        auto add_file_block_column = [&](const LocalColumnIndex& projection,

                                         LocalIndex block_position) -> Status {

            DORIS_CHECK(block_position.value() < _data_reader.file_block_layout.size());

            const auto file_column_id = projection.column_id();

            const auto* field = _find_column_definition(_data_reader.file_schema, file_column_id);

            DORIS_CHECK(field != nullptr);

            ColumnDefinition projected_field;

            RETURN_IF_ERROR(project_column_definition(*field, projection, &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);

            return Status::OK();

        };

        for (const auto& projection : file_request->predicate_columns) {

            RETURN_IF_ERROR(add_file_block_column(

                    projection, file_request->local_positions.at(projection.column_id())));

        }

        for (const auto& projection : file_request->non_predicate_columns) {

            RETURN_IF_ERROR(add_file_block_column(

                    projection, file_request->non_predicate_position(projection.column_id())));

        }

        // 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));

        }

        _file_scan_request = std::move(file_request);

        RETURN_IF_ERROR(_init_reader_condition_cache(*_file_scan_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)));

        };

_ZZN5doris6format11TableReader24_append_delete_predicateEPNS0_15FileScanRequestEENKUlRT_E_clISt6vectorIlSaIlEEEEDaS5_

Line

Count

Source

745

12

        auto append_predicate = [&](auto& deleted_rows) {

746

12

            auto delete_predicate = std::make_shared<DeletePredicate>(deleted_rows);

747

12

            delete_predicate->add_child(VSlotRef::create_shared(

748

12

                    cast_set<int>(block_position.value()), cast_set<int>(block_position.value()),

749

12

                    -1, std::make_shared<DataTypeInt64>(), ROW_POSITION_COLUMN_NAME));

750

12

            request->delete_conjuncts.push_back(

751

12

                    VExprContext::create_shared(std::move(delete_predicate)));

752

12

        };

_ZZN5doris6format11TableReader24_append_delete_predicateEPNS0_15FileScanRequestEENKUlRT_E_clIN7roaring12Roaring64MapEEEDaS5_

Line

Count

Source

745

4

        auto append_predicate = [&](auto& deleted_rows) {

746

4

            auto delete_predicate = std::make_shared<DeletePredicate>(deleted_rows);

747

4

            delete_predicate->add_child(VSlotRef::create_shared(

748

4

                    cast_set<int>(block_position.value()), cast_set<int>(block_position.value()),

749

4

                    -1, std::make_shared<DataTypeInt64>(), ROW_POSITION_COLUMN_NAME));

750

4

            request->delete_conjuncts.push_back(

751

4

                    VExprContext::create_shared(std::move(delete_predicate)));

752

4

        };

        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();

        _file_scan_request.reset();

        _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);

        size_t idx = 0;

        const auto& mappings = _data_reader.column_mapper->mappings();

        for (const auto& mapping : mappings) {

            ColumnPtr column;

            RETURN_IF_ERROR(_materialize_mapping_column(mapping, &_data_reader.block_template, rows,

                                                        &column, idx + 1 == mappings.size()));

            block->replace_by_position(idx, IColumn::mutate(std::move(column)));

            idx++;

        }

        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;