// 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_v2/parquet/parquet_scan.h"

#include <parquet/encoding.h>

#include <algorithm>

#include <limits>

#include <memory>

#include <ranges>

#include <set>

#include <unordered_set>

#include <utility>

#include "common/exception.h"

#include "common/status.h"

#include "core/assert_cast.h"

#include "core/block/block.h"

#include "core/column/column_vector.h"

#include "exprs/vcompound_pred.h"

#include "exprs/vexpr_context.h"

#include "format_v2/parquet/parquet_column_schema.h"

#include "format_v2/parquet/parquet_file_context.h"

#include "format_v2/parquet/parquet_statistics.h"

namespace doris::format::parquet {

namespace {

int64_t column_start_offset(const ::parquet::ColumnChunkMetaData& column_metadata) {

    return column_metadata.has_dictionary_page()

                   ? cast_set<int64_t>(column_metadata.dictionary_page_offset())

                   : cast_set<int64_t>(column_metadata.data_page_offset());

}

bool is_dictionary_data_encoding(::parquet::Encoding::type encoding) {

    return encoding == ::parquet::Encoding::PLAIN_DICTIONARY ||

           encoding == ::parquet::Encoding::RLE_DICTIONARY;

}

bool is_level_encoding(::parquet::Encoding::type encoding) {

    return encoding == ::parquet::Encoding::RLE || encoding == ::parquet::Encoding::BIT_PACKED;

}

bool is_data_page_type(::parquet::PageType::type page_type) {

    return page_type == ::parquet::PageType::DATA_PAGE ||

           page_type == ::parquet::PageType::DATA_PAGE_V2;

}

bool is_fully_dictionary_encoded_chunk(const ::parquet::ColumnChunkMetaData& column_metadata) {

    if (!column_metadata.has_dictionary_page()) {

        return false;

    }

    const auto& encoding_stats = column_metadata.encoding_stats();

    if (!encoding_stats.empty()) {

        bool has_dictionary_data_page = false;

        for (const auto& encoding_stat : encoding_stats) {

            if (!is_data_page_type(encoding_stat.page_type) || encoding_stat.count <= 0) {

                continue;

            }

            if (!is_dictionary_data_encoding(encoding_stat.encoding)) {

                return false;

            }

            has_dictionary_data_page = true;

        }

        return has_dictionary_data_page;

    }

    bool has_dictionary_encoding = false;

    for (const auto encoding : column_metadata.encodings()) {

        if (is_dictionary_data_encoding(encoding)) {

            has_dictionary_encoding = true;

            continue;

        }

        if (!is_level_encoding(encoding)) {

            return false;

        }

    }

    return has_dictionary_encoding;

}

bool supports_row_level_dictionary_filter(const ParquetColumnSchema& column_schema,

                                          const ::parquet::ColumnChunkMetaData& column_metadata) {

    if (column_schema.kind != ParquetColumnSchemaKind::PRIMITIVE ||

        column_schema.descriptor == nullptr || column_schema.type == nullptr ||

        column_schema.max_repetition_level > 0) {

        return false;

    }

    if (!column_schema.type_descriptor.is_string_like ||

        column_metadata.type() != ::parquet::Type::BYTE_ARRAY) {

        return false;

    }

    // Row-level dictionary filtering consumes dictionary ids from DATA_PAGE payloads. It is exact

    // only when every data page is dictionary encoded. Mixed dictionary/plain chunks are left on

    // the normal decoded-value path, matching the safety rule used by StarRocks and Doris v1.

    return is_fully_dictionary_encoded_chunk(column_metadata);

}

void collect_all_leaf_column_ids(const ParquetColumnSchema& column_schema,

                                 std::unordered_set<int>* leaf_column_ids) {

    DORIS_CHECK(leaf_column_ids != nullptr);

    if (column_schema.kind == ParquetColumnSchemaKind::PRIMITIVE) {

        if (column_schema.leaf_column_id >= 0) {

            leaf_column_ids->insert(column_schema.leaf_column_id);

        }

        return;

    }

    for (const auto& child : column_schema.children) {

        DORIS_CHECK(child != nullptr);

        collect_all_leaf_column_ids(*child, leaf_column_ids);

    }

}

void collect_projected_leaf_column_ids(const ParquetColumnSchema& column_schema,

                                       const format::LocalColumnIndex& projection,

                                       std::unordered_set<int>* leaf_column_ids) {

    DORIS_CHECK(leaf_column_ids != nullptr);

    if (projection.project_all_children || projection.children.empty()) {

        collect_all_leaf_column_ids(column_schema, leaf_column_ids);

        return;

    }

    for (const auto& child_projection : projection.children) {

        const auto child_it =

                std::ranges::find_if(column_schema.children, [&](const auto& child_schema) {

                    return child_schema->local_id == child_projection.local_id();

                });

        DORIS_CHECK(child_it != column_schema.children.end());

        collect_projected_leaf_column_ids(**child_it, child_projection, leaf_column_ids);

    }

}

bool is_row_group_outside_range(const ::parquet::FileMetaData& metadata,

                                const ParquetScanRange& scan_range, int row_group_idx) {

    if (scan_range.size < 0) {

        return false;

    }

    const int64_t range_start_offset = scan_range.start_offset;

    const int64_t range_end_offset = range_start_offset + scan_range.size;

    DORIS_CHECK(range_start_offset >= 0);

    DORIS_CHECK(range_end_offset >= range_start_offset);

    if (range_start_offset == 0 &&

        (scan_range.file_size < 0 || range_end_offset >= scan_range.file_size)) {

        return false;

    }

    auto row_group_metadata = metadata.RowGroup(row_group_idx);

    DORIS_CHECK(row_group_metadata != nullptr);

    DORIS_CHECK(row_group_metadata->num_columns() > 0);

    const auto first_column = row_group_metadata->ColumnChunk(0);

    const auto last_column = row_group_metadata->ColumnChunk(row_group_metadata->num_columns() - 1);

    DORIS_CHECK(first_column != nullptr);

    DORIS_CHECK(last_column != nullptr);

    const int64_t row_group_start_offset = column_start_offset(*first_column);

    const int64_t row_group_end_offset =

            column_start_offset(*last_column) + last_column->total_compressed_size();

    const int64_t row_group_mid_offset =

            row_group_start_offset + (row_group_end_offset - row_group_start_offset) / 2;

    return row_group_mid_offset < range_start_offset || row_group_mid_offset >= range_end_offset;

}

std::vector<format::LocalColumnIndex> request_scan_columns(const format::FileScanRequest& request) {

    std::vector<format::LocalColumnIndex> scan_columns;

    scan_columns.reserve(request.predicate_columns.size() + request.non_predicate_columns.size());

    scan_columns.insert(scan_columns.end(), request.predicate_columns.begin(),

                        request.predicate_columns.end());

    scan_columns.insert(scan_columns.end(), request.non_predicate_columns.begin(),

                        request.non_predicate_columns.end());

    return scan_columns;

}

std::vector<ParquetPageCacheRange> build_row_group_prefetch_ranges(

        const ::parquet::FileMetaData& metadata,

        const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema,

        const std::vector<format::LocalColumnIndex>& scan_columns, int row_group_idx) {

    std::unordered_set<int> leaf_column_ids;

    for (const auto& projection : scan_columns) {

        const auto local_id = projection.local_id();

        if (local_id == format::ROW_POSITION_COLUMN_ID ||

            local_id == format::GLOBAL_ROWID_COLUMN_ID) {

            continue;

        }

        DORIS_CHECK(local_id >= 0 && local_id < static_cast<int32_t>(file_schema.size()));

        DORIS_CHECK(file_schema[local_id] != nullptr);

        // Prefetch and merge-reader ranges must be physical leaf chunks, not Doris logical slots.

        // Example: for a struct column s<a:int,b:string>, projecting only s.a should include only

        // the Parquet leaf chunk of a. Projecting the whole struct includes both a and b.

        collect_projected_leaf_column_ids(*file_schema[local_id], projection, &leaf_column_ids);

    }

    auto row_group_metadata = metadata.RowGroup(row_group_idx);

    DORIS_CHECK(row_group_metadata != nullptr);

    std::vector<int> ordered_leaf_column_ids(leaf_column_ids.begin(), leaf_column_ids.end());

    std::ranges::sort(ordered_leaf_column_ids);

    std::vector<ParquetPageCacheRange> ranges;

    ranges.reserve(ordered_leaf_column_ids.size());

    for (const auto leaf_column_id : ordered_leaf_column_ids) {

        DORIS_CHECK(leaf_column_id >= 0 && leaf_column_id < row_group_metadata->num_columns());

        auto column_metadata = row_group_metadata->ColumnChunk(leaf_column_id);

        DORIS_CHECK(column_metadata != nullptr);

        const int64_t offset = column_start_offset(*column_metadata);

        const int64_t size = column_metadata->total_compressed_size();

        DORIS_CHECK(offset >= 0);

        if (size > 0) {

            ranges.push_back(ParquetPageCacheRange {.offset = offset, .size = size});

        }

    }

    return ranges;

}

Status select_row_groups_by_scan_range(const ::parquet::FileMetaData& metadata,

                                       const ParquetScanRange& scan_range,

                                       std::vector<int64_t>* row_group_first_rows,

                                       std::vector<int>* selected_row_groups) {

    DORIS_CHECK(row_group_first_rows != nullptr);

    DORIS_CHECK(selected_row_groups != nullptr);

    row_group_first_rows->assign(metadata.num_row_groups(), 0);

    selected_row_groups->clear();

    selected_row_groups->reserve(metadata.num_row_groups());

    int64_t next_row_group_first_row = 0;

    for (int row_group_idx = 0; row_group_idx < metadata.num_row_groups(); ++row_group_idx) {

        (*row_group_first_rows)[row_group_idx] = next_row_group_first_row;

        auto row_group_metadata = metadata.RowGroup(row_group_idx);

        DORIS_CHECK(row_group_metadata != nullptr);

        const int64_t row_group_rows = row_group_metadata->num_rows();

        if (row_group_rows < 0) {

            return Status::Corruption("Invalid negative row count in parquet row group {}",

                                      row_group_idx);

        }

        next_row_group_first_row += row_group_rows;

        if (!is_row_group_outside_range(metadata, scan_range, row_group_idx)) {

            selected_row_groups->push_back(row_group_idx);

        }

    }

    return Status::OK();

}

Status build_row_group_read_plans(

        const ::parquet::FileMetaData& metadata, ::parquet::ParquetFileReader* file_reader,

        const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema,

        const format::FileScanRequest& request, const std::vector<int>& selected_row_groups,

        const std::vector<int64_t>& row_group_first_rows, RowGroupScanPlan* plan,

        const cctz::time_zone* timezone, const RuntimeState* runtime_state) {

    DORIS_CHECK(plan != nullptr);

    plan->row_groups.reserve(selected_row_groups.size());

    for (const auto row_group_idx : selected_row_groups) {

        DORIS_CHECK(row_group_idx >= 0);

        DORIS_CHECK(static_cast<size_t>(row_group_idx) < row_group_first_rows.size());

        auto row_group_metadata = metadata.RowGroup(row_group_idx);

        DORIS_CHECK(row_group_metadata != nullptr);

        const int64_t row_group_rows = row_group_metadata->num_rows();

        if (row_group_rows == 0) {

            continue;

        }

        RowGroupReadPlan row_group_plan;

        row_group_plan.row_group_id = row_group_idx;

        row_group_plan.first_file_row = row_group_first_rows[row_group_idx];

        row_group_plan.row_group_rows = row_group_rows;

        RETURN_IF_ERROR(select_row_group_ranges_by_page_index(

                file_reader, file_schema, request, row_group_idx, row_group_rows,

                &row_group_plan.selected_ranges, &row_group_plan.page_skip_plans,

                &plan->pruning_stats, timezone, runtime_state));

        if (row_group_plan.selected_ranges.empty()) {

            continue;

        }

        plan->pruning_stats.selected_row_ranges += row_group_plan.selected_ranges.size();

        plan->row_groups.push_back(std::move(row_group_plan));

    }

    return Status::OK();

}

} // namespace

Status plan_parquet_row_groups(const ::parquet::FileMetaData& metadata,

                               ::parquet::ParquetFileReader* file_reader,

                               const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema,

                               const format::FileScanRequest& request,

                               const ParquetScanRange& scan_range, bool enable_bloom_filter,

                               RowGroupScanPlan* plan, const cctz::time_zone* timezone,

                               const RuntimeState* runtime_state) {

    DORIS_CHECK(plan != nullptr);

    plan->row_groups.clear();

    plan->pruning_stats = ParquetPruningStats {};

    // Row-group planning flow:

    //

    //     parquet footer row groups

    //              |

    //              v

    //     split byte-range candidates

    //              |

    //              v

    //     row-group metadata pruning

    //     statistics/ZoneMap -> dictionary -> bloom filter

    //              |

    //              v

    //     page-index pruning per selected row group

    //              |

    //              v

    //     RowGroupReadPlan with selected row ranges

    //

    // Metadata pruning removes whole row groups before readers are opened. Page index pruning runs

    // only for remaining row groups and produces selected row ranges; the scan scheduler later skips

    // gaps between those ranges, while row-level VExpr conjuncts still run on loaded batches for

    // correctness.

    std::vector<int64_t> row_group_first_rows;

    std::vector<int> scan_range_selected_row_groups;

    RETURN_IF_ERROR(select_row_groups_by_scan_range(metadata, scan_range, &row_group_first_rows,

                                                    &scan_range_selected_row_groups));

    std::vector<int> metadata_selected_row_groups;

    RETURN_IF_ERROR(select_row_groups_by_metadata(

            metadata, file_reader, file_schema, request, &scan_range_selected_row_groups,

            &metadata_selected_row_groups, enable_bloom_filter, &plan->pruning_stats, timezone,

            runtime_state));

    RETURN_IF_ERROR(build_row_group_read_plans(metadata, file_reader, file_schema, request,

                                               metadata_selected_row_groups, row_group_first_rows,

                                               plan, timezone, runtime_state));

    plan->pruning_stats.selected_row_groups = plan->row_groups.size();

    return Status::OK();

}

namespace {

using DictionaryResidualConjunct = std::pair<VExprContextSPtr, VExprSPtr>;

using DictionaryResidualConjuncts = std::vector<DictionaryResidualConjunct>;

void update_counter_if_not_null(RuntimeProfile::Counter* counter, int64_t value) {

    if (counter != nullptr) {

        COUNTER_UPDATE(counter, value);

    }

}

uint16_t apply_filter_to_selection(const IColumn::Filter& filter, SelectionVector* selection,

                                   uint16_t selected_rows) {

    uint16_t new_selected_rows = 0;

    for (uint16_t selection_idx = 0; selection_idx < selected_rows; ++selection_idx) {

        const auto row_idx = selection->get_index(selection_idx);

        if (filter[row_idx] != 0) {

            selection->set_index(new_selected_rows++, static_cast<SelectionVector::Index>(row_idx));

        }

    }

    return new_selected_rows;

}

Status execute_compact_filter_conjuncts(const VExprContextSPtrs& conjuncts, size_t rows,

                                        Block* file_block, IColumn::Filter* compact_filter,

                                        bool* can_filter_all) {

    DORIS_CHECK(compact_filter != nullptr);

    DORIS_CHECK(can_filter_all != nullptr);

    compact_filter->resize_fill(rows, 1);

    *can_filter_all = false;

    for (const auto& conjunct : conjuncts) {

        DORIS_CHECK(conjunct != nullptr);

        IColumn::Filter filter(rows, 1);

        bool conjunct_can_filter_all = false;

        RETURN_IF_ERROR(conjunct->execute_filter(file_block, filter.data(), rows, false,

                                                 &conjunct_can_filter_all));

        if (conjunct_can_filter_all) {

            std::ranges::fill(*compact_filter, 0);

            *can_filter_all = true;

            break;

        }

        for (size_t row = 0; row < rows; ++row) {

            (*compact_filter)[row] &= filter[row];

        }

    }

    return Status::OK();

}

Status execute_compact_dictionary_residual_conjuncts(const DictionaryResidualConjuncts& conjuncts,

                                                     size_t rows, Block* file_block,

                                                     IColumn::Filter* compact_filter,

                                                     bool* can_filter_all) {

    DORIS_CHECK(compact_filter != nullptr);

    DORIS_CHECK(can_filter_all != nullptr);

    compact_filter->resize_fill(rows, 1);

    *can_filter_all = false;

    for (const auto& [owner_context, residual_expr] : conjuncts) {

        DORIS_CHECK(owner_context != nullptr);

        DORIS_CHECK(residual_expr != nullptr);

        IColumn::Filter filter(rows, 1);

        bool conjunct_can_filter_all = false;

        RETURN_IF_ERROR(residual_expr->execute_filter(owner_context.get(), file_block,

                                                      filter.data(), rows, false,

                                                      &conjunct_can_filter_all));

        if (conjunct_can_filter_all) {

            std::ranges::fill(*compact_filter, 0);

            *can_filter_all = true;

            break;

        }

        for (size_t row = 0; row < rows; ++row) {

            (*compact_filter)[row] &= filter[row];

        }

    }

    return Status::OK();

}

Status execute_compact_delete_conjuncts(const VExprContextSPtrs& delete_conjuncts, size_t rows,

                                        Block* file_block, IColumn::Filter* compact_filter,

                                        bool* can_filter_all) {

    DORIS_CHECK(compact_filter != nullptr);

    DORIS_CHECK(can_filter_all != nullptr);

    compact_filter->resize_fill(rows, 1);

    *can_filter_all = false;

    for (const auto& delete_conjunct : delete_conjuncts) {

        DORIS_CHECK(delete_conjunct != nullptr);

        int result_column_id = -1;

        RETURN_IF_ERROR(delete_conjunct->root()->execute(delete_conjunct.get(), file_block,

                                                         &result_column_id));

        DORIS_CHECK(result_column_id >= 0 &&

                    result_column_id < static_cast<int>(file_block->columns()));

        const auto& delete_filter = assert_cast<const ColumnUInt8&>(

                                            *file_block->get_by_position(result_column_id).column)

                                            .get_data();

        DORIS_CHECK(delete_filter.size() == rows);

        bool has_kept_row = false;

        for (size_t row = 0; row < rows; ++row) {

            (*compact_filter)[row] &= !delete_filter[row];

            has_kept_row |= (*compact_filter)[row] != 0;

        }

        file_block->erase(result_column_id);

        if (!has_kept_row) {

            *can_filter_all = true;

            break;

        }

    }

    return Status::OK();

}

Status execute_filter_conjuncts(const format::FileScanRequest& request, int64_t batch_rows,

                                Block* file_block, SelectionVector* selection,

                                uint16_t* selected_rows) {

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

        if (*selected_rows == 0) {

            break;

        }

        DORIS_CHECK(conjunct != nullptr);

        IColumn::Filter filter(static_cast<size_t>(batch_rows), 1);

        bool can_filter_all = false;

        RETURN_IF_ERROR(conjunct->execute_filter(file_block, filter.data(),

                                                 static_cast<size_t>(batch_rows), false,

                                                 &can_filter_all));

        *selected_rows =

                can_filter_all ? 0 : apply_filter_to_selection(filter, selection, *selected_rows);

    }

