// 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_reader.h"

#include <gen_cpp/Metrics_types.h>

#include <gen_cpp/PlanNodes_types.h>

#include <gen_cpp/parquet_types.h>

#include <glog/logging.h>

#include <algorithm>

#include <functional>

#include <utility>

#include "common/config.h"

#include "common/status.h"

#include "core/block/block.h"

#include "core/block/column_with_type_and_name.h"

#include "core/column/column.h"

#include "core/data_type/define_primitive_type.h"

#include "core/typeid_cast.h"

#include "core/types.h"

#include "exec/scan/file_scanner.h"

#include "exprs/vbloom_predicate.h"

#include "exprs/vdirect_in_predicate.h"

#include "exprs/vexpr.h"

#include "exprs/vexpr_context.h"

#include "exprs/vin_predicate.h"

#include "exprs/vruntimefilter_wrapper.h"

#include "exprs/vslot_ref.h"

#include "exprs/vtopn_pred.h"

#include "format/column_type_convert.h"

#include "format/parquet/parquet_block_split_bloom_filter.h"

#include "format/parquet/parquet_common.h"

#include "format/parquet/parquet_predicate.h"

#include "format/parquet/parquet_thrift_util.h"

#include "format/parquet/schema_desc.h"

#include "format/parquet/vparquet_file_metadata.h"

#include "format/parquet/vparquet_group_reader.h"

#include "format/parquet/vparquet_page_index.h"

#include "information_schema/schema_scanner.h"

#include "io/file_factory.h"

#include "io/fs/buffered_reader.h"

#include "io/fs/file_reader.h"

#include "io/fs/file_reader_writer_fwd.h"

#include "io/fs/tracing_file_reader.h"

#include "runtime/descriptors.h"

#include "util/slice.h"

#include "util/string_util.h"

#include "util/timezone_utils.h"

namespace cctz {

class time_zone;

} // namespace cctz

namespace doris {

class RowDescriptor;

class RuntimeState;

class SlotDescriptor;

class TupleDescriptor;

namespace io {

struct IOContext;

enum class FileCachePolicy : uint8_t;

} // namespace io

class Block;

} // namespace doris

namespace doris {

#include "common/compile_check_begin.h"

ParquetReader::ParquetReader(RuntimeProfile* profile, const TFileScanRangeParams& params,

                             const TFileRangeDesc& range, size_t batch_size,

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

                             FileMetaCache* meta_cache, bool enable_lazy_mat)

        : _profile(profile),

          _scan_params(params),

          _scan_range(range),

          _batch_size(std::max(batch_size, _MIN_BATCH_SIZE)),

          _range_start_offset(range.start_offset),

          _range_size(range.size),

          _ctz(ctz),

          _io_ctx(io_ctx),

          _state(state),

          _enable_lazy_mat(enable_lazy_mat),

          _enable_filter_by_min_max(

                  state == nullptr ? true

                                   : state->query_options().enable_parquet_filter_by_min_max),

          _enable_filter_by_bloom_filter(

                  state == nullptr ? true

                                   : state->query_options().enable_parquet_filter_by_bloom_filter) {

    _meta_cache = meta_cache;

    _init_profile();

    _init_system_properties();

    _init_file_description();

}

ParquetReader::ParquetReader(RuntimeProfile* profile, const TFileScanRangeParams& params,

                             const TFileRangeDesc& range, size_t batch_size,

                             const cctz::time_zone* ctz,

                             std::shared_ptr<io::IOContext> io_ctx_holder, RuntimeState* state,

                             FileMetaCache* meta_cache, bool enable_lazy_mat)

        : _profile(profile),

          _scan_params(params),

          _scan_range(range),

          _batch_size(std::max(batch_size, _MIN_BATCH_SIZE)),

          _range_start_offset(range.start_offset),

          _range_size(range.size),

          _ctz(ctz),

          _io_ctx(io_ctx_holder ? io_ctx_holder.get() : nullptr),

          _io_ctx_holder(std::move(io_ctx_holder)),

          _state(state),

          _enable_lazy_mat(enable_lazy_mat),

          _enable_filter_by_min_max(

                  state == nullptr ? true

                                   : state->query_options().enable_parquet_filter_by_min_max),

          _enable_filter_by_bloom_filter(

                  state == nullptr ? true

                                   : state->query_options().enable_parquet_filter_by_bloom_filter) {

    _meta_cache = meta_cache;

    _init_profile();

    _init_system_properties();

    _init_file_description();

}

ParquetReader::ParquetReader(const TFileScanRangeParams& params, const TFileRangeDesc& range,

                             io::IOContext* io_ctx, RuntimeState* state, FileMetaCache* meta_cache,

                             bool enable_lazy_mat)

        : _profile(nullptr),

          _scan_params(params),

          _scan_range(range),

          _io_ctx(io_ctx),

          _state(state),

          _enable_lazy_mat(enable_lazy_mat),

          _enable_filter_by_min_max(

                  state == nullptr ? true

                                   : state->query_options().enable_parquet_filter_by_min_max),

          _enable_filter_by_bloom_filter(

                  state == nullptr ? true

                                   : state->query_options().enable_parquet_filter_by_bloom_filter) {

    _meta_cache = meta_cache;

    _init_system_properties();

    _init_file_description();

}

ParquetReader::ParquetReader(const TFileScanRangeParams& params, const TFileRangeDesc& range,

                             std::shared_ptr<io::IOContext> io_ctx_holder, RuntimeState* state,

                             FileMetaCache* meta_cache, bool enable_lazy_mat)

        : _profile(nullptr),

          _scan_params(params),

          _scan_range(range),

          _io_ctx(io_ctx_holder ? io_ctx_holder.get() : nullptr),

