// Licensed to the Apache Software Foundation (ASF) under one

// or more contributor license agreements.  See the NOTICE file

// distributed with this work for additional information

// regarding copyright ownership.  The ASF licenses this file

// to you under the Apache License, Version 2.0 (the

// "License"); you may not use this file except in compliance

// with the License.  You may obtain a copy of the License at

//

//   http://www.apache.org/licenses/LICENSE-2.0

//

// Unless required by applicable law or agreed to in writing,

// software distributed under the License is distributed on an

// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY

// KIND, either express or implied.  See the License for the

// specific language governing permissions and limitations

// under the License.

#include "format/parquet/vparquet_group_reader.h"

#include <gen_cpp/Exprs_types.h>

#include <gen_cpp/Opcodes_types.h>

#include <gen_cpp/Types_types.h>

#include <gen_cpp/parquet_types.h>

#include <string.h>

#include <algorithm>

#include <boost/iterator/iterator_facade.hpp>

#include <memory>

#include <ostream>

#include "common/config.h"

#include "common/consts.h"

#include "common/logging.h"

#include "common/object_pool.h"

#include "common/status.h"

#include "core/assert_cast.h"

#include "core/block/block.h"

#include "core/block/column_with_type_and_name.h"

#include "core/column/column_const.h"

#include "core/column/column_nullable.h"

#include "core/column/column_string.h"

#include "core/column/column_struct.h"

#include "core/column/column_vector.h"

#include "core/custom_allocator.h"

#include "core/data_type/data_type.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/data_type/define_primitive_type.h"

#include "core/pod_array.h"

#include "core/types.h"

#include "exprs/create_predicate_function.h"

#include "exprs/hybrid_set.h"

#include "exprs/vdirect_in_predicate.h"

#include "exprs/vectorized_fn_call.h"

#include "exprs/vexpr.h"

#include "exprs/vexpr_context.h"

#include "exprs/vliteral.h"

#include "exprs/vslot_ref.h"

#include "format/parquet/schema_desc.h"

#include "format/parquet/vparquet_column_reader.h"

#include "format/table/iceberg_reader.h"

#include "runtime/descriptors.h"

#include "runtime/runtime_state.h"

#include "runtime/thread_context.h"

#include "storage/segment/column_reader.h"

namespace cctz {

class time_zone;

} // namespace cctz

namespace doris {

class RuntimeState;

namespace io {

struct IOContext;

} // namespace io

} // namespace doris

namespace doris {

#include "common/compile_check_begin.h"

namespace {

Status build_iceberg_rowid_column(const DataTypePtr& type, const std::string& file_path,

                                  const std::vector<rowid_t>& row_ids, int32_t partition_spec_id,

                                  const std::string& partition_data_json,

                                  MutableColumnPtr* column_out) {

    if (type == nullptr || column_out == nullptr) {

        return Status::InvalidArgument("Invalid iceberg rowid column type or output column");

    }

    MutableColumnPtr column = type->create_column();

    ColumnNullable* nullable_col = check_and_get_column<ColumnNullable>(column.get());

    ColumnStruct* struct_col = nullptr;

    if (nullable_col != nullptr) {

        struct_col =

                check_and_get_column<ColumnStruct>(nullable_col->get_nested_column_ptr().get());

    } else {

        struct_col = check_and_get_column<ColumnStruct>(column.get());

    }

    if (struct_col == nullptr || struct_col->tuple_size() < 4) {

        return Status::InternalError("Invalid iceberg rowid column structure");

    }

    size_t num_rows = row_ids.size();

    auto& file_path_col = struct_col->get_column(0);

    auto& row_pos_col = struct_col->get_column(1);

    auto& spec_id_col = struct_col->get_column(2);

    auto& partition_data_col = struct_col->get_column(3);

    file_path_col.reserve(num_rows);

    row_pos_col.reserve(num_rows);

    spec_id_col.reserve(num_rows);

    partition_data_col.reserve(num_rows);

    for (size_t i = 0; i < num_rows; ++i) {

        file_path_col.insert_data(file_path.data(), file_path.size());

    }

    for (size_t i = 0; i < num_rows; ++i) {

        int64_t row_pos = static_cast<int64_t>(row_ids[i]);

        row_pos_col.insert_data(reinterpret_cast<const char*>(&row_pos), sizeof(row_pos));

    }

    for (size_t i = 0; i < num_rows; ++i) {

        int32_t spec_id = partition_spec_id;

        spec_id_col.insert_data(reinterpret_cast<const char*>(&spec_id), sizeof(spec_id));

    }

    for (size_t i = 0; i < num_rows; ++i) {

        partition_data_col.insert_data(partition_data_json.data(), partition_data_json.size());

    }

    if (nullable_col != nullptr) {

        nullable_col->get_null_map_data().resize_fill(num_rows, 0);

    }

    *column_out = std::move(column);

    return Status::OK();

}

} // namespace

const std::vector<int64_t> RowGroupReader::NO_DELETE = {};

static constexpr uint32_t MAX_DICT_CODE_PREDICATE_TO_REWRITE = std::numeric_limits<uint32_t>::max();

RowGroupReader::RowGroupReader(io::FileReaderSPtr file_reader,

                               const std::vector<std::string>& read_columns,

                               const int32_t row_group_id, const tparquet::RowGroup& row_group,

                               const cctz::time_zone* ctz, io::IOContext* io_ctx,

                               const PositionDeleteContext& position_delete_ctx,

                               const LazyReadContext& lazy_read_ctx, RuntimeState* state,

                               const std::set<uint64_t>& column_ids,

                               const std::set<uint64_t>& filter_column_ids)

