[fix](parquet) Handle empty value sections

### What problem does this PR solve?

Issue Number: close #66430

Related PR: None

Problem Summary:
A compressed Parquet DataPageV2 can legally contain definition levels
but no physical values when every nullable value is NULL. The legacy
reader passed the resulting null, zero-length slice to the BOOLEAN
PLAIN decoder and aborted the BE in BatchedBitReader::Reset().

Use a dedicated empty-value-section decoder for these pages. Accept the
page when definition levels require no physical values, and return
Corruption when values are required but absent. Update page progress
only after decoding succeeds so failures leave reader state unchanged.

### Release note

Fix a BE crash when reading Parquet pages with empty value sections.

### Check List (For Author)

- Test <!-- At least one of them must be included. -->
    - [ ] Regression test
    - [x] Unit Test
        - ./run-be-ut.sh --run --filter=ParquetColumnChunkReaderTest.* -j 2
        - All 11 tests passed with ASAN.
    - [x] Manual test (add detailed scripts or steps below)
        - ./build.sh --be -j 2
        - BE build succeeded.
    - [ ] No need to test or manual test. Explain why:
        - [ ] This is a refactor/code format and no logic has been changed.
        - [ ] Previous test can cover this change.
        - [ ] No code files have been changed.
        - [ ] Other reason <!-- Add your reason? -->

- Behavior changed:
    - [ ] No.
    - [x] Yes. Valid all-NULL Parquet pages are accepted. Pages whose
      definition levels require missing physical values return
      Corruption instead of aborting the BE.

- Does this need documentation?
    - [x] No.
    - [ ] Yes. <!-- Add document PR link here. eg: apache/doris-website#1214 -->

### Check List (For Reviewer who merge this PR)

- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR should merge into -->

Author: acarofpigs

be/src/format/parquet/vparquet_column_chunk_reader.cpp