    return Status::OK();

}

Status execute_delete_conjuncts(const format::FileScanRequest& request, int64_t batch_rows,

                                Block* file_block, SelectionVector* selection,

                                uint16_t* selected_rows) {

    for (const auto& delete_conjunct : request.delete_conjuncts) {

        if (*selected_rows == 0) {

            break;

        }

        DORIS_CHECK(delete_conjunct != nullptr);

        int result_column_id = -1;

        RETURN_IF_ERROR(delete_conjunct->root()->execute(delete_conjunct.get(), file_block,

                                                         &result_column_id));

        DORIS_CHECK(result_column_id >= 0 &&

                    result_column_id < static_cast<int>(file_block->columns()));

        const auto& delete_filter = assert_cast<const ColumnUInt8&>(

                                            *file_block->get_by_position(result_column_id).column)

                                            .get_data();

        DORIS_CHECK(delete_filter.size() == static_cast<size_t>(batch_rows));

        IColumn::Filter keep_filter(static_cast<size_t>(batch_rows), 1);

        bool has_kept_row = false;

        for (size_t row = 0; row < static_cast<size_t>(batch_rows); ++row) {

            keep_filter[row] = !delete_filter[row];

            has_kept_row |= keep_filter[row] != 0;

        }

        file_block->erase(result_column_id);

        *selected_rows =

                !has_kept_row ? 0

                              : apply_filter_to_selection(keep_filter, selection, *selected_rows);

    }

    return Status::OK();

}

} // namespace

