branch-4.2: [fix](be) Stop preallocating compression-sized t-digest b…

be/src/util/tdigest.h

@@ -144,8 +144,13 @@ class TDigest {
             : _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 @@ class TDigest {
             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());

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());
 }
 

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