Multiseries iterator

pp/bare_bones/tests/vector_tests.cpp

@@ -108,7 +108,7 @@ class BareBonesVectorEraseFixture : public testing::Test {
   }
 };
 
-TEST_F(BareBonesVectorEraseFixture, EraseLastItem) {
+TEST_F(BareBonesVectorEraseFixture, EraseLastItemByRange) {
   // Arrange
 
   // Act
@@ -120,7 +120,19 @@ TEST_F(BareBonesVectorEraseFixture, EraseLastItem) {
   EXPECT_EQ("2", *vector_[1]);
 }
 
-TEST_F(BareBonesVectorEraseFixture, EraseFirstItem) {
+TEST_F(BareBonesVectorEraseFixture, EraseLastItem) {
+  // Arrange
+
+  // Act
+  vector_.erase(vector_.end() - 1);
+
+  // Assert
+  EXPECT_EQ(2U, vector_.size());
+  EXPECT_EQ("1", *vector_[0]);
+  EXPECT_EQ("2", *vector_[1]);
+}
+
+TEST_F(BareBonesVectorEraseFixture, EraseFirstItemByRange) {
   // Arrange
 
   // Act
@@ -132,7 +144,19 @@ TEST_F(BareBonesVectorEraseFixture, EraseFirstItem) {
   EXPECT_EQ("3", *vector_[1]);
 }
 
-TEST_F(BareBonesVectorEraseFixture, EraseSecondItem) {
+TEST_F(BareBonesVectorEraseFixture, EraseFirstItem) {
+  // Arrange
+
+  // Act
+  vector_.erase(vector_.begin());
+
+  // Assert
+  EXPECT_EQ(2U, vector_.size());
+  EXPECT_EQ("2", *vector_[0]);
+  EXPECT_EQ("3", *vector_[1]);
+}
+
+TEST_F(BareBonesVectorEraseFixture, EraseSecondItemByRange) {
   // Arrange
 
   // Act
@@ -144,6 +168,18 @@ TEST_F(BareBonesVectorEraseFixture, EraseSecondItem) {
   EXPECT_EQ("3", *vector_[1]);
 }
 
+TEST_F(BareBonesVectorEraseFixture, EraseSecondItem) {
+  // Arrange
+
+  // Act
+  vector_.erase(vector_.begin() + 1);
+
+  // Assert
+  EXPECT_EQ(2U, vector_.size());
+  EXPECT_EQ("1", *vector_[0]);
+  EXPECT_EQ("3", *vector_[1]);
+}
+
 TEST_F(BareBonesVectorEraseFixture, EraseAllItems) {
   // Arrange
 

pp/bare_bones/vector.h

@@ -109,6 +109,14 @@ class GenericVector {
     derived()->set_size(0);
   }
 
+  PROMPP_ALWAYS_INLINE iterator erase(iterator it) noexcept {
+    if (it == end()) [[unlikely]] {
+      return it;
+    }
+
+    return erase(it, it + 1);
+  }
+
   PROMPP_ALWAYS_INLINE iterator erase(iterator first, iterator last) noexcept {
     assert(first >= begin());
     assert(last <= end());

pp/entrypoint/go_constants.cpp

@@ -16,6 +16,7 @@ static_assert(sizeof(PromPP::Prometheus::Relabel::InnerSeries) == Sizeof_InnerSe
 
 static_assert(sizeof(entrypoint::series_data::SamplesIterator) == Sizeof_SerializedDataSamplesIterator);
 static_assert(sizeof(entrypoint::series_data::AggregationIterator) == Sizeof_SerializedDataAggregationIterator);
+static_assert(sizeof(entrypoint::series_data::MultiSeriesDecodeIterator) == Sizeof_MultiSeriesDecodeIterator);
 
 static_assert(sizeof(metrics::Storage::Iterator) == Sizeof_MetricsIterator);
 

pp/entrypoint/go_constants.h

@@ -7,6 +7,7 @@
 
 #define Sizeof_SerializedDataSamplesIterator 152
 #define Sizeof_SerializedDataAggregationIterator 208
+#define Sizeof_MultiSeriesDecodeIterator 48
 
 #define Sizeof_MetricsIterator 24
 

pp/entrypoint/multiseries_iterator.cpp

@@ -0,0 +1,33 @@
+#include "multiseries_iterator.h"
+
+#include "series_data/serialization.h"
+
+extern "C" void prompp_series_data_serialization_serialized_data_multi_series_iterator_ctor(void* args) {
+  struct Arguments {
+    entrypoint::series_data::MultiSeriesDecodeIterator* iterator;
+    entrypoint::series_data::SerializedDataPtr serialized_data;
+    PromPP::Primitives::Go::SliceView<uint32_t> series_ids;
+  };
+
+  const auto in = static_cast<Arguments*>(args);
+  in->serialized_data->construct_multi_series_iterator(in->iterator, in->series_ids.span());
+}
+
+extern "C" void prompp_series_data_serialization_serialized_data_multi_series_iterator_reset(void* args) {
+  struct Arguments {
+    entrypoint::series_data::MultiSeriesDecodeIterator* iterator;
+    entrypoint::series_data::SerializedDataPtr serialized_data;
+    PromPP::Primitives::Go::SliceView<uint32_t> series_ids;
+  };
+
+  const auto in = static_cast<Arguments*>(args);
+  in->serialized_data->reset_multi_series_iterator(*in->iterator, in->series_ids.span());
+}
+
+extern "C" void prompp_series_data_serialization_serialized_data_multi_series_iterator_next(void* iterator) {
+  ++(*static_cast<entrypoint::series_data::MultiSeriesDecodeIterator*>(iterator));
+}
+
+extern "C" void prompp_series_data_serialization_serialized_data_multi_series_iterator_dtor(void* iterator) {
+  std::destroy_at(static_cast<entrypoint::series_data::MultiSeriesDecodeIterator*>(iterator));
+}

pp/entrypoint/multiseries_iterator.h

@@ -0,0 +1,43 @@
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+/**
+ * @brief Construct a multi-series decode iterator over the given series ids.
+ *
+ * @param args {
+ *     iterator uintptr // pointer to storage of size Sizeof_MultiSeriesDecodeIterator (placement new).
+ *     serializedData uintptr // pointer to serialized data.
+ *     seriesIDs []uint32 // slice view of series ids to use in iterator.
+ * }
+ */
+void prompp_series_data_serialization_serialized_data_multi_series_iterator_ctor(void* args);
+
+/**
+ * @brief Reset a multi-series decode iterator into the given series ids.
+ *
+ * @param args {
+ *     iterator uintptr // pointer to a constructed MultiSeriesDecodeIterator.
+ *     serializedData uintptr // pointer to serialized data.
+ *     seriesIDs []uint32 // slice view of series ids to use in iterator.
+ * }
+ */
+void prompp_series_data_serialization_serialized_data_multi_series_iterator_reset(void* args);
+
+/**
+ * @brief Advance multi-series decode iterator.
+ *
+ * @param iterator uintptr // pointer to multi-series decode iterator
+ */
+void prompp_series_data_serialization_serialized_data_multi_series_iterator_next(void* iterator);
+
+/**
+ * @brief Destroy multi-series decode iterator (call before reusing).
+ *
+ * @param iterator uintptr // pointer to multi-series decode iterator
+ */
+void prompp_series_data_serialization_serialized_data_multi_series_iterator_dtor(void* iterator);
+
+#ifdef __cplusplus
+}  // extern "C"
+#endif

pp/entrypoint/series_data/aggregation_iterator.h

@@ -1,6 +1,7 @@
 #pragma once
 
 #include "prometheus/promql/window_function.h"
+#include "select_hints.h"
 #include "series_data/decoder/decorator/changes_iterator.h"
 #include "series_data/decoder/decorator/delta_iterator.h"
 #include "series_data/decoder/decorator/downsampling_decode_iterator.h"
@@ -193,11 +194,6 @@ class AggregationIterator {
   Type type_;
 };
 
-struct SelectHints {
-  ::series_data::decoder::decorator::WindowFunctionParameters function_parameters;
-  PromPP::Prometheus::promql::WindowFunction window_function{PromPP::Prometheus::promql::WindowFunction::kNone};
-};
-
 PROMPP_ALWAYS_INLINE AggregationIterator create_aggregation_iterator(::series_data::serialization::SerializedDataView::SeriesIterator&& iterator,
                                                                      const SelectHints& select_hints,
                                                                      PromPP::Primitives::Timestamp downsampling_ms) {

pp/entrypoint/series_data/multiseries_decode_iterator.h

@@ -0,0 +1,167 @@
+#pragma once
+
+#include "select_hints.h"
+#include "series_data/decoder/decorator/last_over_step.h"
+#include "series_data/decoder/decorator/lookback_delta_iterator.h"
+#include "series_data/decoder/decorator/max_over_time.h"
+#include "series_data/decoder/decorator/min_over_time.h"
+#include "series_data/decoder/decorator/multiseries_iterator.h"
+#include "series_data/decoder/decorator/sum_over_time.h"
+#include "series_data/decoder/universal_decode_iterator.h"
+#include "series_data/serialization/serialized_data.h"
+
+namespace entrypoint::series_data {
+
+class MultiSeriesDecodeIterator {
+ public:
+  using DecodeIteratorSentinel = ::series_data::decoder::DecodeIteratorSentinel;
+  using UniversalDecodeIterator = ::series_data::decoder::UniversalDecodeIterator;
+  using WindowBoundaryCalculator = ::series_data::decoder::decorator::StepLookbackDeltaWindowCalculator;
+
+  template <class Iterator, class SampleHandler>
+  using MultiSeriesIterator = ::series_data::decoder::decorator::MultiSeriesIterator<Iterator, SampleHandler, WindowBoundaryCalculator>;
+  using SeriesIterator = ::series_data::serialization::SerializedDataView::SeriesIterator;
+
+  using LastOverTimeWithStaleNansIterator = ::series_data::decoder::decorator::LastOverTimeWithStaleNansIterator<SeriesIterator>;
+
+  using Iterator = ::series_data::decoder::decorator::LookbackDeltaIterator<LastOverTimeWithStaleNansIterator>;
+
+  using FindMinElement = ::series_data::decoder::decorator::FindMinElement;
+  using FindMaxElement = ::series_data::decoder::decorator::FindMaxElement;
+  using SumOfElements = ::series_data::decoder::decorator::SumOfElements;
+
+  using MinMultiSeriesIterator = MultiSeriesIterator<Iterator, FindMinElement>;
+  using MaxMultiSeriesIterator = MultiSeriesIterator<Iterator, FindMaxElement>;
+  using SumMultiSeriesIterator = MultiSeriesIterator<Iterator, SumOfElements>;
+
+  enum class Type : uint8_t {
+    kMin = 0,
+    kMax,
+    kSum,
+  };
+
+  DECODE_ITERATOR_TYPE_TRAITS();
+
+#define DEFINE_CONSTRUCTOR(MultiSeriesIteratorType, field, type)                                    \
+  template <class... Args>                                                                          \
+  explicit MultiSeriesDecodeIterator(std::in_place_type_t<MultiSeriesIteratorType>, Args&&... args) \
+      : iterator_{.field{std::forward<Args>(args)...}}, type_{Type::type} {}
+
+  DEFINE_CONSTRUCTOR(MinMultiSeriesIterator, min, kMin)
+  DEFINE_CONSTRUCTOR(MaxMultiSeriesIterator, max, kMax)
+  DEFINE_CONSTRUCTOR(SumMultiSeriesIterator, sum, kSum)
+
+#undef DEFINE_CONSTRUCTOR
+
+  template <class Visitor>
+  PROMPP_ALWAYS_INLINE decltype(auto) visit(Visitor&& visitor) const {
+    switch (type_) {
+      case Type::kMin: {
+        return std::forward<Visitor>(visitor)(iterator_.min);
+      }
+
+      case Type::kMax: {
+        return std::forward<Visitor>(visitor)(iterator_.max);
+      }
+
+      default: {
+        return std::forward<Visitor>(visitor)(iterator_.sum);
+      }
+    }
+  }
+
+  template <class Visitor>
+  PROMPP_ALWAYS_INLINE decltype(auto) visit(Visitor&& visitor) {
+    return const_cast<const MultiSeriesDecodeIterator*>(this)->visit(
+        [&]<typename Iterator>(const Iterator& iterator) PROMPP_LAMBDA_INLINE { return std::forward<Visitor>(visitor)(const_cast<Iterator&>(iterator)); });
+  }
+
+  ~MultiSeriesDecodeIterator() {
+    visit([](const auto& iterator) PROMPP_LAMBDA_INLINE { std::destroy_at(&iterator); });
+  }
+
+  PROMPP_ALWAYS_INLINE const ::series_data::encoder::Sample& operator*() const {
+    return visit([](const auto& iterator) PROMPP_LAMBDA_INLINE -> const auto& { return *iterator; });
+  }
+  PROMPP_ALWAYS_INLINE const ::series_data::encoder::Sample* operator->() const {
+    return visit([](const auto& iterator) PROMPP_LAMBDA_INLINE -> const auto* { return iterator.operator->(); });
+  }
+
+  PROMPP_ALWAYS_INLINE bool operator==(const DecodeIteratorSentinel& sentinel) const {
+    return visit([&sentinel](const auto& iterator) PROMPP_LAMBDA_INLINE { return iterator == sentinel; });
+  }
+
+  PROMPP_ALWAYS_INLINE MultiSeriesDecodeIterator& operator++() {
+    visit([]<typename Iterator>(Iterator& iterator) PROMPP_LAMBDA_INLINE { ++iterator; });
+    return *this;
+  }
+
+  template <class IteratorsGenerator>
+  PROMPP_ALWAYS_INLINE void reset(const ::series_data::decoder::decorator::WindowFunctionParameters& parameters, IteratorsGenerator&& iterators_generator) {
+    visit([&](auto& iterator) PROMPP_LAMBDA_INLINE { iterator.reset(parameters, std::forward<IteratorsGenerator>(iterators_generator)); });
+  }
+
+  PROMPP_ALWAYS_INLINE static void create_series_iterators(const SelectHints& select_hints,
+                                                           std::span<const uint32_t> series_ids,
+                                                           ::series_data::serialization::SerializedDataView data_view,
+                                                           BareBones::Vector<Iterator>& iterators) {
+    const auto initial_interval = WindowBoundaryCalculator::initial_window(select_hints.function_parameters);
+
+    iterators.reserve(series_ids.size());
+    data_view.enumerate_series([&](const auto& chunk, uint32_t chunk_id) {
+      if (std::ranges::find(series_ids, chunk.label_set_id) != series_ids.end()) {
+        iterators.emplace_back(LastOverTimeWithStaleNansIterator(data_view.create_series_iterator(chunk_id), initial_interval),
+                               select_hints.function_parameters.lookback_delta);
+      }
+    });
+  }
+
+  [[nodiscard]] PROMPP_ALWAYS_INLINE Type type() const noexcept { return type_; }
+
+ private:
+  union Iterators {
+    ~Iterators() {}
+
+    MinMultiSeriesIterator min;
+    MaxMultiSeriesIterator max;
+    SumMultiSeriesIterator sum;
+  } iterator_;
+
+  Type type_;
+};
+
+PROMPP_ALWAYS_INLINE void construct_multi_series_decode_iterator(MultiSeriesDecodeIterator* iterator,
+                                                                 const SelectHints& select_hints,
+                                                                 std::span<const uint32_t> series_ids,
+                                                                 ::series_data::serialization::SerializedDataView data_view) {
+  const auto create_series_iterators = [&] {
+    BareBones::Vector<MultiSeriesDecodeIterator::Iterator> iterators;
+    MultiSeriesDecodeIterator::create_series_iterators(select_hints, series_ids, data_view, iterators);
+    return iterators;
+  };
+
+  switch (select_hints.window_function) {
+    using enum PromPP::Prometheus::promql::WindowFunction;
+
+    case kMin: {
+      std::construct_at(iterator, std::in_place_type<MultiSeriesDecodeIterator::MinMultiSeriesIterator>, create_series_iterators(),
+                        select_hints.function_parameters);
+      break;
+    }
+
+    case kMax: {
+      std::construct_at(iterator, std::in_place_type<MultiSeriesDecodeIterator::MaxMultiSeriesIterator>, create_series_iterators(),
+                        select_hints.function_parameters);
+      break;
+    }
+
+    case kSum:
+    default: {
+      std::construct_at(iterator, std::in_place_type<MultiSeriesDecodeIterator::SumMultiSeriesIterator>, create_series_iterators(),
+                        select_hints.function_parameters);
+      break;
+    }
+  }
+}
+
+}  // namespace entrypoint::series_data

pp/entrypoint/series_data/select_hints.h

@@ -0,0 +1,13 @@
+#pragma once
+
+#include "prometheus/promql/window_function.h"
+#include "series_data/decoder/decorator/window_function_iterator.h"
+
+namespace entrypoint::series_data {
+
+struct SelectHints {
+  ::series_data::decoder::decorator::WindowFunctionParameters function_parameters;
+  PromPP::Prometheus::promql::WindowFunction window_function{PromPP::Prometheus::promql::WindowFunction::kNone};
+};
+
+}  // namespace entrypoint::series_data