uint16_t apply_compact_filter_to_selection(const IColumn::Filter& filter,

                                           SelectionVector* selection, uint16_t selected_rows) {

    DORIS_CHECK(selection != nullptr);

    DORIS_CHECK(filter.size() == selected_rows);

    uint16_t new_selected_rows = 0;

    for (uint16_t selection_idx = 0; selection_idx < selected_rows; ++selection_idx) {

        if (filter[selection_idx] != 0) {

            selection->set_index(new_selected_rows++, static_cast<SelectionVector::Index>(

                                                              selection->get_index(selection_idx)));

        }

    }

    return new_selected_rows;

}

IColumn::Filter selection_to_filter(const SelectionVector& selection, uint16_t selected_rows,

                                    int64_t batch_rows) {

    IColumn::Filter filter(static_cast<size_t>(batch_rows), 0);

    for (uint16_t selection_idx = 0; selection_idx < selected_rows; ++selection_idx) {

        filter[selection.get_index(selection_idx)] = 1;

    }

    return filter;

}

Status execute_batch_filters(const format::FileScanRequest& request, int64_t batch_rows,

                             Block* file_block, SelectionVector* selection, uint16_t* selected_rows,

                             int64_t* conjunct_filtered_rows) {

    if (request.conjuncts.empty() && request.delete_conjuncts.empty()) {

        return Status::OK();

    }

    const auto selected_rows_before_conjunct = *selected_rows;

    RETURN_IF_ERROR(

            execute_filter_conjuncts(request, batch_rows, file_block, selection, selected_rows));

    if (conjunct_filtered_rows != nullptr) {

        *conjunct_filtered_rows += static_cast<int64_t>(selected_rows_before_conjunct) -

                                   static_cast<int64_t>(*selected_rows);

    }

    if (*selected_rows == 0) {

        return Status::OK();

    }

    return execute_delete_conjuncts(request, batch_rows, file_block, selection, selected_rows);

}