          _io_ctx_holder(std::move(io_ctx_holder)),

          _state(state),

          _enable_lazy_mat(enable_lazy_mat),

          _enable_filter_by_min_max(

                  state == nullptr ? true

                                   : state->query_options().enable_parquet_filter_by_min_max),

          _enable_filter_by_bloom_filter(

                  state == nullptr ? true

                                   : state->query_options().enable_parquet_filter_by_bloom_filter) {

    _meta_cache = meta_cache;

    _init_system_properties();

    _init_file_description();

}

ParquetReader::~ParquetReader() {

    _close_internal();

}

#ifdef BE_TEST

// for unit test

void ParquetReader::set_file_reader(io::FileReaderSPtr file_reader) {

    _file_reader = file_reader;

    _tracing_file_reader = file_reader;

}

#endif

void ParquetReader::_init_profile() {

    if (_profile != nullptr) {

        static const char* parquet_profile = "ParquetReader";

        ADD_TIMER_WITH_LEVEL(_profile, parquet_profile, 1);

        _parquet_profile.filtered_row_groups = ADD_CHILD_COUNTER_WITH_LEVEL(

                _profile, "RowGroupsFiltered", TUnit::UNIT, parquet_profile, 1);

        _parquet_profile.filtered_row_groups_by_min_max = ADD_CHILD_COUNTER_WITH_LEVEL(

                _profile, "RowGroupsFilteredByMinMax", TUnit::UNIT, parquet_profile, 1);

        _parquet_profile.filtered_row_groups_by_bloom_filter = ADD_CHILD_COUNTER_WITH_LEVEL(

                _profile, "RowGroupsFilteredByBloomFilter", TUnit::UNIT, parquet_profile, 1);

        _parquet_profile.to_read_row_groups = ADD_CHILD_COUNTER_WITH_LEVEL(

                _profile, "RowGroupsReadNum", TUnit::UNIT, parquet_profile, 1);

        _parquet_profile.total_row_groups = ADD_CHILD_COUNTER_WITH_LEVEL(

                _profile, "RowGroupsTotalNum", TUnit::UNIT, parquet_profile, 1);

        _parquet_profile.filtered_group_rows = ADD_CHILD_COUNTER_WITH_LEVEL(

                _profile, "FilteredRowsByGroup", TUnit::UNIT, parquet_profile, 1);

        _parquet_profile.filtered_page_rows = ADD_CHILD_COUNTER_WITH_LEVEL(

                _profile, "FilteredRowsByPage", TUnit::UNIT, parquet_profile, 1);

        _parquet_profile.lazy_read_filtered_rows = ADD_CHILD_COUNTER_WITH_LEVEL(

                _profile, "FilteredRowsByLazyRead", TUnit::UNIT, parquet_profile, 1);

        _parquet_profile.filtered_bytes = ADD_CHILD_COUNTER_WITH_LEVEL(

                _profile, "FilteredBytes", TUnit::BYTES, parquet_profile, 1);

        _parquet_profile.raw_rows_read = ADD_CHILD_COUNTER_WITH_LEVEL(

                _profile, "RawRowsRead", TUnit::UNIT, parquet_profile, 1);

        _parquet_profile.column_read_time =

                ADD_CHILD_TIMER_WITH_LEVEL(_profile, "ColumnReadTime", parquet_profile, 1);

        _parquet_profile.parse_meta_time =

                ADD_CHILD_TIMER_WITH_LEVEL(_profile, "ParseMetaTime", parquet_profile, 1);

        _parquet_profile.parse_footer_time =

                ADD_CHILD_TIMER_WITH_LEVEL(_profile, "ParseFooterTime", parquet_profile, 1);

        _parquet_profile.file_reader_create_time =

                ADD_CHILD_TIMER_WITH_LEVEL(_profile, "FileReaderCreateTime", parquet_profile, 1);

        _parquet_profile.open_file_num =

                ADD_CHILD_COUNTER_WITH_LEVEL(_profile, "FileNum", TUnit::UNIT, parquet_profile, 1);

        _parquet_profile.page_index_read_calls =

                ADD_COUNTER_WITH_LEVEL(_profile, "PageIndexReadCalls", TUnit::UNIT, 1);

        _parquet_profile.page_index_filter_time =

                ADD_CHILD_TIMER_WITH_LEVEL(_profile, "PageIndexFilterTime", parquet_profile, 1);

        _parquet_profile.read_page_index_time =

                ADD_CHILD_TIMER_WITH_LEVEL(_profile, "PageIndexReadTime", parquet_profile, 1);

        _parquet_profile.parse_page_index_time =

                ADD_CHILD_TIMER_WITH_LEVEL(_profile, "PageIndexParseTime", parquet_profile, 1);

        _parquet_profile.row_group_filter_time =

                ADD_CHILD_TIMER_WITH_LEVEL(_profile, "RowGroupFilterTime", parquet_profile, 1);

        _parquet_profile.file_footer_read_calls =

                ADD_COUNTER_WITH_LEVEL(_profile, "FileFooterReadCalls", TUnit::UNIT, 1);

        _parquet_profile.file_footer_hit_cache =

                ADD_COUNTER_WITH_LEVEL(_profile, "FileFooterHitCache", TUnit::UNIT, 1);

        _parquet_profile.decompress_time =

                ADD_CHILD_TIMER_WITH_LEVEL(_profile, "DecompressTime", parquet_profile, 1);

        _parquet_profile.decompress_cnt = ADD_CHILD_COUNTER_WITH_LEVEL(

                _profile, "DecompressCount", TUnit::UNIT, parquet_profile, 1);

        _parquet_profile.page_read_counter = ADD_CHILD_COUNTER_WITH_LEVEL(

                _profile, "PageReadCount", TUnit::UNIT, parquet_profile, 1);

        _parquet_profile.page_cache_write_counter = ADD_CHILD_COUNTER_WITH_LEVEL(

                _profile, "PageCacheWriteCount", TUnit::UNIT, parquet_profile, 1);

        _parquet_profile.page_cache_compressed_write_counter = ADD_CHILD_COUNTER_WITH_LEVEL(

                _profile, "PageCacheCompressedWriteCount", TUnit::UNIT, parquet_profile, 1);

        _parquet_profile.page_cache_decompressed_write_counter = ADD_CHILD_COUNTER_WITH_LEVEL(

                _profile, "PageCacheDecompressedWriteCount", TUnit::UNIT, parquet_profile, 1);

        _parquet_profile.page_cache_hit_counter = ADD_CHILD_COUNTER_WITH_LEVEL(

                _profile, "PageCacheHitCount", TUnit::UNIT, parquet_profile, 1);

        _parquet_profile.page_cache_missing_counter = ADD_CHILD_COUNTER_WITH_LEVEL(

                _profile, "PageCacheMissingCount", TUnit::UNIT, parquet_profile, 1);

        _parquet_profile.page_cache_compressed_hit_counter = ADD_CHILD_COUNTER_WITH_LEVEL(

                _profile, "PageCacheCompressedHitCount", TUnit::UNIT, parquet_profile, 1);

        _parquet_profile.page_cache_decompressed_hit_counter = ADD_CHILD_COUNTER_WITH_LEVEL(

                _profile, "PageCacheDecompressedHitCount", TUnit::UNIT, parquet_profile, 1);

        _parquet_profile.decode_header_time =

                ADD_CHILD_TIMER_WITH_LEVEL(_profile, "PageHeaderDecodeTime", parquet_profile, 1);

        _parquet_profile.read_page_header_time =

                ADD_CHILD_TIMER_WITH_LEVEL(_profile, "PageHeaderReadTime", parquet_profile, 1);

        _parquet_profile.decode_value_time =

                ADD_CHILD_TIMER_WITH_LEVEL(_profile, "DecodeValueTime", parquet_profile, 1);

        _parquet_profile.decode_dict_time =

                ADD_CHILD_TIMER_WITH_LEVEL(_profile, "DecodeDictTime", parquet_profile, 1);

        _parquet_profile.decode_level_time =

                ADD_CHILD_TIMER_WITH_LEVEL(_profile, "DecodeLevelTime", parquet_profile, 1);

        _parquet_profile.decode_null_map_time =

                ADD_CHILD_TIMER_WITH_LEVEL(_profile, "DecodeNullMapTime", parquet_profile, 1);

        _parquet_profile.skip_page_header_num = ADD_CHILD_COUNTER_WITH_LEVEL(

                _profile, "SkipPageHeaderNum", TUnit::UNIT, parquet_profile, 1);

        _parquet_profile.parse_page_header_num = ADD_CHILD_COUNTER_WITH_LEVEL(

                _profile, "ParsePageHeaderNum", TUnit::UNIT, parquet_profile, 1);

        _parquet_profile.predicate_filter_time =

                ADD_CHILD_TIMER_WITH_LEVEL(_profile, "PredicateFilterTime", parquet_profile, 1);

        _parquet_profile.dict_filter_rewrite_time =

                ADD_CHILD_TIMER_WITH_LEVEL(_profile, "DictFilterRewriteTime", parquet_profile, 1);

        _parquet_profile.bloom_filter_read_time =

                ADD_CHILD_TIMER_WITH_LEVEL(_profile, "BloomFilterReadTime", parquet_profile, 1);

    }

}

Status ParquetReader::close() {

    _close_internal();

    return Status::OK();

}

void ParquetReader::_close_internal() {

    if (!_closed) {

        _closed = true;

    }

}

Status ParquetReader::_open_file() {

    if (UNLIKELY(_io_ctx && _io_ctx->should_stop)) {

        return Status::EndOfFile("stop");

    }

    if (_file_reader == nullptr) {

        SCOPED_RAW_TIMER(&_reader_statistics.file_reader_create_time);

        ++_reader_statistics.open_file_num;

        _file_description.mtime =

                _scan_range.__isset.modification_time ? _scan_range.modification_time : 0;

        io::FileReaderOptions reader_options =

                FileFactory::get_reader_options(_state, _file_description);

        _file_reader = DORIS_TRY(io::DelegateReader::create_file_reader(

                _profile, _system_properties, _file_description, reader_options,

                io::DelegateReader::AccessMode::RANDOM, _io_ctx));

        _tracing_file_reader = _io_ctx ? std::make_shared<io::TracingFileReader>(

                                                 _file_reader, _io_ctx->file_reader_stats)

                                       : _file_reader;

    }

    if (_file_metadata == nullptr) {

        SCOPED_RAW_TIMER(&_reader_statistics.parse_footer_time);

        if (_tracing_file_reader->size() <= sizeof(PARQUET_VERSION_NUMBER)) {

            // Some system may generate parquet file with only 4 bytes: PAR1

            // Should consider it as empty file.

            return Status::EndOfFile("open file failed, empty parquet file {} with size: {}",

                                     _scan_range.path, _tracing_file_reader->size());

        }

        size_t meta_size = 0;

        bool enable_mapping_varbinary = _scan_params.__isset.enable_mapping_varbinary

                                                ? _scan_params.enable_mapping_varbinary

                                                : false;

        bool enable_mapping_timestamp_tz = _scan_params.__isset.enable_mapping_timestamp_tz

                                                   ? _scan_params.enable_mapping_timestamp_tz

                                                   : false;

        if (_meta_cache == nullptr) {

            // wrap _file_metadata with unique ptr, so that it can be released finally.

            RETURN_IF_ERROR(parse_thrift_footer(_tracing_file_reader, &_file_metadata_ptr,

                                                &meta_size, _io_ctx, enable_mapping_varbinary,

                                                enable_mapping_timestamp_tz));

            _file_metadata = _file_metadata_ptr.get();

            // parse magic number & parse meta data

            _reader_statistics.file_footer_read_calls += 1;

        } else {

            const auto& file_meta_cache_key =

                    FileMetaCache::get_key(_tracing_file_reader, _file_description);

            if (!_meta_cache->lookup(file_meta_cache_key, &_meta_cache_handle)) {

                RETURN_IF_ERROR(parse_thrift_footer(_tracing_file_reader, &_file_metadata_ptr,

                                                    &meta_size, _io_ctx, enable_mapping_varbinary,

                                                    enable_mapping_timestamp_tz));

                // _file_metadata_ptr.release() : move control of _file_metadata to _meta_cache_handle

                _meta_cache->insert(file_meta_cache_key, _file_metadata_ptr.release(),

                                    &_meta_cache_handle);

                _file_metadata = _meta_cache_handle.data<FileMetaData>();

                _reader_statistics.file_footer_read_calls += 1;

            } else {

                _reader_statistics.file_footer_hit_cache++;

            }

            _file_metadata = _meta_cache_handle.data<FileMetaData>();

        }

        if (_file_metadata == nullptr) {

            return Status::InternalError("failed to get file meta data: {}",

                                         _file_description.path);

        }

    }

    return Status::OK();

}

Status ParquetReader::get_file_metadata_schema(const FieldDescriptor** ptr) {

    RETURN_IF_ERROR(_open_file());

    DCHECK(_file_metadata != nullptr);

    *ptr = &_file_metadata->schema();

    return Status::OK();

}

void ParquetReader::_init_system_properties() {

    if (_scan_range.__isset.file_type) {

        // for compatibility

        _system_properties.system_type = _scan_range.file_type;

    } else {

        _system_properties.system_type = _scan_params.file_type;

    }

    _system_properties.properties = _scan_params.properties;

    _system_properties.hdfs_params = _scan_params.hdfs_params;

    if (_scan_params.__isset.broker_addresses) {

        _system_properties.broker_addresses.assign(_scan_params.broker_addresses.begin(),

                                                   _scan_params.broker_addresses.end());

    }

}

void ParquetReader::_init_file_description() {

    _file_description.path = _scan_range.path;

    _file_description.file_size = _scan_range.__isset.file_size ? _scan_range.file_size : -1;

    if (_scan_range.__isset.fs_name) {

        _file_description.fs_name = _scan_range.fs_name;

    }

    if (_scan_range.__isset.file_cache_admission) {

        _file_description.file_cache_admission = _scan_range.file_cache_admission;

    }

}

Status ParquetReader::init_reader(

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

        std::unordered_map<std::string, uint32_t>* col_name_to_block_idx,

        const VExprContextSPtrs& conjuncts,

        phmap::flat_hash_map<int, std::vector<std::shared_ptr<ColumnPredicate>>>&

                slot_id_to_predicates,

        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,

        std::shared_ptr<TableSchemaChangeHelper::Node> table_info_node_ptr, bool filter_groups,

        const std::set<uint64_t>& column_ids, const std::set<uint64_t>& filter_column_ids) {

    _col_name_to_block_idx = col_name_to_block_idx;

    _tuple_descriptor = tuple_descriptor;

    _row_descriptor = row_descriptor;

    _colname_to_slot_id = colname_to_slot_id;

    _not_single_slot_filter_conjuncts = not_single_slot_filter_conjuncts;

    _slot_id_to_filter_conjuncts = slot_id_to_filter_conjuncts;

    _table_info_node_ptr = table_info_node_ptr;

    _filter_groups = filter_groups;

    _column_ids = column_ids;

    _filter_column_ids = filter_column_ids;

    RETURN_IF_ERROR(_open_file());

    _t_metadata = &(_file_metadata->to_thrift());

    if (_file_metadata == nullptr) {

        return Status::InternalError("failed to init parquet reader, please open reader first");

    }

    SCOPED_RAW_TIMER(&_reader_statistics.parse_meta_time);

    _total_groups = _t_metadata->row_groups.size();

    if (_total_groups == 0) {

        return Status::EndOfFile("init reader failed, empty parquet file: " + _scan_range.path);

    }

    _current_row_group_index = RowGroupReader::RowGroupIndex {-1, 0, 0};

    _table_column_names = &all_column_names;

    auto schema_desc = _file_metadata->schema();

    std::map<std::string, std::string> required_file_columns; //file column -> table column

    for (auto table_column_name : all_column_names) {

        if (_table_info_node_ptr->children_column_exists(table_column_name)) {

            required_file_columns.emplace(

                    _table_info_node_ptr->children_file_column_name(table_column_name),

                    table_column_name);

        } else {

            _missing_cols.emplace_back(table_column_name);

        }

    }

    for (int i = 0; i < schema_desc.size(); ++i) {

        const auto& name = schema_desc.get_column(i)->name;

        if (required_file_columns.contains(name)) {

            _read_file_columns.emplace_back(name);

            _read_table_columns.emplace_back(required_file_columns[name]);

            _read_table_columns_set.insert(required_file_columns[name]);

        }

    }

    // build column predicates for column lazy read

    _lazy_read_ctx.conjuncts = conjuncts;

    _lazy_read_ctx.slot_id_to_predicates = slot_id_to_predicates;

    return Status::OK();

}

bool ParquetReader::_exists_in_file(const std::string& expr_name) const {

    // `_read_table_columns_set` is used to ensure that only columns actually read are subject to min-max filtering.

    // This primarily handles cases where partition columns also exist in a file. The reason it's not modified

    // in `_table_info_node_ptr` is that Iceberg、Hudi has inconsistent requirements for this node;

    // Iceberg partition evolution need read partition columns from a file.

    // hudi set `hoodie.datasource.write.drop.partition.columns=false` not need read partition columns from a file.

    return _table_info_node_ptr->children_column_exists(expr_name) &&

           _read_table_columns_set.contains(expr_name);

}

bool ParquetReader::_type_matches(const int cid) const {

    auto* slot = _tuple_descriptor->slots()[cid];

    auto table_col_type = remove_nullable(slot->type());

    const auto& file_col_name = _table_info_node_ptr->children_file_column_name(slot->col_name());

    const auto& file_col_type =

            remove_nullable(_file_metadata->schema().get_column(file_col_name)->data_type);

    return (table_col_type->get_primitive_type() == file_col_type->get_primitive_type()) &&

           !is_complex_type(table_col_type->get_primitive_type());

}

Status ParquetReader::set_fill_columns(

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

                partition_columns,

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

    _lazy_read_ctx.fill_partition_columns = partition_columns;

    _lazy_read_ctx.fill_missing_columns = missing_columns;

    // std::unordered_map<column_name, std::pair<col_id, slot_id>>

    std::unordered_map<std::string, std::pair<uint32_t, int>> predicate_columns;

    // TODO(gabriel): we should try to clear too much structs which are used to represent conjuncts and predicates.

    // visit_slot for lazy mat.

    std::function<void(VExpr * expr)> visit_slot = [&](VExpr* expr) {

        if (expr->is_slot_ref()) {

            VSlotRef* slot_ref = static_cast<VSlotRef*>(expr);

            auto expr_name = slot_ref->expr_name();

            predicate_columns.emplace(expr_name,

                                      std::make_pair(slot_ref->column_id(), slot_ref->slot_id()));

            if (slot_ref->column_id() == 0) {

                _lazy_read_ctx.resize_first_column = false;

            }

            return;

        }

        for (auto& child : expr->children()) {

            visit_slot(child.get());

        }

    };

    for (const auto& conjunct : _lazy_read_ctx.conjuncts) {

        auto expr = conjunct->root();

        if (expr->is_rf_wrapper()) {

            // REF: src/runtime_filter/runtime_filter_consumer.cpp

            VRuntimeFilterWrapper* runtime_filter = assert_cast<VRuntimeFilterWrapper*>(expr.get());

            auto filter_impl = runtime_filter->get_impl();

            visit_slot(filter_impl.get());

        } else {

            visit_slot(expr.get());

        }

    }

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

        auto and_pred = AndBlockColumnPredicate::create_unique();

        for (const auto& entry : _lazy_read_ctx.slot_id_to_predicates) {

            for (const auto& pred : entry.second) {

                if (!_exists_in_file(pred->col_name()) || !_type_matches(pred->column_id())) {

                    continue;

                }

                and_pred->add_column_predicate(

                        SingleColumnBlockPredicate::create_unique(pred->clone(pred->column_id())));

            }

        }

        if (and_pred->num_of_column_predicate() > 0) {

            _push_down_predicates.push_back(std::move(and_pred));

        }

    }

