Promql cpp functions

cmd/prometheus/main.go

@@ -2242,6 +2242,20 @@ func readPromPPFeatures(logger log.Logger) {
 		case "shrink_shard_copier":
 			pp_storage.ShrinkShardCopier = true
 			_ = level.Info(logger).Log("msg", "[FEATURE] Shrink shard copier is enabled.")
+
+		case "select_func_optimization":
+			if err := querier.SetSelectFuncOptimize(strings.TrimSpace(fvalue)); err != nil {
+				level.Error(logger).Log(
+					"msg", "[FEATURE] Error parsing select_func_optimization value",
+					"err", err,
+				)
+				continue
+			}
+
+			level.Info(logger).Log(
+				"msg", "[FEATURE] Select function optimization is set.",
+				"optimization", fvalue,
+			)
 		}
 	}
 }

pp-pkg/storage/adapter.go

@@ -311,7 +311,14 @@ func (ar *Adapter) Querier(mint, maxt int64) (storage.Querier, error) {
 
 		queriers = append(
 			queriers,
-			querier.NewQuerier(head, querier.NewNoOpShardedDeduplicator, mint, maxt, nil, ar.storageQuerierMetrics),
+			querier.NewQuerierWithOutSelectFuncOptimize(
+				head,
+				querier.NewNoOpShardedDeduplicator,
+				mint,
+				maxt,
+				nil,
+				ar.storageQuerierMetrics,
+			),
 		)
 	}
 

pp-pkg/storage/batch_storage_test.go