namespace {

int64_t count_range_rows(const std::vector<RowRange>& ranges) {

    int64_t rows = 0;

    for (const auto& range : ranges) {

        rows += range.length;

    }

    return rows;

}

void append_intersection(const RowRange& left, const RowRange& right,

                         std::vector<RowRange>* result) {

    const int64_t start = std::max(left.start, right.start);

    const int64_t end = std::min(left.start + left.length, right.start + right.length);

    if (start < end) {

        result->push_back(RowRange {.start = start, .length = end - start});

    }

}

std::vector<RowRange> filter_ranges_by_condition_cache(const std::vector<RowRange>& ranges,

                                                       const std::vector<bool>& cache,

                                                       int64_t row_group_first_row,

                                                       int64_t base_granule) {

    std::vector<RowRange> result;

    if (cache.empty()) {

        return ranges;

    }

    // Cache coordinates are file-global granules; RowRange coordinates are row-group-relative.

    // Walk every selected range in order and split it by granule. Granules covered by the bitmap

    // are kept only when the bit is true. Granules outside the bitmap are kept conservatively, so

    // an undersized or old-format cache entry cannot skip valid rows.

    for (const auto& range : ranges) {

        const int64_t global_start = row_group_first_row + range.start;

        const int64_t global_end = global_start + range.length;

        for (int64_t granule = global_start / ConditionCacheContext::GRANULE_SIZE;

             granule <= (global_end - 1) / ConditionCacheContext::GRANULE_SIZE; ++granule) {

            const int64_t cache_idx = granule - base_granule;

            const bool keep = cache_idx < 0 || static_cast<size_t>(cache_idx) >= cache.size() ||

                              cache[static_cast<size_t>(cache_idx)];

            if (!keep) {

                continue;

            }

            const int64_t granule_start = granule * ConditionCacheContext::GRANULE_SIZE;

            const int64_t granule_end = granule_start + ConditionCacheContext::GRANULE_SIZE;

            const RowRange file_granule_range {.start = granule_start - row_group_first_row,

                                               .length = granule_end - granule_start};

            append_intersection(range, file_granule_range, &result);

        }

    }

    return result;

}

} // namespace