    const FieldDescriptor& schema = _file_metadata->schema();

    auto check_iceberg_row_lineage_column_idx = [&](const auto& col_name) -> int {

        if (_row_lineage_columns != nullptr) {

            if (col_name == IcebergTableReader::ROW_LINEAGE_ROW_ID) {

                return _row_lineage_columns->row_id_column_idx;

            } else if (col_name == IcebergTableReader::ROW_LINEAGE_LAST_UPDATED_SEQ_NUMBER) {

                return _row_lineage_columns->last_updated_sequence_number_column_idx;

            }

        }

        return -1;

    };

    for (auto& read_table_col : _read_table_columns) {

        _lazy_read_ctx.all_read_columns.emplace_back(read_table_col);

        auto file_column_name = _table_info_node_ptr->children_file_column_name(read_table_col);

        PrimitiveType column_type =

                schema.get_column(file_column_name)->data_type->get_primitive_type();

        if (is_complex_type(column_type)) {

            _lazy_read_ctx.has_complex_type = true;

        }

        if (predicate_columns.size() > 0) {

            auto iter = predicate_columns.find(read_table_col);

            if (iter == predicate_columns.end()) {

                if (auto row_lineage_idx = check_iceberg_row_lineage_column_idx(read_table_col);

                    row_lineage_idx != -1) {

                    _lazy_read_ctx.predicate_columns.first.emplace_back(read_table_col);

                    // row lineage column can not dict filter.

                    int slot_id = 0;

                    for (auto slot : _tuple_descriptor->slots()) {

                        if (slot->col_name_lower_case() == read_table_col) {

                            slot_id = slot->id();

                        }

                    }

                    _lazy_read_ctx.predicate_columns.second.emplace_back(slot_id);

                    _lazy_read_ctx.all_predicate_col_ids.emplace_back(row_lineage_idx);

                } else {

                    _lazy_read_ctx.lazy_read_columns.emplace_back(read_table_col);

                }

            } else {

                _lazy_read_ctx.predicate_columns.first.emplace_back(iter->first);

                _lazy_read_ctx.predicate_columns.second.emplace_back(iter->second.second);

                _lazy_read_ctx.all_predicate_col_ids.emplace_back(iter->second.first);

            }

        }

    }

