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

#include <unordered_map>

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

#include "exprs/runtime_filter_expr.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/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 "storage/index/zone_map/zonemap_eval_context.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 {

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, 1UL)),

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

}

void ParquetReader::set_batch_size(size_t batch_size) {

    if (_batch_size == batch_size) {

        return;

    }

    _batch_size = batch_size;

}

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, 1UL)),

          _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_expr_zonemap = ADD_CHILD_COUNTER_WITH_LEVEL(

                _profile, "ParquetExprZoneMapFilteredRowGroups", 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.expr_zonemap_unusable = ADD_CHILD_COUNTER_WITH_LEVEL(

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

        _parquet_profile.in_zonemap_point_check = ADD_CHILD_COUNTER_WITH_LEVEL(

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

        _parquet_profile.in_zonemap_range_only = ADD_CHILD_COUNTER_WITH_LEVEL(

                _profile, "InZoneMapRangeOnlyCount", TUnit::UNIT, 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.convert_time =

                ADD_CHILD_TIMER_WITH_LEVEL(_profile, "ConvertTime", 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) {

        _current_group_reader.reset();

        _tracing_file_reader.reset();

        _file_reader.reset();

        _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 ? _state->query_options() : _default_query_options, _file_description);

        if (_io_ctx_holder) {

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

                    _profile, _system_properties, _file_description, reader_options,

                    io::DelegateReader::AccessMode::RANDOM,

                    std::static_pointer_cast<const io::IOContext>(_io_ctx_holder)));

        } else {

            _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 && _io_ctx->file_reader_stats

                                       ? 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::on_before_init_reader(ReaderInitContext* ctx) {

    _column_descs = ctx->column_descs;

    _fill_col_name_to_block_idx = ctx->col_name_to_block_idx;

    RETURN_IF_ERROR(_extract_partition_values(*ctx->range, ctx->tuple_descriptor,

                                              _fill_partition_values,

                                              &_fill_partition_value_is_null));

    for (const auto& desc : *ctx->column_descs) {

        if (desc.category == ColumnCategory::REGULAR ||

            desc.category == ColumnCategory::GENERATED) {

            ctx->column_names.push_back(desc.name);

        } else if (desc.category == ColumnCategory::SYNTHESIZED &&

                   desc.name.starts_with(BeConsts::GLOBAL_ROWID_COL)) {

            auto topn_row_id_column_iter = _create_topn_row_id_column_iterator();

            this->register_synthesized_column_handler(

                    desc.name,

                    [iter = std::move(topn_row_id_column_iter), this, &desc](

                            Block* block, size_t rows) -> Status {

                        return fill_topn_row_id(iter, desc.name, block, rows);

                    });

            continue;

        }

    }

    // Build table_info_node from Parquet file metadata with case-insensitive recursive matching.

    // File is already opened by init_reader before this hook, so metadata is available.

    // tuple_descriptor may be null in unit tests that only set column_descs.

    if (ctx->tuple_descriptor != nullptr) {

        const FieldDescriptor* field_desc = nullptr;

        RETURN_IF_ERROR(get_file_metadata_schema(&field_desc));

        RETURN_IF_ERROR(TableSchemaChangeHelper::BuildTableInfoUtil::by_parquet_name(

                ctx->tuple_descriptor, *field_desc, ctx->table_info_node));

    }

    return Status::OK();

}

Status ParquetReader::_open_file_reader(ReaderInitContext* /*ctx*/) {

    return _open_file();

}

Status ParquetReader::_do_init_reader(ReaderInitContext* base_ctx) {

    auto* ctx = checked_context_cast<ParquetInitContext>(base_ctx);

    _col_name_to_block_idx = base_ctx->col_name_to_block_idx;

    _tuple_descriptor = ctx->tuple_descriptor;

    _row_descriptor = ctx->row_descriptor;

    _colname_to_slot_id = ctx->colname_to_slot_id;

    _not_single_slot_filter_conjuncts = ctx->not_single_slot_filter_conjuncts;

    _slot_id_to_filter_conjuncts = ctx->slot_id_to_filter_conjuncts;

    _filter_groups = ctx->filter_groups;

    _table_info_node_ptr = base_ctx->table_info_node;

    _column_ids = base_ctx->column_ids;

    _filter_column_ids = base_ctx->filter_column_ids;

    // _open_file_reader (called by init_reader NVI before hooks) must have opened the file.

    DCHECK(_file_metadata != nullptr)

            << "ParquetReader::_do_init_reader called without _open_file_reader";

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

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

    // Compute missing columns and file↔table column mapping.

    // This runs in _do_init_reader (not on_before_init_reader) because table-format readers

    // (Iceberg, Paimon, Hive, Hudi) override on_before_init_reader completely.

    if (has_column_descs()) {

        _fill_missing_cols.clear();

        _fill_missing_defaults.clear();

        for (const auto& col_name : base_ctx->column_names) {

            // A projected column that is not present at all in the table-side schema tree means the

            // schema info from FE (e.g. the paimon/iceberg history_schema_info, or the hive/orc name

            // map) is inconsistent with the scan projection. Fail this query loudly instead of aborting

            // the whole BE process via the children_column_exists DCHECK (or an std::out_of_range from

            // children.at() in a release build). A column that IS known but missing from the data file

            // keeps its key (add_not_exist_children) and is correctly classified as fill-missing below.

            if (!_table_info_node_ptr->has_children_column(col_name)) {

                return Status::InternalError(

                        "schema mapping is missing projected column '{}'; the schema info from FE "

                        "is "

                        "inconsistent with the scan projection (file: {})",

                        col_name, _scan_range.path);

            }

            if (!_table_info_node_ptr->children_column_exists(col_name)) {

                _fill_missing_cols.insert(col_name);

            }

        }

        if (_column_descs && !_fill_missing_cols.empty()) {

            for (const auto& desc : *_column_descs) {

                if (_fill_missing_cols.contains(desc.name) &&

                    !_fill_partition_values.contains(desc.name)) {

                    _fill_missing_defaults[desc.name] = desc.default_expr;

                }

            }

        }

    }

    // Resolve file-column ↔ table-column mapping in file-schema order.

    // _init_read_columns handles both normal path (missing cols populated above)

    // and standalone path (_fill_missing_cols empty, _table_info_node_ptr may be null).

    _init_read_columns(base_ctx->column_names);

    // build column predicates for column lazy read

    if (ctx->conjuncts != nullptr) {

        _lazy_read_ctx.conjuncts = *ctx->conjuncts;

    }

    if (ctx->slot_id_to_predicates != nullptr) {

        _lazy_read_ctx.slot_id_to_predicates = *ctx->slot_id_to_predicates;

    }

    // ---- Inlined set_fill_columns logic (partition/missing/synthesized classification) ----

    // 1. Collect predicate columns from conjuncts for lazy materialization

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

    _collect_predicate_columns_from_conjuncts(predicate_columns);

    // 2. Classify read/partition/missing/synthesized columns into lazy vs predicate groups

    _classify_columns_for_lazy_read(predicate_columns, _fill_partition_values,

                                    _fill_missing_defaults);

    // 3. Populate col_names vectors for ColumnProcessor path

    for (auto& kv : _lazy_read_ctx.predicate_partition_columns) {

        _lazy_read_ctx.predicate_partition_col_names.emplace_back(kv.first);

    }

    for (auto& kv : _lazy_read_ctx.predicate_missing_columns) {

        _lazy_read_ctx.predicate_missing_col_names.emplace_back(kv.first);

    }

    for (auto& kv : _lazy_read_ctx.partition_columns) {

        _lazy_read_ctx.partition_col_names.emplace_back(kv.first);

    }

    for (auto& kv : _lazy_read_ctx.missing_columns) {

        _lazy_read_ctx.missing_col_names.emplace_back(kv.first);

    }

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

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

    }

    return Status::OK();

}

void ParquetReader::_init_read_columns(const std::vector<std::string>& column_names) {

    // Build file_col_name → table_col_name map, skipping missing columns.

    // Must iterate file schema in physical order so that _generate_random_access_ranges

    // sees monotonically increasing chunk offsets.

    auto schema_desc = _file_metadata->schema();

    std::map<std::string, std::string> required_file_columns;

    for (const auto& col_name : column_names) {

        if (_fill_missing_cols.contains(col_name)) {

            continue;

        }

        std::string file_col = col_name;

        if (_table_info_node_ptr && _table_info_node_ptr->children_column_exists(col_name)) {

            file_col = _table_info_node_ptr->children_file_column_name(col_name);

        }

        required_file_columns[file_col] = col_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]);

        }

    }

}

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

           !converter::requires_datetimev2_precision_conversion(file_col_type, table_col_type);

}