void ParquetScanScheduler::set_plan(RowGroupScanPlan plan) {

    _row_group_plans = std::move(plan.row_groups);

    _condition_cache_filtered_rows = 0;

    _predicate_filtered_rows = 0;

    reset();

}

void ParquetScanScheduler::set_condition_cache_context(std::shared_ptr<ConditionCacheContext> ctx) {

    _condition_cache_ctx = std::move(ctx);

    if (!_condition_cache_ctx || !_condition_cache_ctx->filter_result || _row_group_plans.empty()) {

        return;

    }

    if (!_condition_cache_ctx->is_hit) {

        _condition_cache_ctx->base_granule =

                _row_group_plans.front().first_file_row / ConditionCacheContext::GRANULE_SIZE;

        const auto& last_plan = _row_group_plans.back();

        const int64_t end_granule = (last_plan.first_file_row + last_plan.row_group_rows +

                                     ConditionCacheContext::GRANULE_SIZE - 1) /

                                    ConditionCacheContext::GRANULE_SIZE;

        DORIS_CHECK(end_granule > _condition_cache_ctx->base_granule);

        _condition_cache_ctx->num_granules =

                std::min(_condition_cache_ctx->filter_result->size(),

                         static_cast<size_t>(end_granule - _condition_cache_ctx->base_granule));

        return;

    }

    std::vector<RowGroupReadPlan> filtered_plans;

    filtered_plans.reserve(_row_group_plans.size());

    for (auto& plan : _row_group_plans) {

        const int64_t old_rows = count_range_rows(plan.selected_ranges);

        plan.selected_ranges = filter_ranges_by_condition_cache(

                plan.selected_ranges, *_condition_cache_ctx->filter_result, plan.first_file_row,

                _condition_cache_ctx->base_granule);

        const int64_t new_rows = count_range_rows(plan.selected_ranges);

        _condition_cache_filtered_rows += old_rows - new_rows;

        if (!plan.selected_ranges.empty()) {

            filtered_plans.push_back(std::move(plan));

        }

    }

    _row_group_plans = std::move(filtered_plans);

    reset();

}

void ParquetScanScheduler::reset() {

    _next_row_group_plan_idx = 0;

    _raw_rows_read = 0;

    reset_current_row_group();

}