    if (_row_id_column_iterator_pair.first != nullptr) {

        _lazy_read_ctx.all_predicate_col_ids.emplace_back(_row_id_column_iterator_pair.second);

    }

    for (auto& kv : _lazy_read_ctx.fill_partition_columns) {

        auto iter = predicate_columns.find(kv.first);

        if (iter == predicate_columns.end()) {

            _lazy_read_ctx.partition_columns.emplace(kv.first, kv.second);

        } else {

            _lazy_read_ctx.predicate_partition_columns.emplace(kv.first, kv.second);

            _lazy_read_ctx.all_predicate_col_ids.emplace_back(iter->second.first);

        }

    }

    for (auto& kv : _lazy_read_ctx.fill_missing_columns) {

        auto iter = predicate_columns.find(kv.first);

        if (iter != predicate_columns.end()) {

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

            if (_slot_id_to_filter_conjuncts->find(iter->second.second) !=

                _slot_id_to_filter_conjuncts->end()) {

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

                    _lazy_read_ctx.missing_columns_conjuncts.emplace_back(ctx);

                }

            }

            _lazy_read_ctx.predicate_missing_columns.emplace(kv.first, kv.second);

            _lazy_read_ctx.all_predicate_col_ids.emplace_back(iter->second.first);

        } else if (auto row_lineage_idx = check_iceberg_row_lineage_column_idx(kv.first);

                   row_lineage_idx != -1) {

            _lazy_read_ctx.predicate_missing_columns.emplace(kv.first, kv.second);

            _lazy_read_ctx.all_predicate_col_ids.emplace_back(row_lineage_idx);

        } else {

            _lazy_read_ctx.missing_columns.emplace(kv.first, kv.second);

        }

    }

    if (_enable_lazy_mat && _lazy_read_ctx.predicate_columns.first.size() > 0 &&

        _lazy_read_ctx.lazy_read_columns.size() > 0) {

        _lazy_read_ctx.can_lazy_read = true;

    }

    if (!_lazy_read_ctx.can_lazy_read) {

        for (auto& kv : _lazy_read_ctx.predicate_partition_columns) {

            _lazy_read_ctx.partition_columns.emplace(kv.first, kv.second);

        }

        for (auto& kv : _lazy_read_ctx.predicate_missing_columns) {

            _lazy_read_ctx.missing_columns.emplace(kv.first, kv.second);

        }

    }

    if (_filter_groups && (_total_groups == 0 || _t_metadata->num_rows == 0 || _range_size < 0)) {

        return Status::EndOfFile("No row group to read");

    }

    _fill_all_columns = true;

    return Status::OK();

}

