diff --git a/be/src/util/tdigest.h b/be/src/util/tdigest.h index a37a606db0111a..7a6f5cfe56946e 100644 --- a/be/src/util/tdigest.h +++ 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&& processed, std::vector&& 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()); diff --git a/be/test/exprs/aggregate/agg_percentile_test.cpp b/be/test/exprs/aggregate/agg_percentile_test.cpp index 97f74de21fca2d..d8497cba9cc809 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 8d7c0408f5a8fc..1ebed599444c15 100644 --- a/be/test/util/tdigest_test.cpp +++ b/be/test/util/tdigest_test.cpp @@ -22,6 +22,7 @@ #include #include +#include #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> 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(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(i), digests[i]->quantile(0.5f)); + } +} + } // namespace doris