be/src/format_v2/parquet/parquet_statistics.cpp

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

#include <algorithm>
#include <cmath>
#include <cstddef>
#include <cstring>
#include <exception>
#include <limits>
#include <map>
#include <memory>
#include <optional>
#include <ranges>
#include <set>
#include <string>
#include <type_traits>
#include <utility>
#include <vector>

#include "common/cast_set.h"
#include "common/config.h"
#include "core/data_type/data_type.h"
#include "core/data_type/data_type_nullable.h"
#include "core/data_type_serde/data_type_serde.h"
#include "core/field.h"
#include "exprs/expr_zonemap_filter.h"
#include "exprs/vectorized_fn_call.h"
#include "exprs/vexpr_context.h"
#include "exprs/vliteral.h"
#include "exprs/vslot_ref.h"
#include "format_v2/parquet/parquet_column_schema.h"
#include "format_v2/parquet/parquet_file_context.h"
#include "format_v2/parquet/reader/native/block_split_bloom_filter.h"
#include "format_v2/parquet/reader/native_column_reader.h"
#include "format_v2/timestamp_statistics.h"
#include "runtime/runtime_profile.h"
#include "storage/index/bloom_filter/bloom_filter.h"
#include "storage/index/zone_map/zone_map_index.h"
#include "storage/index/zone_map/zonemap_eval_context.h"
#include "util/thrift_util.h"
#include "util/unaligned.h"

namespace doris::format::parquet {

namespace detail {

Status validate_native_bloom_filter_layout(int64_t offset, uint32_t header_size,
                                           int64_t payload_size, int64_t declared_length,
                                           size_t file_size) {
    if (offset < 0 || header_size == 0 || payload_size < segment_v2::BloomFilter::MINIMUM_BYTES ||
        payload_size > segment_v2::BloomFilter::MAXIMUM_BYTES || payload_size % 32 != 0) {
        return Status::Corruption(
                "Invalid Parquet Bloom filter layout: offset {}, header {}, payload {}", offset,
                header_size, payload_size);
    }
    const uint64_t unsigned_offset = static_cast<uint64_t>(offset);
    const uint64_t total_size = static_cast<uint64_t>(header_size) + payload_size;
    if (unsigned_offset > file_size || total_size > file_size - unsigned_offset) {
        return Status::Corruption("Parquet Bloom filter range exceeds file size {}", file_size);
    }
    if (declared_length >= 0) {
        const uint64_t unsigned_declared_length = static_cast<uint64_t>(declared_length);
        if (unsigned_declared_length < total_size ||
            unsigned_declared_length > file_size - unsigned_offset) {
            return Status::Corruption(
                    "Parquet Bloom filter requires {} bytes, metadata declares {}, file has {}",
                    total_size, declared_length, file_size - unsigned_offset);
        }
    }
    return Status::OK();
}

bool has_supported_type_defined_order(const tparquet::FileMetaData& metadata, int leaf_column_id) {
    return leaf_column_id >= 0 && metadata.__isset.column_orders &&
           leaf_column_id < static_cast<int>(metadata.column_orders.size()) &&
           metadata.column_orders[leaf_column_id].__isset.TYPE_ORDER;
}

tparquet::Statistics sanitize_native_footer_statistics(const ParquetTypeDescriptor& type_descriptor,
                                                       const tparquet::Statistics& statistics,
                                                       bool has_type_defined_order) {
    auto sanitized = statistics;
    if (!has_type_defined_order || !sanitized.__isset.min_value || !sanitized.__isset.max_value) {
        sanitized.__isset.min_value = false;
        sanitized.__isset.max_value = false;
        sanitized.min_value.clear();
        sanitized.max_value.clear();
    }
    const bool binary = type_descriptor.physical_type == tparquet::Type::BYTE_ARRAY ||
                        type_descriptor.physical_type == tparquet::Type::FIXED_LEN_BYTE_ARRAY;
    if (!sanitized.__isset.min || !sanitized.__isset.max ||
        (binary && sanitized.min != sanitized.max)) {
        sanitized.__isset.min = false;
        sanitized.__isset.max = false;
        sanitized.min.clear();
        sanitized.max.clear();
    }
    return sanitized;
}

bool can_use_native_footer_min_max(const ParquetTypeDescriptor& type_descriptor,
                                   const tparquet::Statistics& statistics,
                                   bool has_type_defined_order) {
    // Inexact bounds remain useful for pruning, but returning them as aggregate values changes the
    // query result. Missing exactness fields are legacy-compatible; only an explicit false rejects.
    if ((statistics.__isset.is_min_value_exact && !statistics.is_min_value_exact) ||
        (statistics.__isset.is_max_value_exact && !statistics.is_max_value_exact)) {
        return false;
    }
    const auto sanitized =
            sanitize_native_footer_statistics(type_descriptor, statistics, has_type_defined_order);
    return (sanitized.__isset.min_value && sanitized.__isset.max_value) ||
           (sanitized.__isset.min && sanitized.__isset.max);
}

} // namespace detail

namespace {

bool build_native_page_statistics(const tparquet::ColumnIndex& column_index,
                                  const ParquetColumnSchema& column_schema, size_t page_idx,
                                  int64_t page_rows, ParquetColumnStatistics* page_statistics,
                                  const cctz::time_zone* timezone);

enum class ParquetRowGroupPruneReason {
    NONE,         // cannot prune; must read
    STATISTICS,   // excluded by ZoneMap statistics
    DICTIONARY,   // excluded by dictionary
    BLOOM_FILTER, // excluded by bloom filter
};

Status read_native_bloom_filter(const tparquet::ColumnMetaData& metadata,
                                const io::FileReaderSPtr& file, io::IOContext* io_ctx,
                                std::unique_ptr<native::BlockSplitBloomFilter>* result) {
    if (result == nullptr || file == nullptr || !metadata.__isset.bloom_filter_offset) {
        return Status::NotSupported("Parquet Bloom filter is unavailable");
    }
    constexpr size_t MAX_BLOOM_HEADER_BYTES = 64;
    if (metadata.bloom_filter_offset < 0 ||
        (metadata.__isset.bloom_filter_length && metadata.bloom_filter_length <= 0)) {
        return Status::Corruption("Invalid Parquet Bloom filter offset or declared length");
    }
    const uint64_t bloom_offset = static_cast<uint64_t>(metadata.bloom_filter_offset);
    if (bloom_offset >= file->size()) {
        return Status::Corruption("Parquet Bloom filter offset exceeds file size {}", file->size());
    }
    const size_t available = file->size() - bloom_offset;
    const size_t declared_available =
            metadata.__isset.bloom_filter_length
                    ? std::min<size_t>(metadata.bloom_filter_length, available)
                    : available;
    const size_t header_read_size = std::min(declared_available, MAX_BLOOM_HEADER_BYTES);
    std::vector<uint8_t> header_buffer(header_read_size);
    size_t bytes_read = 0;
    RETURN_IF_ERROR(file->read_at(metadata.bloom_filter_offset,
                                  Slice(header_buffer.data(), header_buffer.size()), &bytes_read,
                                  io_ctx));
    tparquet::BloomFilterHeader header;
    uint32_t header_size = cast_set<uint32_t>(bytes_read);
    const auto deserialize_status =
            deserialize_thrift_msg(header_buffer.data(), &header_size, true, &header);
    if (!deserialize_status.ok()) {
        // Keep invalid on-disk metadata distinguishable from transient read failures in profiles.
        return Status::Corruption("Malformed Parquet Bloom filter header");
    }
    if (!header.algorithm.__isset.BLOCK || !header.compression.__isset.UNCOMPRESSED ||
        !header.hash.__isset.XXHASH || header.numBytes <= 0) {
        return Status::NotSupported("Unsupported Parquet Bloom filter encoding");
    }

    // Validate the complete split-block layout before allocating or adding footer-controlled
    // offsets; BloomFilter::init() otherwise receives a truncated or oversized backing buffer.
    RETURN_IF_ERROR(detail::validate_native_bloom_filter_layout(
            metadata.bloom_filter_offset, header_size, header.numBytes,
            metadata.__isset.bloom_filter_length ? metadata.bloom_filter_length : -1,
            file->size()));

    std::vector<uint8_t> data(cast_set<size_t>(header.numBytes));
    RETURN_IF_ERROR(file->read_at(static_cast<size_t>(metadata.bloom_filter_offset) + header_size,
                                  Slice(data.data(), data.size()), &bytes_read, io_ctx));
    if (bytes_read != data.size()) {
        return Status::Corruption("Truncated Parquet Bloom filter payload");
    }
    auto bloom_filter = std::make_unique<native::BlockSplitBloomFilter>();
    RETURN_IF_ERROR(bloom_filter->init(reinterpret_cast<const char*>(data.data()), data.size(),
                                       segment_v2::HashStrategyPB::XX_HASH_64));
    *result = std::move(bloom_filter);
    return Status::OK();
}

bool bloom_logical_type_supported(const ParquetColumnSchema& column_schema) {
    if (column_schema.type == nullptr) {
        return false;
    }
    switch (remove_nullable(column_schema.type)->get_primitive_type()) {
    case TYPE_BOOLEAN:
    case TYPE_INT:
    case TYPE_BIGINT:
    case TYPE_FLOAT:
    case TYPE_DOUBLE:
    case TYPE_STRING:
        return true;
    default:
        return false;
    }
}

DecodedTimeUnit decoded_time_unit(ParquetTimeUnit time_unit) {
    switch (time_unit) {
    case ParquetTimeUnit::MILLIS:
        return DecodedTimeUnit::MILLIS;
    case ParquetTimeUnit::MICROS:
        return DecodedTimeUnit::MICROS;
    case ParquetTimeUnit::NANOS:
        return DecodedTimeUnit::NANOS;
    default:
        return DecodedTimeUnit::UNKNOWN;
    }
}

Status read_decoded_field(const ParquetColumnSchema& column_schema, DecodedColumnView view,
                          Field* field, const cctz::time_zone* timezone) {
    DORIS_CHECK(column_schema.type != nullptr);
    DORIS_CHECK(field != nullptr);
    constexpr uint8_t not_null = 0;
    view.row_count = 1;
    view.null_map = &not_null;
    view.time_unit = decoded_time_unit(column_schema.type_descriptor.time_unit);
    view.logical_integer_bit_width = column_schema.type_descriptor.integer_bit_width;
    view.logical_integer_is_signed = !column_schema.type_descriptor.is_unsigned_integer;
    view.decimal_precision = column_schema.type_descriptor.decimal_precision;
    view.decimal_scale = column_schema.type_descriptor.decimal_scale;
    view.fixed_length = column_schema.type_descriptor.fixed_length;
    view.timestamp_is_adjusted_to_utc = column_schema.timestamp_is_adjusted_to_utc.value_or(
            column_schema.type_descriptor.timestamp_is_adjusted_to_utc);
    view.timezone = column_schema.timestamp_is_adjusted_to_utc.has_value() &&
                                    !*column_schema.timestamp_is_adjusted_to_utc
                            ? nullptr
                            : timezone;
    // Statistics are pruning proofs, not row materialization. A malformed non-NULL bound must
    // disable pruning instead of being converted to NULL under permissive scan semantics.
    view.enable_strict_mode = true;
    RETURN_IF_ERROR(column_schema.type->get_serde()->read_field_from_decoded_value(
            *column_schema.type, field, view));
    if (field->is_null()) {
        return Status::DataQualityError("Non-NULL Parquet statistic decoded as NULL");
    }
    return Status::OK();
}

template <typename NativeType>
bool set_decoded_field(const ParquetColumnSchema& column_schema, DecodedValueKind value_kind,
                       const NativeType& value, Field* field, const cctz::time_zone* timezone) {
    DecodedColumnView view;
    view.value_kind = value_kind;
    view.values = reinterpret_cast<const uint8_t*>(&value);
    return read_decoded_field(column_schema, view, field, timezone).ok();
}

int64_t floor_timestamp_seconds(int64_t value, ParquetTimeUnit time_unit) {
    int64_t units_per_second = 1;
    switch (time_unit) {
    case ParquetTimeUnit::MILLIS:
        units_per_second = 1000;
        break;
    case ParquetTimeUnit::MICROS:
        units_per_second = 1000000;
        break;
    case ParquetTimeUnit::NANOS:
        units_per_second = 1000000000;
        break;
    default:
        DORIS_CHECK(false);
    }
    return format::floor_epoch_seconds(value, units_per_second);
}

bool timestamp_min_max_is_safe(const ParquetColumnSchema& column_schema, int64_t min_value,
                               int64_t max_value, const cctz::time_zone* timezone) {
    if (min_value > max_value) {
        return false;
    }
    if (!column_schema.type_descriptor.is_timestamp ||
        !column_schema.type_descriptor.timestamp_is_adjusted_to_utc || timezone == nullptr ||
        remove_nullable(column_schema.type)->get_primitive_type() == TYPE_TIMESTAMPTZ) {
        // TIMESTAMPTZ keeps the original UTC ordering, so local civil-time rollback does not make
        // its converted min/max non-monotonic.
        return true;
    }
    return format::utc_timestamp_range_is_monotonic(
            floor_timestamp_seconds(min_value, column_schema.type_descriptor.time_unit),
            floor_timestamp_seconds(max_value, column_schema.type_descriptor.time_unit), *timezone);
}

template <typename NativeType>
bool valid_min_max(const NativeType& min_value, const NativeType& max_value) {
    if constexpr (std::is_floating_point_v<NativeType>) {
        // Parquet requires readers to ignore min/max statistics if either bound is NaN.
        if (std::isnan(min_value) || std::isnan(max_value)) {
            return false;
        }
    }
    return true;
}

bool decoded_min_max_is_ordered(const ParquetColumnStatistics& column_statistics) {
    return !(column_statistics.max_value < column_statistics.min_value);
}

bool set_decoded_binary_field(const ParquetColumnSchema& column_schema, DecodedValueKind value_kind,
                              const StringRef& value, Field* field,
                              const cctz::time_zone* timezone) {
    std::vector<StringRef> binary_values {value};
    DecodedColumnView view;
    view.value_kind = value_kind;
    view.binary_values = &binary_values;
    return read_decoded_field(column_schema, view, field, timezone).ok();
}

template <typename T>
T load_predicate_value(const char* data) {
    T value;
    memcpy(&value, data, sizeof(T));
    return value;
}

std::optional<int64_t> load_predicate_integral_value(const char* buf, size_t size) {
    switch (size) {
    case sizeof(int8_t):
        return static_cast<int64_t>(load_predicate_value<int8_t>(buf));
    case sizeof(int16_t):
        return static_cast<int64_t>(load_predicate_value<int16_t>(buf));
    case sizeof(int32_t):
        return static_cast<int64_t>(load_predicate_value<int32_t>(buf));
    case sizeof(int64_t):
        return load_predicate_value<int64_t>(buf);
    default:
        return std::nullopt;
    }
}

bool logical_integer_fits_physical_int32(const ParquetTypeDescriptor& type_descriptor,
                                         int64_t value) {
    const int bit_width =
            type_descriptor.integer_bit_width > 0 ? type_descriptor.integer_bit_width : 32;
    if (type_descriptor.is_unsigned_integer) {
        const uint64_t max_value = bit_width >= 32 ? std::numeric_limits<uint32_t>::max()
                                                   : ((uint64_t {1} << bit_width) - 1);
        return value >= 0 && static_cast<uint64_t>(value) <= max_value;
    }
    const int64_t min_value = bit_width >= 32 ? std::numeric_limits<int32_t>::min()
                                              : -(int64_t {1} << (bit_width - 1));
    const int64_t max_value = bit_width >= 32 ? std::numeric_limits<int32_t>::max()
                                              : ((int64_t {1} << (bit_width - 1)) - 1);
    return value >= min_value && value <= max_value;
}

std::optional<int32_t> convert_logical_integer_to_physical_int32(
        const ParquetTypeDescriptor& type_descriptor, int64_t value) {
    if (!logical_integer_fits_physical_int32(type_descriptor, value)) {
        return std::nullopt;
    }
    if (!type_descriptor.is_unsigned_integer) {
        return static_cast<int32_t>(value);
    }
    const auto unsigned_value = static_cast<uint32_t>(value);
    int32_t physical_value;
    memcpy(&physical_value, &unsigned_value, sizeof(physical_value));
    return physical_value;
}

class NativeParquetBloomFilterAdapter final : public segment_v2::BloomFilter {
public:
    NativeParquetBloomFilterAdapter(const ParquetColumnSchema& column_schema,
                                    const segment_v2::BloomFilter& bloom_filter)
            : _column_schema(column_schema), _bloom_filter(bloom_filter) {}