// init file reader and file metadata for parsing schema

Status ParquetReader::init_schema_reader() {

    RETURN_IF_ERROR(_open_file());

    _t_metadata = &(_file_metadata->to_thrift());

    return Status::OK();

}

Status ParquetReader::get_parsed_schema(std::vector<std::string>* col_names,

                                        std::vector<DataTypePtr>* col_types) {

    _total_groups = _t_metadata->row_groups.size();

    auto schema_desc = _file_metadata->schema();

    for (int i = 0; i < schema_desc.size(); ++i) {

        // Get the Column Reader for the boolean column

        col_names->emplace_back(schema_desc.get_column(i)->name);

        col_types->emplace_back(make_nullable(schema_desc.get_column(i)->data_type));

    }

    return Status::OK();

}

void ParquetReader::set_iceberg_rowid_params(const std::string& file_path,

                                             int32_t partition_spec_id,

                                             const std::string& partition_data_json,

                                             int row_id_column_pos) {

    _iceberg_rowid_params.enabled = true;

    _iceberg_rowid_params.file_path = file_path;

    _iceberg_rowid_params.partition_spec_id = partition_spec_id;

    _iceberg_rowid_params.partition_data_json = partition_data_json;

    _iceberg_rowid_params.row_id_column_pos = row_id_column_pos;

    if (_current_group_reader != nullptr) {

        _current_group_reader->set_iceberg_rowid_params(_iceberg_rowid_params);

    }

}