void ParquetScanScheduler::reset_current_row_group() {

    _current_row_group.reset();

    _current_predicate_columns.clear();

    _current_non_predicate_columns.clear();

    _current_dictionary_filters.clear();

    _current_dictionary_residual_conjuncts.clear();

    _current_row_group_rows = 0;

    _current_row_group_id = -1;

    _current_row_group_rows_read = 0;

    _current_row_group_first_row = 0;

    _current_selected_ranges.clear();

    _current_range_idx = 0;

    _current_range_rows_read = 0;

    // Readers are row-group scoped. If every remaining row was filtered, no future output can

    // observe the non-predicate readers' position, so dropping them together with their pending lag

    // avoids a useless end-of-row-group SkipRecords call. Example: predicate readers advance from 0

    // to 10,000 while lazy readers stay at 0; clearing both readers here is sufficient because the

    // next row group constructs a new set starting at its own row 0.

    _pending_non_predicate_skip_rows = 0;

    _current_predicate_prefetched = false;

    _current_non_predicate_prefetched = false;

    _current_merge_range_active = false;

}

Status ParquetScanScheduler::open_next_row_group(

        ParquetFileContext& file_context,

        const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema,

        const format::FileScanRequest& request, bool* has_row_group) {

    *has_row_group = false;

    if (_next_row_group_plan_idx >= _row_group_plans.size()) {

        return Status::OK();

    }

    const RowGroupReadPlan& row_group_plan = _row_group_plans[_next_row_group_plan_idx++];

    const int row_group_idx = row_group_plan.row_group_id;

    // Row-level dictionary filters read dictionary pages before Arrow RecordReaders are created.

    // Keep that probe on the base reader: MergeRangeFileReader expects each registered range to be

    // consumed as one forward pass, while the later RecordReader opens the same column chunk again

    // for the data-page stream.

    file_context.reset_random_access_ranges();

    _current_merge_range_active = false;

    try {

        _current_row_group = file_context.file_reader->RowGroup(row_group_idx);

    } catch (const ::parquet::ParquetException& e) {

        return Status::Corruption("Failed to open parquet row group {}: {}", row_group_idx,

                                  e.what());

    } catch (const std::exception& e) {

        return Status::InternalError("Failed to open parquet row group {}: {}", row_group_idx,

                                     e.what());

    }

    auto row_group_metadata = file_context.metadata->RowGroup(row_group_idx);

    DORIS_CHECK(row_group_metadata != nullptr);

    _current_row_group_rows = row_group_metadata->num_rows();

    DORIS_CHECK(_current_row_group_rows == row_group_plan.row_group_rows);

    DORIS_CHECK(_current_row_group_rows > 0);

    _current_row_group_id = row_group_idx;

    DORIS_CHECK(!row_group_plan.selected_ranges.empty());

    _current_row_group_first_row = row_group_plan.first_file_row;

    _current_row_group_rows_read = 0;

    _current_selected_ranges = row_group_plan.selected_ranges;

    _current_range_idx = 0;

    _current_range_rows_read = 0;

    _current_predicate_columns.clear();

    _current_non_predicate_columns.clear();

    _current_dictionary_filters.clear();

    RETURN_IF_ERROR(prepare_current_dictionary_filters(file_context, file_schema, request,

                                                       row_group_idx, *row_group_metadata));

    _current_merge_range_active =

            prepare_current_row_group_reader(file_context, file_schema, request, row_group_idx);

    ParquetColumnReaderFactory column_reader_factory(

            _current_row_group, file_context.schema->num_columns(), &row_group_plan.page_skip_plans,

            _page_skip_profile, _timezone, _enable_strict_mode,

            _scan_profile.column_reader_profile);

    for (const auto& col : request.predicate_columns) {

        const auto local_id = col.local_id();

        if (local_id == format::ROW_POSITION_COLUMN_ID) {

            _current_predicate_columns[local_id] =

                    column_reader_factory.create_row_position_column_reader(

                            _current_row_group_first_row);

            continue;

        }

        if (local_id == format::GLOBAL_ROWID_COLUMN_ID) {

            DORIS_CHECK(_global_rowid_context.has_value());

            _current_predicate_columns[local_id] =

                    column_reader_factory.create_global_rowid_column_reader(

                            *_global_rowid_context, _current_row_group_first_row);

            continue;

        }

        DORIS_CHECK(local_id >= 0 && local_id < static_cast<int32_t>(file_schema.size()));

        const auto& column_schema = file_schema[local_id];

        DORIS_CHECK(column_schema != nullptr);

        std::unique_ptr<ParquetColumnReader> column_reader;

        RETURN_IF_ERROR(

                column_reader_factory.create(*column_schema, &col, &column_reader,

                                             _current_dictionary_filters.contains(local_id)));

        _current_predicate_columns[local_id] = std::move(column_reader);

    }

    // Start warming filter-column chunks as soon as their row group is selected. Parquet v2 still

    // reads through Arrow's random-access reader; this prefetch only warms Doris file cache blocks

    // in the background and never changes the row/column materialization order.

    if (!_current_merge_range_active) {

        prefetch_current_row_group_columns(file_context, file_schema, request.predicate_columns,

                                           &_current_predicate_prefetched);

    }

    for (const auto& col : request.non_predicate_columns) {

        const auto local_id = col.local_id();

        if (local_id == format::ROW_POSITION_COLUMN_ID) {

            _current_non_predicate_columns[local_id] =

                    column_reader_factory.create_row_position_column_reader(

                            _current_row_group_first_row);

            continue;

        }

        if (local_id == format::GLOBAL_ROWID_COLUMN_ID) {

            DORIS_CHECK(_global_rowid_context.has_value());

            _current_non_predicate_columns[local_id] =

                    column_reader_factory.create_global_rowid_column_reader(

                            *_global_rowid_context, _current_row_group_first_row);

            continue;

        }

        DORIS_CHECK(local_id >= 0 && local_id < static_cast<int32_t>(file_schema.size()));

        const auto& column_schema = file_schema[local_id];

        DORIS_CHECK(column_schema != nullptr);

        std::unique_ptr<ParquetColumnReader> column_reader;

        RETURN_IF_ERROR(column_reader_factory.create(*column_schema, &col, &column_reader));

        _current_non_predicate_columns[local_id] = std::move(column_reader);

    }

    if (!_current_merge_range_active && request.conjuncts.empty() &&

        request.delete_conjuncts.empty()) {

        // With no row-level filters there is no lazy-read decision to wait for, so start warming

        // output chunks immediately after their readers are created. Filtered scans still defer

        // this until at least one row survives the predicate phase.

        prefetch_current_row_group_columns(file_context, file_schema, request.non_predicate_columns,

                                           &_current_non_predicate_prefetched);

    }

    *has_row_group = true;

    return Status::OK();

}