        : _file_reader(file_reader),

          _read_table_columns(read_columns),

          _row_group_id(row_group_id),

          _row_group_meta(row_group),

          _remaining_rows(row_group.num_rows),

          _ctz(ctz),

          _io_ctx(io_ctx),

          _position_delete_ctx(position_delete_ctx),

          _lazy_read_ctx(lazy_read_ctx),

          _state(state),

          _obj_pool(new ObjectPool()),

          _column_ids(column_ids),

          _filter_column_ids(filter_column_ids) {}

RowGroupReader::~RowGroupReader() {

    _column_readers.clear();

    _obj_pool->clear();

}

Status RowGroupReader::init(

        const FieldDescriptor& schema, RowRanges& row_ranges,

        std::unordered_map<int, tparquet::OffsetIndex>& col_offsets,

        const TupleDescriptor* tuple_descriptor, const RowDescriptor* row_descriptor,

        const std::unordered_map<std::string, int>* colname_to_slot_id,

        const VExprContextSPtrs* not_single_slot_filter_conjuncts,

        const std::unordered_map<int, VExprContextSPtrs>* slot_id_to_filter_conjuncts) {

    _tuple_descriptor = tuple_descriptor;

    _row_descriptor = row_descriptor;

    _col_name_to_slot_id = colname_to_slot_id;

    _slot_id_to_filter_conjuncts = slot_id_to_filter_conjuncts;

    _read_ranges = row_ranges;

    _filter_read_ranges_by_condition_cache();

    _remaining_rows = _read_ranges.count();

    if (_read_table_columns.empty()) {

        // Query task that only select columns in path.

        return Status::OK();

    }

    const size_t MAX_GROUP_BUF_SIZE = config::parquet_rowgroup_max_buffer_mb << 20;

    const size_t MAX_COLUMN_BUF_SIZE = config::parquet_column_max_buffer_mb << 20;

    size_t max_buf_size =

            std::min(MAX_COLUMN_BUF_SIZE, MAX_GROUP_BUF_SIZE / _read_table_columns.size());

    for (const auto& read_table_col : _read_table_columns) {

        auto read_file_col = _table_info_node_ptr->children_file_column_name(read_table_col);

        auto* field = schema.get_column(read_file_col);

        std::unique_ptr<ParquetColumnReader> reader;

        RETURN_IF_ERROR(ParquetColumnReader::create(

                _file_reader, field, _row_group_meta, _read_ranges, _ctz, _io_ctx, reader,

                max_buf_size, col_offsets, _state, false, _column_ids, _filter_column_ids));

        if (reader == nullptr) {

            VLOG_DEBUG << "Init row group(" << _row_group_id << ") reader failed";

            return Status::Corruption("Init row group reader failed");

        }

        _column_readers[read_table_col] = std::move(reader);

    }

    bool disable_dict_filter = false;

    if (not_single_slot_filter_conjuncts != nullptr && !not_single_slot_filter_conjuncts->empty()) {

        disable_dict_filter = true;

        _filter_conjuncts.insert(_filter_conjuncts.end(), not_single_slot_filter_conjuncts->begin(),

                                 not_single_slot_filter_conjuncts->end());

    }

    // Check if single slot can be filtered by dict.

    if (_slot_id_to_filter_conjuncts && !_slot_id_to_filter_conjuncts->empty()) {

        const std::vector<std::string>& predicate_col_names =

                _lazy_read_ctx.predicate_columns.first;

        const std::vector<int>& predicate_col_slot_ids = _lazy_read_ctx.predicate_columns.second;

        for (size_t i = 0; i < predicate_col_names.size(); ++i) {

            const std::string& predicate_col_name = predicate_col_names[i];

            int slot_id = predicate_col_slot_ids[i];

            if (predicate_col_name == IcebergTableReader::ROW_LINEAGE_ROW_ID ||

                predicate_col_name == IcebergTableReader::ROW_LINEAGE_LAST_UPDATED_SEQ_NUMBER) {

                // row lineage column can not dict filter.

                if (_slot_id_to_filter_conjuncts->find(slot_id) !=

                    _slot_id_to_filter_conjuncts->end()) {

                    for (auto& ctx : _slot_id_to_filter_conjuncts->at(slot_id)) {

                        _filter_conjuncts.push_back(ctx);

                    }

                }

                continue;

            }

            auto predicate_file_col_name =

                    _table_info_node_ptr->children_file_column_name(predicate_col_name);

            auto field = schema.get_column(predicate_file_col_name);

            if (!disable_dict_filter && !_lazy_read_ctx.has_complex_type &&

                _can_filter_by_dict(

                        slot_id, _row_group_meta.columns[field->physical_column_index].meta_data)) {

                _dict_filter_cols.emplace_back(std::make_pair(predicate_col_name, slot_id));

            } else {

                if (_slot_id_to_filter_conjuncts->find(slot_id) !=

                    _slot_id_to_filter_conjuncts->end()) {

                    for (auto& ctx : _slot_id_to_filter_conjuncts->at(slot_id)) {

                        _filter_conjuncts.push_back(ctx);

                    }

                }

            }

        }

        // Add predicate_partition_columns in _slot_id_to_filter_conjuncts(single slot conjuncts)

        // to _filter_conjuncts, others should be added from not_single_slot_filter_conjuncts.

        for (auto& kv : _lazy_read_ctx.predicate_partition_columns) {

            auto& [value, slot_desc] = kv.second;

            auto iter = _slot_id_to_filter_conjuncts->find(slot_desc->id());

            if (iter != _slot_id_to_filter_conjuncts->end()) {

                for (auto& ctx : iter->second) {

                    _filter_conjuncts.push_back(ctx);

                }

            }

        }

        //For check missing column :   missing column == xx, missing column is null,missing column is not null.

        _filter_conjuncts.insert(_filter_conjuncts.end(),

                                 _lazy_read_ctx.missing_columns_conjuncts.begin(),

                                 _lazy_read_ctx.missing_columns_conjuncts.end());

        RETURN_IF_ERROR(_rewrite_dict_predicates());

    }

    // _state is nullptr in some ut.

    if (_state && _state->enable_adjust_conjunct_order_by_cost()) {

        std::ranges::sort(_filter_conjuncts, [](const auto& a, const auto& b) {

            return a->execute_cost() < b->execute_cost();

        });

    }

    return Status::OK();

}

bool RowGroupReader::_can_filter_by_dict(int slot_id,

                                         const tparquet::ColumnMetaData& column_metadata) {

    SlotDescriptor* slot = nullptr;

    const std::vector<SlotDescriptor*>& slots = _tuple_descriptor->slots();

    for (auto each : slots) {

        if (each->id() == slot_id) {

            slot = each;

            break;

        }

    }

    if (!is_string_type(slot->type()->get_primitive_type()) &&

        !is_var_len_object(slot->type()->get_primitive_type())) {

        return false;

    }

    if (column_metadata.type != tparquet::Type::BYTE_ARRAY) {

        return false;

    }

    if (!is_dictionary_encoded(column_metadata)) {

        return false;

    }

    if (_slot_id_to_filter_conjuncts->find(slot_id) == _slot_id_to_filter_conjuncts->end()) {

        return false;

    }

    // TODO: The current implementation of dictionary filtering does not take into account

    //  the implementation of NULL values because the dictionary itself does not contain

    //  NULL value encoding. As a result, many NULL-related functions or expressions

    //  cannot work properly, such as is null, is not null, coalesce, etc.

    //  Here we check if the predicate expr is IN or BINARY_PRED.

    //  Implementation of NULL value dictionary filtering will be carried out later.

    return std::ranges::all_of(_slot_id_to_filter_conjuncts->at(slot_id), [&](const auto& ctx) {

        return (ctx->root()->node_type() == TExprNodeType::IN_PRED ||

                ctx->root()->node_type() == TExprNodeType::BINARY_PRED) &&

               ctx->root()->children()[0]->node_type() == TExprNodeType::SLOT_REF;

    });

}

// This function is copied from

// https://github.com/apache/impala/blob/master/be/src/exec/parquet/hdfs-parquet-scanner.cc#L1717

bool RowGroupReader::is_dictionary_encoded(const tparquet::ColumnMetaData& column_metadata) {

    // The Parquet spec allows for column chunks to have mixed encodings

    // where some data pages are dictionary-encoded and others are plain

    // encoded. For example, a Parquet file writer might start writing

    // a column chunk as dictionary encoded, but it will switch to plain

    // encoding if the dictionary grows too large.

    //

    // In order for dictionary filters to skip the entire row group,

    // the conjuncts must be evaluated on column chunks that are entirely

    // encoded with the dictionary encoding. There are two checks

    // available to verify this:

    // 1. The encoding_stats field on the column chunk metadata provides

    //    information about the number of data pages written in each

    //    format. This allows for a specific check of whether all the

    //    data pages are dictionary encoded.

    // 2. The encodings field on the column chunk metadata lists the

    //    encodings used. If this list contains the dictionary encoding

    //    and does not include unexpected encodings (i.e. encodings not

    //    associated with definition/repetition levels), then it is entirely

    //    dictionary encoded.

    if (column_metadata.__isset.encoding_stats) {

        // Condition #1 above

        for (const tparquet::PageEncodingStats& enc_stat : column_metadata.encoding_stats) {

            if (enc_stat.page_type == tparquet::PageType::DATA_PAGE &&

                (enc_stat.encoding != tparquet::Encoding::PLAIN_DICTIONARY &&

                 enc_stat.encoding != tparquet::Encoding::RLE_DICTIONARY) &&

                enc_stat.count > 0) {

                return false;

            }

        }

    } else {

        // Condition #2 above

        bool has_dict_encoding = false;

        bool has_nondict_encoding = false;

        for (const tparquet::Encoding::type& encoding : column_metadata.encodings) {

            if (encoding == tparquet::Encoding::PLAIN_DICTIONARY ||

                encoding == tparquet::Encoding::RLE_DICTIONARY) {

                has_dict_encoding = true;

            }

            // RLE and BIT_PACKED are used for repetition/definition levels

            if (encoding != tparquet::Encoding::PLAIN_DICTIONARY &&

                encoding != tparquet::Encoding::RLE_DICTIONARY &&

                encoding != tparquet::Encoding::RLE && encoding != tparquet::Encoding::BIT_PACKED) {

                has_nondict_encoding = true;

                break;

            }

        }

        // Not entirely dictionary encoded if:

        // 1. No dictionary encoding listed

        // OR

        // 2. Some non-dictionary encoding is listed

        if (!has_dict_encoding || has_nondict_encoding) {

            return false;

        }

    }

    return true;

}

Status RowGroupReader::next_batch(Block* block, size_t batch_size, size_t* read_rows,

                                  bool* batch_eof) {

    if (_is_row_group_filtered) {

        *read_rows = 0;

        *batch_eof = true;

        return Status::OK();

    }

    // Process external table query task that select columns are all from path.

    if (_read_table_columns.empty()) {

        bool modify_row_ids = false;

        RETURN_IF_ERROR(_read_empty_batch(batch_size, read_rows, batch_eof, &modify_row_ids));

        RETURN_IF_ERROR(

                _fill_partition_columns(block, *read_rows, _lazy_read_ctx.partition_columns));

        RETURN_IF_ERROR(_fill_missing_columns(block, *read_rows, _lazy_read_ctx.missing_columns));

        RETURN_IF_ERROR(_fill_row_id_columns(block, *read_rows, modify_row_ids));

        RETURN_IF_ERROR(_append_iceberg_rowid_column(block, *read_rows, modify_row_ids));

        Status st = VExprContext::filter_block(_lazy_read_ctx.conjuncts, block, block->columns());

        *read_rows = block->rows();

        return st;

    }

    if (_lazy_read_ctx.can_lazy_read) {

        // call _do_lazy_read recursively when current batch is skipped

        return _do_lazy_read(block, batch_size, read_rows, batch_eof);

    } else {

        FilterMap filter_map;

        int64_t batch_base_row = _total_read_rows;

        RETURN_IF_ERROR((_read_column_data(block, _lazy_read_ctx.all_read_columns, batch_size,

                                           read_rows, batch_eof, filter_map)));

        RETURN_IF_ERROR(

                _fill_partition_columns(block, *read_rows, _lazy_read_ctx.partition_columns));

        RETURN_IF_ERROR(_fill_missing_columns(block, *read_rows, _lazy_read_ctx.missing_columns));

        RETURN_IF_ERROR(_fill_row_id_columns(block, *read_rows, false));

        RETURN_IF_ERROR(_append_iceberg_rowid_column(block, *read_rows, false));

#ifndef NDEBUG

        for (auto col : *block) {

            col.column->sanity_check();

            DCHECK(block->rows() == col.column->size())

                    << absl::Substitute("block rows = $0 , column rows = $1, col name = $2",

                                        block->rows(), col.column->size(), col.name);

        }

#endif

        if (block->rows() == 0) {

            RETURN_IF_ERROR(_convert_dict_cols_to_string_cols(block));

            *read_rows = block->rows();

#ifndef NDEBUG

            for (auto col : *block) {

                col.column->sanity_check();

                DCHECK(block->rows() == col.column->size())

                        << absl::Substitute("block rows = $0 , column rows = $1, col name = $2",

                                            block->rows(), col.column->size(), col.name);

            }

#endif

            return Status::OK();

        }

        {

            SCOPED_RAW_TIMER(&_predicate_filter_time);

            RETURN_IF_ERROR(_build_pos_delete_filter(*read_rows));

            std::vector<uint32_t> columns_to_filter;

            int column_to_keep = block->columns();

            columns_to_filter.resize(column_to_keep);

            for (uint32_t i = 0; i < column_to_keep; ++i) {

                columns_to_filter[i] = i;

            }

            if (!_lazy_read_ctx.conjuncts.empty()) {

                std::vector<IColumn::Filter*> filters;

                if (_position_delete_ctx.has_filter) {

                    filters.push_back(_pos_delete_filter_ptr.get());

                }

                IColumn::Filter result_filter(block->rows(), 1);

                bool can_filter_all = false;

                {

                    RETURN_IF_ERROR_OR_CATCH_EXCEPTION(VExprContext::execute_conjuncts(

                            _filter_conjuncts, &filters, block, &result_filter, &can_filter_all));

                }

                // Condition cache MISS: mark granules with surviving rows (non-lazy path)

                if (!can_filter_all) {

                    _mark_condition_cache_granules(result_filter.data(), block->rows(),

                                                   batch_base_row);

                }

                if (can_filter_all) {

                    for (auto& col : columns_to_filter) {

                        std::move(*block->get_by_position(col).column).assume_mutable()->clear();

                    }

                    Block::erase_useless_column(block, column_to_keep);

                    RETURN_IF_ERROR(_convert_dict_cols_to_string_cols(block));

                    return Status::OK();

                }

                RETURN_IF_CATCH_EXCEPTION(

                        Block::filter_block_internal(block, columns_to_filter, result_filter));

                Block::erase_useless_column(block, column_to_keep);

            } else {

                RETURN_IF_CATCH_EXCEPTION(

                        RETURN_IF_ERROR(_filter_block(block, column_to_keep, columns_to_filter)));

            }

            RETURN_IF_ERROR(_convert_dict_cols_to_string_cols(block));

        }

#ifndef NDEBUG

        for (auto col : *block) {

            col.column->sanity_check();

            DCHECK(block->rows() == col.column->size())

                    << absl::Substitute("block rows = $0 , column rows = $1, col name = $2",

                                        block->rows(), col.column->size(), col.name);

        }

#endif

        *read_rows = block->rows();

        return Status::OK();

    }

}

// Maps each batch row to its global parquet file position via _read_ranges, then marks

// the corresponding condition cache granule as true if the filter indicates the row survived.

// batch_seq_start is the number of rows already read sequentially before this batch

// (i.e., _total_read_rows before the batch started).

void RowGroupReader::_mark_condition_cache_granules(const uint8_t* filter_data, size_t num_rows,

                                                    int64_t batch_seq_start) {

    if (!_condition_cache_ctx || _condition_cache_ctx->is_hit) {

        return;

    }

    auto& cache = *_condition_cache_ctx->filter_result;

    for (size_t i = 0; i < num_rows; i++) {

        if (filter_data[i]) {

            // row-group-relative position of this row

            int64_t rg_pos = _read_ranges.get_row_index_by_pos(batch_seq_start + i);

            // global row number in the parquet file

            size_t granule = (_current_row_group_idx.first_row + rg_pos) /

                             ConditionCacheContext::GRANULE_SIZE;

            size_t cache_idx = granule - _condition_cache_ctx->base_granule;

            if (cache_idx < cache.size()) {

                cache[cache_idx] = true;

            }

        }

    }

}

// On condition cache HIT, removes row ranges whose granules have no surviving rows from

// _read_ranges BEFORE column readers are created. This makes ParquetColumnReader skip I/O

// entirely for false-granule rows — both predicate and lazy columns — via its existing

// page/row-skipping infrastructure.

void RowGroupReader::_filter_read_ranges_by_condition_cache() {

    if (!_condition_cache_ctx || !_condition_cache_ctx->is_hit) {

        return;

    }

    auto& filter_result = *_condition_cache_ctx->filter_result;

    if (filter_result.empty()) {

        return;

    }

    auto old_row_count = _read_ranges.count();

    _read_ranges =

            filter_ranges_by_cache(_read_ranges, filter_result, _current_row_group_idx.first_row,

                                   _condition_cache_ctx->base_granule);

    _is_row_group_filtered = _read_ranges.is_empty();

    _condition_cache_filtered_rows += old_row_count - _read_ranges.count();

}

// Filters read_ranges by removing rows whose cache granule is false.

//

// Cache index i maps to global granule (base_granule + i), which covers global file

// rows [(base_granule+i)*GS, (base_granule+i+1)*GS). Since read_ranges uses

// row-group-relative indices and first_row is the global position of the row group's

// first row, global granule g maps to row-group-relative range:

//   [max(0, g*GS - first_row), max(0, (g+1)*GS - first_row))

//

// We build a RowRanges of all false-granule regions (in row-group-relative coordinates),

// then subtract from read_ranges via ranges_exception.

//

// Granules beyond cache.size() are kept conservatively (assumed true).

//

// When base_granule > 0, the cache only covers granules starting from base_granule.

// This happens when a Parquet file is split across multiple scan ranges and this reader

// only processes row groups starting at a non-zero offset in the file.

RowRanges RowGroupReader::filter_ranges_by_cache(const RowRanges& read_ranges,

                                                 const std::vector<bool>& cache, int64_t first_row,

                                                 int64_t base_granule) {

    constexpr int64_t GS = ConditionCacheContext::GRANULE_SIZE;

    RowRanges filtered_ranges;

    for (size_t i = 0; i < cache.size(); i++) {

        if (!cache[i]) {

            int64_t global_granule = base_granule + static_cast<int64_t>(i);

            int64_t rg_from = std::max(static_cast<int64_t>(0), global_granule * GS - first_row);

            int64_t rg_to =

                    std::max(static_cast<int64_t>(0), (global_granule + 1) * GS - first_row);

            if (rg_from < rg_to) {

                filtered_ranges.add(RowRange(rg_from, rg_to));

            }

        }

    }

    RowRanges result;

    RowRanges::ranges_exception(read_ranges, filtered_ranges, &result);

    return result;

}

Status RowGroupReader::_read_column_data(Block* block,

                                         const std::vector<std::string>& table_columns,

                                         size_t batch_size, size_t* read_rows, bool* batch_eof,

                                         FilterMap& filter_map) {

    size_t batch_read_rows = 0;

    bool has_eof = false;

    for (auto& read_col_name : table_columns) {

        auto& column_with_type_and_name =

                block->safe_get_by_position((*_col_name_to_block_idx)[read_col_name]);

        auto& column_ptr = column_with_type_and_name.column;

        auto& column_type = column_with_type_and_name.type;

        bool is_dict_filter = false;

        for (auto& _dict_filter_col : _dict_filter_cols) {

            if (_dict_filter_col.first == read_col_name) {

                MutableColumnPtr dict_column = ColumnInt32::create();

                if (!_col_name_to_block_idx->contains(read_col_name)) {

                    return Status::InternalError(

                            "Wrong read column '{}' in parquet file, block: {}", read_col_name,

                            block->dump_structure());

                }

                if (column_type->is_nullable()) {

                    block->get_by_position((*_col_name_to_block_idx)[read_col_name]).type =

                            std::make_shared<DataTypeNullable>(std::make_shared<DataTypeInt32>());

                    block->replace_by_position(

                            (*_col_name_to_block_idx)[read_col_name],

                            ColumnNullable::create(std::move(dict_column),

                                                   ColumnUInt8::create(dict_column->size(), 0)));

                } else {

                    block->get_by_position((*_col_name_to_block_idx)[read_col_name]).type =

                            std::make_shared<DataTypeInt32>();

                    block->replace_by_position((*_col_name_to_block_idx)[read_col_name],

                                               std::move(dict_column));

                }

                is_dict_filter = true;

                break;

            }

        }

        size_t col_read_rows = 0;

        bool col_eof = false;

        // Should reset _filter_map_index to 0 when reading next column.

        //        select_vector.reset();

        _column_readers[read_col_name]->reset_filter_map_index();

        while (!col_eof && col_read_rows < batch_size) {

            size_t loop_rows = 0;

            RETURN_IF_ERROR(_column_readers[read_col_name]->read_column_data(

                    column_ptr, column_type, _table_info_node_ptr->get_children_node(read_col_name),

                    filter_map, batch_size - col_read_rows, &loop_rows, &col_eof, is_dict_filter));

            VLOG_DEBUG << "[RowGroupReader] column '" << read_col_name

                       << "' loop_rows=" << loop_rows << " col_read_rows_so_far=" << col_read_rows

                       << std::endl;

            col_read_rows += loop_rows;

        }

        VLOG_DEBUG << "[RowGroupReader] column '" << read_col_name

                   << "' read_rows=" << col_read_rows << std::endl;

        if (batch_read_rows > 0 && batch_read_rows != col_read_rows) {

            LOG(WARNING) << "[RowGroupReader] Mismatched read rows among parquet columns. "

                            "previous_batch_read_rows="

                         << batch_read_rows << ", current_column='" << read_col_name

                         << "', current_col_read_rows=" << col_read_rows;

            return Status::Corruption("Can't read the same number of rows among parquet columns");

        }

        batch_read_rows = col_read_rows;

#ifndef NDEBUG

        column_ptr->sanity_check();

#endif

        if (col_eof) {

            has_eof = true;

        }

    }

    *read_rows = batch_read_rows;

    *batch_eof = has_eof;

    return Status::OK();

}

Status RowGroupReader::_do_lazy_read(Block* block, size_t batch_size, size_t* read_rows,

                                     bool* batch_eof) {

    std::unique_ptr<FilterMap> filter_map_ptr = nullptr;

    size_t pre_read_rows;

    bool pre_eof;

    std::vector<uint32_t> columns_to_filter;

    uint32_t origin_column_num = block->columns();

    columns_to_filter.resize(origin_column_num);

    for (uint32_t i = 0; i < origin_column_num; ++i) {

        columns_to_filter[i] = i;

    }

    IColumn::Filter result_filter;

    size_t pre_raw_read_rows = 0;

    while (!_state->is_cancelled()) {

        // read predicate columns

        pre_read_rows = 0;

        pre_eof = false;

        FilterMap filter_map;

        int64_t batch_base_row = _total_read_rows;

        RETURN_IF_ERROR(_read_column_data(block, _lazy_read_ctx.predicate_columns.first, batch_size,

                                          &pre_read_rows, &pre_eof, filter_map));

        if (pre_read_rows == 0) {

            DCHECK_EQ(pre_eof, true);

            break;

        }

        pre_raw_read_rows += pre_read_rows;

        RETURN_IF_ERROR(_fill_partition_columns(block, pre_read_rows,

                                                _lazy_read_ctx.predicate_partition_columns));

        RETURN_IF_ERROR(_fill_missing_columns(block, pre_read_rows,

                                              _lazy_read_ctx.predicate_missing_columns));

        RETURN_IF_ERROR(_fill_row_id_columns(block, pre_read_rows, false));

        RETURN_IF_ERROR(_append_iceberg_rowid_column(block, pre_read_rows, false));

        RETURN_IF_ERROR(_build_pos_delete_filter(pre_read_rows));

#ifndef NDEBUG

        for (auto col : *block) {

            if (col.column->size() == 0) { // lazy read column.

                continue;

            }

            col.column->sanity_check();

            DCHECK(pre_read_rows == col.column->size())

                    << absl::Substitute("pre_read_rows = $0 , column rows = $1, col name = $2",

                                        pre_read_rows, col.column->size(), col.name);

        }

#endif

        bool can_filter_all = false;

        bool resize_first_column = _lazy_read_ctx.resize_first_column;

        if (resize_first_column && _iceberg_rowid_params.enabled) {

            int row_id_idx = block->get_position_by_name(doris::BeConsts::ICEBERG_ROWID_COL);

            if (row_id_idx == 0) {

                resize_first_column = false;

            }

        }

        {

            SCOPED_RAW_TIMER(&_predicate_filter_time);

            // generate filter vector

            if (resize_first_column) {

                // VExprContext.execute has an optimization, the filtering is executed when block->rows() > 0

                // The following process may be tricky and time-consuming, but we have no other way.

                block->get_by_position(0).column->assume_mutable()->resize(pre_read_rows);

            }

            result_filter.assign(pre_read_rows, static_cast<unsigned char>(1));

            std::vector<IColumn::Filter*> filters;

            if (_position_delete_ctx.has_filter) {

                filters.push_back(_pos_delete_filter_ptr.get());

            }

            VExprContextSPtrs filter_contexts;

            for (auto& conjunct : _filter_conjuncts) {

                filter_contexts.emplace_back(conjunct);

            }

            {

                RETURN_IF_ERROR(VExprContext::execute_conjuncts(filter_contexts, &filters, block,

                                                                &result_filter, &can_filter_all));

            }

            // Condition cache MISS: mark granules with surviving rows

            if (!can_filter_all) {

                _mark_condition_cache_granules(result_filter.data(), pre_read_rows, batch_base_row);

            }

            if (resize_first_column) {

                // We have to clean the first column to insert right data.

                block->get_by_position(0).column->assume_mutable()->clear();

            }

        }

        const uint8_t* __restrict filter_map_data = result_filter.data();

        filter_map_ptr = std::make_unique<FilterMap>();

        RETURN_IF_ERROR(filter_map_ptr->init(filter_map_data, pre_read_rows, can_filter_all));

        if (filter_map_ptr->filter_all()) {

            {

                SCOPED_RAW_TIMER(&_predicate_filter_time);

                for (const auto& col : _lazy_read_ctx.predicate_columns.first) {

                    // clean block to read predicate columns

                    block->get_by_position((*_col_name_to_block_idx)[col])

                            .column->assume_mutable()

                            ->clear();

                }

                for (const auto& col : _lazy_read_ctx.predicate_partition_columns) {

                    block->get_by_position((*_col_name_to_block_idx)[col.first])

                            .column->assume_mutable()

                            ->clear();

                }

                for (const auto& col : _lazy_read_ctx.predicate_missing_columns) {

                    block->get_by_position((*_col_name_to_block_idx)[col.first])

                            .column->assume_mutable()

                            ->clear();

                }

                if (_row_id_column_iterator_pair.first != nullptr) {

                    block->get_by_position(_row_id_column_iterator_pair.second)

                            .column->assume_mutable()

                            ->clear();

                }

                if (_iceberg_rowid_params.enabled) {

                    int row_id_idx =

                            block->get_position_by_name(doris::BeConsts::ICEBERG_ROWID_COL);

                    if (row_id_idx >= 0) {

                        block->get_by_position(static_cast<size_t>(row_id_idx))

                                .column->assume_mutable()

                                ->clear();

                    }

                }

                Block::erase_useless_column(block, origin_column_num);

            }

            if (!pre_eof) {

                // If continuous batches are skipped, we can cache them to skip a whole page

                _cached_filtered_rows += pre_read_rows;

                if (pre_raw_read_rows >= config::doris_scanner_row_num) {

                    *read_rows = 0;

                    RETURN_IF_ERROR(_convert_dict_cols_to_string_cols(block));

                    return Status::OK();

                }

            } else { // pre_eof

                // If filter_map_ptr->filter_all() and pre_eof, we can skip whole row group.

                *read_rows = 0;

                *batch_eof = true;

                _lazy_read_filtered_rows += (pre_read_rows + _cached_filtered_rows);

                RETURN_IF_ERROR(_convert_dict_cols_to_string_cols(block));

                return Status::OK();

            }

        } else {

            break;

        }

    }

    if (_state->is_cancelled()) {

        return Status::Cancelled("cancelled");

    }

    if (filter_map_ptr == nullptr) {

        DCHECK_EQ(pre_read_rows + _cached_filtered_rows, 0);

        *read_rows = 0;

        *batch_eof = true;

        return Status::OK();

    }

    FilterMap& filter_map = *filter_map_ptr;

    DorisUniqueBufferPtr<uint8_t> rebuild_filter_map = nullptr;

    if (_cached_filtered_rows != 0) {

        RETURN_IF_ERROR(_rebuild_filter_map(filter_map, rebuild_filter_map, pre_read_rows));

        pre_read_rows += _cached_filtered_rows;

        _cached_filtered_rows = 0;

    }

    // lazy read columns

    size_t lazy_read_rows;

    bool lazy_eof;

    RETURN_IF_ERROR(_read_column_data(block, _lazy_read_ctx.lazy_read_columns, pre_read_rows,

                                      &lazy_read_rows, &lazy_eof, filter_map));

    if (pre_read_rows != lazy_read_rows) {

        return Status::Corruption("Can't read the same number of rows when doing lazy read");

    }

    // pre_eof ^ lazy_eof

    // we set pre_read_rows as batch_size for lazy read columns, so pre_eof != lazy_eof

    // filter data in predicate columns, and remove filter column

    {

        SCOPED_RAW_TIMER(&_predicate_filter_time);

        if (filter_map.has_filter()) {

            std::vector<uint32_t> predicate_columns = _lazy_read_ctx.all_predicate_col_ids;

            if (_iceberg_rowid_params.enabled) {

                int row_id_idx = block->get_position_by_name(doris::BeConsts::ICEBERG_ROWID_COL);

                if (row_id_idx >= 0 &&

                    std::find(predicate_columns.begin(), predicate_columns.end(),

                              static_cast<uint32_t>(row_id_idx)) == predicate_columns.end()) {

                    predicate_columns.push_back(static_cast<uint32_t>(row_id_idx));

                }

            }

            RETURN_IF_CATCH_EXCEPTION(

                    Block::filter_block_internal(block, predicate_columns, result_filter));

            Block::erase_useless_column(block, origin_column_num);

        } else {

            Block::erase_useless_column(block, origin_column_num);

        }

    }

    RETURN_IF_ERROR(_convert_dict_cols_to_string_cols(block));

    size_t column_num = block->columns();

    size_t column_size = 0;

    for (int i = 0; i < column_num; ++i) {

        size_t cz = block->get_by_position(i).column->size();

        if (column_size != 0 && cz != 0) {

            DCHECK_EQ(column_size, cz);

        }

        if (cz != 0) {

            column_size = cz;

        }

    }

    _lazy_read_filtered_rows += pre_read_rows - column_size;

    *read_rows = column_size;

    *batch_eof = pre_eof;

    RETURN_IF_ERROR(_fill_partition_columns(block, column_size, _lazy_read_ctx.partition_columns));

    RETURN_IF_ERROR(_fill_missing_columns(block, column_size, _lazy_read_ctx.missing_columns));

#ifndef NDEBUG

    for (auto col : *block) {

        col.column->sanity_check();

        DCHECK(block->rows() == col.column->size())

                << absl::Substitute("block rows = $0 , column rows = $1, col name = $2",

                                    block->rows(), col.column->size(), col.name);

    }

#endif

    return Status::OK();

}

Status RowGroupReader::_rebuild_filter_map(FilterMap& filter_map,

                                           DorisUniqueBufferPtr<uint8_t>& filter_map_data,

                                           size_t pre_read_rows) const {

    if (_cached_filtered_rows == 0) {

        return Status::OK();

    }

    size_t total_rows = _cached_filtered_rows + pre_read_rows;

    if (filter_map.filter_all()) {

        RETURN_IF_ERROR(filter_map.init(nullptr, total_rows, true));

        return Status::OK();

    }

    filter_map_data = make_unique_buffer<uint8_t>(total_rows);

    auto* map = filter_map_data.get();

    for (size_t i = 0; i < _cached_filtered_rows; ++i) {

        map[i] = 0;

    }

    const uint8_t* old_map = filter_map.filter_map_data();

    if (old_map == nullptr) {

        // select_vector.filter_all() == true is already built.

        for (size_t i = _cached_filtered_rows; i < total_rows; ++i) {

            map[i] = 1;

        }

    } else {

        memcpy(map + _cached_filtered_rows, old_map, pre_read_rows);

    }

    RETURN_IF_ERROR(filter_map.init(map, total_rows, false));

    return Status::OK();

}

Status RowGroupReader::_fill_partition_columns(

        Block* block, size_t rows,

        const std::unordered_map<std::string, std::tuple<std::string, const SlotDescriptor*>>&

                partition_columns) {

    DataTypeSerDe::FormatOptions _text_formatOptions;

    for (const auto& kv : partition_columns) {

        auto doris_column = block->get_by_position((*_col_name_to_block_idx)[kv.first]).column;

        // obtained from block*, it is a mutable object.

        auto* col_ptr = const_cast<IColumn*>(doris_column.get());

        const auto& [value, slot_desc] = kv.second;

        auto _text_serde = slot_desc->get_data_type_ptr()->get_serde();

        Slice slice(value.data(), value.size());

        uint64_t num_deserialized = 0;

        // Be careful when reading empty rows from parquet row groups.

        if (_text_serde->deserialize_column_from_fixed_json(*col_ptr, slice, rows,

                                                            &num_deserialized,

                                                            _text_formatOptions) != Status::OK()) {

            return Status::InternalError("Failed to fill partition column: {}={}",

                                         slot_desc->col_name(), value);

        }

        if (num_deserialized != rows) {

            return Status::InternalError(

                    "Failed to fill partition column: {}={} ."

                    "Number of rows expected to be written : {}, number of rows actually written : "

                    "{}",

                    slot_desc->col_name(), value, num_deserialized, rows);

        }

    }

    return Status::OK();

}

Status RowGroupReader::_fill_missing_columns(

        Block* block, size_t rows,

        const std::unordered_map<std::string, VExprContextSPtr>& missing_columns) {

    for (const auto& kv : missing_columns) {

        if (!_col_name_to_block_idx->contains(kv.first)) {

            return Status::InternalError("Missing column: {} not found in block {}", kv.first,

                                         block->dump_structure());

        }

        if (kv.second == nullptr) {

            // no default column, fill with null

            auto mutable_column = block->get_by_position((*_col_name_to_block_idx)[kv.first])

                                          .column->assume_mutable();

            auto* nullable_column = assert_cast<ColumnNullable*>(mutable_column.get());

            nullable_column->insert_many_defaults(rows);

        } else {

            // fill with default value