From 7a25e1f7f625dc2371e40d41fe97ed1a3b572233 Mon Sep 17 00:00:00 2001 From: Debjit Date: Thu, 26 Dec 2024 14:32:52 +0530 Subject: [PATCH 1/2] Perf improvement - Summary metric --- core/src/detail/ckms_quantiles.cc | 75 ++++++++++++++++++------------- 1 file changed, 44 insertions(+), 31 deletions(-) diff --git a/core/src/detail/ckms_quantiles.cc b/core/src/detail/ckms_quantiles.cc index 395b6ff3..94ce4073 100644 --- a/core/src/detail/ckms_quantiles.cc +++ b/core/src/detail/ckms_quantiles.cc @@ -85,42 +85,42 @@ double CKMSQuantiles::allowableError(int rank) { } bool CKMSQuantiles::insertBatch() { + // If there is no data to insert return false if (buffer_count_ == 0) { return false; } + // Sort the buffer upto buffer_count_ to prepare for inserting items std::sort(buffer_.begin(), buffer_.begin() + buffer_count_); std::size_t start = 0; + + sample_.reserve(buffer_count_); + // If the sample set is empty, add the first item if (sample_.empty()) { sample_.emplace_back(buffer_[0], 1, 0); - ++start; + ++start; // Skip the first item since it's already added to the sample ++count_; } - std::size_t idx = 0; - std::size_t item = idx++; - + // Loop through the buffer and insert the items into the sample set for (std::size_t i = start; i < buffer_count_; ++i) { - double v = buffer_[i]; - while (idx < sample_.size() && sample_[item].value < v) { - item = idx++; - } - - if (sample_[item].value > v) { - --idx; + float value = buffer_[i]; + + auto iterator = std::lower_bound( + sample_.begin(), sample_.end(), value, + [](const Item& item, float val) { return item.value < val; }); + std::size_t idx = std::distance(sample_.begin(), iterator); + + int delta = 0; + if (idx > 0 && idx < sample_.size()) { + delta = static_cast( + std::floor(allowableError(static_cast(idx) + 1))) + + 1; } - int delta; - if (idx - 1 == 0 || idx + 1 == sample_.size()) { - delta = 0; - } else { - delta = static_cast(std::floor(allowableError(idx + 1))) + 1; - } - - sample_.emplace(sample_.begin() + idx, v, 1, delta); - count_++; - item = idx++; + sample_.emplace(iterator, value, 1, delta); + ++count_; } buffer_count_ = 0; @@ -128,24 +128,37 @@ bool CKMSQuantiles::insertBatch() { } void CKMSQuantiles::compress() { + // If there are less than 2 items in the sample set, there's nothing to + // compress if (sample_.size() < 2) { return; } - std::size_t idx = 0; - std::size_t prev; - std::size_t next = idx++; + std::vector compressed_samples; // Vector to hold compressed samples + compressed_samples.reserve( + sample_.size()); // Reserve space to avoid multiple allocations + + // Start with the first sample + compressed_samples.push_back(sample_[0]); - while (idx < sample_.size()) { - prev = next; - next = idx++; + for (std::size_t idx = 1; idx < sample_.size(); ++idx) { + const Item& current_sample = sample_[idx]; + Item& last_compressed_sample = compressed_samples.back(); - if (sample_[prev].g + sample_[next].g + sample_[next].delta <= - allowableError(idx - 1)) { - sample_[next].g += sample_[prev].g; - sample_.erase(sample_.begin() + prev); + // Check if we can compress the current sample into the last compressed + // sample + if (last_compressed_sample.g + current_sample.g + current_sample.delta <= + allowableError(static_cast(compressed_samples.size()) - 1)) { + // Merge current sample into last compressed sample + last_compressed_sample.g += current_sample.g; // Update weight + } else { + // If not compressible, add current sample to compressed samples + compressed_samples.push_back(current_sample); } } + + // Replace old samples with new compressed samples + sample_ = std::move(compressed_samples); } } // namespace prometheus::detail From ace740a4fe2604a7b76e56d80dfdec1917392076 Mon Sep 17 00:00:00 2001 From: Gregor Jasny Date: Sat, 29 Aug 2026 21:34:49 +0200 Subject: [PATCH 2/2] chore: add test cases and fix code --- core/src/detail/ckms_quantiles.cc | 11 +++++++---- core/tests/summary_test.cc | 25 +++++++++++++++++++++++++ 2 files changed, 32 insertions(+), 4 deletions(-) diff --git a/core/src/detail/ckms_quantiles.cc b/core/src/detail/ckms_quantiles.cc index 94ce4073..8e17e811 100644 --- a/core/src/detail/ckms_quantiles.cc +++ b/core/src/detail/ckms_quantiles.cc @@ -105,11 +105,11 @@ bool CKMSQuantiles::insertBatch() { // Loop through the buffer and insert the items into the sample set for (std::size_t i = start; i < buffer_count_; ++i) { - float value = buffer_[i]; + double value = buffer_[i]; auto iterator = std::lower_bound( sample_.begin(), sample_.end(), value, - [](const Item& item, float val) { return item.value < val; }); + [](const Item& item, double val) { return item.value < val; }); std::size_t idx = std::distance(sample_.begin(), iterator); int delta = 0; @@ -149,8 +149,11 @@ void CKMSQuantiles::compress() { // sample if (last_compressed_sample.g + current_sample.g + current_sample.delta <= allowableError(static_cast(compressed_samples.size()) - 1)) { - // Merge current sample into last compressed sample - last_compressed_sample.g += current_sample.g; // Update weight + // Keep current_sample's value/delta (matches original semantics of + // dropping the earlier, smaller-value sample) but combine weights. + int merged_g = last_compressed_sample.g + current_sample.g; + last_compressed_sample = current_sample; + last_compressed_sample.g = merged_g; } else { // If not compressible, add current sample to compressed samples compressed_samples.push_back(current_sample); diff --git a/core/tests/summary_test.cc b/core/tests/summary_test.cc index f3549fb5..e5bf0ee3 100644 --- a/core/tests/summary_test.cc +++ b/core/tests/summary_test.cc @@ -102,6 +102,31 @@ TEST(SummaryTest, construction_with_dynamic_quantile_vector) { summary.Observe(8.0); } +TEST(SummaryTest, compress_keeps_larger_value_on_merge) { + // With q=1.0 the two samples are eligible to be merged in compress(). + // The surviving sample must retain the larger value, not the smaller one. + Summary summary{Summary::Quantiles{{1.0, 0.001}}, std::chrono::hours{1}}; + summary.Observe(1.0); + summary.Observe(100.0); + auto metric = summary.Collect(); + auto s = metric.summary; + ASSERT_EQ(s.quantile.size(), 1U); + EXPECT_DOUBLE_EQ(s.quantile.at(0).value, 100.0); +} + +TEST(SummaryTest, insert_preserves_double_precision) { + // 2^24+1 is not exactly representable as float, so a truncating + // double->float->double round trip would corrupt the stored value. + Summary summary{Summary::Quantiles{{1.0, 0.001}}, std::chrono::hours{1}}; + const double v = 16777217.0; + summary.Observe(1.0); + summary.Observe(v); + auto metric = summary.Collect(); + auto s = metric.summary; + ASSERT_EQ(s.quantile.size(), 1U); + EXPECT_DOUBLE_EQ(s.quantile.at(0).value, v); +} + TEST(SummaryTest, quantile_with_out_of_order_batches) { // Flush large values into the sample first, then insert smaller values so // that insertBatch() hits the --idx path (value < sample_[item].value).