Status ParquetScanScheduler::skip_current_row_group_rows(int64_t rows) {

    DORIS_CHECK(rows >= 0);

    if (rows == 0) {

        return Status::OK();

    }

    if (_scan_profile.range_gap_skipped_rows != nullptr) {

        COUNTER_UPDATE(_scan_profile.range_gap_skipped_rows, rows);

    }

    for (const auto& column_reader : _current_predicate_columns | std::views::values) {

        RETURN_IF_ERROR(column_reader->skip(rows));

    }

    // Keep page-index/condition-cache gaps pending for lazy columns as well. For example, after a

    // fully filtered [0, 32) batch and a pruned [32, 96) gap, predicate readers are at 96 while lazy

    // readers remain at 0; one later skip(96) is cheaper than skip(32) followed by skip(64).

    DORIS_CHECK(_pending_non_predicate_skip_rows <= std::numeric_limits<int64_t>::max() - rows);

    _pending_non_predicate_skip_rows += rows;

    _current_row_group_rows_read += rows;

    return Status::OK();

}

Status ParquetScanScheduler::flush_pending_non_predicate_skip_rows() {

    if (_pending_non_predicate_skip_rows == 0) {

        return Status::OK();

    }

    for (const auto& column_reader : _current_non_predicate_columns | std::views::values) {

        RETURN_IF_ERROR(column_reader->skip(_pending_non_predicate_skip_rows));

    }

    _pending_non_predicate_skip_rows = 0;

    return Status::OK();

}

namespace {

struct PredicateConjunctSchedule {

    std::map<size_t, VExprContextSPtrs> single_column_conjuncts;

    VExprContextSPtrs remaining_conjuncts;

};

PredicateConjunctSchedule build_predicate_conjunct_schedule(

        const format::FileScanRequest& request) {

    std::unordered_set<size_t> predicate_block_positions;

    predicate_block_positions.reserve(request.predicate_columns.size());

    for (const auto& col : request.predicate_columns) {

        const auto position_it = request.local_positions.find(col.column_id());

        DORIS_CHECK(position_it != request.local_positions.end());

        predicate_block_positions.insert(position_it->second.value());

    }

    PredicateConjunctSchedule schedule;

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

        DORIS_CHECK(conjunct != nullptr);

        DORIS_CHECK(conjunct->root() != nullptr);

        if (!conjunct->root()->is_safe_to_execute_on_selected_rows()) {

            // Round-by-round filtering can compact later predicate columns before evaluating

            // remaining expressions. Stateful functions such as random(1) and error-preserving

            // functions such as assert_true() must see the same full batch they saw before this

            // optimization, so any unsafe conjunct disables the per-column schedule for the batch.

            schedule.remaining_conjuncts = request.conjuncts;

            schedule.single_column_conjuncts.clear();

            return schedule;

        }

        std::set<int> referenced_positions;

        conjunct->root()->collect_slot_column_ids(referenced_positions);

        if (referenced_positions.size() != 1) {

            schedule.remaining_conjuncts.push_back(conjunct);

            continue;

        }

        const auto block_position = static_cast<size_t>(*referenced_positions.begin());

        if (!predicate_block_positions.contains(block_position)) {

            schedule.remaining_conjuncts.push_back(conjunct);

            continue;

        }

        schedule.single_column_conjuncts[block_position].push_back(conjunct);

    }

    return schedule;

}

bool can_evaluate_all_with_dictionary(const VExprContextSPtrs& conjuncts) {

    if (conjuncts.empty()) {

        return false;

    }

    return std::ranges::all_of(conjuncts, [](const auto& conjunct) {

        return conjunct != nullptr && conjunct->root() != nullptr &&

               conjunct->root()->can_evaluate_dictionary_filter();

    });

}

bool can_evaluate_dictionary_exactly(const VExprSPtr& expr) {

    DORIS_CHECK(expr != nullptr);

    const auto* compound_pred = dynamic_cast<const VCompoundPred*>(expr.get());

    if (compound_pred == nullptr) {

        return expr->can_evaluate_dictionary_filter();

    }

    if (compound_pred->op() != TExprOpcode::COMPOUND_AND &&

        compound_pred->op() != TExprOpcode::COMPOUND_OR) {

        return false;

    }

    return !expr->children().empty() &&

           std::ranges::all_of(expr->children(), [](const auto& child) {

               return can_evaluate_dictionary_exactly(child);

           });

}

