This is an automated email from the ASF dual-hosted git repository.
HappenLee 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 c9bd8b0894f [fix](be) Stop preallocating compression-sized t-digest
buffers per group (#69284)
c9bd8b0894f is described below
commit c9bd8b0894fb9c3315e5bc740be875034d0e90d2
Author: Jerry Hu <[email protected]>
AuthorDate: Thu Oct 22 11:43:54 2026 +0800
[fix](be) Stop preallocating compression-sized t-digest buffers per group
(#69284)
### What problem does this PR solve?
Issue Number: None
Problem Summary:
`percentile_approx` and `quantile_union` keep one `TDigest` per GROUP BY
group. The
`TDigest(compression, unmerged_size, merged_size)` constructor eagerly
reserved
`2 * compression` processed centroids and `8 * compression + 1`
unprocessed centroids before
any value was stored. At the default `compression = 10000` this costs
800008 bytes per group
(160000 B + 640008 B), independent of how many rows the group actually
holds.
Both allocations exceed the ASAN primary allocator's 128 KiB size-class
limit, so each digest
occupied two secondary-allocator chunks. With roughly 524288 live groups
the process exhausted
the secondary allocator's chunk table (`kMaxNumChunks = 1 << 20`) and
aborted in
`sanitizer_allocator_secondary.h:42`. Release builds map about 480 GB of
virtual address space
for 600K singleton groups in the same scenario.
Fix:
- Let both vectors grow on demand in the constructor. `_max_processed`
and `_max_unprocessed`
are still initialized exactly as before: they are compression thresholds
used by
`is_dirty()`, the buffered `add()` path, and serialization, not capacity
caches.
- Remove the exact-size `reserve(total)` from `_merge_unprocessed`. Once
the constructor no
longer preallocates, `reserve(total)` would reallocate and copy the
whole buffer on every
merge of a small digest, degrading many sequential merges to O(n^2)
copies. `insert()` grows
geometrically instead, keeping amortized O(1).
The compression thresholds, the t-digest algorithm, the numeric results,
and the serialized
format are unchanged.
Tests updated/added:
- New unit test: a default-constructed digest has zero buffer capacity,
allocates less than
4 KiB in total, keeps the historical thresholds (20000 / 80000), and
still returns the right
quantile after a single value.
- New unit test: 1000 singleton groups allocate memory proportional to
their data
(< 1 KiB per group) and each returns the correct median.
- Updated an existing accumulator test that previously asserted the
80001 preallocation; it
now asserts that the accumulator retains spare write-buffer capacity
without encoding the
removed preallocation.
### Release note
Fix backend aborts for `percentile_approx` / `quantile_union` queries
with a very high number
of GROUP BY groups by no longer preallocating compression-sized buffers
per group.
### Check List (For Author)
- Test: Unit Test
- BE unit tests under ASAN: 41 tests from 5 suites passed, covering
`TDigestTest`
(including the two new capacity/memory tests), `QuantileStateTest`,
`AggregateFunctionPercentileApproxArrayTest`,
`AggregateFunctionQuantileStateTest`,
and `AggregateFunctionQuantileStateRangeTest`. Command:
`./run-be-ut.sh --run
--filter='TDigestTest.*:QuantileStateTest.*:AggregateFunctionPercentileApproxArrayTest.*:AggregateFunctionQuantileStateTest.*:*AggregateFunctionQuantileStateRangeTest.*'
-j 64`
- A standalone ASAN program holding 600000 live `TDigest` instances now
runs to
completion; the same program aborted at about 524288 digests before the
fix.
- A negative-control build against the old header confirms the new
assertions fail on the
unfixed code (capacities 20000 / 80001 and 800008 bytes per digest; 1000
singleton
groups allocate 800008000 bytes).
- `clang-format` (v16) passes on the changed files, the repository-wide
format check
passes, and the build-hygiene check passes.
- `clang-tidy` could not run on this machine: its toolchain reports
`'stddef.h' file not found` for the changed files, unrelated to this
change.
- No regression test was added: reproducing the abort requires more than
~524288 live
digests, which no regression cluster can hold, and release builds
neither abort nor
return an error row. The new unit tests assert the exact capacity and
allocation
behavior at the layer where the fix lives.
- Behavior changed: No
- Does this need documentation: No
---
be/src/util/tdigest.h | 18 +++++-----
be/test/exprs/aggregate/agg_percentile_test.cpp | 5 ++-
be/test/util/tdigest_test.cpp | 44 +++++++++++++++++++++++++
3 files changed, 58 insertions(+), 9 deletions(-)
diff --git a/be/src/util/tdigest.h b/be/src/util/tdigest.h
index 1d5202a2d9b..1ce7d4394c2 100644
--- a/be/src/util/tdigest.h
+++ b/be/src/util/tdigest.h
@@ -143,8 +143,13 @@ public:
: _compression(compression),
_max_processed(processed_size(merged_size, compression)),
_max_unprocessed(unprocessed_size(unmerged_size, compression)) {
- _processed.reserve(_max_processed);
- _unprocessed.reserve(_max_unprocessed + 1);
+ // Deliberately do not preallocate. Aggregations keep one digest per
group
+ // (percentile_approx, quantile_union), so reserving 2*compression
processed and
+ // 8*compression unprocessed centroids up front costs O(groups *
compression) memory
+ // before any value is stored: 800KB per group at the default
compression 10000, which
+ // is 480GB of virtual address space for 600K singleton groups. The
buffers grow on
+ // demand instead; _max_processed and _max_unprocessed still bound
them and drive
+ // exactly the same compression thresholds as before.
}
TDigest(std::vector<Centroid>&& processed, std::vector<Centroid>&&
unprocessed,
@@ -630,12 +635,9 @@ private:
return;
}
- size_t total = _unprocessed.size();
- for (const auto& td : tdigests) {
- total += td->_unprocessed.size();
- }
-
- _unprocessed.reserve(total);
+ // Insert directly instead of reserving the exact merged size:
reserve() reallocates
+ // and copies the whole buffer whenever `total` exceeds the current
capacity, so
+ // merging many small digests would cost O(n^2) copies. insert() grows
geometrically.
for (const auto& td : tdigests) {
_unprocessed.insert(_unprocessed.end(), td->_unprocessed.cbegin(),
td->_unprocessed.cend());
diff --git a/be/test/exprs/aggregate/agg_percentile_test.cpp
b/be/test/exprs/aggregate/agg_percentile_test.cpp
index 28c31f0737f..7a1030e837c 100644
--- a/be/test/exprs/aggregate/agg_percentile_test.cpp
+++ b/be/test/exprs/aggregate/agg_percentile_test.cpp
@@ -185,7 +185,10 @@ TEST(AggregateFunctionQuantileStateTest,
GrowingWindowKeepsResultsCompact) {
EXPECT_NEAR(expected[i],
result.get_value_by_percentile(quantiles[i]), 1.0);
}
}
- EXPECT_GE(accumulator._mutable_tdigest().unprocessed().capacity(), 80001);
+ // The accumulator keeps its write buffer for later rows instead of
shrinking like a
+ // result copy: its unprocessed vector still has spare capacity after the
window, while
+ // the retained result digests above are sized exactly.
+ EXPECT_GT(accumulator._mutable_tdigest().unprocessed().capacity(), 0);
EXPECT_FALSE(accumulator._mutable_tdigest().have_unprocessed());
}
diff --git a/be/test/util/tdigest_test.cpp b/be/test/util/tdigest_test.cpp
index 392724e1862..e7c9e9fab99 100644
--- a/be/test/util/tdigest_test.cpp
+++ b/be/test/util/tdigest_test.cpp
@@ -23,6 +23,7 @@
#include <memory>
#include <numeric>
#include <random>
+#include <vector>
#include "gtest/gtest_pred_impl.h"
#include "testutil/test_util.h"
@@ -342,4 +343,47 @@ TEST_F(TDigestTest, Montonicity) {
}
}
+TEST_F(TDigestTest, DefaultConstructorDoesNotPrepareCompressionSizedBuffers) {
+ const auto allocated_bytes = [](const TDigest& digest) {
+ return (digest._processed.capacity() + digest._unprocessed.capacity())
* sizeof(Centroid) +
+ digest._cumulative.capacity() * sizeof(Weight);
+ };
+
+ TDigest digest(10000);
+ // The compression thresholds keep their historical semantics.
+ EXPECT_EQ(20000, digest.max_processed());
+ EXPECT_EQ(80000, digest.max_unprocessed());
+ // A fresh digest must not preallocate O(compression) buffers.
+ EXPECT_EQ(0, digest.processed().capacity());
+ EXPECT_EQ(0, digest.unprocessed().capacity());
+ EXPECT_LT(allocated_bytes(digest), 4 * 1024);
+
+ digest.add(42.0f);
+ EXPECT_EQ(1, digest.unprocessed().size());
+ EXPECT_LT(allocated_bytes(digest), 4 * 1024);
+ EXPECT_FLOAT_EQ(42.0f, digest.quantile(0.5f));
+}
+
+TEST_F(TDigestTest, SingletonGroupsKeepMemoryProportionalToData) {
+ // percentile_approx keeps one digest per group. With the old eager
reserve every
+ // singleton group held 800000 bytes, so enough live groups exhausted
ASAN's
+ // secondary allocator chunk limit.
+ constexpr size_t kGroups = 1000;
+ std::vector<std::unique_ptr<TDigest>> digests;
+ digests.reserve(kGroups);
+ size_t allocated_bytes = 0;
+ for (size_t i = 0; i < kGroups; ++i) {
+ auto digest = TDigest::create_unique(10000.0f);
+ digest->add(static_cast<float>(i));
+ allocated_bytes += (digest->processed().capacity() +
digest->unprocessed().capacity()) *
+ sizeof(Centroid) +
+ digest->_cumulative.capacity() * sizeof(Weight);
+ digests.push_back(std::move(digest));
+ }
+ EXPECT_LT(allocated_bytes, kGroups * 1024);
+ for (size_t i = 0; i < kGroups; ++i) {
+ EXPECT_FLOAT_EQ(static_cast<float>(i), digests[i]->quantile(0.5f));
+ }
+}
+
} // namespace doris
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
