This is an automated email from the ASF dual-hosted git repository.

mrhhsg pushed a commit to branch 4.0.8-safaricom
in repository https://gitbox.apache.org/repos/asf/doris.git


The following commit(s) were added to refs/heads/4.0.8-safaricom by this push:
     new 2318927a6bf [fix](be) Grow t-digest buffers on demand in 4.0
2318927a6bf is described below

commit 2318927a6bfc380293f6c739951eef16e1582dba
Author: Jerry Hu <[email protected]>
AuthorDate: Thu Oct 22 11:43:54 2026 +0800

    [fix](be) Grow t-digest buffers on demand in 4.0
    
    Percentile aggregations keep a t-digest per group. Eagerly reserving
    compression-sized buffers in every digest uses about 800 KiB even when a
    group has only one value. This can cause excessive memory use with many
    sparse groups.
    
    Remove the constructor reservations and the exact-size reserve before
    merging unprocessed centroids. Keep the existing compression thresholds,
    quantile behavior, and serialization format. Add targeted capacity and
    singleton-group memory assertions.
    
    ASAN BE build and 37 related BE unit tests passed. Formatting and final
    diff checks passed. Direct clang-tidy remains blocked by an unchanged
    unmatched NOLINTEND marker in the target base; SQL regression was not run.
    
    Backport of apache/doris#69284 (c9bd8b0894fb9c3315e5bc740be875034d0e90d2).
---
 be/src/util/tdigest.h         | 18 ++++++++++--------
 be/test/util/tdigest_test.cpp | 44 +++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 54 insertions(+), 8 deletions(-)

diff --git a/be/src/util/tdigest.h b/be/src/util/tdigest.h
index 5dad6d5bac9..efe09daddb4 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,
@@ -570,12 +575,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/util/tdigest_test.cpp b/be/test/util/tdigest_test.cpp
index efb40774324..ab656d1f9ec 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"
@@ -237,4 +238,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]

Reply via email to