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 c90d0c8de5f branch-4.2: [fix](be) Stop preallocating compression-sized
t-digest b… (#69470)
c90d0c8de5f is described below
commit c90d0c8de5fd0a407c1243e9e49639a142a24ea7
Author: Jerry Hu <[email protected]>
AuthorDate: Fri Oct 23 10:49:36 2026 +0800
branch-4.2: [fix](be) Stop preallocating compression-sized t-digest b…
(#69470)
…uffers per group #69284
Cherry-picked from #69284 (c9bd8b0894fb9c3315e5bc740be875034d0e90d2).
Conflict resolution: preserved branch-4.2 TDigest camelCase method names
and adapted the new threshold assertions to
maxProcessed()/maxUnprocessed(). Kept the branch-4.2 quantile-state test
structure while replacing its eager-capacity assertion. No unrelated
master changes were imported.
(cherry picked from commit c9bd8b0894fb9c3315e5bc740be875034d0e90d2)
### What problem does this PR solve?
Issue Number: close #xxx
Related PR: #xxx
Problem Summary:
### Release note
None
### Check List (For Author)
- Test <!-- At least one of them must be included. -->
- [ ] Regression test
- [ ] Unit Test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] 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.
- [ ] Yes. <!-- Explain the behavior change -->
- Does this need documentation?
- [ ] No.
- [ ] Yes. <!-- Add document PR link here. eg:
https://github.com/apache/doris-website/pull/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 -->
---
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 a37a606db01..7a6f5cfe569 100644
--- a/be/src/util/tdigest.h
+++ b/be/src/util/tdigest.h
@@ -144,8 +144,13 @@ public:
: _compression(compression),
_max_processed(processedSize(mergedSize, compression)),
_max_unprocessed(unprocessedSize(unmergedSize, 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,
@@ -573,12 +578,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 97f74de21fc..d8497cba9cc 100644
--- a/be/test/exprs/aggregate/agg_percentile_test.cpp
+++ b/be/test/exprs/aggregate/agg_percentile_test.cpp
@@ -117,7 +117,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().haveUnprocessed());
}
diff --git a/be/test/util/tdigest_test.cpp b/be/test/util/tdigest_test.cpp
index 8d7c0408f5a..1ebed599444 100644
--- a/be/test/util/tdigest_test.cpp
+++ b/be/test/util/tdigest_test.cpp
@@ -22,6 +22,7 @@
#include <memory>
#include <random>
+#include <vector>
#include "gtest/gtest_pred_impl.h"
#include "testutil/test_util.h"
@@ -294,4 +295,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.maxProcessed());
+ EXPECT_EQ(80000, digest.maxUnprocessed());
+ // 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]