Status ParquetReader::get_columns(std::unordered_map<std::string, DataTypePtr>* name_to_type,

                                  std::unordered_set<std::string>* missing_cols) {

    const auto& schema_desc = _file_metadata->schema();

    std::unordered_set<std::string> column_names;

    schema_desc.get_column_names(&column_names);

    for (auto& name : column_names) {

        auto field = schema_desc.get_column(name);

        name_to_type->emplace(name, field->data_type);

    }

    for (auto& col : _missing_cols) {

        missing_cols->insert(col);

    }

    return Status::OK();

}

Status ParquetReader::get_next_block(Block* block, size_t* read_rows, bool* eof) {

    if (_current_group_reader == nullptr || _row_group_eof) {

        Status st = _next_row_group_reader();

        if (!st.ok() && !st.is<ErrorCode::END_OF_FILE>()) {

            return st;

        }

        if (_current_group_reader == nullptr || _row_group_eof || st.is<ErrorCode::END_OF_FILE>()) {

            _current_group_reader.reset(nullptr);

            _row_group_eof = true;

            *read_rows = 0;

            *eof = true;

            return Status::OK();

        }

    }

    if (_push_down_agg_type == TPushAggOp::type::COUNT) {

        auto rows = std::min(_current_group_reader->get_remaining_rows(), (int64_t)_batch_size);

        _current_group_reader->set_remaining_rows(_current_group_reader->get_remaining_rows() -

                                                  rows);

        auto mutate_columns = block->mutate_columns();

        for (auto& col : mutate_columns) {

            col->resize(rows);

        }

        block->set_columns(std::move(mutate_columns));

        *read_rows = rows;

        if (_current_group_reader->get_remaining_rows() == 0) {

            _current_group_reader.reset(nullptr);

        }

        return Status::OK();

    }

    // Limit memory per batch for load paths.

    // _load_bytes_per_row is updated after each batch so the *next* call pre-shrinks _batch_size

    // before reading, ensuring the current batch is already within the limit (from call 2 onward).

    const int64_t max_block_bytes =

            (_state != nullptr && _state->query_type() == TQueryType::LOAD &&

             config::load_reader_max_block_bytes > 0)

                    ? config::load_reader_max_block_bytes

                    : 0;

    if (max_block_bytes > 0 && _load_bytes_per_row > 0) {

        _batch_size = std::max((size_t)1,

                               (size_t)((int64_t)max_block_bytes / (int64_t)_load_bytes_per_row));

    }

    SCOPED_RAW_TIMER(&_reader_statistics.column_read_time);

    Status batch_st =

            _current_group_reader->next_batch(block, _batch_size, read_rows, &_row_group_eof);

    if (batch_st.is<ErrorCode::END_OF_FILE>()) {

        block->clear_column_data();

        _current_group_reader.reset(nullptr);

        *read_rows = 0;

        *eof = true;

        return Status::OK();

    }

    if (!batch_st.ok()) {

        return Status::InternalError("Read parquet file {} failed, reason = {}", _scan_range.path,

                                     batch_st.to_string());

    }

    if (max_block_bytes > 0 && *read_rows > 0) {

        _load_bytes_per_row = block->bytes() / *read_rows;

    }

    if (_row_group_eof) {

        auto column_st = _current_group_reader->merged_column_statistics();

        _column_statistics.merge(column_st);

        _reader_statistics.lazy_read_filtered_rows +=

                _current_group_reader->lazy_read_filtered_rows();

        _reader_statistics.predicate_filter_time += _current_group_reader->predicate_filter_time();

        _reader_statistics.dict_filter_rewrite_time +=

                _current_group_reader->dict_filter_rewrite_time();

        if (_io_ctx) {

            _io_ctx->condition_cache_filtered_rows +=

                    _current_group_reader->condition_cache_filtered_rows();

        }

        if (_current_row_group_index.row_group_id + 1 == _total_groups) {

            *eof = true;

        } else {

            *eof = false;

        }

    }

    return Status::OK();

}

