Re: [PR] [improvement](be) Complete native Parquet reader for FileScannerV2 [doris]

[via GitHub]

github-actions[bot] commented on code in PR #65674:
URL: https://github.com/apache/doris/pull/65674#discussion_r3614077027


##########
be/src/format_v2/parquet/reader/native/column_chunk_reader.cpp:
##########
@@ -0,0 +1,1676 @@
+// 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/reader/native/column_chunk_reader.h"
+
+#include <cctz/time_zone.h>
+#include <gen_cpp/parquet_types.h>
+#include <glog/logging.h>
+#include <parquet/metadata.h>
+#include <string.h>
+
+#include <algorithm>
+#include <cstdint>
+#include <limits>
+#include <memory>
+#include <utility>
+
+#include "common/compiler_util.h" // IWYU pragma: keep
+#include "core/column/column.h"
+#include "core/column/column_decimal.h"
+#include "core/column/column_dictionary.h"
+#include "core/column/column_varbinary.h"
+#include "core/column/column_vector.h"
+#include "core/custom_allocator.h"
+#include "core/data_type_serde/data_type_serde.h"
+#include "core/data_type_serde/parquet_timestamp.h"
+#include "exprs/vexpr.h"
+#include "format_v2/parquet/native_schema_desc.h"
+#include "format_v2/parquet/reader/native/decoder.h"
+#include "format_v2/parquet/reader/native/level_decoder.h"
+#include "format_v2/parquet/reader/native/page_reader.h"
+#include "io/fs/buffered_reader.h"
+#include "runtime/runtime_profile.h"
+#include "storage/cache/page_cache.h"
+#include "util/bit_util.h"
+#include "util/block_compression.h"
+#include "util/cpu_info.h"
+#include "util/unaligned.h"
+
+namespace cctz {
+class time_zone;
+} // namespace cctz
+namespace doris {
+namespace io {
+class BufferedStreamReader;
+struct IOContext;
+} // namespace io
+} // namespace doris
+
+namespace doris::format::parquet::native {
+
+bool can_prepare_page_cache_payload(bool session_cache_enabled, bool 
storage_cache_disabled,
+                                    bool cache_available, bool 
header_available) {
+    return session_cache_enabled && !storage_cache_disabled && cache_available 
&& header_available;
+}
+
+Status validate_uncompressed_page_sizes(const tparquet::PageHeader& header,
+                                        tparquet::CompressionCodec::type codec,
+                                        bool data_page_v2_always_compressed) {
+    const bool is_v2 = header.__isset.data_page_header_v2;
+    const bool page_is_compressed =
+            codec != tparquet::CompressionCodec::UNCOMPRESSED &&
+            (!is_v2 || header.data_page_header_v2.is_compressed || 
data_page_v2_always_compressed);
+    if (!page_is_compressed &&
+        UNLIKELY(header.compressed_page_size != 
header.uncompressed_page_size)) {
+        // An uncompressed payload has one physical representation, so 
accepting two lengths makes
+        // cold reads and decompressed cache hits consume different byte 
boundaries.
+        return Status::Corruption(
+                "Uncompressed Parquet page sizes differ: compressed={}, 
uncompressed={}",
+                header.compressed_page_size, header.uncompressed_page_size);
+    }
+    return Status::OK();
+}
+
+ParquetReaderCompat parquet_reader_compat(const std::string& created_by) {
+    if (created_by.empty()) {
+        return {};
+    }
+    const ::parquet::ApplicationVersion version(created_by);
+    return {.parquet_816_padding =
+                    
version.VersionLt(::parquet::ApplicationVersion::PARQUET_816_FIXED_VERSION()),
+            .data_page_v2_always_compressed = version.VersionLt(
+                    
::parquet::ApplicationVersion::PARQUET_CPP_10353_FIXED_VERSION())};
+}
+
+Status compute_column_chunk_range(const tparquet::ColumnMetaData& metadata, 
size_t file_size,
+                                  bool parquet_816_padding, ColumnChunkRange* 
range) {
+    DORIS_CHECK(range != nullptr);
+    int64_t start = metadata.data_page_offset;
+    if (metadata.__isset.dictionary_page_offset && 
metadata.dictionary_page_offset > 0 &&
+        metadata.dictionary_page_offset < start) {
+        // Some writers use dictionary_page_offset=0 as an absence sentinel. 
Range validation must
+        // follow has_dict_page() or the native reader starts at the Parquet 
magic bytes.
+        start = metadata.dictionary_page_offset;
+    }
+    const int64_t length = metadata.total_compressed_size;
+    if (UNLIKELY(start < 0 || length < 0)) {
+        return Status::Corruption("Parquet column chunk has a negative offset 
or length");
+    }
+    const uint64_t unsigned_start = static_cast<uint64_t>(start);
+    const uint64_t unsigned_length = static_cast<uint64_t>(length);
+    if (UNLIKELY(unsigned_start > file_size || unsigned_length > file_size - 
unsigned_start)) {
+        // Thrift range fields are signed and untrusted; validate before 
converting them to the
+        // unsigned stream-reader coordinates so overflow cannot wrap back 
into the file.
+        return Status::Corruption("Parquet column chunk [{}, {}) exceeds file 
size {}", start,
+                                  unsigned_start + unsigned_length, file_size);
+    }
+    size_t bounded_length = static_cast<size_t>(unsigned_length);
+    if (parquet_816_padding) {
+        // parquet-mr before PARQUET-816 under-reported the chunk by up to 100 
bytes. Padding stays
+        // file-bounded and is only enabled for the affected writer versions.
+        bounded_length += std::min<size_t>(100, file_size - unsigned_start - 
unsigned_length);
+    }
+    range->offset = static_cast<size_t>(unsigned_start);
+    range->length = bounded_length;
+    return Status::OK();
+}
+
+bool validate_offset_index(const tparquet::OffsetIndex& index, const 
ColumnChunkRange& chunk_range,
+                           int64_t data_page_offset, int64_t row_count) {
+    if (index.page_locations.empty() || data_page_offset < 0 || row_count < 0 
||
+        index.page_locations.front().first_row_index != 0 ||
+        index.page_locations.front().offset != data_page_offset ||
+        chunk_range.length > std::numeric_limits<size_t>::max() - 
chunk_range.offset) {
+        return false;
+    }
+    // Row indexes alone cannot detect a uniformly shifted OffsetIndex. Anchor 
its first location
+    // to the owning metadata so page-to-row mapping cannot silently move by 
one physical page.
+    const uint64_t chunk_begin = chunk_range.offset;
+    const uint64_t chunk_end = chunk_begin + chunk_range.length;
+    uint64_t previous_end = chunk_begin;
+    int64_t previous_row = -1;
+    for (const auto& location : index.page_locations) {
+        if (location.first_row_index <= previous_row || 
location.first_row_index >= row_count ||
+            location.offset < 0 || location.compressed_page_size <= 0) {
+            return false;
+        }
+        const uint64_t begin = static_cast<uint64_t>(location.offset);
+        const uint64_t size = 
static_cast<uint64_t>(location.compressed_page_size);
+        if (begin < chunk_begin || begin < previous_end || begin > chunk_end ||
+            size > chunk_end - begin) {
+            return false;
+        }
+        previous_row = location.first_row_index;
+        previous_end = begin + size;
+    }
+    return true;
+}
+
+namespace {
+
+Status append_v2_int96_datetime(ColumnDateTimeV2::Container& data,
+                                const ParquetInt96Timestamp& value,
+                                const cctz::time_zone& timezone) {
+    static constexpr int32_t JULIAN_EPOCH_OFFSET_DAYS = 2440588;
+    static constexpr int64_t MICROS_PER_DAY = 86400000000LL;
+    static constexpr int64_t MICROS_PER_SECOND = 1000000LL;
+
+    // Arrow normalized out-of-day INT96 nanos before the native V2 path 
replaced it. Preserve that
+    // legacy-writer compatibility here; rejecting the carrier loses valid 
pre-epoch/year-0 values
+    // already accepted by Doris external-table scans.
+    const __int128 days = static_cast<__int128>(value.julian_day) - 
JULIAN_EPOCH_OFFSET_DAYS;
+    // Truncate the signed nanos field before day normalization. Flooring a 
negative value first
+    // changes the historical result by one microsecond.
+    const __int128 timestamp_micros = days * MICROS_PER_DAY + 
value.nanos_of_day / 1000;
+    if (timestamp_micros < std::numeric_limits<int64_t>::min() ||
+        timestamp_micros > std::numeric_limits<int64_t>::max()) {
+        return Status::DataQualityError("Parquet INT96 timestamp overflows 
microseconds");
+    }
+
+    const int64_t micros = static_cast<int64_t>(timestamp_micros);
+    int64_t epoch_seconds = micros / MICROS_PER_SECOND;
+    int64_t micros_of_second = micros % MICROS_PER_SECOND;
+    if (micros_of_second < 0) {
+        micros_of_second += MICROS_PER_SECOND;
+        --epoch_seconds;
+    }
+    DateV2Value<DateTimeV2ValueType> datetime;
+    datetime.from_unixtime(epoch_seconds, timezone);
+    datetime.set_microsecond(static_cast<uint32_t>(micros_of_second));
+    if (!datetime.is_valid_date()) {
+        return Status::DataQualityError("Parquet INT96 timestamp is outside 
the Doris range");
+    }
+    data.push_back(datetime);
+    return Status::OK();
+}
+
+class V2Int96DateTimeConsumer final : public ParquetFixedValueConsumer {
+public:
+    V2Int96DateTimeConsumer(IColumn& column, const ParquetDecodeContext& 
context,
+                            ParquetMaterializationState* state)
+            : _data(assert_cast<ColumnDateTimeV2&>(column).get_data()), 
_state(state) {
+        static const auto utc = cctz::utc_time_zone();
+        _timezone = context.timezone == nullptr ? &utc : context.timezone;
+    }
+
+    Status consume(const uint8_t* values, size_t num_values, size_t 
value_width) override {
+        DORIS_CHECK_EQ(value_width, sizeof(ParquetInt96Timestamp));
+        const size_t old_size = _data.size();
+        for (size_t row = 0; row < num_values; ++row) {
+            const auto value = unaligned_load<ParquetInt96Timestamp>(
+                    values + row * sizeof(ParquetInt96Timestamp));
+            const auto status = append_v2_int96_datetime(_data, value, 
*_timezone);
+            if (!status.ok()) {
+                if (_state != nullptr && 
_state->mark_conversion_failure(_data.size())) {
+                    _data.emplace_back();
+                    continue;
+                }
+                _data.resize(old_size);
+                return status;
+            }
+        }
+        return Status::OK();
+    }
+
+private:
+    ColumnDateTimeV2::Container& _data;
+    ParquetMaterializationState* _state;
+    const cctz::time_zone* _timezone = nullptr;
+};
+
+class RejectV2Int96BinaryConsumer final : public ParquetBinaryValueConsumer {
+public:
+    Status consume(const StringRef*, size_t) override {
+        return Status::NotSupported("INT96 cannot be decoded from binary 
Parquet values");
+    }
+};
+
+Status read_v2_int96_datetime(IColumn& column, ParquetDecodeSource& source,
+                              const ParquetDecodeContext& context, size_t 
num_values,
+                              ParquetMaterializationState& state) {
+    V2Int96DateTimeConsumer consumer(column, context, &state);
+    if (context.encoding != ParquetValueEncoding::DICTIONARY) {
+        return source.decode_fixed_values(num_values, consumer);
+    }
+    if (state.dictionary_generation != source.dictionary_generation()) {
+        state.typed_dictionary = column.clone_empty();
+        auto* output_null_map = 
state.begin_dictionary_conversion(source.dictionary_size());
+        V2Int96DateTimeConsumer dictionary_consumer(*state.typed_dictionary, 
context, &state);
+        RejectV2Int96BinaryConsumer binary_consumer;
+        const auto dictionary_status =
+                source.decode_dictionary(dictionary_consumer, binary_consumer);
+        state.end_dictionary_conversion(output_null_map);
+        RETURN_IF_ERROR(dictionary_status);
+        DORIS_CHECK_EQ(state.typed_dictionary->size(), 
source.dictionary_size());
+        state.dictionary_generation = source.dictionary_generation();
+    }
+    RETURN_IF_ERROR(source.decode_dictionary_indices(num_values, 
&state.dictionary_indices));
+    DORIS_CHECK_EQ(state.dictionary_indices.size(), num_values);
+    return state.materialize_dictionary(column);
+}
+
+Status read_native_or_serde(IColumn& column, const DataTypeSerDe& serde,
+                            ParquetDecodeSource& source, const 
ParquetDecodeContext& context,
+                            size_t num_values, ParquetMaterializationState& 
state) {
+    if (context.physical_type == ParquetPhysicalType::INT96 &&
+        check_and_get_column<ColumnDateTimeV2>(&column) != nullptr) {
+        return read_v2_int96_datetime(column, source, context, num_values, 
state);
+    }
+    return serde.read_column_from_parquet(column, source, context, num_values, 
state);
+}
+
+Status translate_value_encoding(tparquet::Encoding::type encoding,
+                                ParquetValueEncoding* translated) {
+    DORIS_CHECK(translated != nullptr);
+    switch (encoding) {
+    case tparquet::Encoding::PLAIN:
+        *translated = ParquetValueEncoding::PLAIN;
+        return Status::OK();
+    case tparquet::Encoding::RLE_DICTIONARY:
+    case tparquet::Encoding::PLAIN_DICTIONARY:
+        *translated = ParquetValueEncoding::DICTIONARY;
+        return Status::OK();
+    case tparquet::Encoding::RLE:
+        *translated = ParquetValueEncoding::RLE;
+        return Status::OK();
+    case tparquet::Encoding::BIT_PACKED:
+        *translated = ParquetValueEncoding::BIT_PACKED;
+        return Status::OK();
+    case tparquet::Encoding::DELTA_BINARY_PACKED:
+        *translated = ParquetValueEncoding::DELTA_BINARY_PACKED;
+        return Status::OK();
+    case tparquet::Encoding::DELTA_LENGTH_BYTE_ARRAY:
+        *translated = ParquetValueEncoding::DELTA_LENGTH_BYTE_ARRAY;
+        return Status::OK();
+    case tparquet::Encoding::DELTA_BYTE_ARRAY:
+        *translated = ParquetValueEncoding::DELTA_BYTE_ARRAY;
+        return Status::OK();
+    case tparquet::Encoding::BYTE_STREAM_SPLIT:
+        *translated = ParquetValueEncoding::BYTE_STREAM_SPLIT;
+        return Status::OK();
+    default:
+        return Status::NotSupported("Unsupported Parquet encoding {}",
+                                    tparquet::to_string(encoding));
+    }
+}
+
+template <bool HAS_FILTER>
+Status decode_selected_values(IColumn& column, const DataTypeSerDe& serde, 
Decoder& decoder,
+                              const ParquetDecodeContext& context,
+                              ParquetMaterializationState& state, 
ColumnSelectVector& select_vector,
+                              int64_t* materialization_time) {
+    SCOPED_RAW_TIMER(materialization_time);
+    ColumnSelectVector::DataReadType read_type;
+    while (const size_t run_length = 
select_vector.get_next_run<HAS_FILTER>(&read_type)) {
+        switch (read_type) {
+        case ColumnSelectVector::CONTENT:
+            RETURN_IF_ERROR(
+                    read_native_or_serde(column, serde, decoder, context, 
run_length, state));
+            break;
+        case ColumnSelectVector::NULL_DATA:
+            column.insert_many_defaults(run_length);
+            break;
+        case ColumnSelectVector::FILTERED_CONTENT:
+            RETURN_IF_ERROR(decoder.skip_values(run_length));
+            break;
+        case ColumnSelectVector::FILTERED_NULL:
+            break;
+        }
+    }
+    return Status::OK();
+}
+
+// Presents one sparse page request as an ordinary sequential source to 
DataTypeSerDe. SerDe is
+// entered once per page fragment; the concrete decoder decides whether to 
gather selected spans,
+// batch-decode and compact, or use the cursor-preserving range fallback.
+class SelectedDecodeSource final : public ParquetDecodeSource {
+public:
+    SelectedDecodeSource(Decoder& decoder, const ParquetSelection& selection)
+            : _decoder(decoder), _selection(selection) {}
+
+    Status decode_fixed_values(size_t num_values, ParquetFixedValueConsumer& 
consumer) override {
+        DORIS_CHECK_EQ(num_values, _selection.selected_values);
+        return _decoder.decode_selected_fixed_values(_selection, consumer);
+    }
+
+    Status decode_binary_values(size_t num_values, ParquetBinaryValueConsumer& 
consumer) override {
+        DORIS_CHECK_EQ(num_values, _selection.selected_values);
+        return _decoder.decode_selected_binary_values(_selection, consumer);
+    }
+
+    Status skip_values(size_t num_values) override {
+        return Status::NotSupported("Selected Parquet source cannot be 
skipped, values={}",
+                                    num_values);
+    }
+
+    bool has_dictionary() const override { return _decoder.has_dictionary(); }
+    uint64_t dictionary_generation() const override { return 
_decoder.dictionary_generation(); }
+    size_t dictionary_size() const override { return 
_decoder.dictionary_size(); }
+
+    Status decode_dictionary(ParquetFixedValueConsumer& fixed_consumer,
+                             ParquetBinaryValueConsumer& binary_consumer) 
override {
+        return _decoder.decode_dictionary(fixed_consumer, binary_consumer);
+    }
+
+    Status decode_dictionary_indices(size_t num_values, std::vector<uint32_t>* 
indices) override {
+        DORIS_CHECK_EQ(num_values, _selection.selected_values);
+        return _decoder.decode_selected_dictionary_indices(_selection, 
indices);
+    }
+
+    Status decode_dictionary_values(size_t num_values,
+                                    ParquetDictionaryValueConsumer& consumer) 
override {
+        DORIS_CHECK_EQ(num_values, _selection.selected_values);
+        return _decoder.decode_selected_dictionary_values(_selection, 
consumer);
+    }
+
+    bool prefer_dictionary_index_materialization(size_t dictionary_bytes) 
const override {
+        // Avoid touching a dictionary larger than L2 for rows already removed 
by sparse selection.
+        // Cache-resident or dense batches instead fuse RLE decode and gather, 
eliminating the
+        // page-sized intermediate ID vector.
+        return _selection.selected_values < _selection.total_values &&
+               dictionary_bytes > 
static_cast<size_t>(CpuInfo::get_l2_cache_size());
+    }
+
+private:
+    Decoder& _decoder;
+    const ParquetSelection& _selection;
+};
+
+Status decode_selected_non_null_values(IColumn& column, const DataTypeSerDe& 
serde,
+                                       Decoder& decoder, const 
ParquetDecodeContext& context,
+                                       ParquetMaterializationState& state,
+                                       ColumnSelectVector& select_vector,
+                                       int64_t* materialization_time) {
+    auto& selection = state.selection;
+    selection.ranges.clear();
+    selection.total_values = select_vector.num_values();
+    selection.selected_values = 0;
+
+    size_t cursor = 0;
+    ColumnSelectVector::DataReadType read_type;
+    while (const size_t run_length = 
select_vector.get_next_run<true>(&read_type)) {
+        DORIS_CHECK(read_type == ColumnSelectVector::CONTENT ||
+                    read_type == ColumnSelectVector::FILTERED_CONTENT);
+        if (read_type == ColumnSelectVector::CONTENT) {
+            selection.ranges.push_back({.first = cursor, .count = run_length});
+            selection.selected_values += run_length;
+        }
+        cursor += run_length;
+    }
+    DORIS_CHECK_EQ(cursor, selection.total_values);
+    if (selection.selected_values == 0) {
+        return decoder.skip_values(selection.total_values);
+    }
+
+    SCOPED_RAW_TIMER(materialization_time);
+    SelectedDecodeSource selected_source(decoder, selection);
+    return read_native_or_serde(column, serde, selected_source, context, 
selection.selected_values,
+                                state);
+}
+
+template <typename Visitor>
+bool visit_nullable_expandable_column(IColumn& column, Visitor&& visitor) {
+#define VISIT_COLUMN(TYPE)                                   \
+    if (auto* typed = check_and_get_column<TYPE>(&column)) { \
+        visitor(*typed);                                     \
+        return true;                                         \
+    }
+    VISIT_COLUMN(ColumnUInt8)
+    VISIT_COLUMN(ColumnInt8)
+    VISIT_COLUMN(ColumnInt16)
+    VISIT_COLUMN(ColumnInt32)
+    VISIT_COLUMN(ColumnInt64)
+    VISIT_COLUMN(ColumnInt128)
+    VISIT_COLUMN(ColumnDate)
+    VISIT_COLUMN(ColumnDateTime)
+    VISIT_COLUMN(ColumnDateV2)
+    VISIT_COLUMN(ColumnDateTimeV2)
+    VISIT_COLUMN(ColumnFloat32)
+    VISIT_COLUMN(ColumnFloat64)
+    VISIT_COLUMN(ColumnIPv4)
+    VISIT_COLUMN(ColumnIPv6)
+    VISIT_COLUMN(ColumnTimeV2)
+    VISIT_COLUMN(ColumnTimeStampTz)
+    VISIT_COLUMN(ColumnOffset32)
+    VISIT_COLUMN(ColumnOffset64)
+    VISIT_COLUMN(ColumnDecimal32)
+    VISIT_COLUMN(ColumnDecimal64)
+    VISIT_COLUMN(ColumnDecimal128V2)
+    VISIT_COLUMN(ColumnDecimal128V3)
+    VISIT_COLUMN(ColumnDecimal256)
+    VISIT_COLUMN(ColumnString)
+    VISIT_COLUMN(ColumnString64)
+    VISIT_COLUMN(ColumnVarbinary)
+    VISIT_COLUMN(ColumnDictI32)
+#undef VISIT_COLUMN
+    return false;
+}
+
+template <typename ColumnType>
+void expand_nullable_pod_values(ColumnType& column, size_t old_size, size_t 
compact_values,
+                                const NullMap& selected_nulls) {
+    auto& data = column.get_data();
+    DORIS_CHECK_EQ(data.size(), old_size + compact_values);
+    data.resize(old_size + selected_nulls.size());
+    size_t source = compact_values;
+    for (size_t output = selected_nulls.size(); output > 0;) {
+        --output;
+        if (selected_nulls[output] != 0) {
+            data[old_size + output] = typename ColumnType::value_type {};
+        } else {
+            DORIS_CHECK(source > 0);
+            --source;
+            data[old_size + output] = std::move(data[old_size + source]);
+        }
+    }
+    DORIS_CHECK_EQ(source, 0);
+}
+
+template <typename Offset>
+void expand_nullable_string_values(ColumnStr<Offset>& column, size_t old_size,
+                                   size_t compact_values, const NullMap& 
selected_nulls) {
+    auto& offsets = column.get_offsets();
+    DORIS_CHECK_EQ(offsets.size(), old_size + compact_values);
+    const Offset prefix_end = old_size == 0 ? 0 : offsets[old_size - 1];
+    offsets.resize(old_size + selected_nulls.size());
+    size_t source = compact_values;
+    for (size_t output = selected_nulls.size(); output > 0;) {
+        --output;
+        if (selected_nulls[output] == 0) {
+            DORIS_CHECK(source > 0);
+            --source;
+            offsets[old_size + output] = offsets[old_size + source];
+        } else {
+            offsets[old_size + output] = source == 0 ? prefix_end : 
offsets[old_size + source - 1];
+        }
+    }
+    DORIS_CHECK_EQ(source, 0);
+}
+
+template <typename ColumnType>
+void expand_nullable_values(ColumnType& column, size_t old_size, size_t 
compact_values,
+                            const NullMap& selected_nulls) {
+    expand_nullable_pod_values(column, old_size, compact_values, 
selected_nulls);
+}
+
+template <typename Offset>
+void expand_nullable_values(ColumnStr<Offset>& column, size_t old_size, size_t 
compact_values,
+                            const NullMap& selected_nulls) {
+    expand_nullable_string_values(column, old_size, compact_values, 
selected_nulls);
+}
+
+void remap_nullable_conversion_failures(IColumn::Filter* 
conversion_failure_null_map,
+                                        size_t old_size, size_t compact_values,
+                                        const NullMap& selected_nulls) {
+    if (conversion_failure_null_map == nullptr) {
+        return;
+    }
+    DORIS_CHECK(conversion_failure_null_map->size() >= old_size + 
selected_nulls.size());
+    size_t source = compact_values;
+    // Walk backwards so writing an expanded row cannot overwrite an unread 
compact failure bit.
+    for (size_t output = selected_nulls.size(); output > 0;) {
+        --output;
+        if (selected_nulls[output] != 0) {
+            (*conversion_failure_null_map)[old_size + output] = 1;
+        } else {
+            DORIS_CHECK(source > 0);
+            --source;
+            (*conversion_failure_null_map)[old_size + output] =
+                    (*conversion_failure_null_map)[old_size + source];
+        }
+    }
+    DORIS_CHECK_EQ(source, 0);
+}
+
+Status decode_selected_nullable_values(IColumn& column, const DataTypeSerDe& 
serde,
+                                       Decoder& decoder, const 
ParquetDecodeContext& context,
+                                       ParquetMaterializationState& state,
+                                       ColumnSelectVector& select_vector, 
NullMap& selected_nulls,
+                                       int64_t* materialization_time) {
+    auto& selection = state.selection;
+    selection.ranges.clear();
+    selection.total_values = 0;
+    selection.selected_values = 0;
+    selected_nulls.clear();
+    selected_nulls.reserve(select_vector.num_values() - 
select_vector.num_filtered());
+
+    size_t physical_cursor = 0;
+    ColumnSelectVector::DataReadType read_type;
+    while (const size_t run_length = 
select_vector.get_next_run<true>(&read_type)) {
+        switch (read_type) {
+        case ColumnSelectVector::CONTENT:
+            if (!selection.ranges.empty() &&
+                selection.ranges.back().first + selection.ranges.back().count 
== physical_cursor) {
+                selection.ranges.back().count += run_length;
+            } else {
+                selection.ranges.push_back({.first = physical_cursor, .count = 
run_length});
+            }
+            selection.selected_values += run_length;
+            selected_nulls.resize_fill(selected_nulls.size() + run_length, 0);
+            physical_cursor += run_length;
+            break;
+        case ColumnSelectVector::NULL_DATA:
+            selected_nulls.resize_fill(selected_nulls.size() + run_length, 1);
+            break;
+        case ColumnSelectVector::FILTERED_CONTENT:
+            physical_cursor += run_length;
+            break;
+        case ColumnSelectVector::FILTERED_NULL:
+            break;
+        }
+    }
+    selection.total_values = physical_cursor;
+    DORIS_CHECK_EQ(selection.total_values, select_vector.num_values() - 
select_vector.num_nulls());
+    DORIS_CHECK_EQ(selected_nulls.size(),
+                   select_vector.num_values() - select_vector.num_filtered());
+
+    const size_t old_size = column.size();
+    SCOPED_RAW_TIMER(materialization_time);
+    if (selection.selected_values == 0) {
+        RETURN_IF_ERROR(decoder.skip_values(selection.total_values));
+        column.insert_many_defaults(selected_nulls.size());
+        return Status::OK();
+    }
+
+    if (state.conversion_failure_null_map != nullptr) {
+        DORIS_CHECK(state.conversion_failure_null_map->size() >= old_size + 
selected_nulls.size());
+        memset(state.conversion_failure_null_map->data() + old_size, 0, 
selection.selected_values);
+    }
+    SelectedDecodeSource selected_source(decoder, selection);
+    const auto status = read_native_or_serde(column, serde, selected_source, 
context,
+                                             selection.selected_values, state);
+    if (!status.ok()) {
+        if (state.conversion_failure_null_map != nullptr) {
+            memcpy(state.conversion_failure_null_map->data() + old_size, 
selected_nulls.data(),
+                   selected_nulls.size());
+        }
+        return status;
+    }
+    DORIS_CHECK_EQ(column.size(), old_size + selection.selected_values);
+
+    remap_nullable_conversion_failures(state.conversion_failure_null_map, 
old_size,
+                                       selection.selected_values, 
selected_nulls);
+    const bool expanded = visit_nullable_expandable_column(column, [&](auto& 
typed_column) {
+        // Decode into the final nested column compactly, then expand in 
place. This preserves the
+        // V2 no-intermediate-column invariant while restoring the nullable 
sparse row layout.
+        expand_nullable_values(typed_column, old_size, 
selection.selected_values, selected_nulls);
+    });
+    DORIS_CHECK(expanded);
+    return Status::OK();
+}
+
+class PlainPredicateConsumer final : public ParquetFixedValueConsumer {
+public:
+    PlainPredicateConsumer(const VExprSPtrs& conjuncts, DataTypePtr data_type, 
int column_id,
+                           IColumn::Filter* matches)
+            : _conjuncts(conjuncts),
+              _data_type(std::move(data_type)),
+              _column_id(column_id),
+              _matches(matches) {
+        DORIS_CHECK(_matches != nullptr);
+    }
+
+    Status consume(const uint8_t* values, size_t num_values, size_t 
value_width) override {
+        const size_t old_size = _matches->size();
+        _matches->resize_fill(old_size + num_values, 1);
+        for (const auto& conjunct : _conjuncts) {
+            RETURN_IF_ERROR(conjunct->execute_on_raw_fixed_values(values, 
num_values, value_width,
+                                                                  _data_type, 
_column_id,
+                                                                  
_matches->data() + old_size));
+        }
+        return Status::OK();
+    }
+
+private:
+    const VExprSPtrs& _conjuncts;
+    DataTypePtr _data_type;
+    int _column_id;
+    IColumn::Filter* _matches;
+};
+
+} // namespace
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::ColumnChunkReader(
+        io::BufferedStreamReader* reader, tparquet::ColumnChunk* column_chunk,
+        NativeFieldSchema* field_schema, const tparquet::OffsetIndex* 
offset_index,
+        size_t total_rows, io::IOContext* io_ctx, const 
ParquetPageReadContext& page_read_ctx,
+        const ColumnChunkRange* chunk_range)
+        : _field_schema(field_schema),
+          _max_rep_level(field_schema->repetition_level),
+          _max_def_level(field_schema->definition_level),
+          _stream_reader(reader),
+          _metadata(column_chunk->meta_data),
+          _offset_index(offset_index),
+          _total_rows(total_rows),
+          _io_ctx(io_ctx),
+          _page_read_ctx(page_read_ctx) {
+    if (chunk_range != nullptr) {
+        _chunk_range = *chunk_range;
+        _has_validated_chunk_range = true;
+    }
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::init() {
+    size_t start_offset = _has_validated_chunk_range
+                                  ? _chunk_range.offset
+                                  : (has_dict_page(_metadata) ? 
_metadata.dictionary_page_offset
+                                                              : 
_metadata.data_page_offset);
+    size_t chunk_size =
+            _has_validated_chunk_range ? _chunk_range.length : 
_metadata.total_compressed_size;
+    // create page reader
+    _page_reader = create_page_reader<IN_COLLECTION, OFFSET_INDEX>(
+            _stream_reader, _io_ctx, start_offset, chunk_size, _total_rows, 
_metadata,
+            _page_read_ctx, _offset_index);
+    // get the block compression codec
+    RETURN_IF_ERROR(get_block_compression_codec(_metadata.codec, 
&_block_compress_codec));
+    _state = INITIALIZED;
+    RETURN_IF_ERROR(_parse_first_page_header());
+    return Status::OK();
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::skip_nested_values(
+        const std::vector<level_t>& def_levels, size_t start_index) {
+    size_t no_value_cnt = 0;
+    size_t value_cnt = 0;
+
+    DORIS_CHECK(start_index <= def_levels.size());
+    for (size_t idx = start_index; idx < def_levels.size(); idx++) {
+        level_t def_level = def_levels[idx];
+        if (IN_COLLECTION && def_level < 
_field_schema->repeated_parent_def_level) {
+            no_value_cnt++;
+        } else if (def_level < _field_schema->definition_level) {
+            no_value_cnt++;
+        } else {
+            value_cnt++;
+        }
+    }
+
+    RETURN_IF_ERROR(skip_values(value_cnt, true));
+    RETURN_IF_ERROR(skip_values(no_value_cnt, false));
+    return Status::OK();
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::read_levels(
+        size_t num_values, std::vector<level_t>* rep_levels, 
std::vector<level_t>* def_levels) {
+    DORIS_CHECK(rep_levels != nullptr);
+    DORIS_CHECK(def_levels != nullptr);
+    if (_remaining_num_values < num_values || _remaining_rep_nums < num_values 
||
+        _remaining_def_nums < num_values) {
+        return Status::Corruption(
+                "Parquet level reader requested {} slots with only {}/{}/{} 
remaining", num_values,
+                _remaining_num_values, _remaining_rep_nums, 
_remaining_def_nums);
+    }
+
+    const size_t start_index = def_levels->size();
+    rep_levels->resize(rep_levels->size() + num_values, 0);
+    def_levels->resize(def_levels->size() + num_values, 0);
+    if (_max_rep_level > 0) {
+        const size_t decoded = _rep_level_decoder.get_levels(
+                rep_levels->data() + rep_levels->size() - num_values, 
num_values);
+        if (decoded != num_values) {
+            return Status::Corruption("Parquet repetition level stream ended 
after {} of {} slots",
+                                      decoded, num_values);
+        }
+    }
+    if (_max_def_level > 0) {
+        const size_t decoded = _def_level_decoder.get_levels(
+                def_levels->data() + def_levels->size() - num_values, 
num_values);
+        if (decoded != num_values) {
+            return Status::Corruption("Parquet definition level stream ended 
after {} of {} slots",
+                                      decoded, num_values);
+        }
+    }
+    _remaining_rep_nums -= num_values;
+    _remaining_def_nums -= num_values;
+    return skip_nested_values(*def_levels, start_index);
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+Status ColumnChunkReader<IN_COLLECTION, 
OFFSET_INDEX>::_parse_first_page_header() {
+    while (true) {
+        RETURN_IF_ERROR(_page_reader->parse_page_header());
+        const tparquet::PageHeader* header = nullptr;
+        RETURN_IF_ERROR(_page_reader->get_page_header(&header));
+        if (header->type == tparquet::PageType::DATA_PAGE ||
+            header->type == tparquet::PageType::DATA_PAGE_V2) {
+            _state = INITIALIZED;
+            return parse_page_header();
+        }
+        if (header->type != tparquet::PageType::DICTIONARY_PAGE) {
+            RETURN_IF_ERROR(_page_reader->skip_auxiliary_page());
+            _state = INITIALIZED;
+            continue;
+        }
+        // the first page maybe directory page even if 
_metadata.__isset.dictionary_page_offset == false,
+        // so we should parse the directory page in next_page()
+        RETURN_IF_ERROR(_decode_dict_page());
+        // parse the real first data page
+        RETURN_IF_ERROR(_page_reader->dict_next_page());
+        _state = INITIALIZED;
+        // A dictionary is the only non-data page with decoder state. Any 
following index or
+        // extension pages are skipped by the same pre-data loop.
+    }
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::parse_page_header() {
+    if (_state == HEADER_PARSED || _state == DATA_LOADED) {
+        return Status::OK();
+    }
+    const tparquet::PageHeader* header = nullptr;
+    while (true) {
+        RETURN_IF_ERROR(_page_reader->parse_page_header());
+        RETURN_IF_ERROR(_page_reader->get_page_header(&header));
+        if (header->type == tparquet::PageType::DATA_PAGE ||
+            header->type == tparquet::PageType::DATA_PAGE_V2) {
+            break;
+        }
+        if (header->type == tparquet::PageType::DICTIONARY_PAGE) {
+            return Status::Corruption("Parquet dictionary page appears after 
data pages");
+        }
+        RETURN_IF_ERROR(_page_reader->skip_auxiliary_page());
+    }
+    int32_t page_num_values = _page_reader->is_header_v2() ? 
header->data_page_header_v2.num_values
+                                                           : 
header->data_page_header.num_values;
+    if (page_num_values < 0 || page_num_values > _metadata.num_values ||
+        (!OFFSET_INDEX &&
+         static_cast<uint64_t>(page_num_values) >
+                 static_cast<uint64_t>(_metadata.num_values) - 
_chunk_parsed_values)) {
+        // Page counts are untrusted and feed both level decoders and scratch 
sizing. Bound each
+        // page by the column metadata before converting to unsigned counters.
+        return Status::Corruption("Parquet data page value count {} exceeds 
column total {}",
+                                  page_num_values, _metadata.num_values);
+    }
+    if constexpr (!IN_COLLECTION) {
+        const size_t page_start_row = _page_reader->start_row();
+        const size_t page_end_row = _page_reader->end_row();
+        if (UNLIKELY(page_end_row < page_start_row ||
+                     static_cast<size_t>(page_num_values) != page_end_row - 
page_start_row)) {
+            // Flat columns have exactly one physical value slot per logical 
row. Rejecting a
+            // divergent header/OffsetIndex span prevents every later page 
from shifting rows.
+            return Status::Corruption(
+                    "Parquet flat data page has {} values for logical row 
range [{}, {})",
+                    page_num_values, page_start_row, page_end_row);
+        }
+    }
+    _remaining_rep_nums = page_num_values;
+    _remaining_def_nums = page_num_values;
+    _remaining_num_values = page_num_values;
+
+    // no offset will parse all header.
+    if constexpr (OFFSET_INDEX == false) {
+        _chunk_parsed_values += _remaining_num_values;
+    }
+    _state = HEADER_PARSED;
+    return Status::OK();
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::next_page() {
+    _state = INITIALIZED;
+    RETURN_IF_ERROR(_page_reader->next_page());
+    return Status::OK();
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+Status ColumnChunkReader<IN_COLLECTION, 
OFFSET_INDEX>::_get_uncompressed_levels(
+        const tparquet::DataPageHeaderV2& page_v2, Slice& page_data) {
+    const size_t rl = page_v2.repetition_levels_byte_length;
+    const size_t dl = page_v2.definition_levels_byte_length;
+    if (UNLIKELY(rl > page_data.size || dl > page_data.size - rl)) {
+        // Validate the physical slice again because a cached entry may itself 
be truncated.
+        return Status::Corruption("Parquet data page v2 level bytes exceed 
available payload");
+    }
+    _v2_rep_levels = Slice(page_data.data, rl);
+    _v2_def_levels = Slice(page_data.data + rl, dl);
+    page_data.data += dl + rl;
+    page_data.size -= dl + rl;
+    return Status::OK();
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::load_page_data() {
+    if (_state == DATA_LOADED) {
+        return Status::OK();
+    }
+    if (UNLIKELY(_state != HEADER_PARSED)) {
+        return Status::Corruption("Should parse page header");
+    }
+
+    const tparquet::PageHeader* header = nullptr;
+    RETURN_IF_ERROR(_page_reader->get_page_header(&header));
+    RETURN_IF_ERROR(validate_uncompressed_page_sizes(
+            *header, _metadata.codec, 
_page_read_ctx.data_page_v2_always_compressed));
+    int32_t uncompressed_size = header->uncompressed_page_size;
+    bool page_loaded = false;
+
+    // First, try to reuse a cache handle previously discovered by PageReader
+    // (header-only lookup) to avoid a second lookup here.
+    if (_page_read_ctx.enable_parquet_file_page_cache && 
!config::disable_storage_page_cache &&
+        StoragePageCache::instance() != nullptr) {
+        if (_page_reader->has_page_cache_handle()) {
+            const PageCacheHandle& handle = _page_reader->page_cache_handle();
+            Slice cached = handle.data();
+            size_t header_size = _page_reader->header_bytes().size();
+            size_t levels_size = 0;
+            if (header->__isset.data_page_header_v2) {
+                const tparquet::DataPageHeaderV2& header_v2 = 
header->data_page_header_v2;
+                size_t rl = header_v2.repetition_levels_byte_length;
+                size_t dl = header_v2.definition_levels_byte_length;
+                levels_size = rl + dl;
+                if (UNLIKELY(header_size > cached.size ||
+                             levels_size > cached.size - header_size)) {
+                    return Status::Corruption("Cached Parquet page is shorter 
than its v2 levels");
+                }
+                _v2_rep_levels =
+                        Slice(reinterpret_cast<const uint8_t*>(cached.data) + 
header_size, rl);
+                _v2_def_levels =
+                        Slice(reinterpret_cast<const uint8_t*>(cached.data) + 
header_size + rl, dl);
+            }
+            // payload_slice points to the bytes after header and levels
+            if (UNLIKELY(header_size + levels_size > cached.size)) {
+                return Status::Corruption("Cached Parquet page is shorter than 
its header");
+            }
+            Slice payload_slice(cached.data + header_size + levels_size,
+                                cached.size - header_size - levels_size);
+
+            bool cache_payload_is_decompressed = 
_page_reader->is_cache_payload_decompressed();
+            const size_t expected_payload_size =
+                    cache_payload_is_decompressed
+                            ? 
static_cast<size_t>(header->uncompressed_page_size) - levels_size
+                            : 
static_cast<size_t>(header->compressed_page_size) - levels_size;
+            if (UNLIKELY(payload_slice.size != expected_payload_size)) {
+                return Status::Corruption("Cached Parquet page payload has 
size {}, expected {}",
+                                          payload_slice.size, 
expected_payload_size);
+            }
+
+            if (cache_payload_is_decompressed) {
+                // Cached payload is already uncompressed
+                _page_data = payload_slice;
+            } else {
+                CHECK(_block_compress_codec);
+                // Decompress cached payload into _decompress_buf for decoding
+                size_t uncompressed_payload_size =
+                        header->__isset.data_page_header_v2
+                                ? 
static_cast<size_t>(header->uncompressed_page_size) - levels_size
+                                : 
static_cast<size_t>(header->uncompressed_page_size);
+                _reserve_decompress_buf(uncompressed_payload_size);
+                _page_data = Slice(_decompress_buf.get(), 
uncompressed_payload_size);
+                SCOPED_RAW_TIMER(&_chunk_statistics.decompress_time);
+                _chunk_statistics.decompress_cnt++;
+                
RETURN_IF_ERROR(_block_compress_codec->decompress(payload_slice, &_page_data));
+                if (UNLIKELY(_page_data.size != uncompressed_payload_size)) {
+                    return Status::Corruption("Parquet page decompressed to {} 
bytes, expected {}",
+                                              _page_data.size, 
uncompressed_payload_size);
+                }
+            }
+            // page cache counters were incremented when PageReader did the 
header-only
+            // cache lookup. Do not increment again to avoid double-counting.
+            page_loaded = true;
+        }
+    }
+
+    if (!page_loaded) {
+        const bool prepare_cache_payload = can_prepare_page_cache_payload(
+                _page_read_ctx.enable_parquet_file_page_cache, 
config::disable_storage_page_cache,
+                StoragePageCache::instance() != nullptr, 
!_page_reader->header_bytes().empty());
+        if (_block_compress_codec != nullptr) {
+            Slice compressed_data;
+            RETURN_IF_ERROR(_page_reader->get_page_data(compressed_data));
+            std::vector<uint8_t> level_bytes;
+            if (header->__isset.data_page_header_v2) {
+                const tparquet::DataPageHeaderV2& header_v2 = 
header->data_page_header_v2;
+                // uncompressed_size = rl + dl + uncompressed_data_size
+                // compressed_size = rl + dl + compressed_data_size
+                uncompressed_size -= header_v2.repetition_levels_byte_length +
+                                     header_v2.definition_levels_byte_length;
+                // copy level bytes (rl + dl) so that we can cache header + 
levels + uncompressed payload
+                size_t rl = header_v2.repetition_levels_byte_length;
+                size_t dl = header_v2.definition_levels_byte_length;
+                size_t level_sz = rl + dl;
+                if (prepare_cache_payload && level_sz > 0) {
+                    level_bytes.resize(level_sz);
+                    memcpy(level_bytes.data(), compressed_data.data, level_sz);
+                }
+                // now remove levels from compressed_data for decompression
+                RETURN_IF_ERROR(_get_uncompressed_levels(header_v2, 
compressed_data));
+            }
+            bool is_v2_compressed = header->__isset.data_page_header_v2 &&
+                                    (header->data_page_header_v2.is_compressed 
||
+                                     
_page_read_ctx.data_page_v2_always_compressed);
+            bool page_has_compression = header->__isset.data_page_header || 
is_v2_compressed;
+
+            if (page_has_compression) {
+                // Decompress payload for immediate decoding
+                _reserve_decompress_buf(uncompressed_size);
+                _page_data = Slice(_decompress_buf.get(), uncompressed_size);
+                SCOPED_RAW_TIMER(&_chunk_statistics.decompress_time);
+                _chunk_statistics.decompress_cnt++;
+                
RETURN_IF_ERROR(_block_compress_codec->decompress(compressed_data, 
&_page_data));
+                if (UNLIKELY(_page_data.size != 
static_cast<size_t>(uncompressed_size))) {
+                    return Status::Corruption("Parquet page decompressed to {} 
bytes, expected {}",
+                                              _page_data.size, 
uncompressed_size);
+                }
+
+                // Decide whether to cache decompressed payload or compressed 
payload based on threshold
+                bool cache_payload_decompressed = should_cache_decompressed(
+                        header, _metadata, 
_page_read_ctx.data_page_v2_always_compressed);
+
+                if (prepare_cache_payload) {
+                    if (cache_payload_decompressed) {
+                        _insert_page_into_cache(level_bytes, _page_data);
+                        
_chunk_statistics.page_cache_decompressed_write_counter += 1;
+                    } else {
+                        if (config::enable_parquet_cache_compressed_pages) {
+                            // cache the compressed payload as-is (header | 
levels | compressed_payload)
+                            _insert_page_into_cache(
+                                    level_bytes, Slice(compressed_data.data, 
compressed_data.size));
+                            
_chunk_statistics.page_cache_compressed_write_counter += 1;
+                        }
+                    }
+                }
+            } else {
+                // no compression on this page, use the data directly
+                _page_data = Slice(compressed_data.data, compressed_data.size);
+                if (prepare_cache_payload) {
+                    _insert_page_into_cache(level_bytes, _page_data);
+                    _chunk_statistics.page_cache_decompressed_write_counter += 
1;
+                }
+            }
+        } else {
+            // For uncompressed page, we may still need to extract v2 levels
+            std::vector<uint8_t> level_bytes;
+            Slice uncompressed_data;
+            RETURN_IF_ERROR(_page_reader->get_page_data(uncompressed_data));
+            if (header->__isset.data_page_header_v2) {
+                const tparquet::DataPageHeaderV2& header_v2 = 
header->data_page_header_v2;
+                size_t rl = header_v2.repetition_levels_byte_length;
+                size_t dl = header_v2.definition_levels_byte_length;
+                size_t level_sz = rl + dl;
+                if (prepare_cache_payload && level_sz > 0) {
+                    level_bytes.resize(level_sz);
+                    memcpy(level_bytes.data(), uncompressed_data.data, 
level_sz);
+                }
+                RETURN_IF_ERROR(_get_uncompressed_levels(header_v2, 
uncompressed_data));
+            }
+            // copy page data out
+            _page_data = Slice(uncompressed_data.data, uncompressed_data.size);
+            // Optionally cache uncompressed data for uncompressed pages
+            if (prepare_cache_payload) {
+                _insert_page_into_cache(level_bytes, _page_data);
+                _chunk_statistics.page_cache_decompressed_write_counter += 1;
+            }
+        }
+    }
+
+    // Initialize repetition level and definition level. Skip when level = 0, 
which means required field.
+    if (_max_rep_level > 0) {
+        SCOPED_RAW_TIMER(&_chunk_statistics.decode_level_time);
+        if (header->__isset.data_page_header_v2) {
+            RETURN_IF_ERROR(_rep_level_decoder.init_v2(_v2_rep_levels, 
_max_rep_level,
+                                                       _remaining_rep_nums));
+        } else {
+            RETURN_IF_ERROR(_rep_level_decoder.init(
+                    &_page_data, 
header->data_page_header.repetition_level_encoding, _max_rep_level,
+                    _remaining_rep_nums));
+        }
+    }
+    if (_max_def_level > 0) {
+        SCOPED_RAW_TIMER(&_chunk_statistics.decode_level_time);
+        if (header->__isset.data_page_header_v2) {
+            RETURN_IF_ERROR(_def_level_decoder.init_v2(_v2_def_levels, 
_max_def_level,
+                                                       _remaining_def_nums));
+        } else {
+            RETURN_IF_ERROR(_def_level_decoder.init(
+                    &_page_data, 
header->data_page_header.definition_level_encoding, _max_def_level,
+                    _remaining_def_nums));
+        }
+    }
+    auto encoding = header->__isset.data_page_header_v2 ? 
header->data_page_header_v2.encoding
+                                                        : 
header->data_page_header.encoding;
+    // change the deprecated encoding to RLE_DICTIONARY
+    if (encoding == tparquet::Encoding::PLAIN_DICTIONARY) {
+        encoding = tparquet::Encoding::RLE_DICTIONARY;
+    }
+    _current_encoding = encoding;
+
+    // Reuse page decoder
+    if (_decoders.find(static_cast<int>(encoding)) != _decoders.end()) {
+        _page_decoder = _decoders[static_cast<int>(encoding)].get();
+    } else {
+        std::unique_ptr<Decoder> page_decoder;
+        RETURN_IF_ERROR(Decoder::get_decoder(_metadata.type, encoding, 
page_decoder));
+        // Set type length
+        page_decoder->set_type_length(_get_type_length());
+        _decoders[static_cast<int>(encoding)] = std::move(page_decoder);
+        _page_decoder = _decoders[static_cast<int>(encoding)].get();
+    }
+    // Encoding headers cannot legitimately advertise more physical values 
than the data page's
+    // logical value count; establish the bound before decoders inspect 
external counts.
+    _page_decoder->set_expected_values(_remaining_num_values);
+    RETURN_IF_ERROR(_page_decoder->set_data(&_page_data));
+
+    _state = DATA_LOADED;
+    return Status::OK();
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::_decode_dict_page() {
+    const tparquet::PageHeader* header = nullptr;
+    RETURN_IF_ERROR(_page_reader->get_page_header(&header));
+    DCHECK_EQ(tparquet::PageType::DICTIONARY_PAGE, header->type);
+    SCOPED_RAW_TIMER(&_chunk_statistics.decode_dict_time);
+
+    // Using the PLAIN_DICTIONARY enum value is deprecated in the Parquet 2.0 
specification.
+    // Prefer using RLE_DICTIONARY in a data page and PLAIN in a dictionary 
page for Parquet 2.0+ files.
+    // refer: https://github.com/apache/parquet-format/blob/master/Encodings.md
+    tparquet::Encoding::type dict_encoding = 
header->dictionary_page_header.encoding;
+    if (dict_encoding != tparquet::Encoding::PLAIN_DICTIONARY &&
+        dict_encoding != tparquet::Encoding::PLAIN) {
+        return Status::InternalError("Unsupported dictionary encoding {}",
+                                     tparquet::to_string(dict_encoding));
+    }
+
+    // Prepare dictionary data
+    int32_t uncompressed_size = header->uncompressed_page_size;
+    RETURN_IF_ERROR(validate_uncompressed_page_sizes(
+            *header, _metadata.codec, 
_page_read_ctx.data_page_v2_always_compressed));
+    auto dict_data = make_unique_buffer<uint8_t>(uncompressed_size);
+    bool dict_loaded = false;
+
+    // Try to load dictionary page from cache
+    if (_page_read_ctx.enable_parquet_file_page_cache && 
!config::disable_storage_page_cache &&
+        StoragePageCache::instance() != nullptr) {
+        if (_page_reader->has_page_cache_handle()) {
+            const PageCacheHandle& handle = _page_reader->page_cache_handle();
+            Slice cached = handle.data();
+            size_t header_size = _page_reader->header_bytes().size();
+            // Dictionary page layout in cache: header | payload (compressed 
or uncompressed)
+            Slice payload_slice(cached.data + header_size, cached.size - 
header_size);
+
+            bool cache_payload_is_decompressed = 
_page_reader->is_cache_payload_decompressed();
+
+            if (cache_payload_is_decompressed) {
+                // Use cached decompressed dictionary data
+                if (UNLIKELY(payload_slice.size != 
static_cast<size_t>(uncompressed_size))) {
+                    return Status::Corruption(
+                            "Cached Parquet dictionary payload has size {}, 
expected {}",
+                            payload_slice.size, uncompressed_size);
+                }
+                memcpy(dict_data.get(), payload_slice.data, 
payload_slice.size);
+                dict_loaded = true;
+            } else {
+                CHECK(_block_compress_codec);
+                // Decompress cached compressed dictionary data
+                Slice dict_slice(dict_data.get(), uncompressed_size);
+                {
+                    SCOPED_RAW_TIMER(&_chunk_statistics.decompress_time);
+                    ++_chunk_statistics.decompress_cnt;
+                    
RETURN_IF_ERROR(_block_compress_codec->decompress(payload_slice, &dict_slice));
+                }
+                if (UNLIKELY(dict_slice.size != 
static_cast<size_t>(uncompressed_size))) {
+                    return Status::Corruption(
+                            "Parquet dictionary decompressed to {} bytes, 
expected {}",
+                            dict_slice.size, uncompressed_size);
+                }
+                dict_loaded = true;
+            }
+
+            // When dictionary page is loaded from cache, we need to skip the 
page data
+            // to update the offset correctly (similar to calling 
get_page_data())
+            if (dict_loaded) {
+                _page_reader->skip_page_data();
+            }
+        }
+    }
+
+    if (!dict_loaded) {
+        // Load and decompress dictionary page from file
+        if (_block_compress_codec != nullptr) {
+            auto dict_num = header->dictionary_page_header.num_values;
+            if (dict_num == 0 && uncompressed_size != 0) {
+                return Status::IOError(
+                        "Dictionary page's num_values is {} but 
uncompressed_size is {}", dict_num,
+                        uncompressed_size);
+            }
+            Slice compressed_data;
+            Slice dict_slice(dict_data.get(), uncompressed_size);
+            if (dict_num != 0) {
+                RETURN_IF_ERROR(_page_reader->get_page_data(compressed_data));
+                // Dictionary probes stop before data pages, so count their 
decompression here or
+                // metadata pruning profiles will report no codec work for the 
scan.
+                {
+                    SCOPED_RAW_TIMER(&_chunk_statistics.decompress_time);
+                    ++_chunk_statistics.decompress_cnt;
+                    RETURN_IF_ERROR(
+                            _block_compress_codec->decompress(compressed_data, 
&dict_slice));
+                }
+                if (UNLIKELY(dict_slice.size != 
static_cast<size_t>(uncompressed_size))) {
+                    return Status::Corruption(
+                            "Parquet dictionary decompressed to {} bytes, 
expected {}",
+                            dict_slice.size, uncompressed_size);
+                }
+            }
+
+            // Decide whether to cache decompressed or compressed dictionary 
based on threshold
+            // If uncompressed_page_size == 0, should_cache_decompressed will 
return true
+            bool cache_payload_decompressed = should_cache_decompressed(
+                    header, _metadata, 
_page_read_ctx.data_page_v2_always_compressed);
+
+            if (_page_read_ctx.enable_parquet_file_page_cache &&
+                !config::disable_storage_page_cache && 
StoragePageCache::instance() != nullptr &&
+                !_page_reader->header_bytes().empty()) {
+                std::vector<uint8_t> empty_levels; // Dictionary pages don't 
have levels
+                if (cache_payload_decompressed) {
+                    // Cache the decompressed dictionary page
+                    // If dict_num == 0, `dict_slice` will be empty
+                    _insert_page_into_cache(empty_levels, dict_slice);
+                    _chunk_statistics.page_cache_decompressed_write_counter += 
1;
+                } else {
+                    if (config::enable_parquet_cache_compressed_pages) {
+                        DCHECK(!compressed_data.empty());
+                        // Cache the compressed dictionary page
+                        _insert_page_into_cache(empty_levels,
+                                                Slice(compressed_data.data, 
compressed_data.size));
+                        _chunk_statistics.page_cache_compressed_write_counter 
+= 1;
+                    }
+                }
+            }
+            // `get_page_data` not called, we should skip the page data
+            // Because `_insert_page_into_cache` will use _page_reader, we 
should exec `skip_page_data` after `_insert_page_into_cache`
+            if (dict_num == 0) {
+                _page_reader->skip_page_data();
+            }
+        } else {
+            Slice dict_slice;
+            RETURN_IF_ERROR(_page_reader->get_page_data(dict_slice));
+            // The data is stored by BufferedStreamReader, we should copy it 
out
+            memcpy(dict_data.get(), dict_slice.data, dict_slice.size);
+
+            // Cache the uncompressed dictionary page
+            if (_page_read_ctx.enable_parquet_file_page_cache &&
+                !config::disable_storage_page_cache && 
StoragePageCache::instance() != nullptr &&
+                !_page_reader->header_bytes().empty()) {
+                std::vector<uint8_t> empty_levels;
+                Slice payload(dict_data.get(), uncompressed_size);
+                _insert_page_into_cache(empty_levels, payload);
+                _chunk_statistics.page_cache_decompressed_write_counter += 1;
+            }
+        }
+    }
+
+    // Cache page decoder
+    std::unique_ptr<Decoder> page_decoder;
+    RETURN_IF_ERROR(
+            Decoder::get_decoder(_metadata.type, 
tparquet::Encoding::RLE_DICTIONARY, page_decoder));
+    // Set type length
+    page_decoder->set_type_length(_get_type_length());
+    // Set the dictionary data
+    RETURN_IF_ERROR(page_decoder->set_dict(dict_data, uncompressed_size,
+                                           
header->dictionary_page_header.num_values));
+    _decoders[static_cast<int>(tparquet::Encoding::RLE_DICTIONARY)] = 
std::move(page_decoder);
+
+    _has_dict = true;
+    return Status::OK();
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+void ColumnChunkReader<IN_COLLECTION, 
OFFSET_INDEX>::_reserve_decompress_buf(size_t size) {
+    if (size > _decompress_buf_size) {
+        _decompress_buf_size = BitUtil::next_power_of_two(size);
+        _decompress_buf = make_unique_buffer<uint8_t>(_decompress_buf_size);
+    }
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+void ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::_insert_page_into_cache(
+        const std::vector<uint8_t>& level_bytes, const Slice& payload) {
+    StoragePageCache::CacheKey key =
+            
_page_reader->make_page_cache_key(_page_reader->header_start_offset());
+    const std::vector<uint8_t>& header_bytes = _page_reader->header_bytes();
+    size_t total = header_bytes.size() + level_bytes.size() + payload.size;
+    auto page = std::make_unique<DataPage>(total, true, segment_v2::DATA_PAGE);
+    size_t pos = 0;
+    memcpy(page->data() + pos, header_bytes.data(), header_bytes.size());
+    pos += header_bytes.size();
+    if (!level_bytes.empty()) {
+        memcpy(page->data() + pos, level_bytes.data(), level_bytes.size());
+        pos += level_bytes.size();
+    }
+    if (payload.size > 0) {
+        memcpy(page->data() + pos, payload.data, payload.size);
+        pos += payload.size;
+    }
+    page->reset_size(total);
+    PageCacheHandle handle;
+    StoragePageCache::instance()->insert(key, page.get(), &handle, 
segment_v2::DATA_PAGE);
+    page.release();
+    _chunk_statistics.page_cache_write_counter += 1;
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::skip_values(size_t 
num_values,
+                                                                   bool 
skip_data) {
+    if (UNLIKELY(_remaining_num_values < num_values)) {
+        return Status::IOError("Skip too many values in current page. {} vs. 
{}",
+                               _remaining_num_values, num_values);
+    }
+    if (skip_data) {
+        SCOPED_RAW_TIMER(&_chunk_statistics.decode_value_time);
+        RETURN_IF_ERROR(_page_decoder->skip_values(num_values));
+    }
+    // Commit logical page progress only after the physical decoder accepted 
the whole request.
+    _remaining_num_values -= num_values;
+    return Status::OK();
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::materialize_values(
+        MutableColumnPtr& doris_column, const DataTypeSerDe& serde, 
ParquetDecodeContext& context,
+        ParquetMaterializationState& state, ColumnSelectVector& select_vector) 
{
+    if (select_vector.num_values() == 0) {
+        return Status::OK();
+    }
+    SCOPED_RAW_TIMER(&_chunk_statistics.decode_value_time);
+    if (UNLIKELY((doris_column->is_column_dictionary() || 
context.dictionary_index_only) &&
+                 !_has_dict)) {
+        return Status::IOError("Not dictionary coded");
+    }
+    if (UNLIKELY(_remaining_num_values < select_vector.num_values())) {
+        return Status::IOError("Decode too many values in current page");
+    }
+    RETURN_IF_ERROR(translate_value_encoding(_current_encoding, 
&context.encoding));
+    Status status;
+    if (select_vector.has_filter()) {
+        if (select_vector.num_nulls() == 0) {
+            ++_chunk_statistics.hybrid_selection_batches;
+            status = decode_selected_non_null_values(*doris_column, serde, 
*_page_decoder, context,
+                                                     state, select_vector,
+                                                     
&_chunk_statistics.materialization_time);
+            _chunk_statistics.hybrid_selection_ranges += 
state.selection.ranges.size();
+        } else if (visit_nullable_expandable_column(*doris_column, [](auto&) 
{})) {
+            // Parquet omits NULL leaf values from the physical stream. For 
example, the logical
+            // DATE sequence [d0, NULL, d1, NULL, d2] is physically encoded as 
[d0, d1, d2]. The
+            // generic fallback follows the logical selection runs, so those 
NULLs split one
+            // contiguous physical span into decode(d0), insert-NULL, 
decode(d1), insert-NULL,
+            // decode(d2). On nullable sparse scans this repeatedly enters 
SerDe and turns decimal
+            // scaling, date conversion, timestamp/timezone conversion, or 
dictionary-ID
+            // materialization into many tiny calls even though the physical 
values are adjacent.
+            //
+            // Build selected non-NULL ranges in physical coordinates instead. 
In the example this
+            // decodes [d0, d1, d2] compactly with one SerDe consumer, records 
[0, 1, 0, 1, 0] as
+            // the logical NULL layout, and expands the final nested column 
backwards so unread
+            // compact values cannot be overwritten. Apply this to every 
expandable V2 scalar,
+            // including ordinary PLAIN primitives: otherwise nullable numeric 
predicates still
+            // fragment a physical span into millions of SerDe calls and 
defeat sparse decoding.
+            ++_chunk_statistics.hybrid_selection_batches;
+            status = decode_selected_nullable_values(
+                    *doris_column, serde, *_page_decoder, context, state, 
select_vector,
+                    _nullable_selection_nulls, 
&_chunk_statistics.materialization_time);
+            _chunk_statistics.hybrid_selection_ranges += 
state.selection.ranges.size();
+        } else {
+            ++_chunk_statistics.hybrid_selection_null_fallback_batches;
+            status = decode_selected_values<true>(*doris_column, serde, 
*_page_decoder, context,
+                                                  state, select_vector,
+                                                  
&_chunk_statistics.materialization_time);
+        }
+    } else {
+        status = decode_selected_values<false>(*doris_column, serde, 
*_page_decoder, context, state,
+                                               select_vector,
+                                               
&_chunk_statistics.materialization_time);
+    }
+    RETURN_IF_ERROR(status);
+    _remaining_num_values -= select_vector.num_values();
+    return Status::OK();
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+bool ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::can_filter_plain_values(
+        const VExprSPtrs& conjuncts, int column_id) const {
+    if (conjuncts.empty() || _current_encoding != tparquet::Encoding::PLAIN ||
+        (_metadata.type != tparquet::Type::INT32 && _metadata.type != 
tparquet::Type::INT64 &&
+         _metadata.type != tparquet::Type::FLOAT && _metadata.type != 
tparquet::Type::DOUBLE)) {
+        return false;
+    }
+    return std::ranges::all_of(conjuncts, [&](const auto& conjunct) {
+        return conjunct != nullptr &&
+               
conjunct->can_execute_on_raw_fixed_values(_field_schema->data_type, column_id);

Review Comment:
   [P1] Exclude logical UINT32 from this raw PLAIN gate unless its physical 
carrier is normalized. A valid Parquet `INT(32,false)` leaf is physical INT32 
but is exposed as Doris BIGINT, so this check accepts a BIGINT predicate and 
the PLAIN decoder passes four-byte values to `execute_on_raw_fixed_values()`, 
which expects eight bytes and returns `Corruption` before fallback. Please 
require an identity-preserving physical/logical width here (or normalize the 
carrier as the Bloom path does) and add a predicate-only PLAIN UINT32 test, 
including a value above `INT32_MAX`.



##########
be/src/format_v2/parquet/reader/native/column_chunk_reader.cpp:
##########
@@ -0,0 +1,1676 @@
+// 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/reader/native/column_chunk_reader.h"
+
+#include <cctz/time_zone.h>
+#include <gen_cpp/parquet_types.h>
+#include <glog/logging.h>
+#include <parquet/metadata.h>
+#include <string.h>
+
+#include <algorithm>
+#include <cstdint>
+#include <limits>
+#include <memory>
+#include <utility>
+
+#include "common/compiler_util.h" // IWYU pragma: keep
+#include "core/column/column.h"
+#include "core/column/column_decimal.h"
+#include "core/column/column_dictionary.h"
+#include "core/column/column_varbinary.h"
+#include "core/column/column_vector.h"
+#include "core/custom_allocator.h"
+#include "core/data_type_serde/data_type_serde.h"
+#include "core/data_type_serde/parquet_timestamp.h"
+#include "exprs/vexpr.h"
+#include "format_v2/parquet/native_schema_desc.h"
+#include "format_v2/parquet/reader/native/decoder.h"
+#include "format_v2/parquet/reader/native/level_decoder.h"
+#include "format_v2/parquet/reader/native/page_reader.h"
+#include "io/fs/buffered_reader.h"
+#include "runtime/runtime_profile.h"
+#include "storage/cache/page_cache.h"
+#include "util/bit_util.h"
+#include "util/block_compression.h"
+#include "util/cpu_info.h"
+#include "util/unaligned.h"
+
+namespace cctz {
+class time_zone;
+} // namespace cctz
+namespace doris {
+namespace io {
+class BufferedStreamReader;
+struct IOContext;
+} // namespace io
+} // namespace doris
+
+namespace doris::format::parquet::native {
+
+bool can_prepare_page_cache_payload(bool session_cache_enabled, bool 
storage_cache_disabled,
+                                    bool cache_available, bool 
header_available) {
+    return session_cache_enabled && !storage_cache_disabled && cache_available 
&& header_available;
+}
+
+Status validate_uncompressed_page_sizes(const tparquet::PageHeader& header,
+                                        tparquet::CompressionCodec::type codec,
+                                        bool data_page_v2_always_compressed) {
+    const bool is_v2 = header.__isset.data_page_header_v2;
+    const bool page_is_compressed =
+            codec != tparquet::CompressionCodec::UNCOMPRESSED &&
+            (!is_v2 || header.data_page_header_v2.is_compressed || 
data_page_v2_always_compressed);
+    if (!page_is_compressed &&
+        UNLIKELY(header.compressed_page_size != 
header.uncompressed_page_size)) {
+        // An uncompressed payload has one physical representation, so 
accepting two lengths makes
+        // cold reads and decompressed cache hits consume different byte 
boundaries.
+        return Status::Corruption(
+                "Uncompressed Parquet page sizes differ: compressed={}, 
uncompressed={}",
+                header.compressed_page_size, header.uncompressed_page_size);
+    }
+    return Status::OK();
+}
+
+ParquetReaderCompat parquet_reader_compat(const std::string& created_by) {
+    if (created_by.empty()) {
+        return {};
+    }
+    const ::parquet::ApplicationVersion version(created_by);
+    return {.parquet_816_padding =
+                    
version.VersionLt(::parquet::ApplicationVersion::PARQUET_816_FIXED_VERSION()),
+            .data_page_v2_always_compressed = version.VersionLt(
+                    
::parquet::ApplicationVersion::PARQUET_CPP_10353_FIXED_VERSION())};
+}
+
+Status compute_column_chunk_range(const tparquet::ColumnMetaData& metadata, 
size_t file_size,
+                                  bool parquet_816_padding, ColumnChunkRange* 
range) {
+    DORIS_CHECK(range != nullptr);
+    int64_t start = metadata.data_page_offset;
+    if (metadata.__isset.dictionary_page_offset && 
metadata.dictionary_page_offset > 0 &&
+        metadata.dictionary_page_offset < start) {
+        // Some writers use dictionary_page_offset=0 as an absence sentinel. 
Range validation must
+        // follow has_dict_page() or the native reader starts at the Parquet 
magic bytes.
+        start = metadata.dictionary_page_offset;
+    }
+    const int64_t length = metadata.total_compressed_size;
+    if (UNLIKELY(start < 0 || length < 0)) {
+        return Status::Corruption("Parquet column chunk has a negative offset 
or length");
+    }
+    const uint64_t unsigned_start = static_cast<uint64_t>(start);
+    const uint64_t unsigned_length = static_cast<uint64_t>(length);
+    if (UNLIKELY(unsigned_start > file_size || unsigned_length > file_size - 
unsigned_start)) {
+        // Thrift range fields are signed and untrusted; validate before 
converting them to the
+        // unsigned stream-reader coordinates so overflow cannot wrap back 
into the file.
+        return Status::Corruption("Parquet column chunk [{}, {}) exceeds file 
size {}", start,
+                                  unsigned_start + unsigned_length, file_size);
+    }
+    size_t bounded_length = static_cast<size_t>(unsigned_length);
+    if (parquet_816_padding) {
+        // parquet-mr before PARQUET-816 under-reported the chunk by up to 100 
bytes. Padding stays
+        // file-bounded and is only enabled for the affected writer versions.
+        bounded_length += std::min<size_t>(100, file_size - unsigned_start - 
unsigned_length);
+    }
+    range->offset = static_cast<size_t>(unsigned_start);
+    range->length = bounded_length;
+    return Status::OK();
+}
+
+bool validate_offset_index(const tparquet::OffsetIndex& index, const 
ColumnChunkRange& chunk_range,
+                           int64_t data_page_offset, int64_t row_count) {
+    if (index.page_locations.empty() || data_page_offset < 0 || row_count < 0 
||
+        index.page_locations.front().first_row_index != 0 ||
+        index.page_locations.front().offset != data_page_offset ||
+        chunk_range.length > std::numeric_limits<size_t>::max() - 
chunk_range.offset) {
+        return false;
+    }
+    // Row indexes alone cannot detect a uniformly shifted OffsetIndex. Anchor 
its first location
+    // to the owning metadata so page-to-row mapping cannot silently move by 
one physical page.
+    const uint64_t chunk_begin = chunk_range.offset;
+    const uint64_t chunk_end = chunk_begin + chunk_range.length;
+    uint64_t previous_end = chunk_begin;
+    int64_t previous_row = -1;
+    for (const auto& location : index.page_locations) {
+        if (location.first_row_index <= previous_row || 
location.first_row_index >= row_count ||
+            location.offset < 0 || location.compressed_page_size <= 0) {
+            return false;
+        }
+        const uint64_t begin = static_cast<uint64_t>(location.offset);
+        const uint64_t size = 
static_cast<uint64_t>(location.compressed_page_size);
+        if (begin < chunk_begin || begin < previous_end || begin > chunk_end ||
+            size > chunk_end - begin) {
+            return false;
+        }
+        previous_row = location.first_row_index;
+        previous_end = begin + size;
+    }
+    return true;
+}
+
+namespace {
+
+Status append_v2_int96_datetime(ColumnDateTimeV2::Container& data,
+                                const ParquetInt96Timestamp& value,
+                                const cctz::time_zone& timezone) {
+    static constexpr int32_t JULIAN_EPOCH_OFFSET_DAYS = 2440588;
+    static constexpr int64_t MICROS_PER_DAY = 86400000000LL;
+    static constexpr int64_t MICROS_PER_SECOND = 1000000LL;
+
+    // Arrow normalized out-of-day INT96 nanos before the native V2 path 
replaced it. Preserve that
+    // legacy-writer compatibility here; rejecting the carrier loses valid 
pre-epoch/year-0 values
+    // already accepted by Doris external-table scans.
+    const __int128 days = static_cast<__int128>(value.julian_day) - 
JULIAN_EPOCH_OFFSET_DAYS;
+    // Truncate the signed nanos field before day normalization. Flooring a 
negative value first
+    // changes the historical result by one microsecond.
+    const __int128 timestamp_micros = days * MICROS_PER_DAY + 
value.nanos_of_day / 1000;
+    if (timestamp_micros < std::numeric_limits<int64_t>::min() ||
+        timestamp_micros > std::numeric_limits<int64_t>::max()) {
+        return Status::DataQualityError("Parquet INT96 timestamp overflows 
microseconds");
+    }
+
+    const int64_t micros = static_cast<int64_t>(timestamp_micros);
+    int64_t epoch_seconds = micros / MICROS_PER_SECOND;
+    int64_t micros_of_second = micros % MICROS_PER_SECOND;
+    if (micros_of_second < 0) {
+        micros_of_second += MICROS_PER_SECOND;
+        --epoch_seconds;
+    }
+    DateV2Value<DateTimeV2ValueType> datetime;
+    datetime.from_unixtime(epoch_seconds, timezone);
+    datetime.set_microsecond(static_cast<uint32_t>(micros_of_second));
+    if (!datetime.is_valid_date()) {
+        return Status::DataQualityError("Parquet INT96 timestamp is outside 
the Doris range");
+    }
+    data.push_back(datetime);
+    return Status::OK();
+}
+
+class V2Int96DateTimeConsumer final : public ParquetFixedValueConsumer {
+public:
+    V2Int96DateTimeConsumer(IColumn& column, const ParquetDecodeContext& 
context,
+                            ParquetMaterializationState* state)
+            : _data(assert_cast<ColumnDateTimeV2&>(column).get_data()), 
_state(state) {
+        static const auto utc = cctz::utc_time_zone();
+        _timezone = context.timezone == nullptr ? &utc : context.timezone;
+    }
+
+    Status consume(const uint8_t* values, size_t num_values, size_t 
value_width) override {
+        DORIS_CHECK_EQ(value_width, sizeof(ParquetInt96Timestamp));
+        const size_t old_size = _data.size();
+        for (size_t row = 0; row < num_values; ++row) {
+            const auto value = unaligned_load<ParquetInt96Timestamp>(
+                    values + row * sizeof(ParquetInt96Timestamp));
+            const auto status = append_v2_int96_datetime(_data, value, 
*_timezone);
+            if (!status.ok()) {
+                if (_state != nullptr && 
_state->mark_conversion_failure(_data.size())) {
+                    _data.emplace_back();
+                    continue;
+                }
+                _data.resize(old_size);
+                return status;
+            }
+        }
+        return Status::OK();
+    }
+
+private:
+    ColumnDateTimeV2::Container& _data;
+    ParquetMaterializationState* _state;
+    const cctz::time_zone* _timezone = nullptr;
+};
+
+class RejectV2Int96BinaryConsumer final : public ParquetBinaryValueConsumer {
+public:
+    Status consume(const StringRef*, size_t) override {
+        return Status::NotSupported("INT96 cannot be decoded from binary 
Parquet values");
+    }
+};
+
+Status read_v2_int96_datetime(IColumn& column, ParquetDecodeSource& source,
+                              const ParquetDecodeContext& context, size_t 
num_values,
+                              ParquetMaterializationState& state) {
+    V2Int96DateTimeConsumer consumer(column, context, &state);
+    if (context.encoding != ParquetValueEncoding::DICTIONARY) {
+        return source.decode_fixed_values(num_values, consumer);
+    }
+    if (state.dictionary_generation != source.dictionary_generation()) {
+        state.typed_dictionary = column.clone_empty();
+        auto* output_null_map = 
state.begin_dictionary_conversion(source.dictionary_size());
+        V2Int96DateTimeConsumer dictionary_consumer(*state.typed_dictionary, 
context, &state);
+        RejectV2Int96BinaryConsumer binary_consumer;
+        const auto dictionary_status =
+                source.decode_dictionary(dictionary_consumer, binary_consumer);
+        state.end_dictionary_conversion(output_null_map);
+        RETURN_IF_ERROR(dictionary_status);
+        DORIS_CHECK_EQ(state.typed_dictionary->size(), 
source.dictionary_size());
+        state.dictionary_generation = source.dictionary_generation();
+    }
+    RETURN_IF_ERROR(source.decode_dictionary_indices(num_values, 
&state.dictionary_indices));
+    DORIS_CHECK_EQ(state.dictionary_indices.size(), num_values);
+    return state.materialize_dictionary(column);
+}
+
+Status read_native_or_serde(IColumn& column, const DataTypeSerDe& serde,
+                            ParquetDecodeSource& source, const 
ParquetDecodeContext& context,
+                            size_t num_values, ParquetMaterializationState& 
state) {
+    if (context.physical_type == ParquetPhysicalType::INT96 &&
+        check_and_get_column<ColumnDateTimeV2>(&column) != nullptr) {
+        return read_v2_int96_datetime(column, source, context, num_values, 
state);
+    }
+    return serde.read_column_from_parquet(column, source, context, num_values, 
state);
+}
+
+Status translate_value_encoding(tparquet::Encoding::type encoding,
+                                ParquetValueEncoding* translated) {
+    DORIS_CHECK(translated != nullptr);
+    switch (encoding) {
+    case tparquet::Encoding::PLAIN:
+        *translated = ParquetValueEncoding::PLAIN;
+        return Status::OK();
+    case tparquet::Encoding::RLE_DICTIONARY:
+    case tparquet::Encoding::PLAIN_DICTIONARY:
+        *translated = ParquetValueEncoding::DICTIONARY;
+        return Status::OK();
+    case tparquet::Encoding::RLE:
+        *translated = ParquetValueEncoding::RLE;
+        return Status::OK();
+    case tparquet::Encoding::BIT_PACKED:
+        *translated = ParquetValueEncoding::BIT_PACKED;
+        return Status::OK();
+    case tparquet::Encoding::DELTA_BINARY_PACKED:
+        *translated = ParquetValueEncoding::DELTA_BINARY_PACKED;
+        return Status::OK();
+    case tparquet::Encoding::DELTA_LENGTH_BYTE_ARRAY:
+        *translated = ParquetValueEncoding::DELTA_LENGTH_BYTE_ARRAY;
+        return Status::OK();
+    case tparquet::Encoding::DELTA_BYTE_ARRAY:
+        *translated = ParquetValueEncoding::DELTA_BYTE_ARRAY;
+        return Status::OK();
+    case tparquet::Encoding::BYTE_STREAM_SPLIT:
+        *translated = ParquetValueEncoding::BYTE_STREAM_SPLIT;
+        return Status::OK();
+    default:
+        return Status::NotSupported("Unsupported Parquet encoding {}",
+                                    tparquet::to_string(encoding));
+    }
+}
+
+template <bool HAS_FILTER>
+Status decode_selected_values(IColumn& column, const DataTypeSerDe& serde, 
Decoder& decoder,
+                              const ParquetDecodeContext& context,
+                              ParquetMaterializationState& state, 
ColumnSelectVector& select_vector,
+                              int64_t* materialization_time) {
+    SCOPED_RAW_TIMER(materialization_time);
+    ColumnSelectVector::DataReadType read_type;
+    while (const size_t run_length = 
select_vector.get_next_run<HAS_FILTER>(&read_type)) {
+        switch (read_type) {
+        case ColumnSelectVector::CONTENT:
+            RETURN_IF_ERROR(
+                    read_native_or_serde(column, serde, decoder, context, 
run_length, state));
+            break;
+        case ColumnSelectVector::NULL_DATA:
+            column.insert_many_defaults(run_length);
+            break;
+        case ColumnSelectVector::FILTERED_CONTENT:
+            RETURN_IF_ERROR(decoder.skip_values(run_length));
+            break;
+        case ColumnSelectVector::FILTERED_NULL:
+            break;
+        }
+    }
+    return Status::OK();
+}
+
+// Presents one sparse page request as an ordinary sequential source to 
DataTypeSerDe. SerDe is
+// entered once per page fragment; the concrete decoder decides whether to 
gather selected spans,
+// batch-decode and compact, or use the cursor-preserving range fallback.
+class SelectedDecodeSource final : public ParquetDecodeSource {
+public:
+    SelectedDecodeSource(Decoder& decoder, const ParquetSelection& selection)
+            : _decoder(decoder), _selection(selection) {}
+
+    Status decode_fixed_values(size_t num_values, ParquetFixedValueConsumer& 
consumer) override {
+        DORIS_CHECK_EQ(num_values, _selection.selected_values);
+        return _decoder.decode_selected_fixed_values(_selection, consumer);
+    }
+
+    Status decode_binary_values(size_t num_values, ParquetBinaryValueConsumer& 
consumer) override {
+        DORIS_CHECK_EQ(num_values, _selection.selected_values);
+        return _decoder.decode_selected_binary_values(_selection, consumer);
+    }
+
+    Status skip_values(size_t num_values) override {
+        return Status::NotSupported("Selected Parquet source cannot be 
skipped, values={}",
+                                    num_values);
+    }
+
+    bool has_dictionary() const override { return _decoder.has_dictionary(); }
+    uint64_t dictionary_generation() const override { return 
_decoder.dictionary_generation(); }
+    size_t dictionary_size() const override { return 
_decoder.dictionary_size(); }
+
+    Status decode_dictionary(ParquetFixedValueConsumer& fixed_consumer,
+                             ParquetBinaryValueConsumer& binary_consumer) 
override {
+        return _decoder.decode_dictionary(fixed_consumer, binary_consumer);
+    }
+
+    Status decode_dictionary_indices(size_t num_values, std::vector<uint32_t>* 
indices) override {
+        DORIS_CHECK_EQ(num_values, _selection.selected_values);
+        return _decoder.decode_selected_dictionary_indices(_selection, 
indices);
+    }
+
+    Status decode_dictionary_values(size_t num_values,
+                                    ParquetDictionaryValueConsumer& consumer) 
override {
+        DORIS_CHECK_EQ(num_values, _selection.selected_values);
+        return _decoder.decode_selected_dictionary_values(_selection, 
consumer);
+    }
+
+    bool prefer_dictionary_index_materialization(size_t dictionary_bytes) 
const override {
+        // Avoid touching a dictionary larger than L2 for rows already removed 
by sparse selection.
+        // Cache-resident or dense batches instead fuse RLE decode and gather, 
eliminating the
+        // page-sized intermediate ID vector.
+        return _selection.selected_values < _selection.total_values &&
+               dictionary_bytes > 
static_cast<size_t>(CpuInfo::get_l2_cache_size());
+    }
+
+private:
+    Decoder& _decoder;
+    const ParquetSelection& _selection;
+};
+
+Status decode_selected_non_null_values(IColumn& column, const DataTypeSerDe& 
serde,
+                                       Decoder& decoder, const 
ParquetDecodeContext& context,
+                                       ParquetMaterializationState& state,
+                                       ColumnSelectVector& select_vector,
+                                       int64_t* materialization_time) {
+    auto& selection = state.selection;
+    selection.ranges.clear();
+    selection.total_values = select_vector.num_values();
+    selection.selected_values = 0;
+
+    size_t cursor = 0;
+    ColumnSelectVector::DataReadType read_type;
+    while (const size_t run_length = 
select_vector.get_next_run<true>(&read_type)) {
+        DORIS_CHECK(read_type == ColumnSelectVector::CONTENT ||
+                    read_type == ColumnSelectVector::FILTERED_CONTENT);
+        if (read_type == ColumnSelectVector::CONTENT) {
+            selection.ranges.push_back({.first = cursor, .count = run_length});
+            selection.selected_values += run_length;
+        }
+        cursor += run_length;
+    }
+    DORIS_CHECK_EQ(cursor, selection.total_values);
+    if (selection.selected_values == 0) {
+        return decoder.skip_values(selection.total_values);
+    }
+
+    SCOPED_RAW_TIMER(materialization_time);
+    SelectedDecodeSource selected_source(decoder, selection);
+    return read_native_or_serde(column, serde, selected_source, context, 
selection.selected_values,
+                                state);
+}
+
+template <typename Visitor>
+bool visit_nullable_expandable_column(IColumn& column, Visitor&& visitor) {
+#define VISIT_COLUMN(TYPE)                                   \
+    if (auto* typed = check_and_get_column<TYPE>(&column)) { \
+        visitor(*typed);                                     \
+        return true;                                         \
+    }
+    VISIT_COLUMN(ColumnUInt8)
+    VISIT_COLUMN(ColumnInt8)
+    VISIT_COLUMN(ColumnInt16)
+    VISIT_COLUMN(ColumnInt32)
+    VISIT_COLUMN(ColumnInt64)
+    VISIT_COLUMN(ColumnInt128)
+    VISIT_COLUMN(ColumnDate)
+    VISIT_COLUMN(ColumnDateTime)
+    VISIT_COLUMN(ColumnDateV2)
+    VISIT_COLUMN(ColumnDateTimeV2)
+    VISIT_COLUMN(ColumnFloat32)
+    VISIT_COLUMN(ColumnFloat64)
+    VISIT_COLUMN(ColumnIPv4)
+    VISIT_COLUMN(ColumnIPv6)
+    VISIT_COLUMN(ColumnTimeV2)
+    VISIT_COLUMN(ColumnTimeStampTz)
+    VISIT_COLUMN(ColumnOffset32)
+    VISIT_COLUMN(ColumnOffset64)
+    VISIT_COLUMN(ColumnDecimal32)
+    VISIT_COLUMN(ColumnDecimal64)
+    VISIT_COLUMN(ColumnDecimal128V2)
+    VISIT_COLUMN(ColumnDecimal128V3)
+    VISIT_COLUMN(ColumnDecimal256)
+    VISIT_COLUMN(ColumnString)
+    VISIT_COLUMN(ColumnString64)
+    VISIT_COLUMN(ColumnVarbinary)
+    VISIT_COLUMN(ColumnDictI32)
+#undef VISIT_COLUMN
+    return false;
+}
+
+template <typename ColumnType>
+void expand_nullable_pod_values(ColumnType& column, size_t old_size, size_t 
compact_values,
+                                const NullMap& selected_nulls) {
+    auto& data = column.get_data();
+    DORIS_CHECK_EQ(data.size(), old_size + compact_values);
+    data.resize(old_size + selected_nulls.size());
+    size_t source = compact_values;
+    for (size_t output = selected_nulls.size(); output > 0;) {
+        --output;
+        if (selected_nulls[output] != 0) {
+            data[old_size + output] = typename ColumnType::value_type {};
+        } else {
+            DORIS_CHECK(source > 0);
+            --source;
+            data[old_size + output] = std::move(data[old_size + source]);
+        }
+    }
+    DORIS_CHECK_EQ(source, 0);
+}
+
+template <typename Offset>
+void expand_nullable_string_values(ColumnStr<Offset>& column, size_t old_size,
+                                   size_t compact_values, const NullMap& 
selected_nulls) {
+    auto& offsets = column.get_offsets();
+    DORIS_CHECK_EQ(offsets.size(), old_size + compact_values);
+    const Offset prefix_end = old_size == 0 ? 0 : offsets[old_size - 1];
+    offsets.resize(old_size + selected_nulls.size());
+    size_t source = compact_values;
+    for (size_t output = selected_nulls.size(); output > 0;) {
+        --output;
+        if (selected_nulls[output] == 0) {
+            DORIS_CHECK(source > 0);
+            --source;
+            offsets[old_size + output] = offsets[old_size + source];
+        } else {
+            offsets[old_size + output] = source == 0 ? prefix_end : 
offsets[old_size + source - 1];
+        }
+    }
+    DORIS_CHECK_EQ(source, 0);
+}
+
+template <typename ColumnType>
+void expand_nullable_values(ColumnType& column, size_t old_size, size_t 
compact_values,
+                            const NullMap& selected_nulls) {
+    expand_nullable_pod_values(column, old_size, compact_values, 
selected_nulls);
+}
+
+template <typename Offset>
+void expand_nullable_values(ColumnStr<Offset>& column, size_t old_size, size_t 
compact_values,
+                            const NullMap& selected_nulls) {
+    expand_nullable_string_values(column, old_size, compact_values, 
selected_nulls);
+}
+
+void remap_nullable_conversion_failures(IColumn::Filter* 
conversion_failure_null_map,
+                                        size_t old_size, size_t compact_values,
+                                        const NullMap& selected_nulls) {
+    if (conversion_failure_null_map == nullptr) {
+        return;
+    }
+    DORIS_CHECK(conversion_failure_null_map->size() >= old_size + 
selected_nulls.size());
+    size_t source = compact_values;
+    // Walk backwards so writing an expanded row cannot overwrite an unread 
compact failure bit.
+    for (size_t output = selected_nulls.size(); output > 0;) {
+        --output;
+        if (selected_nulls[output] != 0) {
+            (*conversion_failure_null_map)[old_size + output] = 1;
+        } else {
+            DORIS_CHECK(source > 0);
+            --source;
+            (*conversion_failure_null_map)[old_size + output] =
+                    (*conversion_failure_null_map)[old_size + source];
+        }
+    }
+    DORIS_CHECK_EQ(source, 0);
+}
+
+Status decode_selected_nullable_values(IColumn& column, const DataTypeSerDe& 
serde,
+                                       Decoder& decoder, const 
ParquetDecodeContext& context,
+                                       ParquetMaterializationState& state,
+                                       ColumnSelectVector& select_vector, 
NullMap& selected_nulls,
+                                       int64_t* materialization_time) {
+    auto& selection = state.selection;
+    selection.ranges.clear();
+    selection.total_values = 0;
+    selection.selected_values = 0;
+    selected_nulls.clear();
+    selected_nulls.reserve(select_vector.num_values() - 
select_vector.num_filtered());
+
+    size_t physical_cursor = 0;
+    ColumnSelectVector::DataReadType read_type;
+    while (const size_t run_length = 
select_vector.get_next_run<true>(&read_type)) {
+        switch (read_type) {
+        case ColumnSelectVector::CONTENT:
+            if (!selection.ranges.empty() &&
+                selection.ranges.back().first + selection.ranges.back().count 
== physical_cursor) {
+                selection.ranges.back().count += run_length;
+            } else {
+                selection.ranges.push_back({.first = physical_cursor, .count = 
run_length});
+            }
+            selection.selected_values += run_length;
+            selected_nulls.resize_fill(selected_nulls.size() + run_length, 0);
+            physical_cursor += run_length;
+            break;
+        case ColumnSelectVector::NULL_DATA:
+            selected_nulls.resize_fill(selected_nulls.size() + run_length, 1);
+            break;
+        case ColumnSelectVector::FILTERED_CONTENT:
+            physical_cursor += run_length;
+            break;
+        case ColumnSelectVector::FILTERED_NULL:
+            break;
+        }
+    }
+    selection.total_values = physical_cursor;
+    DORIS_CHECK_EQ(selection.total_values, select_vector.num_values() - 
select_vector.num_nulls());
+    DORIS_CHECK_EQ(selected_nulls.size(),
+                   select_vector.num_values() - select_vector.num_filtered());
+
+    const size_t old_size = column.size();
+    SCOPED_RAW_TIMER(materialization_time);
+    if (selection.selected_values == 0) {
+        RETURN_IF_ERROR(decoder.skip_values(selection.total_values));
+        column.insert_many_defaults(selected_nulls.size());
+        return Status::OK();
+    }
+
+    if (state.conversion_failure_null_map != nullptr) {
+        DORIS_CHECK(state.conversion_failure_null_map->size() >= old_size + 
selected_nulls.size());
+        memset(state.conversion_failure_null_map->data() + old_size, 0, 
selection.selected_values);
+    }
+    SelectedDecodeSource selected_source(decoder, selection);
+    const auto status = read_native_or_serde(column, serde, selected_source, 
context,
+                                             selection.selected_values, state);
+    if (!status.ok()) {
+        if (state.conversion_failure_null_map != nullptr) {
+            memcpy(state.conversion_failure_null_map->data() + old_size, 
selected_nulls.data(),
+                   selected_nulls.size());
+        }
+        return status;
+    }
+    DORIS_CHECK_EQ(column.size(), old_size + selection.selected_values);
+
+    remap_nullable_conversion_failures(state.conversion_failure_null_map, 
old_size,
+                                       selection.selected_values, 
selected_nulls);
+    const bool expanded = visit_nullable_expandable_column(column, [&](auto& 
typed_column) {
+        // Decode into the final nested column compactly, then expand in 
place. This preserves the
+        // V2 no-intermediate-column invariant while restoring the nullable 
sparse row layout.
+        expand_nullable_values(typed_column, old_size, 
selection.selected_values, selected_nulls);
+    });
+    DORIS_CHECK(expanded);
+    return Status::OK();
+}
+
+class PlainPredicateConsumer final : public ParquetFixedValueConsumer {
+public:
+    PlainPredicateConsumer(const VExprSPtrs& conjuncts, DataTypePtr data_type, 
int column_id,
+                           IColumn::Filter* matches)
+            : _conjuncts(conjuncts),
+              _data_type(std::move(data_type)),
+              _column_id(column_id),
+              _matches(matches) {
+        DORIS_CHECK(_matches != nullptr);
+    }
+
+    Status consume(const uint8_t* values, size_t num_values, size_t 
value_width) override {
+        const size_t old_size = _matches->size();
+        _matches->resize_fill(old_size + num_values, 1);
+        for (const auto& conjunct : _conjuncts) {
+            RETURN_IF_ERROR(conjunct->execute_on_raw_fixed_values(values, 
num_values, value_width,
+                                                                  _data_type, 
_column_id,
+                                                                  
_matches->data() + old_size));
+        }
+        return Status::OK();
+    }
+
+private:
+    const VExprSPtrs& _conjuncts;
+    DataTypePtr _data_type;
+    int _column_id;
+    IColumn::Filter* _matches;
+};
+
+} // namespace
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::ColumnChunkReader(
+        io::BufferedStreamReader* reader, tparquet::ColumnChunk* column_chunk,
+        NativeFieldSchema* field_schema, const tparquet::OffsetIndex* 
offset_index,
+        size_t total_rows, io::IOContext* io_ctx, const 
ParquetPageReadContext& page_read_ctx,
+        const ColumnChunkRange* chunk_range)
+        : _field_schema(field_schema),
+          _max_rep_level(field_schema->repetition_level),
+          _max_def_level(field_schema->definition_level),
+          _stream_reader(reader),
+          _metadata(column_chunk->meta_data),
+          _offset_index(offset_index),
+          _total_rows(total_rows),
+          _io_ctx(io_ctx),
+          _page_read_ctx(page_read_ctx) {
+    if (chunk_range != nullptr) {
+        _chunk_range = *chunk_range;
+        _has_validated_chunk_range = true;
+    }
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::init() {
+    size_t start_offset = _has_validated_chunk_range
+                                  ? _chunk_range.offset
+                                  : (has_dict_page(_metadata) ? 
_metadata.dictionary_page_offset
+                                                              : 
_metadata.data_page_offset);
+    size_t chunk_size =
+            _has_validated_chunk_range ? _chunk_range.length : 
_metadata.total_compressed_size;
+    // create page reader
+    _page_reader = create_page_reader<IN_COLLECTION, OFFSET_INDEX>(
+            _stream_reader, _io_ctx, start_offset, chunk_size, _total_rows, 
_metadata,
+            _page_read_ctx, _offset_index);
+    // get the block compression codec
+    RETURN_IF_ERROR(get_block_compression_codec(_metadata.codec, 
&_block_compress_codec));
+    _state = INITIALIZED;
+    RETURN_IF_ERROR(_parse_first_page_header());
+    return Status::OK();
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::skip_nested_values(
+        const std::vector<level_t>& def_levels, size_t start_index) {
+    size_t no_value_cnt = 0;
+    size_t value_cnt = 0;
+
+    DORIS_CHECK(start_index <= def_levels.size());
+    for (size_t idx = start_index; idx < def_levels.size(); idx++) {
+        level_t def_level = def_levels[idx];
+        if (IN_COLLECTION && def_level < 
_field_schema->repeated_parent_def_level) {
+            no_value_cnt++;
+        } else if (def_level < _field_schema->definition_level) {
+            no_value_cnt++;
+        } else {
+            value_cnt++;
+        }
+    }
+
+    RETURN_IF_ERROR(skip_values(value_cnt, true));
+    RETURN_IF_ERROR(skip_values(no_value_cnt, false));
+    return Status::OK();
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::read_levels(
+        size_t num_values, std::vector<level_t>* rep_levels, 
std::vector<level_t>* def_levels) {
+    DORIS_CHECK(rep_levels != nullptr);
+    DORIS_CHECK(def_levels != nullptr);
+    if (_remaining_num_values < num_values || _remaining_rep_nums < num_values 
||
+        _remaining_def_nums < num_values) {
+        return Status::Corruption(
+                "Parquet level reader requested {} slots with only {}/{}/{} 
remaining", num_values,
+                _remaining_num_values, _remaining_rep_nums, 
_remaining_def_nums);
+    }
+
+    const size_t start_index = def_levels->size();
+    rep_levels->resize(rep_levels->size() + num_values, 0);
+    def_levels->resize(def_levels->size() + num_values, 0);
+    if (_max_rep_level > 0) {
+        const size_t decoded = _rep_level_decoder.get_levels(
+                rep_levels->data() + rep_levels->size() - num_values, 
num_values);
+        if (decoded != num_values) {
+            return Status::Corruption("Parquet repetition level stream ended 
after {} of {} slots",
+                                      decoded, num_values);
+        }
+    }
+    if (_max_def_level > 0) {
+        const size_t decoded = _def_level_decoder.get_levels(
+                def_levels->data() + def_levels->size() - num_values, 
num_values);
+        if (decoded != num_values) {
+            return Status::Corruption("Parquet definition level stream ended 
after {} of {} slots",
+                                      decoded, num_values);
+        }
+    }
+    _remaining_rep_nums -= num_values;
+    _remaining_def_nums -= num_values;
+    return skip_nested_values(*def_levels, start_index);
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+Status ColumnChunkReader<IN_COLLECTION, 
OFFSET_INDEX>::_parse_first_page_header() {
+    while (true) {
+        RETURN_IF_ERROR(_page_reader->parse_page_header());
+        const tparquet::PageHeader* header = nullptr;
+        RETURN_IF_ERROR(_page_reader->get_page_header(&header));
+        if (header->type == tparquet::PageType::DATA_PAGE ||
+            header->type == tparquet::PageType::DATA_PAGE_V2) {
+            _state = INITIALIZED;
+            return parse_page_header();
+        }
+        if (header->type != tparquet::PageType::DICTIONARY_PAGE) {
+            RETURN_IF_ERROR(_page_reader->skip_auxiliary_page());
+            _state = INITIALIZED;
+            continue;
+        }
+        // the first page maybe directory page even if 
_metadata.__isset.dictionary_page_offset == false,
+        // so we should parse the directory page in next_page()
+        RETURN_IF_ERROR(_decode_dict_page());
+        // parse the real first data page
+        RETURN_IF_ERROR(_page_reader->dict_next_page());
+        _state = INITIALIZED;
+        // A dictionary is the only non-data page with decoder state. Any 
following index or
+        // extension pages are skipped by the same pre-data loop.
+    }
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::parse_page_header() {
+    if (_state == HEADER_PARSED || _state == DATA_LOADED) {
+        return Status::OK();
+    }
+    const tparquet::PageHeader* header = nullptr;
+    while (true) {
+        RETURN_IF_ERROR(_page_reader->parse_page_header());
+        RETURN_IF_ERROR(_page_reader->get_page_header(&header));
+        if (header->type == tparquet::PageType::DATA_PAGE ||
+            header->type == tparquet::PageType::DATA_PAGE_V2) {
+            break;
+        }
+        if (header->type == tparquet::PageType::DICTIONARY_PAGE) {
+            return Status::Corruption("Parquet dictionary page appears after 
data pages");
+        }
+        RETURN_IF_ERROR(_page_reader->skip_auxiliary_page());
+    }
+    int32_t page_num_values = _page_reader->is_header_v2() ? 
header->data_page_header_v2.num_values
+                                                           : 
header->data_page_header.num_values;
+    if (page_num_values < 0 || page_num_values > _metadata.num_values ||
+        (!OFFSET_INDEX &&
+         static_cast<uint64_t>(page_num_values) >
+                 static_cast<uint64_t>(_metadata.num_values) - 
_chunk_parsed_values)) {
+        // Page counts are untrusted and feed both level decoders and scratch 
sizing. Bound each
+        // page by the column metadata before converting to unsigned counters.
+        return Status::Corruption("Parquet data page value count {} exceeds 
column total {}",
+                                  page_num_values, _metadata.num_values);
+    }
+    if constexpr (!IN_COLLECTION) {
+        const size_t page_start_row = _page_reader->start_row();
+        const size_t page_end_row = _page_reader->end_row();
+        if (UNLIKELY(page_end_row < page_start_row ||
+                     static_cast<size_t>(page_num_values) != page_end_row - 
page_start_row)) {
+            // Flat columns have exactly one physical value slot per logical 
row. Rejecting a
+            // divergent header/OffsetIndex span prevents every later page 
from shifting rows.
+            return Status::Corruption(
+                    "Parquet flat data page has {} values for logical row 
range [{}, {})",
+                    page_num_values, page_start_row, page_end_row);
+        }
+    }
+    _remaining_rep_nums = page_num_values;
+    _remaining_def_nums = page_num_values;
+    _remaining_num_values = page_num_values;
+
+    // no offset will parse all header.
+    if constexpr (OFFSET_INDEX == false) {
+        _chunk_parsed_values += _remaining_num_values;
+    }
+    _state = HEADER_PARSED;
+    return Status::OK();
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::next_page() {
+    _state = INITIALIZED;
+    RETURN_IF_ERROR(_page_reader->next_page());
+    return Status::OK();
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+Status ColumnChunkReader<IN_COLLECTION, 
OFFSET_INDEX>::_get_uncompressed_levels(
+        const tparquet::DataPageHeaderV2& page_v2, Slice& page_data) {
+    const size_t rl = page_v2.repetition_levels_byte_length;
+    const size_t dl = page_v2.definition_levels_byte_length;
+    if (UNLIKELY(rl > page_data.size || dl > page_data.size - rl)) {
+        // Validate the physical slice again because a cached entry may itself 
be truncated.
+        return Status::Corruption("Parquet data page v2 level bytes exceed 
available payload");
+    }
+    _v2_rep_levels = Slice(page_data.data, rl);
+    _v2_def_levels = Slice(page_data.data + rl, dl);
+    page_data.data += dl + rl;
+    page_data.size -= dl + rl;
+    return Status::OK();
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::load_page_data() {
+    if (_state == DATA_LOADED) {
+        return Status::OK();
+    }
+    if (UNLIKELY(_state != HEADER_PARSED)) {
+        return Status::Corruption("Should parse page header");
+    }
+
+    const tparquet::PageHeader* header = nullptr;
+    RETURN_IF_ERROR(_page_reader->get_page_header(&header));
+    RETURN_IF_ERROR(validate_uncompressed_page_sizes(
+            *header, _metadata.codec, 
_page_read_ctx.data_page_v2_always_compressed));
+    int32_t uncompressed_size = header->uncompressed_page_size;
+    bool page_loaded = false;
+
+    // First, try to reuse a cache handle previously discovered by PageReader
+    // (header-only lookup) to avoid a second lookup here.
+    if (_page_read_ctx.enable_parquet_file_page_cache && 
!config::disable_storage_page_cache &&
+        StoragePageCache::instance() != nullptr) {
+        if (_page_reader->has_page_cache_handle()) {
+            const PageCacheHandle& handle = _page_reader->page_cache_handle();
+            Slice cached = handle.data();
+            size_t header_size = _page_reader->header_bytes().size();
+            size_t levels_size = 0;
+            if (header->__isset.data_page_header_v2) {
+                const tparquet::DataPageHeaderV2& header_v2 = 
header->data_page_header_v2;
+                size_t rl = header_v2.repetition_levels_byte_length;
+                size_t dl = header_v2.definition_levels_byte_length;
+                levels_size = rl + dl;
+                if (UNLIKELY(header_size > cached.size ||
+                             levels_size > cached.size - header_size)) {
+                    return Status::Corruption("Cached Parquet page is shorter 
than its v2 levels");
+                }
+                _v2_rep_levels =
+                        Slice(reinterpret_cast<const uint8_t*>(cached.data) + 
header_size, rl);
+                _v2_def_levels =
+                        Slice(reinterpret_cast<const uint8_t*>(cached.data) + 
header_size + rl, dl);
+            }
+            // payload_slice points to the bytes after header and levels
+            if (UNLIKELY(header_size + levels_size > cached.size)) {
+                return Status::Corruption("Cached Parquet page is shorter than 
its header");
+            }
+            Slice payload_slice(cached.data + header_size + levels_size,
+                                cached.size - header_size - levels_size);
+
+            bool cache_payload_is_decompressed = 
_page_reader->is_cache_payload_decompressed();
+            const size_t expected_payload_size =
+                    cache_payload_is_decompressed
+                            ? 
static_cast<size_t>(header->uncompressed_page_size) - levels_size
+                            : 
static_cast<size_t>(header->compressed_page_size) - levels_size;
+            if (UNLIKELY(payload_slice.size != expected_payload_size)) {
+                return Status::Corruption("Cached Parquet page payload has 
size {}, expected {}",
+                                          payload_slice.size, 
expected_payload_size);
+            }
+
+            if (cache_payload_is_decompressed) {
+                // Cached payload is already uncompressed
+                _page_data = payload_slice;
+            } else {
+                CHECK(_block_compress_codec);
+                // Decompress cached payload into _decompress_buf for decoding
+                size_t uncompressed_payload_size =
+                        header->__isset.data_page_header_v2
+                                ? 
static_cast<size_t>(header->uncompressed_page_size) - levels_size
+                                : 
static_cast<size_t>(header->uncompressed_page_size);
+                _reserve_decompress_buf(uncompressed_payload_size);
+                _page_data = Slice(_decompress_buf.get(), 
uncompressed_payload_size);
+                SCOPED_RAW_TIMER(&_chunk_statistics.decompress_time);
+                _chunk_statistics.decompress_cnt++;
+                
RETURN_IF_ERROR(_block_compress_codec->decompress(payload_slice, &_page_data));
+                if (UNLIKELY(_page_data.size != uncompressed_payload_size)) {
+                    return Status::Corruption("Parquet page decompressed to {} 
bytes, expected {}",
+                                              _page_data.size, 
uncompressed_payload_size);
+                }
+            }
+            // page cache counters were incremented when PageReader did the 
header-only
+            // cache lookup. Do not increment again to avoid double-counting.
+            page_loaded = true;
+        }
+    }
+
+    if (!page_loaded) {
+        const bool prepare_cache_payload = can_prepare_page_cache_payload(
+                _page_read_ctx.enable_parquet_file_page_cache, 
config::disable_storage_page_cache,
+                StoragePageCache::instance() != nullptr, 
!_page_reader->header_bytes().empty());
+        if (_block_compress_codec != nullptr) {
+            Slice compressed_data;
+            RETURN_IF_ERROR(_page_reader->get_page_data(compressed_data));
+            std::vector<uint8_t> level_bytes;
+            if (header->__isset.data_page_header_v2) {
+                const tparquet::DataPageHeaderV2& header_v2 = 
header->data_page_header_v2;
+                // uncompressed_size = rl + dl + uncompressed_data_size
+                // compressed_size = rl + dl + compressed_data_size
+                uncompressed_size -= header_v2.repetition_levels_byte_length +
+                                     header_v2.definition_levels_byte_length;
+                // copy level bytes (rl + dl) so that we can cache header + 
levels + uncompressed payload
+                size_t rl = header_v2.repetition_levels_byte_length;
+                size_t dl = header_v2.definition_levels_byte_length;
+                size_t level_sz = rl + dl;
+                if (prepare_cache_payload && level_sz > 0) {
+                    level_bytes.resize(level_sz);
+                    memcpy(level_bytes.data(), compressed_data.data, level_sz);
+                }
+                // now remove levels from compressed_data for decompression
+                RETURN_IF_ERROR(_get_uncompressed_levels(header_v2, 
compressed_data));
+            }
+            bool is_v2_compressed = header->__isset.data_page_header_v2 &&
+                                    (header->data_page_header_v2.is_compressed 
||
+                                     
_page_read_ctx.data_page_v2_always_compressed);
+            bool page_has_compression = header->__isset.data_page_header || 
is_v2_compressed;
+
+            if (page_has_compression) {
+                // Decompress payload for immediate decoding
+                _reserve_decompress_buf(uncompressed_size);
+                _page_data = Slice(_decompress_buf.get(), uncompressed_size);
+                SCOPED_RAW_TIMER(&_chunk_statistics.decompress_time);
+                _chunk_statistics.decompress_cnt++;
+                
RETURN_IF_ERROR(_block_compress_codec->decompress(compressed_data, 
&_page_data));
+                if (UNLIKELY(_page_data.size != 
static_cast<size_t>(uncompressed_size))) {
+                    return Status::Corruption("Parquet page decompressed to {} 
bytes, expected {}",
+                                              _page_data.size, 
uncompressed_size);
+                }
+
+                // Decide whether to cache decompressed payload or compressed 
payload based on threshold
+                bool cache_payload_decompressed = should_cache_decompressed(
+                        header, _metadata, 
_page_read_ctx.data_page_v2_always_compressed);
+
+                if (prepare_cache_payload) {
+                    if (cache_payload_decompressed) {
+                        _insert_page_into_cache(level_bytes, _page_data);
+                        
_chunk_statistics.page_cache_decompressed_write_counter += 1;
+                    } else {
+                        if (config::enable_parquet_cache_compressed_pages) {
+                            // cache the compressed payload as-is (header | 
levels | compressed_payload)
+                            _insert_page_into_cache(
+                                    level_bytes, Slice(compressed_data.data, 
compressed_data.size));
+                            
_chunk_statistics.page_cache_compressed_write_counter += 1;
+                        }
+                    }
+                }
+            } else {
+                // no compression on this page, use the data directly
+                _page_data = Slice(compressed_data.data, compressed_data.size);
+                if (prepare_cache_payload) {
+                    _insert_page_into_cache(level_bytes, _page_data);
+                    _chunk_statistics.page_cache_decompressed_write_counter += 
1;
+                }
+            }
+        } else {
+            // For uncompressed page, we may still need to extract v2 levels
+            std::vector<uint8_t> level_bytes;
+            Slice uncompressed_data;
+            RETURN_IF_ERROR(_page_reader->get_page_data(uncompressed_data));
+            if (header->__isset.data_page_header_v2) {
+                const tparquet::DataPageHeaderV2& header_v2 = 
header->data_page_header_v2;
+                size_t rl = header_v2.repetition_levels_byte_length;
+                size_t dl = header_v2.definition_levels_byte_length;
+                size_t level_sz = rl + dl;
+                if (prepare_cache_payload && level_sz > 0) {
+                    level_bytes.resize(level_sz);
+                    memcpy(level_bytes.data(), uncompressed_data.data, 
level_sz);
+                }
+                RETURN_IF_ERROR(_get_uncompressed_levels(header_v2, 
uncompressed_data));
+            }
+            // copy page data out
+            _page_data = Slice(uncompressed_data.data, uncompressed_data.size);
+            // Optionally cache uncompressed data for uncompressed pages
+            if (prepare_cache_payload) {
+                _insert_page_into_cache(level_bytes, _page_data);
+                _chunk_statistics.page_cache_decompressed_write_counter += 1;
+            }
+        }
+    }
+
+    // Initialize repetition level and definition level. Skip when level = 0, 
which means required field.
+    if (_max_rep_level > 0) {
+        SCOPED_RAW_TIMER(&_chunk_statistics.decode_level_time);
+        if (header->__isset.data_page_header_v2) {
+            RETURN_IF_ERROR(_rep_level_decoder.init_v2(_v2_rep_levels, 
_max_rep_level,
+                                                       _remaining_rep_nums));
+        } else {
+            RETURN_IF_ERROR(_rep_level_decoder.init(
+                    &_page_data, 
header->data_page_header.repetition_level_encoding, _max_rep_level,
+                    _remaining_rep_nums));
+        }
+    }
+    if (_max_def_level > 0) {
+        SCOPED_RAW_TIMER(&_chunk_statistics.decode_level_time);
+        if (header->__isset.data_page_header_v2) {
+            RETURN_IF_ERROR(_def_level_decoder.init_v2(_v2_def_levels, 
_max_def_level,
+                                                       _remaining_def_nums));
+        } else {
+            RETURN_IF_ERROR(_def_level_decoder.init(
+                    &_page_data, 
header->data_page_header.definition_level_encoding, _max_def_level,
+                    _remaining_def_nums));
+        }
+    }
+    auto encoding = header->__isset.data_page_header_v2 ? 
header->data_page_header_v2.encoding
+                                                        : 
header->data_page_header.encoding;
+    // change the deprecated encoding to RLE_DICTIONARY
+    if (encoding == tparquet::Encoding::PLAIN_DICTIONARY) {
+        encoding = tparquet::Encoding::RLE_DICTIONARY;
+    }
+    _current_encoding = encoding;
+
+    // Reuse page decoder
+    if (_decoders.find(static_cast<int>(encoding)) != _decoders.end()) {
+        _page_decoder = _decoders[static_cast<int>(encoding)].get();
+    } else {
+        std::unique_ptr<Decoder> page_decoder;
+        RETURN_IF_ERROR(Decoder::get_decoder(_metadata.type, encoding, 
page_decoder));
+        // Set type length
+        page_decoder->set_type_length(_get_type_length());
+        _decoders[static_cast<int>(encoding)] = std::move(page_decoder);
+        _page_decoder = _decoders[static_cast<int>(encoding)].get();
+    }
+    // Encoding headers cannot legitimately advertise more physical values 
than the data page's
+    // logical value count; establish the bound before decoders inspect 
external counts.
+    _page_decoder->set_expected_values(_remaining_num_values);
+    RETURN_IF_ERROR(_page_decoder->set_data(&_page_data));

Review Comment:
   [P1] Handle an empty value section for an all-NULL page before initializing 
the encoding-specific decoder. Parquet omits NULLs from the value stream, so 
after V1/V2 definition levels are consumed a positive-row all-NULL page 
legitimately reaches here with empty `_page_data`; dictionary, delta, and 
BOOLEAN RLE `set_data()` implementations then reject it even though 
materialization would request zero physical values. Please establish a 
zero-physical-value decoder state (without accepting truncated nonempty 
streams) and add real V1/V2 level-only page tests for these encodings.



##########
be/src/format_v2/parquet/reader/native_column_reader.cpp:
##########
@@ -0,0 +1,757 @@
+// 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/reader/native_column_reader.h"
+
+#include <algorithm>
+#include <cstddef>
+#include <limits>
+#include <ranges>
+#include <string>
+#include <utility>
+
+#include "common/cast_set.h"
+#include "common/config.h"
+#include "core/assert_cast.h"
+#include "core/column/column_nullable.h"
+#include "core/column/column_string.h"
+#include "core/column/column_vector.h"
+#include "core/data_type/data_type_array.h"
+#include "core/data_type/data_type_map.h"
+#include "core/data_type/data_type_nullable.h"
+#include "core/data_type/data_type_number.h"
+#include "core/data_type/data_type_struct.h"
+#include "format_v2/column_data.h"
+#include "format_v2/parquet/parquet_column_schema.h"
+#include "format_v2/parquet/parquet_file_context.h"
+#include "runtime/runtime_state.h"
+
+namespace doris::format::parquet {
+namespace {
+
+constexpr size_t MAX_RETAINED_BATCH_SCRATCH_BYTES = 4UL << 20;
+
+DataTypePtr projected_type(const ParquetColumnSchema& schema,
+                           const format::LocalColumnIndex* projection) {
+    if (!format::is_partial_projection(projection)) {
+        return schema.type;
+    }
+    switch (schema.kind) {
+    case ParquetColumnSchemaKind::PRIMITIVE:
+        return schema.type;
+    case ParquetColumnSchemaKind::STRUCT: {
+        DataTypes child_types;
+        Strings child_names;
+        child_types.reserve(projection->children.size());
+        child_names.reserve(projection->children.size());
+        for (const auto& child_projection : projection->children) {
+            const auto child_it = std::ranges::find_if(schema.children, 
[&](const auto& child) {
+                return child->local_id == child_projection.local_id();
+            });
+            DORIS_CHECK(child_it != schema.children.end());
+            child_types.push_back(make_nullable(projected_type(**child_it, 
&child_projection)));
+            child_names.push_back((*child_it)->name);
+        }
+        DataTypePtr type = std::make_shared<DataTypeStruct>(child_types, 
child_names);
+        return schema.type->is_nullable() ? make_nullable(type) : type;
+    }
+    case ParquetColumnSchemaKind::LIST: {
+        DORIS_CHECK(schema.children.size() == 1);
+        const auto* child_projection =
+                format::find_child_projection(projection, 
schema.children[0]->local_id);
+        DORIS_CHECK(child_projection != nullptr);
+        DataTypePtr type = std::make_shared<DataTypeArray>(
+                projected_type(*schema.children[0], child_projection));
+        return schema.type->is_nullable() ? make_nullable(type) : type;
+    }
+    case ParquetColumnSchemaKind::MAP: {
+        DORIS_CHECK(schema.children.size() == 2);
+        const auto* value_projection =
+                format::find_child_projection(projection, 
schema.children[1]->local_id);
+        DORIS_CHECK(value_projection != nullptr);
+        DataTypePtr type = std::make_shared<DataTypeMap>(
+                make_nullable(schema.children[0]->type),
+                make_nullable(projected_type(*schema.children[1], 
value_projection)));
+        return schema.type->is_nullable() ? make_nullable(type) : type;
+    }
+    }
+    DORIS_CHECK(false);
+    return nullptr;
+}
+
+const NativeFieldSchema* find_child_field(const NativeFieldSchema& parent,
+                                          const ParquetColumnSchema& child) {
+    auto field_it = std::ranges::find_if(parent.children, [&](const 
NativeFieldSchema& field) {
+        return (child.parquet_field_id >= 0 && field.field_id == 
child.parquet_field_id) ||
+               field.name == child.name;
+    });
+    return field_it == parent.children.end() ? nullptr : &*field_it;
+}
+
+void collect_physical_subtree_ids(const NativeFieldSchema& field, 
std::set<uint64_t>* ids) {
+    DORIS_CHECK(ids != nullptr);
+    ids->insert(field.get_column_id());
+    for (const auto& child : field.children) {
+        collect_physical_subtree_ids(child, ids);
+    }
+}
+
+void collect_projected_ids(const ParquetColumnSchema& schema,
+                           const format::LocalColumnIndex* projection,
+                           const NativeFieldSchema& native_field, 
std::set<uint64_t>* ids) {
+    DORIS_CHECK(ids != nullptr);
+    if (!format::is_partial_projection(projection)) {
+        return;
+    }
+    for (const auto& child_projection : projection->children) {
+        const auto schema_it = std::ranges::find_if(schema.children, [&](const 
auto& child) {
+            return child->local_id == child_projection.local_id();
+        });
+        DORIS_CHECK(schema_it != schema.children.end());
+        const NativeFieldSchema* child_field = find_child_field(native_field, 
**schema_it);
+        DORIS_CHECK(child_field != nullptr);
+        if (format::is_full_projection(&child_projection)) {
+            // A full child path is a request for its complete physical 
subtree. Keeping only the
+            // child group id makes its grandchildren SkipReadingReaders and 
silently defaults
+            // STRUCT fields (or breaks ARRAY/MAP shape invariants).
+            collect_physical_subtree_ids(*child_field, ids);
+        } else {
+            ids->insert(child_field->get_column_id());
+            collect_projected_ids(**schema_it, &child_projection, 
*child_field, ids);
+        }
+    }
+    if (schema.kind == ParquetColumnSchemaKind::MAP) {
+        DORIS_CHECK(!native_field.children.empty());
+        // MAP entry existence and offsets are owned by the key stream even 
for value-only
+        // projections. Keep the key reader live so the native complex reader 
can validate
+        // key/value entry alignment while constructing offsets.
+        ids->insert(native_field.children[0].get_column_id());
+    }
+}
+
+Status append_non_null_dictionary_values(MutableColumnPtr& target, 
MutableColumnPtr values) {
+    DORIS_CHECK(target);
+    DORIS_CHECK(values);
+    const size_t value_count = values->size();
+    if (auto* nullable = check_and_get_column<ColumnNullable>(*target); 
nullable != nullptr) {
+        nullable->get_nested_column().insert_range_from(*values, 0, 
value_count);
+        auto& null_map = nullable->get_null_map_data();
+        null_map.resize_fill(null_map.size() + value_count, 0);
+        return Status::OK();
+    }
+    target->insert_range_from(*values, 0, value_count);
+    return Status::OK();
+}
+
+} // namespace
+
+NativeColumnReader::NativeColumnReader(const ParquetColumnSchema& schema,
+                                       DataTypePtr projected_type,
+                                       ParquetColumnReaderProfile profile)
+        : ParquetColumnReader(schema, std::move(projected_type), profile),
+          _nested(schema.kind != ParquetColumnSchemaKind::PRIMITIVE) {}
+
+NativeColumnReader::~NativeColumnReader() {
+    (void)sync_native_profile();
+}
+
+Status NativeColumnReader::create(
+        const ParquetColumnSchema& column_schema, const 
format::LocalColumnIndex* projection,
+        io::FileReaderSPtr file, const NativeParquetMetadata* metadata, int 
row_group_id,
+        const std::vector<RowRange>& selected_ranges,
+        const std::unordered_map<int, tparquet::OffsetIndex>& offset_indexes,
+        const cctz::time_zone* timezone, io::IOContext* io_ctx, RuntimeState* 
runtime_state,
+        bool enable_page_cache, const std::string& page_cache_file_key,
+        bool enable_dictionary_filter, ParquetColumnReaderProfile profile,
+        std::unique_ptr<ParquetColumnReader>* reader) {
+    if (reader == nullptr) {
+        return Status::InvalidArgument("Native parquet reader result is null");
+    }
+    if (file == nullptr || metadata == nullptr) {
+        return Status::InvalidArgument("Native parquet file context is not 
initialized");
+    }
+    if (row_group_id < 0 ||
+        row_group_id >= 
static_cast<int>(metadata->to_thrift().row_groups.size())) {
+        return Status::InvalidArgument("Invalid native parquet row group {}", 
row_group_id);
+    }
+    const auto& native_schema = metadata->schema();
+    if (column_schema.local_id < 0 || column_schema.local_id >= 
native_schema.size()) {
+        return Status::InvalidArgument("Invalid native parquet top-level 
column id {} for {}",
+                                       column_schema.local_id, 
column_schema.name);
+    }
+    auto* field = 
const_cast<NativeFieldSchema*>(native_schema.get_column(column_schema.local_id));
+    DORIS_CHECK(field != nullptr);
+    if (field->name != column_schema.name &&
+        !(field->field_id >= 0 && field->field_id == 
column_schema.parquet_field_id)) {
+        return Status::Corruption(
+                "Native/metadata parquet schema mismatch at column {}: 
native={}, arrow={}",
+                column_schema.local_id, field->name, column_schema.name);
+    }
+
+    auto type = projected_type(column_schema, projection);
+    std::shared_ptr<NativeSchemaNode> schema_node;
+    RETURN_IF_ERROR(build_native_schema_node(type, column_schema, 
&schema_node));
+    std::set<uint64_t> projected_ids;
+    collect_projected_ids(column_schema, projection, *field, &projected_ids);
+
+    auto native_reader = std::unique_ptr<NativeColumnReader>(
+            new NativeColumnReader(column_schema, std::move(type), profile));
+    RETURN_IF_ERROR(native_reader->init(
+            std::move(file), metadata, row_group_id, field, 
std::move(schema_node),
+            std::move(projected_ids), selected_ranges, offset_indexes, 
timezone, io_ctx,
+            runtime_state, enable_page_cache, page_cache_file_key, 
enable_dictionary_filter));
+    *reader = std::move(native_reader);
+    return Status::OK();
+}
+
+Status NativeColumnReader::init(
+        io::FileReaderSPtr file, const NativeParquetMetadata* metadata, int 
row_group_id,
+        NativeFieldSchema* field, std::shared_ptr<NativeSchemaNode> 
schema_node,
+        std::set<uint64_t> projected_column_ids, const std::vector<RowRange>& 
selected_ranges,
+        const std::unordered_map<int, tparquet::OffsetIndex>& offset_indexes,
+        const cctz::time_zone* timezone, io::IOContext* io_ctx, RuntimeState* 
runtime_state,
+        bool enable_page_cache, const std::string& page_cache_file_key,
+        bool enable_dictionary_filter) {
+    DORIS_CHECK(file != nullptr);
+    DORIS_CHECK(metadata != nullptr);
+    DORIS_CHECK(field != nullptr);
+    DORIS_CHECK(schema_node != nullptr);
+    const auto& row_group = metadata->to_thrift().row_groups[row_group_id];
+    DORIS_CHECK(row_group.num_rows > 0);
+    _row_group_rows = row_group.num_rows;
+    _selected_ranges = selected_ranges;
+    DORIS_CHECK(!_selected_ranges.empty());
+    for (const auto& range : _selected_ranges) {
+        DORIS_CHECK(range.start >= 0);
+        DORIS_CHECK(range.length > 0);
+        DORIS_CHECK(range.start + range.length <= _row_group_rows);
+        _row_ranges.add(segment_v2::RowRange(range.start, range.start + 
range.length));
+    }
+    // Offset indexes are immutable row-group metadata owned by 
ParquetScanScheduler. Sharing them
+    // avoids retaining the full N-column page-location map once per projected 
reader.
+    _offset_indexes = &offset_indexes;
+    _schema_node = std::move(schema_node);
+    _projected_column_ids = std::move(projected_column_ids);
+    _dictionary_filter_enabled = enable_dictionary_filter;
+
+    const size_t max_group_buffer = config::parquet_rowgroup_max_buffer_mb << 
20;
+    const size_t max_column_buffer = config::parquet_column_max_buffer_mb << 
20;
+    const size_t max_buffer_size = std::min(max_group_buffer, 
max_column_buffer);
+    RuntimeState* native_runtime_state = runtime_state;
+    const bool runtime_page_cache_enabled =
+            runtime_state == nullptr ||
+            runtime_state->query_options().enable_parquet_file_page_cache;
+    if (runtime_page_cache_enabled != enable_page_cache) {
+        TQueryOptions query_options =
+                runtime_state == nullptr ? TQueryOptions() : 
runtime_state->query_options();
+        query_options.__set_enable_parquet_file_page_cache(enable_page_cache);
+        _page_cache_runtime_state = RuntimeState::create_unique(query_options, 
TQueryGlobals());
+        native_runtime_state = _page_cache_runtime_state.get();
+    }
+    const auto& thrift_metadata = metadata->to_thrift();
+    const auto compat = native::parquet_reader_compat(
+            thrift_metadata.__isset.created_by ? thrift_metadata.created_by : 
"");
+    RETURN_IF_ERROR(native::ColumnReader::create(
+            std::move(file), field, row_group, _row_ranges, timezone, io_ctx, 
_native_reader,
+            max_buffer_size, *_offset_indexes, native_runtime_state, false, 
_projected_column_ids,
+            _filter_column_ids, page_cache_file_key, compat,
+            runtime_state != nullptr && runtime_state->enable_strict_mode()));
+    DORIS_CHECK(_native_reader != nullptr);
+    _skip_column = _type->create_column();
+    return Status::OK();
+}
+
+Status NativeColumnReader::read_with_filter(int64_t rows, const uint8_t* 
filter_data,
+                                            bool filter_all, MutableColumnPtr& 
column,
+                                            const DataTypePtr& output_type, 
bool dictionary_ids,
+                                            int64_t* rows_read) {
+    DORIS_CHECK(rows >= 0);
+    DORIS_CHECK(column);
+    DORIS_CHECK(output_type != nullptr);
+    DORIS_CHECK(rows_read != nullptr);
+    *rows_read = 0;
+    if (rows == 0) {
+        return Status::OK();
+    }
+
+    native::FilterMap filter;
+    RETURN_IF_ERROR(filter.init(filter_data, static_cast<size_t>(rows), 
filter_all));
+    _native_reader->reset_filter_map_index();
+    ColumnPtr native_column(std::move(column));
+    bool eof = false;
+    int64_t native_calls = 0;
+    int64_t consecutive_empty_calls = 0;
+    while (*rows_read < rows && !eof) {
+        ++native_calls;
+        size_t loop_rows = 0;
+        RETURN_IF_ERROR(_native_reader->read_column_data(
+                native_column, output_type, _schema_node, filter,
+                static_cast<size_t>(rows - *rows_read), &loop_rows, &eof, 
dictionary_ids));
+        if (loop_rows == 0 && !eof) {
+            // A selected RowRanges plan may reject the current data page 
completely. V1 advances
+            // the page cursor and deliberately returns zero rows so the 
caller can request the
+            // next page. Bound consecutive empty transitions by the Row Group 
row count to retain
+            // a deterministic corruption exit if a decoder ever stops 
advancing.
+            if (++consecutive_empty_calls > _row_group_rows + 1) {
+                column = IColumn::mutate(std::move(native_column));
+                return Status::Corruption("Native parquet reader made no 
progress for column {}",
+                                          _name);
+            }
+            continue;
+        }
+        consecutive_empty_calls = 0;
+        *rows_read += static_cast<int64_t>(loop_rows);
+    }
+    column = IColumn::mutate(std::move(native_column));
+    if (_profile.native_read_calls != nullptr) {
+        COUNTER_UPDATE(_profile.native_read_calls, native_calls);
+    }
+    if (_nested && _profile.nested_batches != nullptr) {
+        COUNTER_UPDATE(_profile.nested_batches, 1);
+    }
+    // Retained-capacity inspection walks the native reader tree. Check it 
periodically instead of
+    // on every small batch; row-group destruction is still the hard lifetime 
bound for scratch.
+    constexpr size_t SCRATCH_CHECK_BATCH_INTERVAL = 16;
+    if (++_batches_since_scratch_check >= SCRATCH_CHECK_BATCH_INTERVAL) {
+        
_native_reader->release_batch_scratch(MAX_RETAINED_BATCH_SCRATCH_BYTES);
+        _batches_since_scratch_check = 0;
+    }
+    if (*rows_read != rows) {
+        return Status::Corruption("Native parquet reader returned {} rows, 
expected {} for {}",
+                                  *rows_read, rows, _name);
+    }
+    return Status::OK();
+}
+
+Status NativeColumnReader::read_with_plain_filter(int64_t rows, const uint8_t* 
filter_data,
+                                                  bool filter_all, const 
VExprSPtrs& conjuncts,
+                                                  int column_id, 
IColumn::Filter* row_filter,
+                                                  int64_t* rows_read, bool* 
used_filter) {
+    DORIS_CHECK(rows >= 0);
+    DORIS_CHECK(row_filter != nullptr);
+    DORIS_CHECK(rows_read != nullptr);
+    DORIS_CHECK(used_filter != nullptr);
+    row_filter->clear();
+    *rows_read = 0;
+    *used_filter = false;
+    if (rows == 0) {
+        return Status::OK();
+    }
+
+    native::FilterMap filter;
+    RETURN_IF_ERROR(filter.init(filter_data, static_cast<size_t>(rows), 
filter_all));
+    _native_reader->reset_filter_map_index();
+    bool eof = false;
+    int64_t consecutive_empty_calls = 0;
+    while (*rows_read < rows && !eof) {
+        size_t loop_rows = 0;
+        IColumn::Filter loop_filter;
+        bool loop_used = false;
+        RETURN_IF_ERROR(_native_reader->read_plain_filter(
+                conjuncts, column_id, filter, static_cast<size_t>(rows - 
*rows_read), &loop_filter,
+                &loop_rows, &eof, &loop_used));
+        if (!loop_used) {
+            // A fallback is safe only before this call has consumed a logical 
row. Plain-only
+            // chunk validation makes the decision stable for all later page 
fragments.
+            DORIS_CHECK_EQ(*rows_read, 0);

Review Comment:
   [P1] Return `Corruption` here instead of asserting once the direct PLAIN 
path has made progress. The chunk is classified PLAIN-only from untrusted 
footer metadata, so a two-page file whose footer omits page 2's DELTA encoding 
consumes page 1, then page 2 returns `loop_used == false` with `*rows_read > 0` 
and this `DORIS_CHECK_EQ` terminates the BE. Keep ordinary fallback for the 
zero-progress case, but report an encoding mismatch after progress as a query 
error and cover it with a two-page malformed-metadata test.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org
For additional commands, e-mail: commits-h...@doris.apache.org