@@ -10,14 +10,14 @@ import (
 	"github.com/jonboulle/clockwork"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/prometheus/pp/go/storage/storagetest"
+	"github.com/prometheus/prometheus/storage"
 	"github.com/stretchr/testify/suite"
 
 	"github.com/prometheus/prometheus/model/labels"
 	"github.com/prometheus/prometheus/pp/go/cppbridge"
 	pp_model "github.com/prometheus/prometheus/pp/go/model"
 	pp_storage "github.com/prometheus/prometheus/pp/go/storage"
 	"github.com/prometheus/prometheus/pp/go/storage/catalog"
-	"github.com/prometheus/prometheus/storage"
 )
 
 const (
@@ -202,7 +202,7 @@ func (s *BatchStorageSuite) TestCommit_WithSamplesAdded() {
 		StartTimestampMs: 0,
 		EndTimestampMs:   5000,
 		LabelSetIDs:      []uint32{0},
-	})
+	}, cppbridge.NoDownsampling, &storage.SelectHints{})
 	s.Require().Equal(cppbridge.DataStorageQueryStatusSuccess, queryResult.Status)
 	s.Equal(storagetest.SamplesMap{
 		0: []cppbridge.Sample{

pp/bare_bones/algorithm.h

@@ -1,5 +1,6 @@
 #pragma once
 
+#include <bit>
 #include <utility>
 
 namespace BareBones {
@@ -20,6 +21,12 @@ constexpr bool is_in(const Value& value, Args&&... args) {
   return (... || (value == std::forward<Args>(args)));
 }
 
+template <class ResultType, auto... Values>
+constexpr ResultType build_bitmask() {
+  static_assert((... && std::has_single_bit(static_cast<ResultType>(Values))), "each bitmask value must be a power of two");
+  return (... | static_cast<ResultType>(Values));
+}
+
 template <class Range1, class Range2, class Comparator>
 auto lexicographical_compare_three_way(const Range1& range1, const Range2& range2, Comparator&& comparator) {
   using result_type = decltype(comparator(*range1.begin(), *range2.begin()));

pp/entrypoint/aggregation_iterator.cpp

@@ -0,0 +1,31 @@
+#include "aggregation_iterator.h"
+
+#include "series_data/serialization.h"
+
+extern "C" void prompp_series_data_serialization_serialized_data_aggregation_iterator_ctor(void* args) {
+  struct Arguments {
+    entrypoint::series_data::AggregationIterator* iterator;
+    entrypoint::series_data::SerializedDataPtr serialized_data;
+    uint32_t chunk_ref;
+  };
+
+  const auto in = static_cast<Arguments*>(args);
+  std::construct_at(in->iterator, in->serialized_data->aggregation_iterator(in->chunk_ref));
+}
+
+extern "C" void prompp_series_data_serialization_serialized_data_aggregation_iterator_next(void* iterator) {
+  using series_data::decoder::DecodeIteratorSentinel;
+
+  ++(*static_cast<entrypoint::series_data::AggregationIterator*>(iterator));
+}
+
+extern "C" void prompp_series_data_serialization_serialized_data_aggregation_iterator_reset(void* args) {
+  struct Arguments {
+    entrypoint::series_data::AggregationIterator* iterator;
+    entrypoint::series_data::SerializedDataPtr serialized_data;
+    uint32_t chunk_ref;
+  };
+
+  const Arguments* in = static_cast<Arguments*>(args);
+  *in->iterator = in->serialized_data->aggregation_iterator(in->chunk_ref);
+}

pp/entrypoint/aggregation_iterator.h

@@ -0,0 +1,38 @@
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+/**
+ * @brief Create a aggregation iterator for corresponding chunk_ref.
+ *
+ * @param args {
+ *     serializedData uintptr // pointer to serialized data.
+ *     chunk_ref uint32 // inner chunk id.
+ * }
+ *
+ */
+void prompp_series_data_serialization_serialized_data_aggregation_iterator_ctor(void* args);
+
+/**
+ * @brief Advance aggregation iterator.
+ *
+ * @param iterator uintptr // pointer to aggregation iterator
+ *
+ */
+void prompp_series_data_serialization_serialized_data_aggregation_iterator_next(void* iterator);
+
+/**
+ * @brief Reset a aggregation iterator for corresponding chunk_ref.
+ *
+ * @param args {
+ *     serializedData uintptr // pointer to serialized data.
+ *     iterator uintptr // pointer to aggregation iterator
+ *     chunkRef uint32 // inner chunk id.
+ * }
+ *
+ */
+void prompp_series_data_serialization_serialized_data_aggregation_iterator_reset(void* args);
+
+#ifdef __cplusplus
+}  // extern "C"
+#endif

pp/entrypoint/go_constants.cpp

@@ -1,8 +1,8 @@
 #include "go_constants.h"
 
-#include "head/serialization.h"
 #include "metrics/storage.h"
 #include "prometheus/relabeler.h"
+#include "series_data/serialization.h"
 #include "wal/output_decoder.h"
 #include "wal/segment_samples_storage.h"
 
@@ -14,7 +14,8 @@ static_assert(sizeof(roaring::Roaring) == Sizeof_RoaringBitset);
 
 static_assert(sizeof(PromPP::Prometheus::Relabel::InnerSeries) == Sizeof_InnerSeries);
 
-static_assert(sizeof(entrypoint::head::SerializedDataIterator) == Sizeof_SerializedDataIterator);
+static_assert(sizeof(entrypoint::series_data::SamplesIterator) == Sizeof_SerializedDataSamplesIterator);
+static_assert(sizeof(entrypoint::series_data::AggregationIterator) == Sizeof_SerializedDataAggregationIterator);
 
 static_assert(sizeof(metrics::Storage::Iterator) == Sizeof_MetricsIterator);
 

pp/entrypoint/go_constants.h

@@ -5,7 +5,8 @@
 #define Sizeof_InnerSeries (Sizeof_SizeT + Sizeof_BareBonesVector + Sizeof_RoaringBitset)
 #define Sizeof_GoLabels 16
 
-#define Sizeof_SerializedDataIterator 184
+#define Sizeof_SerializedDataSamplesIterator 152
+#define Sizeof_SerializedDataAggregationIterator 208
 
 #define Sizeof_MetricsIterator 24
 

pp/entrypoint/head_wal.cpp

@@ -5,8 +5,8 @@
 #include "exception.hpp"
 #include "hashdex.hpp"
 #include "head/lss.h"
-#include "head/series_data.h"
 #include "primitives/go_slice.h"
+#include "series_data/series_data.h"
 #include "wal/decoder.h"
 #include "wal/encoder.h"
 #include "wal/wal.h"

pp/entrypoint/samples_iterator.cpp

@@ -0,0 +1,43 @@
+#include "samples_iterator.h"
+
+#include "series_data/serialization.h"
+
+extern "C" void prompp_series_data_serialization_serialized_data_samples_iterator_ctor(void* args) {
+  struct Arguments {
+    entrypoint::series_data::SamplesIterator* iterator;
+    entrypoint::series_data::SerializedDataPtr serialized_data;
+    uint32_t chunk_ref;
+  };
+
+  const auto in = static_cast<Arguments*>(args);
+  std::construct_at(in->iterator, in->serialized_data->samples_iterator(in->chunk_ref));
+}
+
+extern "C" void prompp_series_data_serialization_serialized_data_samples_iterator_next(void* iterator) {
+  using series_data::decoder::DecodeIteratorSentinel;
+
+  ++(*static_cast<entrypoint::series_data::SamplesIterator*>(iterator));
+}
+
+extern "C" void prompp_series_data_serialization_serialized_data_samples_iterator_seek(void* args) {
+  using series_data::decoder::DecodeIteratorSentinel;
+
+  struct Arguments {
+    entrypoint::series_data::SamplesIterator* iterator;
+    int64_t target_timestamp;
+  };
+
+  const Arguments* in = static_cast<Arguments*>(args);
+  in->iterator->seek_to(in->target_timestamp);
+}
+
+extern "C" void prompp_series_data_serialization_serialized_data_samples_iterator_reset(void* args) {
+  struct Arguments {
+    entrypoint::series_data::SamplesIterator* iterator;
+    entrypoint::series_data::SerializedDataPtr serialized_data;
+    uint32_t chunk_ref;
+  };
+
+  const Arguments* in = static_cast<Arguments*>(args);
+  *in->iterator = in->serialized_data->samples_iterator(in->chunk_ref);
+}

pp/entrypoint/samples_iterator.h

@@ -0,0 +1,49 @@
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+/**
+ * @brief Create a samples iterator for corresponding chunk_ref.
+ *
+ * @param args {
+ *     serializedData uintptr // pointer to serialized data.
+ *     chunk_ref uint32 // inner chunk id.
+ * }
+ *
+ */
+void prompp_series_data_serialization_serialized_data_samples_iterator_ctor(void* args);
+
+/**
+ * @brief Advance samples iterator.
+ *
+ * @param iterator uintptr // pointer to samples iterator
+ *
+ */
+void prompp_series_data_serialization_serialized_data_samples_iterator_next(void* iterator);
+
+/**
+ * @brief Advance samples iterator until referenced sample is gte targetTimestamp.
+ *
+ * @param args {
+ *     iterator uintptr // pointer to samples iterator
+ *     targetTimestamp int64 // target timestamp
+ * }
+ *
+ */
+void prompp_series_data_serialization_serialized_data_samples_iterator_seek(void* args);
+
+/**
+ * @brief Reset a samples iterator for corresponding chunk_ref.
+ *
+ * @param args {
+ *     serializedData uintptr // pointer to serialized data.
+ *     iterator uintptr // pointer to samples iterator
+ *     chunkRef uint32 // inner chunk id.
+ * }
+ *
+ */
+void prompp_series_data_serialization_serialized_data_samples_iterator_reset(void* args);
+
+#ifdef __cplusplus
+}  // extern "C"
+#endif