@@ -43,6 +43,35 @@ namespace cctz {
 class time_zone;
 } // namespace cctz
 namespace doris {
+
+namespace {
+
+class EmptyValueSectionDecoder final : public Decoder {
+public:
+    Status decode_values(MutableColumnPtr& doris_column, DataTypePtr&,
+                         ColumnSelectVector& select_vector, bool) override {
+        const size_t physical_values = select_vector.num_values() - select_vector.num_nulls();
+        if (UNLIKELY(physical_values != 0)) {
+            return Status::Corruption(
+                    "Parquet definition levels require {} values from an empty value section",
+                    physical_values);
+        }
+        doris_column->insert_many_defaults(select_vector.num_values() -
+                                           select_vector.num_filtered());
+        return Status::OK();
+    }
+
+    Status skip_values(size_t num_values) override {
+        if (UNLIKELY(num_values != 0)) {
+            return Status::Corruption(
+                    "Parquet definition levels require {} values from an empty value section",
+                    num_values);
+        }
+        return Status::OK();
+    }
+};
+
+} // namespace
 namespace io {
 class BufferedStreamReader;
 struct IOContext;
@@ -358,17 +387,29 @@ Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::load_page_data() {
     }
 
     // Reuse page decoder
+    Decoder* encoding_decoder = nullptr;
     if (_decoders.find(static_cast<int>(encoding)) != _decoders.end()) {
-        _page_decoder = _decoders[static_cast<int>(encoding)].get();
+        encoding_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_decoder = _decoders[static_cast<int>(encoding)].get();
+    }
+    _empty_value_section = _page_data.empty() && _max_def_level > 0;
+    if (_empty_value_section) {
+        // Nullable all-NULL pages legally contain only definition levels. Use a dedicated decoder
+        // so a later non-NULL definition level cannot consume stale state from the previous page.
+        if (_empty_value_decoder == nullptr) {
+            _empty_value_decoder = std::make_unique<EmptyValueSectionDecoder>();
+        }
+        _page_decoder = _empty_value_decoder.get();
+    } else {
+        _page_decoder = encoding_decoder;
+        RETURN_IF_ERROR(_page_decoder->set_data(&_page_data));
     }
-    RETURN_IF_ERROR(_page_decoder->set_data(&_page_data));
 
     _state = DATA_LOADED;
     return Status::OK();
@@ -546,13 +587,13 @@ Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::skip_values(size_t num_va
         return Status::IOError("Skip too many values in current page. {} vs. {}",
                                _remaining_num_values, num_values);
     }
-    _remaining_num_values -= num_values;
     if (skip_data) {
         SCOPED_RAW_TIMER(&_chunk_statistics.decode_value_time);
-        return _page_decoder->skip_values(num_values);
-    } else {
-        return Status::OK();
+        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>
@@ -569,8 +610,10 @@ Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::decode_values(
     if (UNLIKELY(_remaining_num_values < select_vector.num_values())) {
         return Status::IOError("Decode too many values in current page");
     }
+    RETURN_IF_ERROR(
+            _page_decoder->decode_values(doris_column, data_type, select_vector, is_dict_filter));
     _remaining_num_values -= select_vector.num_values();
-    return _page_decoder->decode_values(doris_column, data_type, select_vector, is_dict_filter);
+    return Status::OK();
 }
 
 template <bool IN_COLLECTION, bool OFFSET_INDEX>

be/src/format/parquet/vparquet_column_chunk_reader.h

@@ -277,7 +277,9 @@ class ColumnChunkReader {
     Slice _v2_def_levels;
     bool _dict_checked = false;
     bool _has_dict = false;
+    bool _empty_value_section = false;
     Decoder* _page_decoder = nullptr;
+    std::unique_ptr<Decoder> _empty_value_decoder;
     // Map: encoding -> Decoder
     // Plain or Dictionary encoding. If the dictionary grows too big, the encoding will fall back to the plain encoding
     std::unordered_map<int, std::unique_ptr<Decoder>> _decoders;

be/test/format/parquet/parquet_column_chunk_reader_test.cpp

@@ -27,13 +27,17 @@
 
 #include "core/assert_cast.h"
 #include "core/column/column_string.h"
+#include "core/column/column_vector.h"
+#include "core/data_type/data_type_number.h"
 #include "format/parquet/schema_desc.h"
 #include "format/parquet/vparquet_column_chunk_reader.h"
 #include "format/parquet/vparquet_column_reader.h"
 #include "io/fs/buffered_reader.h"
 #include "io/fs/file_reader.h"
 #include "runtime/runtime_state.h"
+#include "util/block_compression.h"
 #include "util/coding.h"
+#include "util/faststring.h"
 #include "util/thrift_util.h"
 
 namespace doris {
@@ -251,6 +255,52 @@ Status make_plain_fixture(ColumnChunkFixture* fixture, int page_count = 1) {
     return Status::OK();
 }
 
+Status make_empty_boolean_v2_fixture(bool is_null, ColumnChunkFixture* fixture) {
+    BlockCompressionCodec* codec = nullptr;
+    RETURN_IF_ERROR(get_block_compression_codec(segment_v2::CompressionTypePB::SNAPPY, &codec));
+    const uint8_t unused = 0;
+    faststring compressed_values;
+    RETURN_IF_ERROR(codec->compress(Slice(&unused, 0), &compressed_values));
+
+    const std::vector<uint8_t> definition_levels {2, static_cast<uint8_t>(is_null ? 0 : 1)};
+    std::vector<uint8_t> payload = definition_levels;
+    if (compressed_values.size() != 0) {
+        payload.insert(payload.end(), compressed_values.data(),
+                       compressed_values.data() + compressed_values.size());
+    }
+
+    tparquet::DataPageHeaderV2 data_header;
+    data_header.__set_num_values(1);
+    data_header.__set_num_nulls(is_null ? 1 : 0);
+    data_header.__set_num_rows(1);
+    data_header.__set_encoding(tparquet::Encoding::PLAIN);
+    data_header.__set_definition_levels_byte_length(definition_levels.size());
+    data_header.__set_repetition_levels_byte_length(0);
+    data_header.__set_is_compressed(true);
+
+    tparquet::PageHeader header;
+    header.type = tparquet::PageType::DATA_PAGE_V2;
+    header.__set_compressed_page_size(cast_set<int32_t>(payload.size()));
+    header.__set_uncompressed_page_size(cast_set<int32_t>(definition_levels.size()));
+    header.__set_data_page_header_v2(data_header);
+
+    int64_t data_page_offset = 0;
+    int32_t data_page_size = 0;
+    RETURN_IF_ERROR(
+            append_page(&header, payload, fixture->data, &data_page_offset, &data_page_size));
+
+    auto& metadata = fixture->chunk.meta_data;
+    metadata.__set_type(tparquet::Type::BOOLEAN);
+    metadata.__set_codec(tparquet::CompressionCodec::SNAPPY);
+    metadata.__set_num_values(1);
+    metadata.__set_data_page_offset(data_page_offset);
+    metadata.__set_total_compressed_size(data_page_size);
+
+    fixture->field_schema.physical_type = tparquet::Type::BOOLEAN;
+    fixture->field_schema.definition_level = 1;
+    return Status::OK();
+}
+
 void expect_dictionary_values(ColumnChunkReader<false, false>* reader) {
     MutableColumnPtr column = ColumnString::create();
     ASSERT_TRUE(reader->read_dict_values_to_column(column).ok());
@@ -386,6 +436,59 @@ TEST(ParquetColumnChunkReaderTest, FailedDictionaryCheckCanBeRetried) {
     }
 }
 
+TEST(ParquetColumnChunkReaderTest, CompressedV2AllNullBooleanAcceptsEmptyValueSection) {
+    ColumnChunkFixture fixture;
+    ASSERT_TRUE(make_empty_boolean_v2_fixture(true, &fixture).ok());
+    CountingBufferedReader buffered_reader(std::move(fixture.data));
+    ParquetPageReadContext page_read_ctx(false);
+    ColumnChunkReader<false, false> reader(&buffered_reader, &fixture.chunk, &fixture.field_schema,
+                                           nullptr, 1, nullptr, page_read_ctx);
+
+    ASSERT_TRUE(reader.init().ok());
+    ASSERT_TRUE(reader.parse_page_header().ok());
+    ASSERT_TRUE(reader.load_page_data().ok());
+    EXPECT_TRUE(reader.get_page_data().empty());
+    EXPECT_TRUE(reader.skip_values(0).ok());
+
+    FilterMap filter_map;
+    ASSERT_TRUE(filter_map.init(nullptr, 0, false).ok());
+    ColumnSelectVector select_vector;
+    const std::vector<uint16_t> null_map {0, 1};
+    ASSERT_TRUE(select_vector.init(null_map, 1, nullptr, &filter_map, 0).ok());
+    MutableColumnPtr column = ColumnUInt8::create();
+    DataTypePtr data_type = std::make_shared<DataTypeUInt8>();
+    ASSERT_TRUE(reader.decode_values(column, data_type, select_vector, false).ok());
+    ASSERT_EQ(column->size(), 1);
+    EXPECT_EQ(assert_cast<const ColumnUInt8&>(*column).get_data()[0], 0);
+}
+
+TEST(ParquetColumnChunkReaderTest, CompressedV2NonNullBooleanRejectsEmptyValueSection) {
+    ColumnChunkFixture fixture;
+    ASSERT_TRUE(make_empty_boolean_v2_fixture(false, &fixture).ok());
+    CountingBufferedReader buffered_reader(std::move(fixture.data));
+    ParquetPageReadContext page_read_ctx(false);
+    ColumnChunkReader<false, false> reader(&buffered_reader, &fixture.chunk, &fixture.field_schema,
+                                           nullptr, 1, nullptr, page_read_ctx);
+
+    ASSERT_TRUE(reader.init().ok());
+    ASSERT_TRUE(reader.parse_page_header().ok());
+    ASSERT_TRUE(reader.load_page_data().ok());
+    EXPECT_TRUE(reader.skip_values(1).is<ErrorCode::CORRUPTION>());
+    EXPECT_EQ(reader.remaining_num_values(), 1);
+
+    FilterMap filter_map;
+    ASSERT_TRUE(filter_map.init(nullptr, 0, false).ok());
+    ColumnSelectVector select_vector;
+    const std::vector<uint16_t> null_map {1};
+    ASSERT_TRUE(select_vector.init(null_map, 1, nullptr, &filter_map, 0).ok());
+    MutableColumnPtr column = ColumnUInt8::create();
+    DataTypePtr data_type = std::make_shared<DataTypeUInt8>();
+    EXPECT_TRUE(reader.decode_values(column, data_type, select_vector, false)
+                        .is<ErrorCode::CORRUPTION>());
+    EXPECT_EQ(reader.remaining_num_values(), 1);
+    EXPECT_TRUE(column->empty());
+}
+
 TEST(ParquetColumnChunkReaderTest, ScalarDictionaryReadUsesExplicitProbe) {
     ColumnChunkFixture fixture;
     ASSERT_TRUE(make_dictionary_fixture(&fixture).ok());