move diff and merge to use union of all buckets approach

Author: ethandmd

sdk/src/metrics/aggregation/base2_exponential_histogram_aggregation.cc

@@ -7,6 +7,7 @@
 #include <algorithm>
 #include <limits>
 #include <memory>
+#include <iostream>
 #include <mutex>
 #include <utility>
 #include "opentelemetry/common/spin_lock_mutex.h"
@@ -186,6 +187,41 @@ void Base2ExponentialHistogramAggregation::Downscale(uint32_t by) noexcept
   indexer_ = Base2ExponentialHistogramIndexer(point_data_.scale_);
 }
 
+// Merge A and B into a new circular buffer C.
+// Caller must ensure that A and B are used as buckets at the same scale.
+AdaptingCircularBufferCounter MergeBuckets(size_t max_buckets, const AdaptingCircularBufferCounter &A, const AdaptingCircularBufferCounter &B)
+{
+  AdaptingCircularBufferCounter C = AdaptingCircularBufferCounter(max_buckets);
+  C.Clear();
+
+  if (A.Empty() && B.Empty())
+  {
+    return C;
+  }
+  if (A.Empty())
+  {
+    return B;
+  }
+  if (B.Empty())
+  {
+    return A;
+  }
+
+  auto min_index = (std::min)(A.StartIndex(), B.StartIndex());
+  auto max_index = (std::max)(A.EndIndex(), B.EndIndex());
+
+  for (int i = min_index; i <= max_index; i++)
+  {
+    auto count = A.Get(i) + B.Get(i);
+    if (count > 0)
+    {
+      C.Increment(i, count);
+    }
+ }
+
+  return C;
+}
+
 std::unique_ptr<Aggregation> Base2ExponentialHistogramAggregation::Merge(
     const Aggregation &delta) const noexcept
 {
@@ -197,44 +233,50 @@ std::unique_ptr<Aggregation> Base2ExponentialHistogramAggregation::Merge(
   auto high_res        = left.scale_ < right.scale_ ? right : left;
   auto scale_reduction = high_res.scale_ - low_res.scale_;
 
-  if (scale_reduction > 0)
-  {
-    DownscaleBuckets(&high_res.positive_buckets_, scale_reduction);
-    DownscaleBuckets(&high_res.negative_buckets_, scale_reduction);
-    high_res.scale_ -= scale_reduction;
-  }
-
   Base2ExponentialHistogramPointData result_value;
   result_value.count_          = low_res.count_ + high_res.count_;
   result_value.sum_            = low_res.sum_ + high_res.sum_;
   result_value.zero_count_     = low_res.zero_count_ + high_res.zero_count_;
   result_value.scale_          = (std::min)(low_res.scale_, high_res.scale_);
-  result_value.max_buckets_    = low_res.max_buckets_;
+  result_value.max_buckets_    = low_res.max_buckets_ >= high_res.max_buckets_
+                                     ? low_res.max_buckets_
+                                     : high_res.max_buckets_;
   result_value.record_min_max_ = low_res.record_min_max_ && high_res.record_min_max_;
   if (result_value.record_min_max_)
   {
     result_value.min_ = (std::min)(low_res.min_, high_res.min_);
     result_value.max_ = (std::max)(low_res.max_, high_res.max_);
   }
-  if (!high_res.positive_buckets_.Empty())
+
+  if (scale_reduction > 0)
   {
-    for (int i = high_res.positive_buckets_.StartIndex();
-         i <= high_res.positive_buckets_.EndIndex(); i++)
-    {
-      low_res.positive_buckets_.Increment(i, high_res.positive_buckets_.Get(i));
-    }
+    DownscaleBuckets(&high_res.positive_buckets_, scale_reduction);
+    DownscaleBuckets(&high_res.negative_buckets_, scale_reduction);
+    high_res.scale_ -= scale_reduction;
   }
-  result_value.positive_buckets_ = std::move(low_res.positive_buckets_);
 
-  if (!high_res.negative_buckets_.Empty())
+  auto pos_min_index = (std::min)(low_res.positive_buckets_.StartIndex(), high_res.positive_buckets_.StartIndex());
+  auto pos_max_index = (std::max)(low_res.positive_buckets_.EndIndex(), high_res.positive_buckets_.EndIndex());
+  auto neg_min_index = (std::min)(low_res.negative_buckets_.StartIndex(), high_res.negative_buckets_.StartIndex());
+  auto neg_max_index = (std::max)(low_res.negative_buckets_.EndIndex(), high_res.negative_buckets_.EndIndex());
+
+  if (pos_max_index > pos_min_index + result_value.max_buckets_ || neg_max_index > neg_min_index + result_value.max_buckets_)
   {
-    for (int i = high_res.negative_buckets_.StartIndex();
-         i <= high_res.negative_buckets_.EndIndex(); i++)
-    {
-      low_res.negative_buckets_.Increment(i, high_res.negative_buckets_.Get(i));
-    }
+    // We need to downscale the buckets to fit into the new max_buckets_.
+    const uint32_t scale_reduction = GetScaleReduction(pos_min_index, pos_max_index, result_value.max_buckets_);
+    DownscaleBuckets(&low_res.positive_buckets_, scale_reduction);
+    DownscaleBuckets(&high_res.positive_buckets_, scale_reduction);
+    DownscaleBuckets(&low_res.negative_buckets_, scale_reduction);
+    DownscaleBuckets(&high_res.negative_buckets_, scale_reduction);
+    low_res.scale_ -= scale_reduction;
+    high_res.scale_ -= scale_reduction;
+    result_value.scale_ -= scale_reduction;
   }
-  result_value.negative_buckets_ = std::move(low_res.negative_buckets_);
+
+  result_value.positive_buckets_ = MergeBuckets(result_value.max_buckets_, low_res.positive_buckets_,
+      high_res.positive_buckets_);
+  result_value.negative_buckets_ = MergeBuckets(result_value.max_buckets_, low_res.negative_buckets_,
+      high_res.negative_buckets_);
 
   return std::unique_ptr<Base2ExponentialHistogramAggregation>{
       new Base2ExponentialHistogramAggregation(std::move(result_value))};
@@ -258,58 +300,70 @@ std::unique_ptr<Aggregation> Base2ExponentialHistogramAggregation::Diff(
     high_res.scale_ -= scale_reduction;
   }
 
+  auto pos_min_index = (std::min)(left.positive_buckets_.StartIndex(), right.positive_buckets_.StartIndex());
+  auto pos_max_index = (std::max)(left.positive_buckets_.EndIndex(), right.positive_buckets_.EndIndex());
+  auto neg_min_index = (std::min)(left.negative_buckets_.StartIndex(), right.negative_buckets_.StartIndex());
+  auto neg_max_index = (std::max)(left.negative_buckets_.EndIndex(), right.negative_buckets_.EndIndex());
+
+  if (pos_max_index > pos_min_index + low_res.max_buckets_ || neg_max_index > neg_min_index + low_res.max_buckets_)
+  {
+    // We need to downscale the buckets to fit into the new max_buckets_.
+    const uint32_t scale_reduction = GetScaleReduction(pos_min_index, pos_max_index, low_res.max_buckets_);
+    DownscaleBuckets(&left.positive_buckets_, scale_reduction);
+    DownscaleBuckets(&right.positive_buckets_, scale_reduction);
+    DownscaleBuckets(&left.negative_buckets_, scale_reduction);
+    DownscaleBuckets(&right.negative_buckets_, scale_reduction);
+    left.scale_ -= scale_reduction;
+    right.scale_ -= scale_reduction;
+  }
+
   Base2ExponentialHistogramPointData result_value;
   result_value.scale_          = low_res.scale_;
   result_value.max_buckets_    = low_res.max_buckets_;
   result_value.record_min_max_ = false;
   // caution for underflow
-  result_value.count_ = (left.count_ >= right.count_) ? (left.count_ - right.count_) : 0;
-  result_value.sum_   = (left.sum_ >= right.sum_) ? (left.sum_ - right.sum_) : 0.0;
+  // expect right.{sum, count} >= left.{sum, count} since metric points should be monotonically increasing
+  result_value.count_ = (right.count_ >= left.count_) ? (right.count_ - left.count_) : 0.0;
+  result_value.sum_   = (right.sum_ >= left.sum_) ? (right.sum_ - left.sum_) : 0.0;
   result_value.zero_count_ =
-      (left.zero_count_ >= right.zero_count_) ? (left.zero_count_ - right.zero_count_) : 0;
-  if (!high_res.positive_buckets_.Empty())
+      (right.zero_count_ >= left.zero_count_) ? (right.zero_count_ - left.zero_count_) : 0;
+  result_value.positive_buckets_ = AdaptingCircularBufferCounter{right.max_buckets_};
+  result_value.negative_buckets_ = AdaptingCircularBufferCounter{right.max_buckets_};
+
+  if (!left.positive_buckets_.Empty() && !right.positive_buckets_.Empty())
   {
-    for (int i = high_res.positive_buckets_.StartIndex();
-         i <= high_res.positive_buckets_.EndIndex(); i++)
+    for (auto i = pos_min_index; i <= pos_max_index; i++)
     {
-      low_res.positive_buckets_.Increment(i, 0 - high_res.positive_buckets_.Get(i));
+      auto l_cnt = left.positive_buckets_.Get(i);
+      auto r_cnt = right.positive_buckets_.Get(i);
+      // expect right >= left since metric points should be monotonically increasing
+      auto delta = (std::max)((uint64_t)(0), r_cnt - l_cnt);
+      if (l_cnt > 0)
+      {
+        result_value.positive_buckets_.Increment(i, delta);
+      }
     }
   }
-  result_value.positive_buckets_ = std::move(low_res.positive_buckets_);
 
-  if (!high_res.negative_buckets_.Empty())
+  if (!left.negative_buckets_.Empty() && !right.negative_buckets_.Empty())
   {
-    for (int i = high_res.negative_buckets_.StartIndex();
-         i <= high_res.negative_buckets_.EndIndex(); i++)
+    for (auto i = neg_min_index; i <= neg_max_index; i++)
     {
-      low_res.negative_buckets_.Increment(i, 0 - high_res.negative_buckets_.Get(i));
+      auto l_cnt = left.negative_buckets_.Get(i);
+      auto r_cnt = right.negative_buckets_.Get(i);
+      // expect right >= left since metric points should be monotonically increasing
+      auto delta = (std::max)((uint64_t)(0), r_cnt - l_cnt);
+      if (delta > 0)
+      {
+        result_value.negative_buckets_.Increment(i, delta);
+      }
     }
   }
-  result_value.negative_buckets_ = std::move(low_res.negative_buckets_);
 
   return std::unique_ptr<Base2ExponentialHistogramAggregation>{
       new Base2ExponentialHistogramAggregation(std::move(result_value))};
 }
 
-// std::unique_ptr<Aggregation> Base2ExponentialHistogramAggregation::Diff(
-//     const Aggregation &next) const noexcept
-// {
-//   auto curr_value = nostd::get<Base2ExponentialHistogramPointData>(ToPoint());
-//   auto next_value = nostd::get<Base2ExponentialHistogramPointData>(
-//       (static_cast<const Base2ExponentialHistogramAggregation &>(next).ToPoint()));
-
-//   Base2ExponentialHistogramPointData result_value;
-//   result_value.scale_          = curr_value.scale_;
-//   result_value.max_buckets_    = curr_value.max_buckets_;
-//   result_value.record_min_max_ = false;
-//   result_value.count_          = next_value.count_ - curr_value.count_;
-//   result_value.sum_            = next_value.sum_ - curr_value.sum_;
-//   result_value.zero_count_     = next_value.zero_count_ - curr_value.zero_count_;
-
-//   return std::unique_ptr<Base2ExponentialHistogramAggregation>{
-//       new Base2ExponentialHistogramAggregation(std::move(result_value))};
-// }
-
 PointType Base2ExponentialHistogramAggregation::ToPoint() const noexcept
 {
   const std::lock_guard<opentelemetry::common::SpinLockMutex> locked(lock_);