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

};

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;

    std::optional<ColumnDefinition> schema_column = std::nullopt;

    size_t next_file_column_idx = 0;

};

struct ReadProfile {

    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;

};

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;

    // Digest of stable pushed-down predicates. 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;

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

    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; }

    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; }

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

        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;

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

                }

                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;

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

        if (_data_reader.reader) {

            RETURN_IF_ERROR(close_current_reader());

        }

        _current_task.reset();

        _current_file_description.reset();

        _remaining_table_level_count = -1;

        return Status::OK();

    }

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

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

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

        }

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

        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; }

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

            column->resize(rows);

            block->replace_by_position(column_idx, std::move(column));

        }

        return Status::OK();

    }

    Status _read_table_level_count(Block* block, bool* eos) {

        DORIS_CHECK(block != nullptr);

        DORIS_CHECK(eos != nullptr);

        DORIS_CHECK(_push_down_agg_type == TPushAggOp::type::COUNT);

        DORIS_CHECK(_remaining_table_level_count >= 0);

        if (_remaining_table_level_count == 0) {

            _remaining_table_level_count = -1;

            _current_task.reset();

            *eos = true;

            return Status::OK();

        }

        const int64_t batch_size = _runtime_state == nullptr

                                           ? _remaining_table_level_count

                                           : static_cast<int64_t>(_runtime_state->batch_size());

        const auto rows = std::min(_remaining_table_level_count, batch_size);

        RETURN_IF_ERROR(_materialize_count_rows(cast_set<size_t>(rows), block));

        _remaining_table_level_count -= rows;

        *eos = false;

        return Status::OK();

    }

    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

632

12

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

633

12

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

634

12

            delete_predicate->add_child(VSlotRef::create_shared(

635

12

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

636

12

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

637

12

            request->delete_conjuncts.push_back(

638

12

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

639

12

        };

_ZZN5doris6format11TableReader24_append_delete_predicateEPNS0_15FileScanRequestEENKUlRT_E_clIN7roaring12Roaring64MapEEEDaS5_

Line

Count

Source

632

4

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

633

4

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

634

4

            delete_predicate->add_child(VSlotRef::create_shared(

635

4

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

636

4

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

637

4

            request->delete_conjuncts.push_back(

638

4

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

639

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

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

        size_t idx = 0;

        for (const auto& mapping : _data_reader.column_mapper->mappings()) {

            ColumnPtr column;

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

                                                        &column));

            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;

        }

        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;

        const auto 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);

        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) {

            return true;

        }

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