void collect_dictionary_residual_exprs(const VExprContextSPtr& owner_context, const VExprSPtr& expr,

                                       DictionaryResidualConjuncts* residual_conjuncts) {

    DORIS_CHECK(owner_context != nullptr);

    DORIS_CHECK(expr != nullptr);

    DORIS_CHECK(residual_conjuncts != nullptr);

    if (can_evaluate_dictionary_exactly(expr)) {

        return;

    }

    // VCompoundPred dictionary evaluation is a conservative prefilter for AND when only some

    // children are dictionary-aware. Split AND so exact dictionary children are not executed again

    // on materialized rows. Do not split a non-exact OR: its branches cannot be evaluated

    // independently after a dictionary prefilter without changing the original boolean semantics.

    const auto* compound_pred = dynamic_cast<const VCompoundPred*>(expr.get());

    if (compound_pred != nullptr && compound_pred->op() == TExprOpcode::COMPOUND_AND) {

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

            collect_dictionary_residual_exprs(owner_context, child, residual_conjuncts);

        }

        return;

    }

    residual_conjuncts->emplace_back(owner_context, expr);

}

DictionaryResidualConjuncts build_dictionary_residual_conjuncts(

        const VExprContextSPtrs& conjuncts) {

    DictionaryResidualConjuncts residual_conjuncts;

    for (const auto& conjunct : conjuncts) {

        DORIS_CHECK(conjunct != nullptr);

        collect_dictionary_residual_exprs(conjunct, conjunct->root(), &residual_conjuncts);

    }

    return residual_conjuncts;

}

uint16_t count_selected_rows(const IColumn::Filter& filter) {

    uint16_t selected_rows = 0;

    for (const auto value : filter) {

        selected_rows += value != 0;

    }

    return selected_rows;

}

Status filter_read_predicate_columns(Block* file_block, const std::vector<uint32_t>& positions,

                                     const IColumn::Filter& compact_filter) {

    if (positions.empty()) {

        return Status::OK();

    }

    RETURN_IF_CATCH_EXCEPTION(Block::filter_block_internal(file_block, positions, compact_filter));

    return Status::OK();

}

IColumn::Filter build_dictionary_entry_filter(size_t block_position,

                                              const ParquetColumnSchema& column_schema,

                                              const VExprContextSPtrs& conjuncts,

                                              const ParquetDictionaryWords& dict_words) {

    auto fields = dictionary_fields_from_words(dict_words);

    IColumn::Filter dictionary_filter(fields.size(), 1);

    DictionaryEvalContext ctx;

    auto& slot = ctx.slots

                         .emplace(static_cast<int>(block_position),

                                  DictionaryEvalContext::SlotDictionary {

                                          .data_type = column_schema.type, .values = {}})

                         .first->second;

    slot.values.reserve(1);

    for (size_t dict_idx = 0; dict_idx < fields.size(); ++dict_idx) {

        slot.values.clear();

        slot.values.push_back(fields[dict_idx]);

        dictionary_filter[dict_idx] = VExprContext::evaluate_dictionary_filter(conjuncts, ctx) ==

                                                      ZoneMapFilterResult::kNoMatch

                                              ? 0

                                              : 1;

    }

    return dictionary_filter;

}

} // namespace

Status ParquetScanScheduler::prepare_current_dictionary_filters(

        ParquetFileContext& file_context,

        const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema,

        const format::FileScanRequest& request, int row_group_idx,

        const ::parquet::RowGroupMetaData& row_group_metadata) {

    _current_dictionary_filters.clear();

    _current_dictionary_residual_conjuncts.clear();

    if (request.conjuncts.empty()) {

        return Status::OK();

    }

    PredicateConjunctSchedule schedule;

    {

        SCOPED_TIMER(_scan_profile.dict_filter_expr_rewrite_time);

        schedule = build_predicate_conjunct_schedule(request);

    }

    if (schedule.single_column_conjuncts.empty()) {

        return Status::OK();

    }

    SCOPED_TIMER(_scan_profile.dict_filter_rewrite_time);

    for (const auto& col : request.predicate_columns) {

        const auto local_id = col.local_id();

        if (local_id < 0 || local_id >= static_cast<int32_t>(file_schema.size())) {

            continue;

        }

        const auto position_it = request.local_positions.find(col.column_id());

        DORIS_CHECK(position_it != request.local_positions.end());

        const auto block_position = static_cast<size_t>(position_it->second.value());

        const auto conjunct_it = schedule.single_column_conjuncts.find(block_position);

        if (conjunct_it == schedule.single_column_conjuncts.end() ||

            !can_evaluate_all_with_dictionary(conjunct_it->second)) {

            continue;

        }

        update_counter_if_not_null(_scan_profile.dict_filter_candidate_columns, 1);

        // This optimization is deliberately limited to single-column predicates with a dictionary

        // evaluable part. Mixed AND predicates are split so dictionary-covered children run as a

        // dict-id prefilter and residual children keep the normal row-level expression path.

        const auto& column_schema = file_schema[local_id];

        DORIS_CHECK(column_schema != nullptr);

        if (column_schema->leaf_column_id < 0 ||

            column_schema->leaf_column_id >= row_group_metadata.num_columns()) {

            update_counter_if_not_null(_scan_profile.dict_filter_unsupported_columns, 1);

            continue;

        }

        auto column_chunk = row_group_metadata.ColumnChunk(column_schema->leaf_column_id);

        if (column_chunk == nullptr ||