void ParquetReader::_collect_predicate_columns_from_conjuncts(

        std::unordered_map<std::string, std::pair<uint32_t, int>>& predicate_columns) {

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

            RuntimeFilterExpr* runtime_filter = assert_cast<RuntimeFilterExpr*>(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) {

                // Parquet shares _push_down_predicates for row-group/page min-max pruning and

                // bloom-filter evaluation, so this flag currently gates both predicate paths.

                if (!has_column_optimization(pred->col_name(), ColumnOptimizationTypes::MIN_MAX)) {

                    continue;

                }

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

        }

    }

}

void ParquetReader::_classify_columns_for_lazy_read(

        const std::unordered_map<std::string, std::pair<uint32_t, int>>&

                predicate_conjuncts_columns,

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

                partition_columns,

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

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

    auto predicate_columns = predicate_conjuncts_columns;

#ifndef BE_TEST

    for (const auto& [col_name, _] : _generated_col_handlers) {

        int slot_id = -1;

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

            if (slot->col_name() == col_name) {

                slot_id = slot->id();

                break;

            }

        }

        DCHECK(slot_id != -1) << "slot id should not be -1 for generated column: " << col_name;

        auto column_index = _row_descriptor->get_column_id(slot_id);

        if (column_index == 0) {

            _lazy_read_ctx.resize_first_column = false;

        }

        // assume generated columns are only used for predicate push down.

        predicate_columns.emplace(col_name, std::make_pair(column_index, slot_id));

    }

    for (const auto& [col_name, _] : _synthesized_col_handlers) {

        int slot_id = -1;

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

            if (slot->col_name() == col_name) {

                slot_id = slot->id();

                break;

            }

        }

        DCHECK(slot_id != -1) << "slot id should not be -1 for synthesized column: " << col_name;

        auto column_index = _row_descriptor->get_column_id(slot_id);

        if (column_index == 0) {

            _lazy_read_ctx.resize_first_column = false;

        }

        // synthesized columns always fill data on first phase.

        _lazy_read_ctx.all_predicate_col_ids.emplace_back(column_index);

    }

