This is an automated email from the ASF dual-hosted git repository.
gavinchou pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 76bd6f0a138 [fix](rle) Fix infinite loop when a repeated run reaches
2^30 values (#69100)
76bd6f0a138 is described below
commit 76bd6f0a138a17dde4b4b8123038ca5c4679a2c9
Author: Xin Liao <[email protected]>
AuthorDate: Fri Oct 16 16:20:57 2026 +0800
[fix](rle) Fix infinite loop when a repeated run reaches 2^30 values
(#69100)
### What problem does this PR solve?
Problem Summary:
`RleEncoder::FlushRepeatedRun()` wrote `repeat_count_ << 1` into an
`int32_t`. `repeat_count_` was an `int`, so a repeated run of 2^30 or
more values overflowed the indicator into a negative number.
`BitWriter::PutVlqInt()` then took that as an `int32_t` and shifted it
right arithmetically:
```cpp
inline void BitWriter::PutVlqInt(int32_t v) {
while ((v & 0xFFFFFF80) != 0L) {
PutAligned<uint8_t>((v & 0x7F) | 0x80, 1);
v >>= 7;
}
PutAligned<uint8_t>(v & 0x7F, 1);
}
```
An arithmetic right shift of a negative value converges to `-1` and
stays there, and `-1 & 0xFFFFFF80` is non-zero, so **the loop never
exits**. It appends `0xFF` to the output buffer forever, pinning a core
at 100% and growing the buffer without bound.
#### How it was hit
A base compaction in a production cluster wedged on a table with many
nullable `text` columns. Two `BaseCompactionT` threads sat in this stack
for over a month:
```
#0 inline_memcpy (size=348057208)
#2 doris::faststring::GrowArray (newcapacity=502825728)
#4 doris::BitWriter::GetNextBytePtr (num_bytes=1)
#5 doris::BitWriter::PutAligned<unsigned char> (val=255 '\377')
#6 doris::BitWriter::PutVlqInt (v=-1)
bit_stream_utils.inline.h:91
#7 doris::RleEncoder<bool>::FlushRepeatedRun rle_encoding.h:499
#9 doris::segment_v2::NullBitmapBuilder::finish
#11 doris::segment_v2::ScalarColumnWriter::finish
#15 doris::Merger::vertical_compact_one_group (is_key=false,
max_rows_per_segment=1073741824)
#20 doris::CloudBaseCompaction::execute_compact
```
`v=-1` together with `val=255` pins it down exactly: the loop is
emitting `0xFF` with `v` stuck at `-1`.
Two existing behaviours combined to produce a 2^30-value run:
- Vertical compaction derives the segment row limit from
`vertical_compaction_max_segment_size / avg_bytes_per_row`
([`Compaction::get_avg_segment_rows()`](https://github.com/apache/doris/blob/master/be/src/olap/compaction.cpp)).
For this table the average row size integer-divided to `0`, so the
divisor degenerated to `1` and the limit became the config default
`1073741824` = 2^30 rows — visible as `max_rows_per_segment=1073741824`
in the stack above.
- `ScalarColumnWriter::append_nulls()` never checks whether the page is
full, unlike `append_data()`, and NULLs never reach the page builder. So
an all-NULL column accumulated a single null-bitmap run spanning the
whole 2^30-row segment in one page.
Because the stuck tasks kept the base compaction slots occupied
(`max_base_compaction_task_num_per_disk` defaults to 2),
`_generate_cloud_compaction_tasks()` computed `base_n = 0` and stopped
picking any tablet, so **no base compaction ran on that backend again**
and compaction scores climbed to the cap across the instance.
### What this PR changes
1. **`PutVlqInt()` takes `uint32_t`.** A logical shift always converges
to 0, so the loop terminates for every possible input. This is the
backstop that makes a hang impossible regardless of what callers pass.
2. **`repeat_count_` widened to `int64_t`, and long runs are split.**
Runs longer than `MAX_VALUES_PER_REPEATED_RUN` (2^30 - 1) are emitted as
several back-to-back repeated runs of the same value. Readers already
merge adjacent runs carrying the same value (`RleDecoder::GetNextRun`),
so splitting is invisible to them. The cap keeps the indicator inside
`int32_t`'s positive range — exactly the range previous versions could
encode — so **output stays byte-identical for every run length that used
to work**, and no reader change is needed.
3. **`faststring::reserve()` grows geometrically** via the existing
`GrowToAtLeast()` instead of reallocating to exactly the requested
capacity. `BitWriter` reserves a handful of bytes at a time, so exact
growth reallocated and memcpy'd the entire buffer every 8 bytes, making
an n-byte append cost O(n^2). This is what turned the infinite loop into
a thread stuck in `memcpy` with a ~500MB buffer, and it slows down every
other large RLE buffer as well. Measured: reaching 1 MiB now takes 26
reallocations instead of ~131,000.
Deliberately **not** in this PR, each worth its own change:
- `ScalarColumnWriter::append_nulls()` still lets a page grow without
bound.
- `Compaction::get_avg_segment_rows()` still degenerates to the
configured byte size when the average row size rounds down to zero.
### Release note
Fix a BE hang in RLE encoding when a column produces a run of 2^30 or
more identical values (for example an all-NULL column in a very large
segment), which could permanently stall base compaction on that backend.
### Check List (For Author)
- Test
- [x] Unit Test
New tests:
| Test | Covers |
|---|---|
| `TestRle.TestRepeatedRunExceedingIndicatorRange` | Run lengths around
and far beyond 2^30 encode compactly and decode back to the original
length |
| `TestBitStreamUtil.TestPutVlqIntRoundTrip` | `PutVlqInt` round-trips
every `uint32_t`, including values `>= 2^31` that used to hang |
| `FaststringTest.TestReserve_ExponentiallyExpand` | `reserve()` does
not reallocate on every call |
Verified that the new tests actually catch the bug: built the exact
repro (`RleEncoder<bool>::Put(true, 1 << 30)` followed by `Flush()`)
against unpatched master and it never returns (killed by timeout); with
this patch it completes and emits 8 bytes. All pre-existing tests in
`rle_encoding_test`, `bit_stream_utils_test` and `faststring_test` pass
(29 tests, with `LOOP_LESS_OR_MORE` forced to the slow/thorough branch).
Also confirmed no format regression: a short run encoded via `Put(v)` in
a loop and via `Put(v, n)` in one call still produce byte-identical
output.
- Behavior changed:
- [x] No. Encoded output is unchanged for every run length that
previously encoded correctly; only run lengths that used to hang now
produce (correct, readable) output.
- Does this need documentation?
- [x] No.
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 fbaa219d012..270a0635ed3 100644
--- a/be/src/util/bit_stream_utils.h
+++ b/be/src/util/bit_stream_utils.h
@@ -59,7 +59,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 2c3ceb79e44..f24e8ce90f2 100644
--- a/be/src/util/bit_stream_utils.inline.h
+++ b/be/src/util/bit_stream_utils.inline.h
@@ -85,12 +85,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 f96d00e7941..18442456c49 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"
@@ -156,6 +158,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
@@ -212,7 +224,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
@@ -437,7 +449,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;
}
@@ -457,7 +469,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;
}
}
@@ -514,11 +526,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]
