diff --git a/core/src/detail/ckms_quantiles.cc b/core/src/detail/ckms_quantiles.cc index 395b6ff3..8e17e811 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; - } - - int delta; - if (idx - 1 == 0 || idx + 1 == sample_.size()) { - delta = 0; - } else { - delta = static_cast(std::floor(allowableError(idx + 1))) + 1; + double value = buffer_[i]; + + auto iterator = std::lower_bound( + sample_.begin(), sample_.end(), value, + [](const Item& item, double 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; } - sample_.emplace(sample_.begin() + idx, v, 1, delta); - count_++; - item = idx++; + sample_.emplace(iterator, value, 1, delta); + ++count_; } buffer_count_ = 0; @@ -128,24 +128,40 @@ 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++; - - while (idx < sample_.size()) { - prev = next; - next = idx++; - - if (sample_[prev].g + sample_[next].g + sample_[next].delta <= - allowableError(idx - 1)) { - sample_[next].g += sample_[prev].g; - sample_.erase(sample_.begin() + prev); + 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]); + + for (std::size_t idx = 1; idx < sample_.size(); ++idx) { + const Item& current_sample = sample_[idx]; + Item& last_compressed_sample = compressed_samples.back(); + + // 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)) { + // 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); } } + + // Replace old samples with new compressed samples + sample_ = std::move(compressed_samples); } } // namespace prometheus::detail 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).