#endif

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

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

            }

        }

    }

    for (auto& kv : 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 : 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 {

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

        }

    }

}

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

}

Status ParquetReader::_get_columns_impl(

        std::unordered_map<std::string, DataTypePtr>* name_to_type) {

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

    }

    return Status::OK();

}

Status ParquetReader::_do_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();

        }

    }

    // 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 (_deletion_vector != nullptr) {

        _row_group_deletion_vector_rows.clear();

        auto it = _deletion_vector->begin();

        it.move(cast_set<uint64_t>(row_group_index.first_row));

        const auto end = _deletion_vector->end();

        while (it != end && *it < cast_set<uint64_t>(row_group_index.last_row)) {

            _row_group_deletion_vector_rows.push_back(cast_set<int64_t>(*it));

            ++it;

        }

        // Iceberg may plan a DV together with older position delete files. Preserve both sources

        // while keeping only the current row group's temporary expansion.

        if (_delete_rows != nullptr) {

            const auto position_begin = std::lower_bound(_delete_rows->begin(), _delete_rows->end(),

                                                         row_group_index.first_row);

            const auto position_end =

                    std::lower_bound(position_begin, _delete_rows->end(), row_group_index.last_row);

            _row_group_deletion_vector_rows.insert(_row_group_deletion_vector_rows.end(),

                                                   position_begin, position_end);

            std::ranges::sort(_row_group_deletion_vector_rows);

            auto unique_end = std::ranges::unique(_row_group_deletion_vector_rows).begin();

            _row_group_deletion_vector_rows.erase(unique_end,

                                                  _row_group_deletion_vector_rows.end());

        }

        return RowGroupReader::PositionDeleteContext(

                _row_group_deletion_vector_rows, row_group.num_rows, row_group_index.first_row, 0,

                cast_set<int64_t>(_row_group_deletion_vector_rows.size()));

    }

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

            _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 =