    void add_bytes(const char*, size_t) override { DORIS_CHECK(false); }

    bool test_bytes(const char* buf, size_t size) const override {
        if (buf == nullptr ||
            _column_schema.type_descriptor.physical_type != tparquet::Type::INT32) {
            return _bloom_filter.test_bytes(buf, size);
        }
        const auto logical_value = load_predicate_integral_value(buf, size);
        if (!logical_value.has_value()) {
            return true;
        }
        const auto physical_value = convert_logical_integer_to_physical_int32(
                _column_schema.type_descriptor, *logical_value);
        if (!physical_value.has_value()) {
            return false;
        }
        // Native file Bloom bytes are hashed from the Parquet physical carrier, not the wider
        // Doris logical literal used by VExpr (for example UINT32 is exposed as BIGINT).
        return _bloom_filter.test_bytes(reinterpret_cast<const char*>(&*physical_value),
                                        sizeof(*physical_value));
    }

    void set_has_null(bool has_null) override { DORIS_CHECK(!has_null); }
    bool has_null() const override { return false; }
    void add_hash(uint64_t) override { DORIS_CHECK(false); }
    bool test_hash(uint64_t hash) const override { return _bloom_filter.test_hash(hash); }

private:
    const ParquetColumnSchema& _column_schema;
    const segment_v2::BloomFilter& _bloom_filter;
};

bool bloom_filter_supported(const ParquetColumnSchema& column_schema) {
    if (!bloom_logical_type_supported(column_schema)) {
        return false;
    }
    switch (column_schema.type_descriptor.physical_type) {
    case tparquet::Type::BOOLEAN:
    case tparquet::Type::INT32:
    case tparquet::Type::INT64:
    case tparquet::Type::FLOAT:
    case tparquet::Type::DOUBLE:
    case tparquet::Type::BYTE_ARRAY:
        return true;
    case tparquet::Type::FIXED_LEN_BYTE_ARRAY:
        return column_schema.type_descriptor.is_string_like &&
               column_schema.type_descriptor.fixed_length > 0;
    default:
        return false;
    }
}

const ParquetColumnSchema* resolve_local_leaf_schema(
        const std::vector<std::unique_ptr<ParquetColumnSchema>>& schema,
        const format::LocalColumnId file_column_id) {
    if (!file_column_id.is_valid() || file_column_id.value() >= static_cast<int>(schema.size())) {
        return nullptr;
    }
    const ParquetColumnSchema* column_schema = schema[file_column_id.value()].get();
    if (column_schema == nullptr || column_schema->kind != ParquetColumnSchemaKind::PRIMITIVE ||
        column_schema->leaf_column_id < 0 || column_schema->max_repetition_level > 0) {
        return nullptr;
    }
    return column_schema;
}

const ParquetColumnSchema* resolve_bloom_filter_leaf_schema(
        const std::vector<std::unique_ptr<ParquetColumnSchema>>& schema,
        const format::LocalColumnId file_column_id, const expr_zonemap::BloomFilterProbe& probe) {
    if (probe.path.empty()) {
        return resolve_local_leaf_schema(schema, file_column_id);
    }
    if (!file_column_id.is_valid() || file_column_id.value() >= static_cast<int>(schema.size())) {
        return nullptr;
    }
    const ParquetColumnSchema* column_schema = schema[file_column_id.value()].get();
    // A nested predicate must bind to its exact localized primitive path. Falling back to a
    // sibling leaf's Bloom filter could turn absence in that sibling into an invalid row-group skip.
    for (const auto& path_element : probe.path) {
        if (column_schema == nullptr) {
            return nullptr;
        }
        if (path_element.kind == expr_zonemap::BloomFilterPathKind::STRUCT_FIELD) {
            if (column_schema->kind != ParquetColumnSchemaKind::STRUCT) {
                return nullptr;
            }
            const ParquetColumnSchema* field_schema = nullptr;
            if (!path_element.field_name.empty()) {
                auto field = std::ranges::find_if(column_schema->children, [&](const auto& child) {
                    return child != nullptr && child->name == path_element.field_name;
                });
                if (field != column_schema->children.end()) {
                    field_schema = field->get();
                }
            } else if (path_element.field_ordinal >= 0 &&
                       path_element.field_ordinal <
                               static_cast<int32_t>(column_schema->children.size())) {
                field_schema = column_schema->children[path_element.field_ordinal].get();
            }
            column_schema = field_schema;
        } else {
            if (column_schema->kind != ParquetColumnSchemaKind::LIST ||
                column_schema->children.size() != 1) {
                return nullptr;
            }
            column_schema = column_schema->children[0].get();
        }
    }
    if (column_schema == nullptr || column_schema->kind != ParquetColumnSchemaKind::PRIMITIVE ||
        column_schema->leaf_column_id < 0) {
        return nullptr;
    }
    return column_schema;
}

std::optional<format::LocalColumnId> file_column_id_by_block_position(
        const format::FileScanRequest& request, int block_position) {
    for (const auto& [file_column_id, local_index] : request.local_positions) {
        if (local_index.value() == block_position) {
            return file_column_id;
        }
    }
    return std::nullopt;
}

enum class VariantComparisonOp { EQ, NE, LT, LE, GT, GE };

struct VariantShreddedPredicate {
    int slot_index = -1;
    std::vector<std::string> path;
    DataTypePtr comparison_type;
    DataTypePtr literal_type;
    Field literal;
    VariantComparisonOp op = VariantComparisonOp::EQ;
};

std::string callable_name(const VExprSPtr& expr) {
    if (const auto function = std::dynamic_pointer_cast<VectorizedFnCall>(expr);
        function != nullptr) {
        return function->function_name();
    }
    return expr == nullptr ? std::string {} : expr->expr_name();
}

std::optional<VariantComparisonOp> variant_comparison_op(std::string_view name) {
    if (name == "eq") {
        return VariantComparisonOp::EQ;
    }
    if (name == "ne") {
        return VariantComparisonOp::NE;
    }
    if (name == "lt") {
        return VariantComparisonOp::LT;
    }
    if (name == "le") {
        return VariantComparisonOp::LE;
    }
    if (name == "gt") {
        return VariantComparisonOp::GT;
    }
    if (name == "ge") {
        return VariantComparisonOp::GE;
    }
    return std::nullopt;
}

VariantComparisonOp reverse_variant_comparison(VariantComparisonOp op) {
    switch (op) {
    case VariantComparisonOp::EQ:
    case VariantComparisonOp::NE:
        return op;
    case VariantComparisonOp::LT:
        return VariantComparisonOp::GT;
    case VariantComparisonOp::LE:
        return VariantComparisonOp::GE;
    case VariantComparisonOp::GT:
        return VariantComparisonOp::LT;
    case VariantComparisonOp::GE:
        return VariantComparisonOp::LE;
    }
    __builtin_unreachable();
}

std::optional<std::pair<Field, DataTypePtr>> variant_literal(const VExprSPtr& expr) {
    const auto literal = std::dynamic_pointer_cast<VLiteral>(expr);
    if (literal == nullptr || !literal->get_column_ptr() || literal->get_column_ptr()->empty()) {
        return std::nullopt;
    }
    Field value;
    literal->get_column_ptr()->get(0, value);
    if (value.is_null()) {
        return std::nullopt;
    }
    return std::make_pair(std::move(value), literal->get_data_type());
}

std::optional<VariantShreddedPredicate> extract_variant_shredded_predicate(
        const VExprContextSPtr& conjunct) {
    if (conjunct == nullptr || conjunct->root() == nullptr ||
        conjunct->root()->get_num_children() != 2) {
        return std::nullopt;
    }
    auto op = variant_comparison_op(callable_name(conjunct->root()));
    if (!op.has_value()) {
        return std::nullopt;
    }

    VExprSPtr value_expr;
    std::optional<std::pair<Field, DataTypePtr>> literal;
    if ((literal = variant_literal(conjunct->root()->get_child(1))).has_value()) {
        value_expr = conjunct->root()->get_child(0);
    } else if ((literal = variant_literal(conjunct->root()->get_child(0))).has_value()) {
        value_expr = conjunct->root()->get_child(1);
        op = reverse_variant_comparison(*op);
    } else {
        return std::nullopt;
    }

    const auto comparison_type = value_expr->data_type();
    while (value_expr->node_type() == TExprNodeType::CAST_EXPR &&
           value_expr->get_num_children() == 1) {
        if (!expr_zonemap::data_types_compatible(value_expr->data_type(), comparison_type)) {
            // Every removed cast must preserve the comparison domain. Otherwise bounds for the
            // raw typed leaf could skip rows whose value changes in an intermediate narrowing cast.
            return std::nullopt;
        }
        value_expr = value_expr->get_child(0);
    }

    std::vector<std::string> reverse_path;
    while (callable_name(value_expr) == "element_at" && value_expr->get_num_children() == 2) {
        const auto key = variant_literal(value_expr->get_child(1));
        if (!key.has_value() || key->first.get_type() != TYPE_STRING) {
            // Repeated array shredding has no single scalar page range, so only object keys are
            // eligible for this file-level optimization.
            return std::nullopt;
        }
        reverse_path.push_back(key->first.get<TYPE_STRING>());
        value_expr = value_expr->get_child(0);
    }
    const auto slot = std::dynamic_pointer_cast<VSlotRef>(value_expr);
    if (slot == nullptr || reverse_path.empty() || comparison_type == nullptr ||
        remove_nullable(slot->data_type())->get_primitive_type() != TYPE_VARIANT ||
        !expr_zonemap::data_types_compatible(comparison_type, literal->second)) {
        return std::nullopt;
    }
    std::ranges::reverse(reverse_path);
    return VariantShreddedPredicate {.slot_index = slot->column_id(),
                                     .path = std::move(reverse_path),
                                     .comparison_type = comparison_type,
                                     .literal_type = literal->second,
                                     .literal = std::move(literal->first),
                                     .op = *op};
}

VExprContextSPtrs metadata_pruning_conjuncts(const format::FileScanRequest& request) {
    const size_t safe_count =
            std::min(request.metadata_pruning_safe_conjunct_count, request.conjuncts.size());
    return VExprContextSPtrs(request.conjuncts.begin(), request.conjuncts.begin() + safe_count);
}

bool has_variant_shredded_filter(const format::FileScanRequest& request) {
    const auto conjuncts = metadata_pruning_conjuncts(request);
    return std::ranges::any_of(conjuncts, [](const auto& conjunct) {
        return extract_variant_shredded_predicate(conjunct).has_value();
    });
}

const ParquetColumnSchema* child_named(const ParquetColumnSchema& parent, std::string_view name) {
    const auto it = std::ranges::find_if(parent.children, [&](const auto& child) {
        return child != nullptr && child->name == name;
    });
    return it == parent.children.end() ? nullptr : it->get();
}

struct ResolvedVariantShredding {
    std::vector<const ParquetColumnSchema*> fallback_values;
    const ParquetColumnSchema* typed_value = nullptr;
};

bool metadata_cast_is_order_preserving(const DataTypePtr& source, const DataTypePtr& target) {
    if (expr_zonemap::data_types_compatible(source, target)) {
        return true;
    }
    const auto source_type = remove_nullable(source);
    const auto target_type = remove_nullable(target);
    const auto source_primitive = source_type->get_primitive_type();
    const auto target_primitive = target_type->get_primitive_type();
    // Metadata bounds may cross only exact widening domains. This mirrors the residual CAST while
    // excluding rounding, overflow, and narrowing cases that could reverse a pruning decision.
    if (source_primitive == TYPE_FLOAT && target_primitive == TYPE_DOUBLE) {
        return true;
    }
    if (is_int(source_primitive) && source_primitive != TYPE_LARGEINT &&
        is_decimalv3(target_primitive)) {
        const uint32_t required_integer_digits = source_primitive == TYPE_TINYINT    ? 3
                                                 : source_primitive == TYPE_SMALLINT ? 5
                                                 : source_primitive == TYPE_INT      ? 10
                                                                                     : 19;
        return target_type->get_precision() >= target_type->get_scale() &&
               target_type->get_precision() - target_type->get_scale() >= required_integer_digits;
    }
    if (is_decimalv3(source_primitive) && is_decimalv3(target_primitive)) {
        const uint32_t source_integer_digits =
                source_type->get_precision() - source_type->get_scale();
        const uint32_t target_integer_digits =
                target_type->get_precision() - target_type->get_scale();
        return target_integer_digits >= source_integer_digits &&
               target_type->get_scale() >= source_type->get_scale();
    }
    return false;
}

std::optional<Field> cast_metadata_field(const Field& value, const DataTypePtr& source,
                                         const DataTypePtr& target) {
    if (expr_zonemap::data_types_compatible(source, target)) {
        return value;
    }
    const auto source_type = remove_nullable(source);
    const auto target_type = remove_nullable(target);
    if (source_type->get_primitive_type() == TYPE_FLOAT &&
        target_type->get_primitive_type() == TYPE_DOUBLE) {
        return Field::create_field<TYPE_DOUBLE>(static_cast<double>(value.get<TYPE_FLOAT>()));
    }
    try {
        auto source_column = source_type->create_column();
        source_column->insert(value);
        DataTypeSerDe::FormatOptions options = DataTypeSerDe::get_default_format_options();
        options.converted_from_string = true;
        std::string text = source_type->to_string(*source_column, 0, options);
        StringRef input(text.data(), text.size());
        auto target_column = target_type->create_column();
        if (!target_type->get_serde()
                     ->from_string_strict_mode(input, *target_column, options)
                     .ok() ||
            target_column->size() != 1) {
            return std::nullopt;
        }
        Field result;
        target_column->get(0, result);
        return result;
    } catch (...) {
        return std::nullopt;
    }
}

std::optional<ParquetColumnStatistics> normalize_variant_statistics(
        const VariantShreddedPredicate& predicate, const ParquetColumnSchema& typed_value,
        const ParquetColumnStatistics& statistics) {
    if (!statistics.has_min_max ||
        expr_zonemap::data_types_compatible(typed_value.type, predicate.comparison_type)) {
        return statistics;
    }
    auto min_value =
            cast_metadata_field(statistics.min_value, typed_value.type, predicate.comparison_type);
    auto max_value =
            cast_metadata_field(statistics.max_value, typed_value.type, predicate.comparison_type);
    if (!min_value.has_value() || !max_value.has_value()) {
        return std::nullopt;
    }
    auto normalized = statistics;
    normalized.min_value = std::move(*min_value);
    normalized.max_value = std::move(*max_value);
    return normalized;
}

std::optional<ResolvedVariantShredding> resolve_variant_shredding(
        const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema,
        const format::FileScanRequest& request, const VariantShreddedPredicate& predicate) {
    const auto local_id = file_column_id_by_block_position(request, predicate.slot_index);
    if (!local_id.has_value() || local_id->value() < 0 ||
        local_id->value() >= static_cast<int>(file_schema.size())) {
        return std::nullopt;
    }
    const ParquetColumnSchema* wrapper = file_schema[local_id->value()].get();
    if (wrapper == nullptr || wrapper->kind != ParquetColumnSchemaKind::VARIANT) {
        return std::nullopt;
    }
    std::vector<const ParquetColumnSchema*> fallbacks;
    for (const auto& component : predicate.path) {
        const auto* typed_object = child_named(*wrapper, "typed_value");
        if (typed_object == nullptr || typed_object->kind != ParquetColumnSchemaKind::STRUCT) {
            return std::nullopt;
        }
        wrapper = child_named(*typed_object, component);
        if (wrapper == nullptr || wrapper->kind != ParquetColumnSchemaKind::STRUCT) {
            return std::nullopt;
        }
        const auto* fallback = child_named(*wrapper, "value");
        if (fallback == nullptr || fallback->kind != ParquetColumnSchemaKind::PRIMITIVE) {
            return std::nullopt;
        }
        // A residual at any wrapper can contribute to the descendant path, so deepest-leaf
        // bounds are safe only when every fallback encountered during descent is absent.
        fallbacks.push_back(fallback);
    }
    const auto* typed = child_named(*wrapper, "typed_value");
    const auto typed_primitive = typed == nullptr || typed->type == nullptr
                                         ? INVALID_TYPE
                                         : remove_nullable(typed->type)->get_primitive_type();
    if (fallbacks.empty() || typed == nullptr ||
        typed->kind != ParquetColumnSchemaKind::PRIMITIVE || typed->max_repetition_level != 0 ||
        // Parquet float statistics do not prove that a page contains no NaN. Min/max pruning in
        // the presence of NaN is not order preserving, so keep those pages until such proof exists.
        typed_primitive == TYPE_FLOAT || typed_primitive == TYPE_DOUBLE ||
        !metadata_cast_is_order_preserving(typed->type, predicate.comparison_type) ||
        !expr_zonemap::data_types_compatible(predicate.comparison_type, predicate.literal_type)) {
        return std::nullopt;
    }
    return ResolvedVariantShredding {.fallback_values = std::move(fallbacks), .typed_value = typed};
}

bool fallback_is_all_null(const tparquet::RowGroup& row_group,
                          const ParquetColumnSchema& fallback) {
    if (fallback.max_repetition_level != 0 || fallback.leaf_column_id < 0 ||
        fallback.leaf_column_id >= static_cast<int>(row_group.columns.size())) {
        return false;
    }
    const auto& chunk = row_group.columns[fallback.leaf_column_id];
    return row_group.num_rows >= 0 && chunk.__isset.meta_data &&
           chunk.meta_data.num_values == row_group.num_rows && chunk.meta_data.__isset.statistics &&
           chunk.meta_data.statistics.__isset.null_count &&
           chunk.meta_data.statistics.null_count == chunk.meta_data.num_values;
}

bool fallbacks_are_all_null(const tparquet::RowGroup& row_group,
                            const std::vector<const ParquetColumnSchema*>& fallbacks) {
    return std::ranges::all_of(fallbacks, [&](const auto* fallback) {
        return fallback != nullptr && fallback_is_all_null(row_group, *fallback);
    });
}

bool variant_statistics_exclude(const VariantShreddedPredicate& predicate,
                                const ParquetColumnStatistics& statistics) {
    if (!statistics.has_any_statistics()) {
        return false;
    }
    if (!statistics.has_not_null) {
        return true;
    }
    if (!statistics.has_min_max) {
        return false;
    }
    const auto& literal = predicate.literal;
    switch (predicate.op) {
    case VariantComparisonOp::EQ:
        return literal < statistics.min_value || statistics.max_value < literal;
    case VariantComparisonOp::NE:
        return statistics.min_value == literal && statistics.max_value == literal;
    case VariantComparisonOp::LT:
        return statistics.min_value >= literal;
    case VariantComparisonOp::LE:
        return statistics.min_value > literal;
    case VariantComparisonOp::GT:
        return statistics.max_value <= literal;
    case VariantComparisonOp::GE:
        return statistics.max_value < literal;
    }
    __builtin_unreachable();
}

bool has_expr_zonemap_filter(const format::FileScanRequest& request, const RuntimeState*) {
    // FileScannerV2 metadata pruning is a fixed part of its scan pipeline and must not inherit
    // the legacy scanner's expression ZoneMap session gate.
    for (const auto& conjunct : metadata_pruning_conjuncts(request)) {
        if (conjunct != nullptr && conjunct->root() != nullptr &&
            conjunct->root()->can_evaluate_zonemap_filter()) {
            return true;
        }
    }
    return has_variant_shredded_filter(request);
}

std::set<int> collect_expr_zonemap_slot_indexes(const VExprContextSPtrs& conjuncts) {
    std::set<int> slot_indexes;
    for (const auto& conjunct : conjuncts) {
        if (conjunct != nullptr && conjunct->root() != nullptr &&
            conjunct->root()->can_evaluate_zonemap_filter()) {
            conjunct->root()->collect_slot_column_ids(slot_indexes);
        }
    }
    return slot_indexes;
}

template <typename SlotIndexSelector>
std::map<int, VExprContextSPtrs> collect_conjuncts_by_single_slot(
        const VExprContextSPtrs& conjuncts, SlotIndexSelector slot_index_selector) {
    std::map<int, VExprContextSPtrs> conjuncts_by_slot;
    for (const auto& conjunct : conjuncts) {
        const auto slot_index = slot_index_selector(conjunct);
        if (slot_index >= 0) {
            conjuncts_by_slot[slot_index].push_back(conjunct);
        }
    }
    return conjuncts_by_slot;
}

std::shared_ptr<segment_v2::ZoneMap> make_zonemap_from_statistics(
        const ParquetColumnStatistics& statistics) {
    if (!statistics.has_null_count && !statistics.has_min_max) {
        return nullptr;
    }
    segment_v2::ZoneMap zone_map;
    zone_map.has_null = statistics.has_null;
    zone_map.has_not_null = statistics.has_not_null;
    if (!statistics.has_not_null) {
        return std::make_shared<segment_v2::ZoneMap>(std::move(zone_map));
    }
    if (!statistics.has_min_max) {
        // Null counts remain trustworthy when min/max decoding fails (for example, because a
        // floating-point bound is NaN). pass_all prevents range pruning without discarding the
        // has_null/has_not_null flags needed by IS NULL and IS NOT NULL predicates.
        zone_map.pass_all = true;
        return std::make_shared<segment_v2::ZoneMap>(std::move(zone_map));
    }
    zone_map.min_value = statistics.min_value;
    zone_map.max_value = statistics.max_value;
    return std::make_shared<segment_v2::ZoneMap>(std::move(zone_map));
}

void add_slot_zonemap(ZoneMapEvalContext* ctx, int slot_index, const DataTypePtr& data_type,
                      std::shared_ptr<segment_v2::ZoneMap> zone_map) {
    DORIS_CHECK(ctx != nullptr);
    ZoneMapEvalContext::SlotZoneMap slot_zone_map;
    slot_zone_map.data_type = data_type;
    slot_zone_map.zone_map = std::move(zone_map);
    const auto primitive_type = remove_nullable(data_type)->get_primitive_type();
    slot_zone_map.floating_nan_count_unknown =
            primitive_type == TYPE_FLOAT || primitive_type == TYPE_DOUBLE;
    ctx->slots.emplace(slot_index, std::move(slot_zone_map));
}

void accumulate_zonemap_stats(const ZoneMapEvalContext& ctx, ParquetPruningStats* pruning_stats) {
    if (pruning_stats == nullptr) {
        return;
    }
    pruning_stats->expr_zonemap_unusable_evals += ctx.stats.unusable_zonemap_eval_count;
    pruning_stats->in_zonemap_point_check_count += ctx.stats.in_zonemap_point_check_count;
    pruning_stats->in_zonemap_range_only_count += ctx.stats.in_zonemap_range_only_count;
}

} // namespace

bool can_use_parquet_page_index(const format::FileScanRequest& request,
                                const RuntimeState* runtime_state) {
    return config::enable_parquet_page_index && has_expr_zonemap_filter(request, runtime_state);
}

std::shared_ptr<segment_v2::ZoneMap> ParquetStatisticsUtils::MakeZoneMap(
        const ParquetColumnStatistics& statistics) {
    return make_zonemap_from_statistics(statistics);
}

ParquetColumnStatistics ParquetStatisticsUtils::TransformColumnStatistics(
        const ParquetColumnSchema& column_schema, const tparquet::Statistics* statistics,
        int64_t column_value_count, const cctz::time_zone* timezone) {
    ParquetColumnStatistics result;
    if (statistics == nullptr || column_value_count < 0) {
        return result;
    }

    if (statistics->__isset.null_count && statistics->null_count > column_value_count) {
        // An impossible null count makes all derived min/max and all-null flags untrustworthy;
        // disable pruning instead of turning corrupt footer metadata into false negatives.
        return result;
    }

    const bool has_null_count = statistics->__isset.null_count && statistics->null_count >= 0;
    const int64_t null_count = has_null_count ? statistics->null_count : 0;
    const bool has_not_null = has_null_count ? column_value_count > null_count : true;
    const std::string* min_value = statistics->__isset.min_value
                                           ? &statistics->min_value
                                           : (statistics->__isset.min ? &statistics->min : nullptr);
    const std::string* max_value = statistics->__isset.max_value
                                           ? &statistics->max_value
                                           : (statistics->__isset.max ? &statistics->max : nullptr);

    tparquet::ColumnIndex index;
    index.__set_null_pages({!has_not_null});
    index.__set_null_counts({null_count});
    if (min_value != nullptr && max_value != nullptr) {
        index.__set_min_values({*min_value});
        index.__set_max_values({*max_value});
    }
    // Footer statistics and page indexes share the same little-endian physical encoding. Reusing
    // one decoder keeps native row-group and page pruning identical for logical types and NaNs.
    if (!build_native_page_statistics(index, column_schema, 0, column_value_count, &result,
                                      timezone)) {
        return {};
    }
    if (!has_null_count) {
        result.has_null_count = false;
        result.has_null = true;
    }
    return result;
}

bool ParquetStatisticsUtils::NativeBloomFilterExcludes(
        const ParquetColumnSchema& column_schema, int slot_index,
        const VExprContextSPtrs& conjuncts, const segment_v2::BloomFilter& bloom_filter) {
    if (!bloom_filter_supported(column_schema)) {
        return false;
    }
    NativeParquetBloomFilterAdapter adapter(column_schema, bloom_filter);
    BloomFilterEvalContext ctx;
    ctx.slots.emplace(slot_index, BloomFilterEvalContext::SlotBloomFilter {
                                          .data_type = column_schema.type,
                                          .bloom_filter = &adapter,
                                  });
    return VExprContext::evaluate_bloom_filter(conjuncts, ctx) == ZoneMapFilterResult::kNoMatch;
}

namespace {

void collect_filtered_leaf_ids(const ParquetColumnSchema& column_schema,
                               const format::LocalColumnIndex* projection,
                               std::set<int>* leaf_column_ids) {
    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_schema : column_schema.children) {
        if (!format::is_child_projected(projection, child_schema->local_id)) {
            continue;
        }
        // The leaf set must match the physical projection. A complete Variant projection naturally
        // reaches every sibling; a validated typed-leaf projection reads only retained children.
        const auto* child_projection =
                format::find_child_projection(projection, child_schema->local_id);
        collect_filtered_leaf_ids(*child_schema, child_projection, leaf_column_ids);
    }
}

bool native_metadata_predicate_is_type_safe(const ParquetColumnSchema& column_schema) {
    DORIS_CHECK(column_schema.type != nullptr);
    // Raw VARBINARY file slots may feed table-side STRING casts. Footer/page metadata is still in
    // the pre-cast domain, so using it for a rewritten table predicate can cause false negatives.
    if (remove_nullable(column_schema.type)->get_primitive_type() == TYPE_VARBINARY) {
        return false;
    }
    // UUID readers render canonical text, so their physical 16-byte bounds are not STRING bounds.
    return !column_schema.type_descriptor.is_uuid;
}

bool variant_metadata_predicate_is_type_safe(const ParquetColumnSchema& column_schema) {
    if (!native_metadata_predicate_is_type_safe(column_schema)) {
        return false;
    }
    const auto& descriptor = column_schema.type_descriptor;
    // An ordinary raw-binary STRING slot preserves its bytes, but Variant reconstruction renders
    // the binary identity before the residual STRING cast and therefore changes the domain.
    return !descriptor.is_string_like || descriptor.is_string_annotation;
}

bool check_native_statistics(const tparquet::FileMetaData& metadata,
                             const tparquet::RowGroup& row_group,
                             const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema,
                             const format::FileScanRequest& request,
                             ParquetPruningStats* pruning_stats, const cctz::time_zone* timezone) {
    const auto conjuncts = metadata_pruning_conjuncts(request);
    const auto slot_indexes = collect_expr_zonemap_slot_indexes(conjuncts);
    if (slot_indexes.empty()) {
        return false;
    }
    ZoneMapEvalContext ctx;
    for (const int slot_index : slot_indexes) {
        const auto file_column_id = file_column_id_by_block_position(request, slot_index);
        if (!file_column_id.has_value()) {
            continue;
        }
        const auto* column_schema = resolve_local_leaf_schema(file_schema, *file_column_id);
        if (column_schema == nullptr || column_schema->type == nullptr ||
            !native_metadata_predicate_is_type_safe(*column_schema) ||
            column_schema->leaf_column_id >= static_cast<int>(row_group.columns.size())) {
            continue;
        }
        const auto& chunk = row_group.columns[column_schema->leaf_column_id];
        std::shared_ptr<segment_v2::ZoneMap> zone_map;
        if (chunk.__isset.meta_data) {
            const auto& column_metadata = chunk.meta_data;
            std::optional<tparquet::Statistics> safe_statistics;
            if (column_metadata.__isset.statistics) {
                safe_statistics = detail::sanitize_native_footer_statistics(
                        column_schema->type_descriptor, column_metadata.statistics,
                        detail::has_supported_type_defined_order(metadata,
                                                                 column_schema->leaf_column_id));
            }
            zone_map = ParquetStatisticsUtils::MakeZoneMap(
                    ParquetStatisticsUtils::TransformColumnStatistics(
                            *column_schema,
                            safe_statistics.has_value() ? &*safe_statistics : nullptr,
                            column_metadata.num_values, timezone));
        }
        add_slot_zonemap(&ctx, slot_index, column_schema->type, std::move(zone_map));
    }
    const auto result = VExprContext::evaluate_zonemap_filter(conjuncts, ctx);
    accumulate_zonemap_stats(ctx, pruning_stats);
    return result == ZoneMapFilterResult::kNoMatch;
}

bool check_shredded_variant_statistics(
        const tparquet::FileMetaData& metadata, const tparquet::RowGroup& row_group,
        const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema,
        const format::FileScanRequest& request, const cctz::time_zone* timezone) {
    for (const auto& conjunct : metadata_pruning_conjuncts(request)) {
        const auto predicate = extract_variant_shredded_predicate(conjunct);
        if (!predicate.has_value()) {
            continue;
        }
        const auto shredding = resolve_variant_shredding(file_schema, request, *predicate);
        if (!shredding.has_value() || shredding->typed_value->leaf_column_id < 0 ||
            shredding->typed_value->leaf_column_id >= static_cast<int>(row_group.columns.size()) ||
            !fallbacks_are_all_null(row_group, shredding->fallback_values) ||
            !variant_metadata_predicate_is_type_safe(*shredding->typed_value) ||
            !detail::has_supported_type_defined_order(metadata,
                                                      shredding->typed_value->leaf_column_id)) {
            continue;
        }
        const auto& chunk = row_group.columns[shredding->typed_value->leaf_column_id];
        if (!chunk.__isset.meta_data) {
            continue;
        }
        const auto& column_metadata = chunk.meta_data;
        if (column_metadata.num_values != row_group.num_rows) {
            continue;
        }
        std::optional<tparquet::Statistics> safe_statistics;
        if (column_metadata.__isset.statistics) {
            safe_statistics = detail::sanitize_native_footer_statistics(
                    shredding->typed_value->type_descriptor, column_metadata.statistics, true);
        }
        const auto statistics = ParquetStatisticsUtils::TransformColumnStatistics(
                *shredding->typed_value, safe_statistics.has_value() ? &*safe_statistics : nullptr,
                column_metadata.num_values, timezone);
        const auto normalized =
                normalize_variant_statistics(*predicate, *shredding->typed_value, statistics);
        if (normalized.has_value() && variant_statistics_exclude(*predicate, *normalized)) {
            return true;
        }
    }
    return false;
}

bool is_native_dictionary_data_encoding(tparquet::Encoding::type encoding) {
    return encoding == tparquet::Encoding::PLAIN_DICTIONARY ||
           encoding == tparquet::Encoding::RLE_DICTIONARY;
}

bool is_native_level_encoding(tparquet::Encoding::type encoding) {
    return encoding == tparquet::Encoding::RLE || encoding == tparquet::Encoding::BIT_PACKED;
}

bool is_native_dictionary_encoded_chunk(const tparquet::ColumnMetaData& metadata) {
    if (!metadata.__isset.dictionary_page_offset || metadata.dictionary_page_offset < 0) {
        return false;
    }
    if (metadata.__isset.encoding_stats && !metadata.encoding_stats.empty()) {
        bool has_dictionary_data_page = false;
        for (const auto& encoding_stat : metadata.encoding_stats) {
            if ((encoding_stat.page_type != tparquet::PageType::DATA_PAGE &&
                 encoding_stat.page_type != tparquet::PageType::DATA_PAGE_V2) ||
                encoding_stat.count <= 0) {
                continue;
            }
            if (!is_native_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 : metadata.encodings) {
        if (is_native_dictionary_data_encoding(encoding)) {
            has_dictionary_encoding = true;
        } else if (!is_native_level_encoding(encoding)) {
            return false;
        }
    }
    return has_dictionary_encoding;
}

const format::LocalColumnIndex* find_request_projection(const format::FileScanRequest& request,
                                                        format::LocalColumnId file_column_id) {
    for (const auto& projection : request.predicate_columns) {
        if (projection.local_id() == file_column_id.value()) {
            return &projection;
        }
    }
    for (const auto& projection : request.non_predicate_columns) {
        if (projection.local_id() == file_column_id.value()) {
            return &projection;
        }
    }
    return nullptr;
}

ParquetRowGroupPruneReason native_dictionary_prune_reason(
        const tparquet::RowGroup& row_group, int row_group_idx,
        const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema,
        const format::FileScanRequest& request, const cctz::time_zone* timezone,
        ParquetFileContext* file_context, const ParquetColumnReaderProfile& column_reader_profile) {
    if (file_context == nullptr || file_context->native_metadata == nullptr) {
        return ParquetRowGroupPruneReason::NONE;
    }
    const auto conjuncts_by_slot = collect_conjuncts_by_single_slot(
            metadata_pruning_conjuncts(request), expr_zonemap::single_slot_dictionary_index);
    for (const auto& [slot_index, conjuncts] : conjuncts_by_slot) {
        const auto file_column_id = file_column_id_by_block_position(request, slot_index);
        if (!file_column_id.has_value()) {
            continue;
        }
        const auto* column_schema = resolve_local_leaf_schema(file_schema, *file_column_id);
        const auto* projection = find_request_projection(request, *file_column_id);
        if (column_schema == nullptr || projection == nullptr || column_schema->type == nullptr ||
            !column_schema->type_descriptor.is_string_like ||
            column_schema->leaf_column_id >= static_cast<int>(row_group.columns.size())) {
            continue;
        }
        if (!native_metadata_predicate_is_type_safe(*column_schema)) {
            // The file-local VARBINARY may feed a table-side STRING cast. Pruning before that cast
            // can compare different Field kinds and incorrectly discard a matching row group.
            continue;
        }
        const auto& chunk = row_group.columns[column_schema->leaf_column_id];
        if (!chunk.__isset.meta_data ||
            (chunk.meta_data.type != tparquet::Type::BYTE_ARRAY &&
             chunk.meta_data.type != tparquet::Type::FIXED_LEN_BYTE_ARRAY) ||
            !is_native_dictionary_encoded_chunk(chunk.meta_data)) {
            continue;
        }
        std::unique_ptr<ParquetColumnReader> reader;
        const std::vector<RowRange> ranges {{0, row_group.num_rows}};
        const std::unordered_map<int, tparquet::OffsetIndex> offset_indexes;
        // Metadata pruning uses the real native reader, so its page work must be attributed to the
        // scan profile even when the row group is eliminated before execution readers are built.
        const auto status = NativeColumnReader::create(
                *column_schema, projection, file_context->native_file,
                file_context->native_metadata, row_group_idx, ranges, offset_indexes, timezone,
                std::nullopt, file_context->native_io_ctx, nullptr,
                file_context->native_page_cache_enabled, file_context->native_page_cache_file_key,
                true, column_reader_profile, &reader);
        if (!status.ok() || reader == nullptr) {
            continue;
        }
        auto dictionary_result = reader->dictionary_values();
        if (!dictionary_result.has_value()) {
            continue;
        }
        auto dictionary = std::move(dictionary_result).value();
        std::vector<Field> values(dictionary->size());
        for (size_t value_idx = 0; value_idx < dictionary->size(); ++value_idx) {
            dictionary->get(value_idx, values[value_idx]);
        }
        DictionaryEvalContext ctx;
        ctx.slots.emplace(slot_index, DictionaryEvalContext::SlotDictionary {
                                              .data_type = column_schema->type,
                                              .values = std::move(values),
                                      });
        if (VExprContext::evaluate_dictionary_filter(conjuncts, ctx) ==
            ZoneMapFilterResult::kNoMatch) {
            return ParquetRowGroupPruneReason::DICTIONARY;
        }
    }
    return ParquetRowGroupPruneReason::NONE;
}

ParquetRowGroupPruneReason native_bloom_filter_prune_reason(
        const tparquet::RowGroup& row_group,
        const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema,
        const format::FileScanRequest& request, ParquetFileContext* file_context,
        ParquetPruningStats* pruning_stats) {
    if (file_context == nullptr || file_context->native_file == nullptr) {
        return ParquetRowGroupPruneReason::NONE;
    }
    struct BloomProbeGroup {
        const ParquetColumnSchema* column_schema = nullptr;
        int slot_index = -1;
        VExprContextSPtrs conjuncts;
    };
    struct LeafBloomProbeGroup {
        int leaf_column_id = -1;
        std::vector<BloomProbeGroup> probes;
    };
    // The vector preserves first-probe order, while the map only deduplicates repeated leaves.
    // This avoids reading a potentially large later payload before an earlier probe can prune.
    std::vector<LeafBloomProbeGroup> probes_by_first_use;
    std::map<int, size_t> group_index_by_leaf;
    const auto add_probe = [&](const ParquetColumnSchema& column_schema, int slot_index,
                               VExprContextSPtrs conjuncts) {
        if (column_schema.type == nullptr ||
            !native_metadata_predicate_is_type_safe(column_schema) ||
            !bloom_filter_supported(column_schema) || column_schema.leaf_column_id < 0 ||
            column_schema.leaf_column_id >= static_cast<int>(row_group.columns.size())) {
            return;
        }
        const auto [group_it, inserted] = group_index_by_leaf.try_emplace(
                column_schema.leaf_column_id, probes_by_first_use.size());
        if (inserted) {
            probes_by_first_use.push_back(
                    {.leaf_column_id = column_schema.leaf_column_id, .probes = {}});
        }
        probes_by_first_use[group_it->second].probes.push_back({.column_schema = &column_schema,
                                                                .slot_index = slot_index,
                                                                .conjuncts = std::move(conjuncts)});
    };

    const auto pruning_conjuncts = metadata_pruning_conjuncts(request);
    // Resolve direct and nested probes in one conjunct-order pass. Deduplication happens only
    // after first use so a later Bloom payload cannot be read before an earlier pruning probe.
    for (const auto& conjunct : pruning_conjuncts) {
        if (conjunct == nullptr || conjunct->root() == nullptr ||
            !conjunct->root()->can_evaluate_bloom_filter()) {
            continue;
        }
        auto probe = expr_zonemap::extract_bloom_filter_predicate_probe(conjunct->root());
        if (!probe.has_value()) {
            continue;
        }
        const auto file_column_id = file_column_id_by_block_position(request, probe->slot_index);
        if (!file_column_id.has_value()) {
            continue;
        }
        const auto* column_schema =
                probe->path.empty()
                        ? resolve_local_leaf_schema(file_schema, *file_column_id)
                        : resolve_bloom_filter_leaf_schema(file_schema, *file_column_id, *probe);
        if (column_schema == nullptr ||
            !expr_zonemap::data_types_compatible(column_schema->type, probe->value_type)) {
            continue;
        }
        add_probe(*column_schema, probe->slot_index, {conjunct});
    }

    for (const auto& leaf_group : probes_by_first_use) {
        const int leaf_column_id = leaf_group.leaf_column_id;
        if (pruning_stats != nullptr) {
            ++pruning_stats->bloom_filter_probe_attempts;
        }
        const auto& chunk = row_group.columns[leaf_column_id];
        if (!chunk.__isset.meta_data) {
            if (pruning_stats != nullptr) {
                ++pruning_stats->bloom_filter_conservative_fallbacks;
            }
            continue;
        }
        std::unique_ptr<native::BlockSplitBloomFilter> bloom_filter;
        Status bloom_status;
        int64_t timer_sink = 0;
        {
            SCOPED_RAW_TIMER(pruning_stats == nullptr ? &timer_sink
                                                      : &pruning_stats->bloom_filter_read_time);
            bloom_status = read_native_bloom_filter(row_group.columns[leaf_column_id].meta_data,
                                                    file_context->native_file,
                                                    file_context->native_io_ctx, &bloom_filter);
            if (!bloom_status.ok()) {
                bloom_filter.reset();
            }
        }
        if (bloom_filter == nullptr) {
            if (pruning_stats != nullptr) {
                ++pruning_stats->bloom_filter_conservative_fallbacks;
                if (bloom_status.is<ErrorCode::CORRUPTION>()) {
                    ++pruning_stats->bloom_filter_corrupt_rejections;
                }
            }
            continue;
        }
        if (pruning_stats != nullptr) {
            ++pruning_stats->bloom_filter_probe_successes;
        }
        // Keep at most one decoded payload live while reusing it for every predicate on this leaf.
        for (const auto& probe : leaf_group.probes) {
            if (ParquetStatisticsUtils::NativeBloomFilterExcludes(
                        *probe.column_schema, probe.slot_index, probe.conjuncts, *bloom_filter)) {
                return ParquetRowGroupPruneReason::BLOOM_FILTER;
            }
        }
    }
    return ParquetRowGroupPruneReason::NONE;
}

int64_t native_requested_compressed_bytes(
        const tparquet::RowGroup& row_group,
        const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema,
        const format::FileScanRequest& request) {
    std::set<int> leaf_column_ids;
    auto collect_projection = [&](const format::LocalColumnIndex& projection) {
        const int32_t local_id = projection.local_id();
        if (local_id < 0 || local_id >= static_cast<int32_t>(file_schema.size()) ||
            file_schema[local_id] == nullptr) {
            return;
        }
        collect_filtered_leaf_ids(*file_schema[local_id], &projection, &leaf_column_ids);
    };
    for (const auto& projection : request.predicate_columns) {
        collect_projection(projection);
    }
    for (const auto& projection : request.non_predicate_columns) {
        collect_projection(projection);
    }
    int64_t bytes = 0;
    for (const int leaf_column_id : leaf_column_ids) {
        if (leaf_column_id < 0 || leaf_column_id >= static_cast<int>(row_group.columns.size())) {
            continue;
        }
        const auto& chunk = row_group.columns[leaf_column_id];
        if (chunk.__isset.meta_data && chunk.meta_data.total_compressed_size > 0) {
            bytes += chunk.meta_data.total_compressed_size;
        }
    }
    return bytes;
}

} // namespace

Status select_row_groups_by_metadata(
        const tparquet::FileMetaData& metadata,
        const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema,
        const format::FileScanRequest& request, const std::vector<int>* candidate_row_groups,
        std::vector<int>* selected_row_groups, bool enable_bloom_filter,
        ParquetPruningStats* pruning_stats, const cctz::time_zone* timezone,
        const RuntimeState* runtime_state, ParquetFileContext* file_context,
        const ParquetColumnReaderProfile& column_reader_profile,
        ParquetMetadataProbeMode probe_mode) {
    int64_t timer_sink = 0;
    SCOPED_RAW_TIMER(pruning_stats == nullptr ? &timer_sink
                                              : &pruning_stats->row_group_filter_time);
    if (selected_row_groups == nullptr) {
        return Status::InvalidArgument("selected_row_groups is null");
    }
    selected_row_groups->clear();
    const size_t candidate_size = candidate_row_groups == nullptr ? metadata.row_groups.size()
                                                                  : candidate_row_groups->size();
    if (pruning_stats != nullptr) {
        pruning_stats->total_row_groups = cast_set<int64_t>(candidate_size);
    }
    const bool contains_variant =
            file_context != nullptr ? file_context->contains_variant
                                    : std::ranges::any_of(file_schema, [](const auto& column) {
                                          DORIS_CHECK(column != nullptr);
                                          return column->contains_variant;
                                      });
    selected_row_groups->reserve(candidate_size);
    for (size_t candidate_idx = 0; candidate_idx < candidate_size; ++candidate_idx) {
        const int row_group_idx = candidate_row_groups == nullptr
                                          ? static_cast<int>(candidate_idx)
                                          : (*candidate_row_groups)[candidate_idx];
        if (row_group_idx < 0 || row_group_idx >= static_cast<int>(metadata.row_groups.size())) {
            // Candidate ids originate in external split metadata; a corrupt id must not terminate
            // the BE while planning an otherwise recoverable file scan.
            return Status::Corruption("Invalid Parquet row group candidate {} for {} row groups",
                                      row_group_idx, metadata.row_groups.size());
        }
        const auto& row_group = metadata.row_groups[row_group_idx];
        if (row_group.num_rows < 0) {
            return Status::Corruption("Parquet row group {} has negative row count {}",
                                      row_group_idx, row_group.num_rows);
        }
        if (row_group.num_rows == 0) {
            // Native metadata probes construct positive row ranges; empty groups contribute no
            // rows and must be discarded before dictionary, statistics, or Bloom reader setup.
            continue;
        }
        ParquetRowGroupPruneReason prune_reason = ParquetRowGroupPruneReason::NONE;
        if (probe_mode != ParquetMetadataProbeMode::EXPENSIVE_ONLY &&
            has_expr_zonemap_filter(request, runtime_state) &&
            (check_native_statistics(metadata, row_group, file_schema, request, pruning_stats,
                                     timezone) ||
             (contains_variant && check_shredded_variant_statistics(
                                          metadata, row_group, file_schema, request, timezone)))) {
            prune_reason = ParquetRowGroupPruneReason::STATISTICS;
        }
        if (probe_mode != ParquetMetadataProbeMode::FOOTER_ONLY &&
            prune_reason == ParquetRowGroupPruneReason::NONE) {
            prune_reason =
                    native_dictionary_prune_reason(row_group, row_group_idx, file_schema, request,
                                                   timezone, file_context, column_reader_profile);
        }
        if (probe_mode != ParquetMetadataProbeMode::FOOTER_ONLY &&
            prune_reason == ParquetRowGroupPruneReason::NONE && enable_bloom_filter) {
            prune_reason = native_bloom_filter_prune_reason(row_group, file_schema, request,
                                                            file_context, pruning_stats);
        }
        if (prune_reason == ParquetRowGroupPruneReason::NONE) {
            selected_row_groups->push_back(row_group_idx);
            continue;
        }
        if (pruning_stats != nullptr) {
            pruning_stats->filtered_group_rows += row_group.num_rows;
            pruning_stats->filtered_bytes +=
                    native_requested_compressed_bytes(row_group, file_schema, request);
            if (prune_reason == ParquetRowGroupPruneReason::STATISTICS) {
                ++pruning_stats->filtered_row_groups_by_statistics;
            } else if (prune_reason == ParquetRowGroupPruneReason::DICTIONARY) {
                ++pruning_stats->filtered_row_groups_by_dictionary;
            } else {
                ++pruning_stats->filtered_row_groups_by_bloom_filter;
            }
        }
    }
    return Status::OK();
}

namespace {

std::vector<RowRange> intersect_ranges(const std::vector<RowRange>& left,
                                       const std::vector<RowRange>& right) {
    std::vector<RowRange> result;
    size_t left_idx = 0;
    size_t right_idx = 0;
    while (left_idx < left.size() && right_idx < right.size()) {
        const int64_t left_start = left[left_idx].start;
        const int64_t left_end = left_start + left[left_idx].length;
        const int64_t right_start = right[right_idx].start;
        const int64_t right_end = right_start + right[right_idx].length;
        const int64_t start = std::max(left_start, right_start);
        const int64_t end = std::min(left_end, right_end);
        if (start < end) {
            result.push_back(RowRange {start, end - start});
        }
        if (left_end < right_end) {
            ++left_idx;
        } else {
            ++right_idx;
        }
    }
    return result;
}

std::vector<RowRange> union_ranges(const std::vector<RowRange>& left,
                                   const std::vector<RowRange>& right) {
    std::vector<RowRange> result;
    result.reserve(left.size() + right.size());
    auto append = [&](const RowRange& range) {
        if (range.length == 0) {
            return;
        }
        if (!result.empty()) {
            auto& previous = result.back();
            const int64_t previous_end = previous.start + previous.length;
            if (range.start <= previous_end) {
                previous.length =
                        std::max(previous_end, range.start + range.length) - previous.start;
                return;
            }
        }
        result.push_back(range);
    };
    size_t left_idx = 0;
    size_t right_idx = 0;
    while (left_idx < left.size() || right_idx < right.size()) {
        if (right_idx == right.size() ||
            (left_idx < left.size() && left[left_idx].start <= right[right_idx].start)) {
            append(left[left_idx++]);
        } else {
            append(right[right_idx++]);
        }
    }
    return result;
}

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_row_range(const RowRange& range, std::vector<RowRange>* ranges) {
    if (range.length == 0) {
        return;
    }
    if (!ranges->empty()) {
        auto& previous = ranges->back();
        if (previous.start + previous.length == range.start) {
            previous.length += range.length;
            return;
        }
    }
    ranges->push_back(range);
}

bool ranges_intersect(const std::vector<RowRange>& ranges, const RowRange& range) {
    const int64_t range_end = range.start + range.length;
    for (const auto& selected_range : ranges) {
        const int64_t selected_end = selected_range.start + selected_range.length;
        if (selected_end <= range.start) {
            continue;
        }
        if (selected_range.start >= range_end) {
            return false;
        }
        return true;
    }
    return false;
}

void collect_leaf_schemas(const ParquetColumnSchema& column_schema,
                          const format::LocalColumnIndex* projection,
                          std::vector<const ParquetColumnSchema*>* leaf_schemas) {
    if (column_schema.kind == ParquetColumnSchemaKind::PRIMITIVE) {
        leaf_schemas->push_back(&column_schema);
        return;
    }
    for (const auto& child_schema : column_schema.children) {
        if (column_schema.kind != ParquetColumnSchemaKind::VARIANT &&
            !format::is_child_projected(projection, child_schema->local_id)) {
            continue;
        }
        // A logical Variant projection materializes every physical sibling; build skip plans for
        // that identical leaf set so shredded columns cannot drift to different row positions.
        const auto* child_projection =
                column_schema.kind == ParquetColumnSchemaKind::VARIANT
                        ? nullptr
                        : format::find_child_projection(projection, child_schema->local_id);
        collect_leaf_schemas(*child_schema, child_projection, leaf_schemas);
    }
}

void collect_request_leaf_schemas(
        const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema,
        const format::FileScanRequest& request,
        std::vector<const ParquetColumnSchema*>* leaf_schemas) {
    std::set<int> seen_leaf_ids;
    auto collect_projection = [&](const format::LocalColumnIndex& projection) {
        const int32_t local_id = projection.local_id();
        if (local_id < 0 || local_id >= static_cast<int32_t>(file_schema.size())) {
            return;
        }
        std::vector<const ParquetColumnSchema*> projection_leaf_schemas;
        collect_leaf_schemas(*file_schema[local_id], &projection, &projection_leaf_schemas);
        for (const auto* leaf_schema : projection_leaf_schemas) {
            DORIS_CHECK(leaf_schema != nullptr);
            if (seen_leaf_ids.insert(leaf_schema->leaf_column_id).second) {
                leaf_schemas->push_back(leaf_schema);
            }
        }
    };
    for (const auto& projection : request.predicate_columns) {
        collect_projection(projection);
    }
    for (const auto& projection : request.non_predicate_columns) {
        collect_projection(projection);
    }
}

template <typename ValueType>
bool set_native_page_scalar_min_max(const tparquet::ColumnIndex& column_index,
                                    const ParquetColumnSchema& column_schema, size_t page_idx,
                                    DecodedValueKind kind, ParquetColumnStatistics* page_statistics,
                                    const cctz::time_zone* timezone) {
    if (page_idx >= column_index.min_values.size() || page_idx >= column_index.max_values.size() ||
        column_index.min_values[page_idx].size() != sizeof(ValueType) ||
        column_index.max_values[page_idx].size() != sizeof(ValueType)) {
        return false;
    }
    const auto min_value = unaligned_load<ValueType>(column_index.min_values[page_idx].data());
    const auto max_value = unaligned_load<ValueType>(column_index.max_values[page_idx].data());
    if constexpr (std::is_integral_v<ValueType>) {
        if (remove_nullable(column_schema.type)->get_primitive_type() == TYPE_TIMEV2) {
            int64_t units_per_day = 0;
            switch (column_schema.type_descriptor.time_unit) {
            case ParquetTimeUnit::MILLIS:
                units_per_day = 86400000;
                break;
            case ParquetTimeUnit::MICROS:
                units_per_day = 86400000000;
                break;
            case ParquetTimeUnit::NANOS:
                units_per_day = 86400000000000;
                break;
            default:
                return false;
            }
            // TIME statistics are pruning proofs. Validate the raw carrier before rescaling so an
            // invalid bound at or beyond 24:00 cannot publish a misleading ZoneMap.
            if (min_value < 0 || max_value < 0 || min_value >= units_per_day ||
                max_value >= units_per_day) {
                return false;
            }
        }
    }
    if constexpr (std::is_same_v<ValueType, int64_t>) {
        if (!timestamp_min_max_is_safe(column_schema, min_value, max_value, timezone)) {
            return false;
        }
    }
    if (!valid_min_max(min_value, max_value)) {
        return true;
    }
    if (!set_decoded_field(column_schema, kind, min_value, &page_statistics->min_value, timezone) ||
        !set_decoded_field(column_schema, kind, max_value, &page_statistics->max_value, timezone)) {
        return false;
    }
    if (decoded_min_max_is_ordered(*page_statistics)) {
        page_statistics->has_min_max = true;
    }
    return true;
}

bool set_native_page_boolean_min_max(const tparquet::ColumnIndex& column_index,
                                     const ParquetColumnSchema& column_schema, size_t page_idx,
                                     ParquetColumnStatistics* page_statistics,
                                     const cctz::time_zone* timezone) {
    if (page_idx >= column_index.min_values.size() || page_idx >= column_index.max_values.size() ||
        column_index.min_values[page_idx].size() != 1 ||
        column_index.max_values[page_idx].size() != 1) {
        return false;
    }
    // Parquet BOOLEAN statistics use the same one-bit value representation as PLAIN pages; bits
    // outside the value bit are padding and must not change false into true.
    const uint8_t min_value = static_cast<uint8_t>(column_index.min_values[page_idx][0]) & 1;
    const uint8_t max_value = static_cast<uint8_t>(column_index.max_values[page_idx][0]) & 1;
    if (!valid_min_max(min_value, max_value)) {
        return true;
    }
    if (!set_decoded_field(column_schema, DecodedValueKind::BOOL, min_value,
                           &page_statistics->min_value, timezone) ||
        !set_decoded_field(column_schema, DecodedValueKind::BOOL, max_value,
                           &page_statistics->max_value, timezone)) {
        return false;
    }
    if (decoded_min_max_is_ordered(*page_statistics)) {
        page_statistics->has_min_max = true;
    }
    return true;
}

bool build_native_page_statistics(const tparquet::ColumnIndex& column_index,
                                  const ParquetColumnSchema& column_schema, size_t page_idx,
                                  int64_t page_rows, ParquetColumnStatistics* page_statistics,
                                  const cctz::time_zone* timezone) {
    DORIS_CHECK(page_statistics != nullptr);
    *page_statistics = {};
    if (!column_index.__isset.null_counts || page_idx >= column_index.null_pages.size() ||
        page_idx >= column_index.null_counts.size()) {
        return false;
    }
    const int64_t null_count = column_index.null_counts[page_idx];
    const bool all_null = column_index.null_pages[page_idx];
    if (page_rows < 0 || null_count < 0 || null_count > page_rows ||
        all_null != (null_count == page_rows)) {
        // The caller supplies the exact flat page or row-group span. Contradictory optional null
        // metadata must disable pruning instead of turning a partial span into an all-null proof.
        return false;
    }
    page_statistics->has_null_count = true;
    page_statistics->has_null = null_count > 0;
    page_statistics->has_not_null = !all_null;
    if (!page_statistics->has_not_null) {
        return true;
    }
    switch (column_schema.type_descriptor.physical_type) {
    case tparquet::Type::BOOLEAN:
        return set_native_page_boolean_min_max(column_index, column_schema, page_idx,
                                               page_statistics, timezone);
    case tparquet::Type::INT32:
        return set_native_page_scalar_min_max<int32_t>(
                column_index, column_schema, page_idx,
                decoded_value_kind(column_schema.type_descriptor), page_statistics, timezone);
    case tparquet::Type::INT64:
        return set_native_page_scalar_min_max<int64_t>(
                column_index, column_schema, page_idx,
                decoded_value_kind(column_schema.type_descriptor), page_statistics, timezone);
    case tparquet::Type::FLOAT:
        return set_native_page_scalar_min_max<float>(column_index, column_schema, page_idx,
                                                     DecodedValueKind::FLOAT, page_statistics,
                                                     timezone);
    case tparquet::Type::DOUBLE:
        return set_native_page_scalar_min_max<double>(column_index, column_schema, page_idx,
                                                      DecodedValueKind::DOUBLE, page_statistics,
                                                      timezone);
    case tparquet::Type::BYTE_ARRAY:
    case tparquet::Type::FIXED_LEN_BYTE_ARRAY: {
        if (page_idx >= column_index.min_values.size() ||
            page_idx >= column_index.max_values.size()) {
            return false;
        }
        const auto& min_value = column_index.min_values[page_idx];
        const auto& max_value = column_index.max_values[page_idx];
        const bool fixed =
                column_schema.type_descriptor.physical_type == tparquet::Type::FIXED_LEN_BYTE_ARRAY;
        if (fixed &&
            (column_schema.type_descriptor.fixed_length <= 0 ||
             min_value.size() != static_cast<size_t>(column_schema.type_descriptor.fixed_length) ||
             max_value.size() != static_cast<size_t>(column_schema.type_descriptor.fixed_length))) {
            return false;
        }
        const auto kind = fixed ? DecodedValueKind::FIXED_BINARY : DecodedValueKind::BINARY;
        if (!set_decoded_binary_field(column_schema, kind,
                                      StringRef(min_value.data(), min_value.size()),
                                      &page_statistics->min_value, timezone) ||
            !set_decoded_binary_field(column_schema, kind,
                                      StringRef(max_value.data(), max_value.size()),
                                      &page_statistics->max_value, timezone)) {
            return false;
        }
        if (decoded_min_max_is_ordered(*page_statistics)) {
            page_statistics->has_min_max = true;
        }
        return true;
    }
    default:
        return false;
    }
}

RowRange native_page_row_range(const tparquet::OffsetIndex& offset_index, size_t page_idx,
                               int64_t row_group_rows) {
    const auto& locations = offset_index.page_locations;
    const int64_t start = locations[page_idx].first_row_index;
    const int64_t end = page_idx + 1 == locations.size() ? row_group_rows
                                                         : locations[page_idx + 1].first_row_index;
    return {.start = start, .length = end - start};
}

class NativePageIndexPredicateEvaluator {
public:
    NativePageIndexPredicateEvaluator(
            const tparquet::FileMetaData& metadata,
            const std::unordered_map<int, NativeParquetPageIndex>& page_indexes,
            const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema,
            const format::FileScanRequest& request, int64_t row_group_rows,
            ParquetPruningStats* pruning_stats, const cctz::time_zone* timezone)
            : _metadata(metadata),
              _page_indexes(page_indexes),
              _file_schema(file_schema),
              _request(request),
              _row_group_rows(row_group_rows),
              _pruning_stats(pruning_stats),
              _timezone(timezone) {}

    std::optional<std::vector<RowRange>> evaluate(const VExprSPtr& expr) const {
        if (expr == nullptr || !expr->can_evaluate_zonemap_filter()) {
            return std::nullopt;
        }
        if (expr->op() == TExprOpcode::COMPOUND_AND) {
            return evaluate_compound(expr, true);
        }
        if (expr->op() == TExprOpcode::COMPOUND_OR) {
            return evaluate_compound(expr, false);
        }
        return evaluate_leaf(expr);
    }

private:
    struct SlotPageZoneMaps {
        DataTypePtr data_type;
        std::vector<RowRange> ranges;
        std::vector<std::shared_ptr<segment_v2::ZoneMap>> zone_maps;
    };

    std::optional<std::vector<RowRange>> evaluate_compound(const VExprSPtr& expr,
                                                           bool is_and) const {
        std::optional<std::vector<RowRange>> ranges;
        for (const auto& child : expr->children()) {
            if (!child->can_evaluate_zonemap_filter()) {
                if (!is_and) {
                    return std::nullopt;
                }
                continue;
            }
            auto child_ranges = evaluate(child);
            if (!child_ranges.has_value()) {
                // An unavailable AND child can be ignored, while an unavailable OR branch must
                // retain the complete range so metadata pruning cannot create a false negative.
                if (!is_and) {
                    return std::nullopt;
                }
                continue;
            }
            if (!ranges.has_value()) {
                ranges = std::move(*child_ranges);
            } else if (is_and) {
                ranges = intersect_ranges(*ranges, *child_ranges);
            } else {
                ranges = union_ranges(*ranges, *child_ranges);
            }
            if (is_and && ranges->empty()) {
                return ranges;
            }
        }
        return ranges;
    }

    std::optional<std::vector<RowRange>> evaluate_leaf(const VExprSPtr& expr) const {
        std::set<int> slot_indexes;
        expr->collect_slot_column_ids(slot_indexes);
        if (slot_indexes.size() != 1) {
            return std::nullopt;
        }
        const int slot_index = *slot_indexes.begin();
        const auto* pages = load_slot_pages(slot_index);
        if (pages == nullptr) {
            return std::nullopt;
        }

        std::vector<RowRange> ranges;
        for (size_t page_idx = 0; page_idx < pages->ranges.size(); ++page_idx) {
            ZoneMapEvalContext ctx;
            add_slot_zonemap(&ctx, slot_index, pages->data_type, pages->zone_maps[page_idx]);
            if (expr->evaluate_zonemap_filter(ctx) != ZoneMapFilterResult::kNoMatch) {
                append_row_range(pages->ranges[page_idx], &ranges);
            }
            accumulate_zonemap_stats(ctx, _pruning_stats);
        }
        return ranges;
    }

    const SlotPageZoneMaps* load_slot_pages(int slot_index) const {
        const auto cached = _slot_page_zone_maps.find(slot_index);
        if (cached != _slot_page_zone_maps.end()) {
            return cached->second.has_value() ? &*cached->second : nullptr;
        }
        const auto file_column_id = file_column_id_by_block_position(_request, slot_index);
        if (!file_column_id.has_value()) {
            _slot_page_zone_maps.emplace(slot_index, std::nullopt);
            return nullptr;
        }
        const auto* column_schema = resolve_local_leaf_schema(_file_schema, *file_column_id);
        if (column_schema == nullptr || column_schema->type == nullptr ||
            !native_metadata_predicate_is_type_safe(*column_schema) ||
            !detail::has_supported_type_defined_order(_metadata, column_schema->leaf_column_id)) {
            _slot_page_zone_maps.emplace(slot_index, std::nullopt);
            return nullptr;
        }
        const auto index_it = _page_indexes.find(column_schema->leaf_column_id);
        if (index_it == _page_indexes.end()) {
            _slot_page_zone_maps.emplace(slot_index, std::nullopt);
            return nullptr;
        }

        const auto& indexes = index_it->second;
        SlotPageZoneMaps pages;
        pages.data_type = column_schema->type;
        pages.ranges.reserve(indexes.offset_index.page_locations.size());
        pages.zone_maps.reserve(indexes.offset_index.page_locations.size());
        for (size_t page_idx = 0; page_idx < indexes.offset_index.page_locations.size();
             ++page_idx) {
            const auto page_range =
                    native_page_row_range(indexes.offset_index, page_idx, _row_group_rows);
            ParquetColumnStatistics statistics;
            if (!build_native_page_statistics(indexes.column_index, *column_schema, page_idx,
                                              page_range.length, &statistics, _timezone)) {
                _slot_page_zone_maps.emplace(slot_index, std::nullopt);
                return nullptr;
            }
            pages.ranges.push_back(page_range);
            pages.zone_maps.push_back(ParquetStatisticsUtils::MakeZoneMap(statistics));
        }
        const auto inserted = _slot_page_zone_maps.emplace(slot_index, std::move(pages));
        return &*inserted.first->second;
    }

    const tparquet::FileMetaData& _metadata;
    const std::unordered_map<int, NativeParquetPageIndex>& _page_indexes;
    const std::vector<std::unique_ptr<ParquetColumnSchema>>& _file_schema;
    const format::FileScanRequest& _request;
    int64_t _row_group_rows;
    ParquetPruningStats* _pruning_stats;
    const cctz::time_zone* _timezone;
    mutable std::unordered_map<int, std::optional<SlotPageZoneMaps>> _slot_page_zone_maps;
};

} // namespace

Status select_row_group_ranges_by_native_page_index(
        const tparquet::FileMetaData& metadata, const tparquet::RowGroup& row_group,
        const std::unordered_map<int, NativeParquetPageIndex>& page_indexes,
        const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema,
        const format::FileScanRequest& request, int64_t row_group_rows,
        std::vector<RowRange>* selected_ranges, std::map<int, ParquetPageSkipPlan>* page_skip_plans,
        ParquetPruningStats* pruning_stats, const cctz::time_zone* timezone,
        const RuntimeState* runtime_state) {
    int64_t filter_time_sink = 0;
    SCOPED_RAW_TIMER(pruning_stats == nullptr ? &filter_time_sink
                                              : &pruning_stats->page_index_filter_time);
    DORIS_CHECK(selected_ranges != nullptr);
    selected_ranges->clear();
    selected_ranges->push_back({.start = 0, .length = row_group_rows});
    if (page_skip_plans != nullptr) {
        page_skip_plans->clear();
    }
    if (row_group_rows <= 0 || !config::enable_parquet_page_index ||
        !has_expr_zonemap_filter(request, runtime_state) || page_indexes.empty()) {
        return Status::OK();
    }
    if (pruning_stats != nullptr) {
        ++pruning_stats->page_index_read_calls;
    }

    std::map<int, VExprContextSPtrs> conjuncts_by_slot;
    VExprContextSPtrs multi_slot_conjuncts;
    // Compound predicates must honor the same metadata-pruning fence as single-slot predicates.
    for (const auto& conjunct : metadata_pruning_conjuncts(request)) {
        const auto slot_index = expr_zonemap::single_slot_zonemap_index(conjunct);
        if (slot_index >= 0) {
            conjuncts_by_slot[slot_index].push_back(conjunct);
        } else if (conjunct != nullptr && conjunct->root() != nullptr &&
                   conjunct->root()->can_evaluate_zonemap_filter()) {
            multi_slot_conjuncts.push_back(conjunct);
        }
    }
    for (const auto& [slot_index, conjuncts] : conjuncts_by_slot) {
        const auto file_column_id = file_column_id_by_block_position(request, slot_index);
        if (!file_column_id.has_value()) {
            continue;
        }
        const auto* column_schema = resolve_local_leaf_schema(file_schema, *file_column_id);
        if (column_schema == nullptr || column_schema->type == nullptr ||
            !native_metadata_predicate_is_type_safe(*column_schema) ||
            !detail::has_supported_type_defined_order(metadata, column_schema->leaf_column_id)) {
            continue;
        }
        const auto index_it = page_indexes.find(column_schema->leaf_column_id);
        if (index_it == page_indexes.end()) {
            continue;
        }
        const auto& indexes = index_it->second;
        std::vector<RowRange> filter_ranges;
        bool usable = true;
        for (size_t page_idx = 0; page_idx < indexes.offset_index.page_locations.size();
             ++page_idx) {
            const auto page_range =
                    native_page_row_range(indexes.offset_index, page_idx, row_group_rows);
            ParquetColumnStatistics statistics;
            if (!build_native_page_statistics(indexes.column_index, *column_schema, page_idx,
                                              page_range.length, &statistics, timezone)) {
                usable = false;
                break;
            }
            ZoneMapEvalContext ctx;
            add_slot_zonemap(&ctx, slot_index, column_schema->type,
                             ParquetStatisticsUtils::MakeZoneMap(statistics));
            if (VExprContext::evaluate_zonemap_filter(conjuncts, ctx) !=
                ZoneMapFilterResult::kNoMatch) {
                append_row_range(page_range, &filter_ranges);
            }
            accumulate_zonemap_stats(ctx, pruning_stats);
        }
        if (!usable) {
            continue;
        }
        *selected_ranges = intersect_ranges(*selected_ranges, filter_ranges);
        if (selected_ranges->empty()) {
            if (pruning_stats != nullptr) {
                pruning_stats->filtered_page_rows += row_group_rows;
                ++pruning_stats->filtered_row_groups_by_page_index;
            }
            return Status::OK();
        }
    }

    NativePageIndexPredicateEvaluator evaluator(metadata, page_indexes, file_schema, request,
                                                row_group_rows, pruning_stats, timezone);
    for (const auto& conjunct : multi_slot_conjuncts) {
        auto conjunct_ranges = evaluator.evaluate(conjunct->root());
        if (!conjunct_ranges.has_value()) {
            continue;
        }
        *selected_ranges = intersect_ranges(*selected_ranges, *conjunct_ranges);
        if (selected_ranges->empty()) {
            if (pruning_stats != nullptr) {
                pruning_stats->filtered_page_rows += row_group_rows;
                ++pruning_stats->filtered_row_groups_by_page_index;
            }
            return Status::OK();
        }
    }

    for (const auto& conjunct : metadata_pruning_conjuncts(request)) {
        const auto predicate = extract_variant_shredded_predicate(conjunct);
        if (!predicate.has_value()) {
            continue;
        }
        const auto shredding = resolve_variant_shredding(file_schema, request, *predicate);
        if (!shredding.has_value() || shredding->typed_value->leaf_column_id < 0 ||
            !fallbacks_are_all_null(row_group, shredding->fallback_values) ||
            !variant_metadata_predicate_is_type_safe(*shredding->typed_value) ||
            !detail::has_supported_type_defined_order(metadata,
                                                      shredding->typed_value->leaf_column_id)) {
            continue;
        }
        const auto index_it = page_indexes.find(shredding->typed_value->leaf_column_id);
        if (index_it == page_indexes.end()) {
            continue;
        }
        const auto& indexes = index_it->second;
        std::vector<RowRange> filter_ranges;
        bool usable = true;
        for (size_t page_idx = 0; page_idx < indexes.offset_index.page_locations.size();
             ++page_idx) {
            const auto page_range =
                    native_page_row_range(indexes.offset_index, page_idx, row_group_rows);
            ParquetColumnStatistics statistics;
            if (!build_native_page_statistics(indexes.column_index, *shredding->typed_value,
                                              page_idx, page_range.length, &statistics, timezone)) {
                usable = false;
                break;
            }
            const auto normalized =
                    normalize_variant_statistics(*predicate, *shredding->typed_value, statistics);
            if (!normalized.has_value()) {
                usable = false;
                break;
            }
            if (!variant_statistics_exclude(*predicate, *normalized)) {
                append_row_range(page_range, &filter_ranges);
            }
        }
        if (!usable) {
            continue;
        }
        *selected_ranges = intersect_ranges(*selected_ranges, filter_ranges);
        if (selected_ranges->empty()) {
            if (pruning_stats != nullptr) {
                pruning_stats->filtered_page_rows += row_group_rows;
                ++pruning_stats->filtered_row_groups_by_page_index;
            }
            return Status::OK();
        }
    }

    if (page_skip_plans != nullptr) {
        std::vector<const ParquetColumnSchema*> leaves;
        collect_request_leaf_schemas(file_schema, request, &leaves);
        for (const auto* leaf : leaves) {
            const auto index_it = page_indexes.find(leaf->leaf_column_id);
            if (index_it == page_indexes.end() || leaf->max_repetition_level != 0) {
                continue;
            }
            const auto& offset_index = index_it->second.offset_index;
            ParquetPageSkipPlan skip_plan;
            skip_plan.leaf_column_id = leaf->leaf_column_id;
            skip_plan.skipped_pages.resize(offset_index.page_locations.size());
            skip_plan.skipped_page_compressed_sizes.resize(offset_index.page_locations.size());
            for (size_t page_idx = 0; page_idx < offset_index.page_locations.size(); ++page_idx) {
                const auto range = native_page_row_range(offset_index, page_idx, row_group_rows);
                if (range.length == 0 || ranges_intersect(*selected_ranges, range)) {
                    continue;
                }
                skip_plan.skipped_pages[page_idx] = 1;
                skip_plan.skipped_page_compressed_sizes[page_idx] =
                        offset_index.page_locations[page_idx].compressed_page_size;
                append_row_range(range, &skip_plan.skipped_ranges);
            }
            if (!skip_plan.empty()) {
                page_skip_plans->emplace(skip_plan.leaf_column_id, std::move(skip_plan));
            }
        }
    }
    if (pruning_stats != nullptr) {
        pruning_stats->filtered_page_rows += row_group_rows - count_range_rows(*selected_ranges);
    }
    return Status::OK();
}

} // namespace doris::format::parquet

Coverage Report

Created: 2026-08-16 13:33