[Fix](parquet-reader) Fix definition level rle decode dead loop in parquet-reader.

Author: kaka11chen

be/src/util/bit_stream_utils.h

@@ -145,6 +145,10 @@ class BitReader {
 
     bool is_initialized() const { return buffer_ != nullptr; }
 
+    const uint8_t* buffer() const { return buffer_; }
+
+    int max_bytes() const { return max_bytes_; }
+
 private:
     // Used by SeekToBit() and GetValue() to fetch the
     // the next word into buffer_.

be/src/util/rle_encoding.h

@@ -120,6 +120,8 @@ class RleDecoder {
     // Get current repeated value, make sure that count equals repeated_count()
     T get_repeated_value(size_t count);
 
+    const BitReader& bit_reader() const { return bit_reader_; }
+
 private:
     bool ReadHeader();
 

be/src/vec/exec/format/parquet/level_decoder.h

@@ -56,6 +56,8 @@ class LevelDecoder {
 
     inline void rewind_one() { _rle_decoder.RewindOne(); }
 
+    const RleDecoder<level_t>& rle_decoder() const { return _rle_decoder; }
+
 private:
     tparquet::Encoding::type _encoding;
     level_t _bit_width = 0;
@@ -65,4 +67,4 @@ class LevelDecoder {
     BitReader _bit_packed_decoder;
 };
 
-} // namespace doris::vectorized
+} // namespace doris::vectorized

be/src/vec/exec/format/parquet/vparquet_column_reader.cpp

@@ -1,4 +1,5 @@
 // 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
@@ -209,7 +210,15 @@ Status ScalarColumnReader::_skip_values(size_t num_values) {
             level_t def_level = -1;
             size_t loop_skip = def_decoder.get_next_run(&def_level, num_values - skipped);
             if (loop_skip == 0) {
-                continue;
+                std::stringstream ss;
+                auto& bit_reader = def_decoder.rle_decoder().bit_reader();
+                ss << "def_decoder buffer (hex): ";
+                for (size_t i = 0; i < bit_reader.max_bytes(); ++i) {
+                    ss << std::hex << std::setw(2) << std::setfill('0')
+                       << static_cast<int>(bit_reader.buffer()[i]) << " ";
+                }
+                LOG(WARNING) << ss.str();
+                return Status::InternalError("Failed to decode definition level.");
             }
             if (def_level == 0) {
                 null_size += loop_skip;
@@ -254,7 +263,15 @@ Status ScalarColumnReader::_read_values(size_t num_values, ColumnPtr& doris_colu
                 level_t def_level;
                 size_t loop_read = def_decoder.get_next_run(&def_level, num_values - has_read);
                 if (loop_read == 0) {
-                    continue;
+                    std::stringstream ss;
+                    auto& bit_reader = def_decoder.rle_decoder().bit_reader();
+                    ss << "def_decoder buffer (hex): ";
+                    for (size_t i = 0; i < bit_reader.max_bytes(); ++i) {
+                        ss << std::hex << std::setw(2) << std::setfill('0')
+                           << static_cast<int>(bit_reader.buffer()[i]) << " ";
+                    }
+                    LOG(WARNING) << ss.str();
+                    return Status::InternalError("Failed to decode definition level.");
                 }
                 bool is_null = def_level == 0;
                 if (!(prev_is_null ^ is_null)) {