RowGroupReader::PositionDeleteContext ParquetReader::_get_position_delete_ctx(

        const tparquet::RowGroup& row_group, const RowGroupReader::RowGroupIndex& row_group_index) {

    if (_delete_rows == nullptr) {

        return RowGroupReader::PositionDeleteContext(row_group.num_rows, row_group_index.first_row);

    }

    const int64_t* delete_rows = &(*_delete_rows)[0];

    const int64_t* delete_rows_end = delete_rows + _delete_rows->size();

    const int64_t* start_pos = std::lower_bound(delete_rows + _delete_rows_index, delete_rows_end,

                                                row_group_index.first_row);

    int64_t start_index = start_pos - delete_rows;

    const int64_t* end_pos = std::lower_bound(start_pos, delete_rows_end, row_group_index.last_row);

    int64_t end_index = end_pos - delete_rows;

    _delete_rows_index = end_index;

    return RowGroupReader::PositionDeleteContext(*_delete_rows, row_group.num_rows,

                                                 row_group_index.first_row, start_index, end_index);

}

Status ParquetReader::_next_row_group_reader() {

    if (_current_group_reader != nullptr) {

        _current_group_reader->collect_profile_before_close();

    }

    RowRanges candidate_row_ranges;

    while (++_current_row_group_index.row_group_id < _total_groups) {

        const auto& row_group = _t_metadata->row_groups[_current_row_group_index.row_group_id];

        _current_row_group_index.first_row = _current_row_group_index.last_row;

        _current_row_group_index.last_row = _current_row_group_index.last_row + row_group.num_rows;

        if (_filter_groups && _is_misaligned_range_group(row_group)) {

            continue;

        }

        candidate_row_ranges.clear();

        // The range of lines to be read is determined by the push down predicate.

        RETURN_IF_ERROR(_process_min_max_bloom_filter(

                _current_row_group_index, row_group, _push_down_predicates, &candidate_row_ranges));

        std::function<int64_t(const FieldSchema*)> column_compressed_size =

                [&row_group, &column_compressed_size](const FieldSchema* field) -> int64_t {

            if (field->physical_column_index >= 0) {

                int parquet_col_id = field->physical_column_index;

                if (row_group.columns[parquet_col_id].__isset.meta_data) {

                    return row_group.columns[parquet_col_id].meta_data.total_compressed_size;

                }

                return 0;

            }

            int64_t size = 0;

            for (const FieldSchema& child : field->children) {

                size += column_compressed_size(&child);

            }

            return size;

        };

        int64_t group_size = 0; // only calculate the needed columns

        for (auto& read_col : _read_file_columns) {

            const FieldSchema* field = _file_metadata->schema().get_column(read_col);

            group_size += column_compressed_size(field);

        }

        _reader_statistics.read_rows += candidate_row_ranges.count();

        if (_io_ctx) {

            _io_ctx->file_reader_stats->read_rows += candidate_row_ranges.count();

        }

        if (candidate_row_ranges.count() != 0) {

            // need read this row group.

            _reader_statistics.read_row_groups++;

            _reader_statistics.filtered_page_rows +=

                    row_group.num_rows - candidate_row_ranges.count();

            break;

        } else {

            // this row group be filtered.

            _reader_statistics.filtered_row_groups++;

            _reader_statistics.filtered_bytes += group_size;

            _reader_statistics.filtered_group_rows += row_group.num_rows;

        }

    }

    if (_current_row_group_index.row_group_id == _total_groups) {

        _row_group_eof = true;

        _current_group_reader.reset(nullptr);

        return Status::EndOfFile("No next RowGroupReader");

    }

    // process page index and generate the ranges to read

    auto& row_group = _t_metadata->row_groups[_current_row_group_index.row_group_id];

    RowGroupReader::PositionDeleteContext position_delete_ctx =

            _get_position_delete_ctx(row_group, _current_row_group_index);

    io::FileReaderSPtr group_file_reader;

    if (typeid_cast<io::InMemoryFileReader*>(_file_reader.get())) {

        // InMemoryFileReader has the ability to merge small IO

        group_file_reader = _file_reader;

    } else {

        size_t avg_io_size = 0;

        const std::vector<io::PrefetchRange> io_ranges =

                _generate_random_access_ranges(_current_row_group_index, &avg_io_size);

        int64_t merged_read_slice_size = -1;

        if (_state != nullptr && _state->query_options().__isset.merge_read_slice_size) {

            merged_read_slice_size = _state->query_options().merge_read_slice_size;

        }

        // The underlying page reader will prefetch data in column.

        // Using both MergeRangeFileReader and BufferedStreamReader simultaneously would waste a lot of memory.

        group_file_reader =

                avg_io_size < io::MergeRangeFileReader::SMALL_IO

                        ? std::make_shared<io::MergeRangeFileReader>(

                                  _profile, _file_reader, io_ranges, merged_read_slice_size)

                        : _file_reader;

    }

    _current_group_reader.reset(new RowGroupReader(

            _io_ctx ? std::make_shared<io::TracingFileReader>(group_file_reader,

                                                              _io_ctx->file_reader_stats)

                    : group_file_reader,

            _read_table_columns, _current_row_group_index.row_group_id, row_group, _ctz, _io_ctx,

            position_delete_ctx, _lazy_read_ctx, _state, _column_ids, _filter_column_ids));

    if (_iceberg_rowid_params.enabled) {

        _current_group_reader->set_iceberg_rowid_params(_iceberg_rowid_params);

    }

    _row_group_eof = false;

    _current_group_reader->set_current_row_group_idx(_current_row_group_index);

    _current_group_reader->set_row_id_column_iterator(_row_id_column_iterator_pair);

    _current_group_reader->set_row_lineage_columns(_row_lineage_columns);

    _current_group_reader->set_col_name_to_block_idx(_col_name_to_block_idx);

    if (_condition_cache_ctx) {

        _current_group_reader->set_condition_cache_context(_condition_cache_ctx);

    }

    _current_group_reader->_table_info_node_ptr = _table_info_node_ptr;

    return _current_group_reader->init(_file_metadata->schema(), candidate_row_ranges, _col_offsets,

                                       _tuple_descriptor, _row_descriptor, _colname_to_slot_id,

                                       _not_single_slot_filter_conjuncts,

                                       _slot_id_to_filter_conjuncts);

}

