This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new 900935edf39 branch-4.2: [fix](rle) Fix infinite loop when a repeated
run reaches 2^30 values #69100 (#69193)
900935edf39 is described below
commit 900935edf39cd8ea0675c21b54799db55733978f
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Fri Oct 16 23:40:55 2026 +0800
branch-4.2: [fix](rle) Fix infinite loop when a repeated run reaches 2^30
values #69100 (#69193)
Cherry-picked from #69100
Co-authored-by: Xin Liao <[email protected]>
Co-authored-by: Xin Liao <[email protected]>
---
be/src/util/bit_stream_utils.h | 4 ++-
be/src/util/bit_stream_utils.inline.h | 8 +++---
be/src/util/faststring.h | 5 +++-
be/src/util/rle_encoding.h | 36 +++++++++++++++++++-----
be/test/util/bit_stream_utils_test.cpp | 25 +++++++++++++++++
be/test/util/faststring_test.cpp | 24 ++++++++++++++++
be/test/util/rle_encoding_test.cpp | 50 ++++++++++++++++++++++++++++++++++
7 files changed, 139 insertions(+), 13 deletions(-)
diff --git a/be/src/util/bit_stream_utils.h b/be/src/util/bit_stream_utils.h
index dc194bf5f8c..2bc82689bd7 100644
--- a/be/src/util/bit_stream_utils.h
+++ b/be/src/util/bit_stream_utils.h
@@ -60,7 +60,9 @@ public:
// Write a Vlq encoded int to the buffer. The value is written byte
aligned.
// For more details on vlq: en.wikipedia.org/wiki/Variable-length_quantity
- void PutVlqInt(int32_t v);
+ // Takes an unsigned value on purpose: the encoding shifts `v` right until
it
+ // is exhausted, which only terminates for a logical shift.
+ void PutVlqInt(uint32_t v);
// Get the index to the next aligned byte and advance the underlying
buffer by num_bytes.
size_t GetByteIndexAndAdvance(int num_bytes) {
diff --git a/be/src/util/bit_stream_utils.inline.h
b/be/src/util/bit_stream_utils.inline.h
index dba339a1b52..da233efd76a 100644
--- a/be/src/util/bit_stream_utils.inline.h
+++ b/be/src/util/bit_stream_utils.inline.h
@@ -86,12 +86,12 @@ void BitWriter::PutAligned(T val, int num_bytes) {
memcpy(ptr, &val, num_bytes);
}
-inline void BitWriter::PutVlqInt(int32_t v) {
- while ((v & 0xFFFFFF80) != 0L) {
- PutAligned<uint8_t>((v & 0x7F) | 0x80, 1);
+inline void BitWriter::PutVlqInt(uint32_t v) {
+ while ((v & 0xFFFFFF80U) != 0U) {
+ PutAligned<uint8_t>(static_cast<uint8_t>((v & 0x7F) | 0x80), 1);
v >>= 7;
}
- PutAligned<uint8_t>(v & 0x7F, 1);
+ PutAligned<uint8_t>(static_cast<uint8_t>(v & 0x7F), 1);
}
inline BitReader::BitReader(const uint8_t* buffer, int buffer_len)
diff --git a/be/src/util/faststring.h b/be/src/util/faststring.h
index 738444553d4..4e2a4c5d74f 100644
--- a/be/src/util/faststring.h
+++ b/be/src/util/faststring.h
@@ -106,7 +106,10 @@ public:
if (newcapacity <= capacity_) [[likely]] {
return;
}
- GrowArray(newcapacity);
+ // Grow by at least 50% rather than to exactly `newcapacity`. Callers
such
+ // as BitWriter reserve a few bytes at a time, and reallocating on
every
+ // call makes appending n bytes cost O(n^2) in memcpy.
+ GrowToAtLeast(newcapacity);
}
// Append the given data to the string, resizing capacity as necessary.
diff --git a/be/src/util/rle_encoding.h b/be/src/util/rle_encoding.h
index 1685ab85f1e..3489fdfd586 100644
--- a/be/src/util/rle_encoding.h
+++ b/be/src/util/rle_encoding.h
@@ -18,6 +18,8 @@
#include <glog/logging.h>
+#include <algorithm>
+#include <cstdint>
#include <limits> // IWYU pragma: keep
#include "common/cast_set.h"
@@ -157,6 +159,16 @@ public:
Clear();
}
+ // The largest number of values a single repeated run can describe.
+ //
+ // A repeated run is written as a vlq-encoded indicator holding
+ // `run_length << 1` (see FlushRepeatedRun) and read back into a uint32_t
by
+ // BitReader::GetVlqInt. Capping the run length here keeps the indicator
+ // inside int32_t's positive range, which is exactly the range every
released
+ // version could already encode, so the output stays byte-identical for any
+ // run length that used to work.
+ static constexpr int64_t MAX_VALUES_PER_REPEATED_RUN = (int64_t {1} << 30)
- 1;
+
// Reserve 'num_bytes' bytes for a plain encoded header, set each
// byte with 'val': this is used for the RLE-encoded data blocks in
// order to be able to able to store the initial ordinal position
@@ -213,7 +225,7 @@ private:
// if we are in a literal run. If the repeat_count_ get high enough, we
switch
// to encoding repeated runs.
uint64_t current_value_;
- int repeat_count_;
+ int64_t repeat_count_;
// Number of literals in the current run. This does not include the
literals
// that might be in buffered_values_. Only after we've got a group big
enough
@@ -438,7 +450,7 @@ void RleEncoder<T>::Put(T value, size_t run_length) {
// Fast path: if this is a continuation of the current repeated run and
// we've already buffered enough values, just increment repeat_count_
if (current_value_ == value && repeat_count_ >= 8 && run_length > 0)
[[likely]] {
- repeat_count_ += run_length;
+ repeat_count_ += static_cast<int64_t>(run_length);
return;
}
@@ -458,7 +470,7 @@ void RleEncoder<T>::Put(T value, size_t run_length) {
// After flushing, if we still have a repeated run and more
values,
// we can add them directly to repeat_count_
if (repeat_count_ >= 8 && run_length > 0) {
- repeat_count_ += run_length;
+ repeat_count_ += static_cast<int64_t>(run_length);
return;
}
}
@@ -515,11 +527,21 @@ void RleEncoder<T>::FlushLiteralRun(bool
update_indicator_byte) {
template <typename T>
void RleEncoder<T>::FlushRepeatedRun() {
DCHECK_GT(repeat_count_, 0);
- // The lsb of 0 indicates this is a repeated run
- int32_t indicator_value = repeat_count_ << 1 | 0;
- bit_writer_.PutVlqInt(indicator_value);
- bit_writer_.PutAligned(current_value_, BitUtil::Ceil(bit_width_, 8));
+ // A run longer than MAX_VALUES_PER_REPEATED_RUN cannot be described by a
+ // single indicator, so emit it as several back-to-back repeated runs of
the
+ // same value. Decoders stitch those back together transparently
+ // (RleDecoder::GetNextRun merges adjacent runs carrying the same value),
so
+ // splitting is invisible to readers.
+ while (repeat_count_ > 0) {
+ const int64_t run_length = std::min(repeat_count_,
MAX_VALUES_PER_REPEATED_RUN);
+ // The lsb of 0 indicates this is a repeated run
+ bit_writer_.PutVlqInt(static_cast<uint32_t>(run_length) << 1);
+ bit_writer_.PutAligned(current_value_, BitUtil::Ceil(bit_width_, 8));
+ repeat_count_ -= run_length;
+ }
num_buffered_values_ = 0;
+ // The loop above drains repeat_count_; assign it anyway so the
postcondition
+ // holds even if it somehow entered non-positive.
repeat_count_ = 0;
}
diff --git a/be/test/util/bit_stream_utils_test.cpp
b/be/test/util/bit_stream_utils_test.cpp
index 3a078f004c0..55976ce4ee3 100644
--- a/be/test/util/bit_stream_utils_test.cpp
+++ b/be/test/util/bit_stream_utils_test.cpp
@@ -218,4 +218,29 @@ TEST(TestBitStreamUtil, TestUint64) {
reader.GetValue(16, &v4);
EXPECT_EQ(v4, 126);
}
+// PutVlqInt() must terminate for every uint32_t. It used to take an int32_t
and
+// shift it right arithmetically, so any value with the sign bit set (>= 2^31)
+// made the loop spin forever appending 0xFF.
+TEST(TestBitStreamUtil, TestPutVlqIntRoundTrip) {
+ const int max_vlq_byte_len = BitReader::MAX_VLQ_BYTE_LEN;
+ const std::vector<uint32_t> values = {
+ 0, 1, 0x7F, 0x80, 0x3FFF, 0x4000,
+ 0x1FFFFF, 0x200000, 0xFFFFFFF, 0x10000000, 0x7FFFFFFE,
0x7FFFFFFF,
+ 0x80000000, 0x80000001, 0xFFFFFFFE, 0xFFFFFFFF,
+ };
+
+ for (uint32_t value : values) {
+ faststring buffer;
+ BitWriter writer(&buffer);
+ writer.PutVlqInt(value);
+ writer.Flush();
+ EXPECT_LE(writer.bytes_written(), max_vlq_byte_len) << "value=" <<
value;
+
+ BitReader reader(buffer.data(), buffer.size());
+ uint32_t decoded = 0;
+ ASSERT_TRUE(reader.GetVlqInt(&decoded)) << "value=" << value;
+ EXPECT_EQ(value, decoded) << "value=" << value;
+ }
+}
+
} // namespace doris
diff --git a/be/test/util/faststring_test.cpp b/be/test/util/faststring_test.cpp
index 1abcd7e97fb..ea708337ab2 100644
--- a/be/test/util/faststring_test.cpp
+++ b/be/test/util/faststring_test.cpp
@@ -135,4 +135,28 @@ TEST_F(FaststringTest, TestAppend_ExponentiallyExpand) {
}
}
+// reserve() must grow geometrically. BitWriter reserves a handful of bytes at
a
+// time, so reallocating to the exact requested size made appending n bytes
cost
+// O(n^2) in memcpy -- enough to stall a thread outright once the buffer
reached
+// hundreds of megabytes.
+TEST_F(FaststringTest, TestReserve_ExponentiallyExpand) {
+ constexpr size_t kTargetSize = 1 * 1024 * 1024;
+
+ faststring s;
+ size_t last_capacity = s.capacity();
+ int reallocations = 0;
+ for (size_t i = 0; i < kTargetSize; ++i) {
+ s.reserve(i + 1);
+ if (s.capacity() != last_capacity) {
+ ++reallocations;
+ last_capacity = s.capacity();
+ }
+ }
+
+ EXPECT_GE(s.capacity(), kTargetSize);
+ // Growing 1.5x at a time from kInitialCapacity needs ~26 steps to pass 1
MiB.
+ // Exact growth would instead reallocate on (almost) every one of the 1M
calls.
+ EXPECT_LT(reallocations, 64);
+}
+
} // namespace doris
diff --git a/be/test/util/rle_encoding_test.cpp
b/be/test/util/rle_encoding_test.cpp
index 5af34168869..59b9368edce 100644
--- a/be/test/util/rle_encoding_test.cpp
+++ b/be/test/util/rle_encoding_test.cpp
@@ -698,6 +698,56 @@ TEST_F(TestRle, TestPutFlushThenFastPath) {
}
}
+// Regression test for a hang on repeated runs that a single run indicator
+// cannot describe.
+//
+// FlushRepeatedRun() used to write `repeat_count_ << 1` into an int32_t, so a
+// run of 2^30 or more values overflowed into a negative indicator.
+// BitWriter::PutVlqInt() then took an int32_t and shifted it right
+// arithmetically, which sticks at -1 instead of reaching 0 -- an infinite loop
+// appending 0xFF to the output buffer. In the wild this showed up as a base
+// compaction thread spinning forever while flushing the null bitmap of a
+// nullable column whose page held one run of 2^30 null rows.
+TEST_F(TestRle, TestRepeatedRunExceedingIndicatorRange) {
+ constexpr int64_t kMaxRun = RleEncoder<bool>::MAX_VALUES_PER_REPEATED_RUN;
+ const std::vector<int64_t> run_lengths = {
+ kMaxRun - 1,
+ kMaxRun,
+ kMaxRun + 1,
+ int64_t {1} << 30, // smallest run length that used to overflow
+ (int64_t {1} << 30) + 12345,
+ int64_t {1} << 31,
+ 3 * (int64_t {1} << 30),
+ };
+
+ for (int64_t run_length : run_lengths) {
+ faststring buffer;
+ RleEncoder<bool> encoder(&buffer, 1);
+ // Put() extends an established run in O(1), so this stays cheap even
for
+ // billions of values.
+ encoder.Put(true, run_length);
+ const int encoded_len = encoder.Flush();
+
+ // A split run stays compact: one short header per emitted run.
+ const int64_t expected_runs = (run_length + kMaxRun - 1) / kMaxRun;
+ EXPECT_LE(encoded_len, expected_runs * 8 + 8) << "run_length=" <<
run_length;
+
+ // GetNextRun() merges adjacent repeated runs carrying the same value,
so
+ // splitting is invisible to readers: this must read back as one run of
+ // the original length.
+ RleDecoder<bool> decoder(buffer.data(), encoded_len, 1);
+ int64_t decoded = 0;
+ while (decoded < run_length) {
+ bool val = false;
+ size_t run = decoder.GetNextRun(&val,
static_cast<size_t>(run_length - decoded));
+ ASSERT_GT(run, 0u) << "run_length=" << run_length << " decoded="
<< decoded;
+ ASSERT_TRUE(val) << "run_length=" << run_length << " decoded=" <<
decoded;
+ decoded += static_cast<int64_t>(run);
+ }
+ EXPECT_EQ(decoded, run_length) << "run_length=" << run_length;
+ }
+}
+
// Benchmark for Put function performance with consecutive equal values
// This test compares:
// 1. Put(value) called N times in a loop
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