std::vector<io::PrefetchRange> ParquetReader::_generate_random_access_ranges(

        const RowGroupReader::RowGroupIndex& group, size_t* avg_io_size) {

    std::vector<io::PrefetchRange> result;

    int64_t last_chunk_end = -1;

    size_t total_io_size = 0;

    std::function<void(const FieldSchema*, const tparquet::RowGroup&)> scalar_range =

            [&](const FieldSchema* field, const tparquet::RowGroup& row_group) {

                if (_column_ids.empty() ||

                    _column_ids.find(field->get_column_id()) != _column_ids.end()) {

                    if (field->data_type->get_primitive_type() == TYPE_ARRAY) {

                        scalar_range(&field->children[0], row_group);

                    } else if (field->data_type->get_primitive_type() == TYPE_MAP) {

                        scalar_range(&field->children[0], row_group);

                        scalar_range(&field->children[1], row_group);

                    } else if (field->data_type->get_primitive_type() == TYPE_STRUCT) {

                        for (int i = 0; i < field->children.size(); ++i) {

                            scalar_range(&field->children[i], row_group);

                        }

                    } else {

                        const tparquet::ColumnChunk& chunk =

                                row_group.columns[field->physical_column_index];

                        auto& chunk_meta = chunk.meta_data;

                        int64_t chunk_start = has_dict_page(chunk_meta)

                                                      ? chunk_meta.dictionary_page_offset

                                                      : chunk_meta.data_page_offset;

                        int64_t chunk_end = chunk_start + chunk_meta.total_compressed_size;

                        DCHECK_GE(chunk_start, last_chunk_end);

                        result.emplace_back(chunk_start, chunk_end);

                        total_io_size += chunk_meta.total_compressed_size;

                        last_chunk_end = chunk_end;

                    }

                }

            };

    const tparquet::RowGroup& row_group = _t_metadata->row_groups[group.row_group_id];

    for (const auto& read_col : _read_file_columns) {

        const FieldSchema* field = _file_metadata->schema().get_column(read_col);

        scalar_range(field, row_group);

    }

    if (!result.empty()) {

        *avg_io_size = total_io_size / result.size();

    }

    return result;

}

bool ParquetReader::_is_misaligned_range_group(const tparquet::RowGroup& row_group) const {

    int64_t start_offset = _get_column_start_offset(row_group.columns[0].meta_data);

    auto& last_column = row_group.columns[row_group.columns.size() - 1].meta_data;

    int64_t end_offset = _get_column_start_offset(last_column) + last_column.total_compressed_size;

    int64_t row_group_mid = start_offset + (end_offset - start_offset) / 2;

    if (!(row_group_mid >= _range_start_offset &&

          row_group_mid < _range_start_offset + _range_size)) {

        return true;

    }

    return false;

}

int64_t ParquetReader::get_total_rows() const {

    if (!_t_metadata) return 0;

    if (!_filter_groups) return _t_metadata->num_rows;

    int64_t total = 0;

    for (const auto& rg : _t_metadata->row_groups) {

        if (!_is_misaligned_range_group(rg)) {