Agg series set

cmd/prometheus/main.go

@@ -2240,7 +2240,7 @@ func readPromPPFeatures(logger log.Logger) {
 			remotewriter.DefaultSampleAgeLimit = defaultSampleAgeLimit
 
 		case "select_func_optimization":
-			if err := querier.SetSelectFuncOptimize(strings.TrimSpace(fvalue)); err != nil {
+			if err := selectFuncOptimization(strings.TrimSpace(fvalue)); err != nil {
 				level.Error(logger).Log(
 					"msg", "[FEATURE] Error parsing select_func_optimization value",
 					"err", err,
@@ -2255,3 +2255,15 @@ func readPromPPFeatures(logger log.Logger) {
 		}
 	}
 }
+
+// selectFuncOptimization sets the select function optimization.
+func selectFuncOptimization(fvalue string) error {
+	for opt := range strings.SplitSeq(fvalue, "|") {
+		if err := querier.SetSelectFuncOptimize(opt); err != nil {
+			querier.SetDefaultOptimizeType() // reset to default if error
+			return err
+		}
+	}
+
+	return nil
+}

pp-pkg/storage/adapter.go

@@ -36,6 +36,7 @@ type Adapter struct {
 	hashdexLimits         atomic.Value // stores cppbridge.WALHashdexLimits
 	transparentState      *cppbridge.StateV2
 	mergeOutOfOrderChunks func()
+	scrapeInterval        atomic.Int64
 
 	// stat
 	activeQuerierMetrics  *querier.Metrics
@@ -132,7 +133,10 @@ func (ar *Adapter) AppendSnappyProtobuf(
 	state *cppbridge.StateV2,
 	commitToWal bool,
 ) error {
-	hx, err := cppbridge.NewWALSnappyProtobufHashdex(compressedData.Bytes(), ar.hashdexLimits.Load().(cppbridge.WALHashdexLimits))
+	hx, err := cppbridge.NewWALSnappyProtobufHashdex(
+		compressedData.Bytes(),
+		ar.hashdexLimits.Load().(cppbridge.WALHashdexLimits),
+	)
 	compressedData.Destroy()
 	if err != nil {
 		return err
@@ -247,11 +251,12 @@ func (ar *Adapter) ChunkQuerier(mint, maxt int64) (storage.ChunkQuerier, error)
 // Label limit fields follow the same 0 = no limit semantics as scrape config.
 func (ar *Adapter) ApplyConfig(cfg *config.Config) error {
 	limits := cppbridge.WALHashdexLimits{
-		MaxLabelNamesPerTimeseries: uint32(cfg.GlobalConfig.LabelLimit),
-		MaxLabelNameLength:         uint32(cfg.GlobalConfig.LabelNameLengthLimit),
-		MaxLabelValueLength:        uint32(cfg.GlobalConfig.LabelValueLengthLimit),
+		MaxLabelNamesPerTimeseries: uint32(cfg.GlobalConfig.LabelLimit),            // #nosec G115 // no overflow
+		MaxLabelNameLength:         uint32(cfg.GlobalConfig.LabelNameLengthLimit),  // #nosec G115 // no overflow
+		MaxLabelValueLength:        uint32(cfg.GlobalConfig.LabelValueLengthLimit), // #nosec G115 // no overflow
 	}
 	ar.hashdexLimits.Store(limits)
+	ar.scrapeInterval.Store(int64(cfg.GlobalConfig.ScrapeInterval))
 	return nil
 }
 
@@ -269,7 +274,7 @@ func (ar *Adapter) HeadQuerier(mint, maxt int64) (storage.Querier, error) {
 		querier.NewNoOpShardedDeduplicator,
 		mint,
 		maxt,
-		nil,
+		ar.scrapeInterval.Load(),
 		ar.activeQuerierMetrics,
 	), nil
 }
@@ -296,7 +301,14 @@ func (ar *Adapter) Querier(mint, maxt int64) (storage.Querier, error) {
 	ahead := ar.proxy.Get()
 	queriers = append(
 		queriers,
-		querier.NewQuerier(ahead, querier.NewNoOpShardedDeduplicator, mint, maxt, nil, ar.activeQuerierMetrics),
+		querier.NewQuerier(
+			ahead,
+			querier.NewNoOpShardedDeduplicator,
+			mint,
+			maxt,
+			ar.scrapeInterval.Load(),
+			ar.activeQuerierMetrics,
+		),
 	)
 
 	for _, head := range ar.proxy.Heads() {
@@ -316,7 +328,7 @@ func (ar *Adapter) Querier(mint, maxt int64) (storage.Querier, error) {
 				querier.NewNoOpShardedDeduplicator,
 				mint,
 				maxt,
-				nil,
+				ar.scrapeInterval.Load(),
 				ar.storageQuerierMetrics,
 			),
 		)

pp-pkg/storage/batch_storage.go

@@ -84,11 +84,23 @@ func (bs *BatchStorage) Commit(ctx context.Context) error {
 // CommitWithState adds aggregated series from [pp_storage.TransactionHead] to the [Head] with [cppbridge.StateV2].
 func (bs *BatchStorage) CommitWithState(ctx context.Context, state *cppbridge.StateV2) error {
 	s := bs.transactionHead.Shards()[0]
-	_, err := bs.adapter.AppendGoHeadHashdex(ctx, cppbridge.NewGoHeadHashdex(s.LSS().Target(), s.DataStorage().Raw()), state, false)
+	_, err := bs.adapter.AppendGoHeadHashdex(
+		ctx,
+		cppbridge.NewGoHeadHashdex(s.LSS().Target(), s.DataStorage().Raw()),
+		state,
+		false,
+	)
 	return err
 }
 
 // Querier calls f() with the given parameters. Returns a [querier.Querier].
 func (bs *BatchStorage) Querier(mint, maxt int64) (storage.Querier, error) {
-	return querier.NewQuerier(bs.transactionHead, querier.NewNoOpShardedDeduplicator, mint, maxt, nil, nil), nil
+	return querier.NewQuerier(
+		bs.transactionHead,
+		querier.NewNoOpShardedDeduplicator,
+		mint,
+		maxt,
+		bs.adapter.scrapeInterval.Load(),
+		nil,
+	), nil
 }

pp/entrypoint/series_data_data_storage.cpp

@@ -219,27 +219,31 @@ extern "C" void prompp_series_data_data_storage_query_final(void* args) {
   }
 }
 
-extern "C" void prompp_series_data_data_storage_query_first_timestamps(void* args, void* res) {
+extern "C" void prompp_series_data_data_storage_query_first_timestamps(void* args) {
   using PromPP::Primitives::Timestamp;
   using series_data::Decoder;
 
   struct Arguments {
     DataStoragePtr data_storage;
+    Timestamp not_found_timestamp_value;
     SliceView<LabelSetID> series_ids;
-  };
-
-  struct Result {
     Slice<Timestamp> timestamps;
   };
 
   const auto in = static_cast<Arguments*>(args);
-  const auto out = static_cast<Result*>(res);
 
-  assert(in->series_ids.size() == out->timestamps.size());
+  assert(in->series_ids.size() == in->timestamps.size());
   const auto& storage = *in->data_storage;
 
-  std::ranges::transform(in->series_ids, out->timestamps.begin(),
-                         [&storage](uint32_t series_id) { return Decoder::get_series_min_timestamp(storage, series_id); });
+  std::ranges::transform(in->series_ids, in->timestamps.begin(), [&storage, in](uint32_t series_id) {
+    if (storage.open_chunks.size() > series_id) [[likely]] {
+      if (!storage.open_chunks[series_id].is_empty()) [[likely]] {
+        return Decoder::get_series_min_timestamp(storage, series_id);
+      }
+    }
+
+    return in->not_found_timestamp_value;
+  });
 }
 
 extern "C" void prompp_series_data_data_storage_allocated_memory(void* args, void* res) {
@@ -497,4 +501,4 @@ extern "C" void prompp_series_data_data_storage_loader_dtor(void* args) {
   };
 
   static_cast<Arguments*>(args)->~Arguments();
-}
+}

pp/entrypoint/series_data_data_storage.h

@@ -144,14 +144,13 @@ void prompp_series_data_data_storage_instant_query(void* args, void* res);
  * @brief Get the first sample timestamp per series
  *
  * @param args {
- *        dataStorage uintptr  // pointer to constructed data storage
- *        seriesIds   []uint32 // series ids
- * }
- * @param res {
- *        timestamps []int64  // same length as seriesIds; filled from storage
+ *        dataStorage            uintptr  // pointer to constructed data storage
+ *        notFoundTimestampValue int64    // timestamp value to return if series is not found in storage
+ *        seriesIds              []uint32 // series ids
+ *        timestamps             []int64  // same length as seriesIds; filled from storage
  * }
  */
-void prompp_series_data_data_storage_query_first_timestamps(void* args, void* res);
+void prompp_series_data_data_storage_query_first_timestamps(void* args);
 
 /**
  * @brief finishes all Queriers after data load.

pp/go/cppbridge/data_storage.go

@@ -121,8 +121,8 @@ func (ds *DataStorage) InstantQuery(targetTimestamp int64, labelSetIDs []uint32,
 }
 
 // QueryFirstTimestamps fills timestamps with the first sample timestamp (Prometheus ms) for each series in seriesIDs
-func (ds *DataStorage) QueryFirstTimestamps(seriesIDs []uint32, timestamps []int64) {
-	seriesDataDataStorageQueryFirstTimestamps(ds.dataStorage, seriesIDs, timestamps)
+func (ds *DataStorage) QueryFirstTimestamps(seriesIDs []uint32, timestamps []int64, notFoundTimestampValue int64) {
+	seriesDataDataStorageQueryFirstTimestamps(ds.dataStorage, notFoundTimestampValue, seriesIDs, timestamps)
 	runtime.KeepAlive(ds)
 }
 

pp/go/cppbridge/entrypoint.go

@@ -1573,24 +1573,22 @@ func primitivesLSSQueryLabelValues(lss uintptr, label_name string, matchers []mo
 	return res.status, res.values
 }
 
-func primitivesLSSGetLabelNameIDs(lss uintptr, names []string) []uint32 {
+func primitivesLSSGetLabelNameIDs(lss uintptr, names []string, nameIDs []uint32) {
 	args := struct {
 		lss   uintptr
 		names []string
 	}{lss, names}
 
 	res := struct {
 		outIDs []uint32
-	}{make([]uint32, len(names))}
+	}{nameIDs}
 
 	testGC()
 	fastcgo.UnsafeCall2(
 		C.prompp_primitives_lss_get_label_name_ids,
 		uintptr(unsafe.Pointer(&args)),
 		uintptr(unsafe.Pointer(&res)),
 	)
-
-	return res.outIDs
 }
 
 func primitivesLSSCreateSnapshotLSS(lss uintptr) uintptr {
@@ -2145,20 +2143,23 @@ func seriesDataDataStorageInstantQuery(dataStorage uintptr, labelSetIDs []uint32
 	return res
 }
 
-func seriesDataDataStorageQueryFirstTimestamps(dataStorage uintptr, seriesIDs []uint32, timestamps []int64) {
+func seriesDataDataStorageQueryFirstTimestamps(
+	dataStorage uintptr,
+	notFoundTimestampValue int64,
+	seriesIDs []uint32,
+	timestamps []int64,
+) {
 	args := struct {
-		dataStorage uintptr
-		seriesIDs   []uint32
-	}{dataStorage, seriesIDs}
-	res := struct {
-		timestamps []int64
-	}{timestamps}
+		dataStorage            uintptr
+		notFoundTimestampValue int64
+		seriesIDs              []uint32
+		timestamps             []int64
+	}{dataStorage, notFoundTimestampValue, seriesIDs, timestamps}
 
 	testGC()
-	fastcgo.UnsafeCall2(
+	fastcgo.UnsafeCall1(
 		C.prompp_series_data_data_storage_query_first_timestamps,
 		uintptr(unsafe.Pointer(&args)),
-		uintptr(unsafe.Pointer(&res)),
 	)
 }
 

pp/go/cppbridge/entrypoint.h

@@ -1507,14 +1507,13 @@ void prompp_series_data_data_storage_instant_query(void* args, void* res);
  * @brief Get the first sample timestamp per series
  *
  * @param args {
- *        dataStorage uintptr  // pointer to constructed data storage
- *        seriesIds   []uint32 // series ids
- * }
- * @param res {
- *        timestamps []int64  // same length as seriesIds; filled from storage
+ *        dataStorage            uintptr  // pointer to constructed data storage
+ *        notFoundTimestampValue int64    // timestamp value to return if series is not found in storage
+ *        seriesIds              []uint32 // series ids
+ *        timestamps             []int64  // same length as seriesIds; filled from storage
  * }
  */
-void prompp_series_data_data_storage_query_first_timestamps(void* args, void* res);
+void prompp_series_data_data_storage_query_first_timestamps(void* args);
 
 /**
  * @brief finishes all Queriers after data load.

pp/go/cppbridge/head_test.go

@@ -7,7 +7,6 @@ import (
 
 	"github.com/prometheus/prometheus/pp/go/storage/querier"
 	"github.com/prometheus/prometheus/storage"
-	"github.com/stretchr/testify/require"
 
 	"github.com/prometheus/prometheus/pp/go/cppbridge"
 	"github.com/prometheus/prometheus/pp/go/model"
@@ -183,7 +182,7 @@ func (s *HeadSuite) TestInstantQuery() {
 	// Arrange
 	dataStorage := cppbridge.NewDataStorage()
 	encoder := cppbridge.NewHeadEncoderWithDataStorage(dataStorage)
-	var series = []struct {
+	series := []struct {
 		SeriesID uint32
 		cppbridge.Sample
 	}{
@@ -213,19 +212,19 @@ func (s *HeadSuite) TestInstantQuery() {
 	result := dataStorage.InstantQuery(targetTimestamp, seriesIDs, uintptr(unsafe.Pointer(unsafe.SliceData(instantSeries))))
 
 	// Assert
-	require.Equal(s.T(), cppbridge.DataStorageQueryStatusSuccess, result.Status)
+	s.Require().Equal(cppbridge.DataStorageQueryStatusSuccess, result.Status)
 
 	s.Equal(defaultTimestamp, instantSeries[0].Timestamp)
-	s.Equal(series[2].Sample, cppbridge.Sample{Timestamp: instantSeries[1].Timestamp, Value: instantSeries[1].Value})
-	s.Equal(series[5].Sample, cppbridge.Sample{Timestamp: instantSeries[2].Timestamp, Value: instantSeries[2].Value})
-	s.Equal(series[6].Sample, cppbridge.Sample{Timestamp: instantSeries[3].Timestamp, Value: instantSeries[3].Value})
+	s.Equal(cppbridge.Sample{Timestamp: instantSeries[1].Timestamp, Value: instantSeries[1].Value}, series[2].Sample)
+	s.Equal(cppbridge.Sample{Timestamp: instantSeries[2].Timestamp, Value: instantSeries[2].Value}, series[5].Sample)
+	s.Equal(cppbridge.Sample{Timestamp: instantSeries[3].Timestamp, Value: instantSeries[3].Value}, series[6].Sample)
 }
 
 func (s *HeadSuite) TestQueryFirstTimestampsWithEmptySeriesIds() {
 	// Arrange
 
 	// Act
-	s.dataStorage.QueryFirstTimestamps(nil, nil)
+	s.dataStorage.QueryFirstTimestamps(nil, nil, 0)
 
 	// Assert
 }
@@ -242,7 +241,7 @@ func (s *HeadSuite) TestQueryFirstTimestamps() {
 
 	// Act
 	timestamps := make([]int64, 2)
-	s.dataStorage.QueryFirstTimestamps([]uint32{1, 0}, timestamps)
+	s.dataStorage.QueryFirstTimestamps([]uint32{1, 0}, timestamps, 0)
 
 	// Assert
 	s.Equal([]int64{2, 5}, timestamps)
@@ -259,12 +258,41 @@ func (s *HeadSuite) TestQueryFirstTimestampsInFinalizedChunk() {
 
 	// Act
 	timestamps := make([]int64, 1)
-	s.dataStorage.QueryFirstTimestamps([]uint32{0}, timestamps)
+	s.dataStorage.QueryFirstTimestamps([]uint32{0}, timestamps, 0)
 
 	// Assert
 	s.Equal([]int64{5}, timestamps)
 }
 
+func (s *HeadSuite) TestQueryFirstTimestampsRotatedLSS() {
+	// Arrange
+	s.lss.FindOrEmplace(model.NewLabelSetBuilder().Set("job", "1").Build())
+	s.lss.FindOrEmplace(model.NewLabelSetBuilder().Set("job", "2").Build())
+
+	// Act
+	timestamps := make([]int64, 2)
+	s.dataStorage.QueryFirstTimestamps([]uint32{1, 0}, timestamps, -1)
+
+	// Assert
+	s.Equal([]int64{-1, -1}, timestamps)
+}
+
+func (s *HeadSuite) TestQueryFirstTimestampsRotatedLSSWithEmptySeries() {
+	// Arrange
+	s.lss.FindOrEmplace(model.NewLabelSetBuilder().Set("job", "1").Build())
+	s.lss.FindOrEmplace(model.NewLabelSetBuilder().Set("job", "2").Build())
+
+	s.encoder.Encode(1, 5, 1.0)
+	s.encoder.Encode(1, 9, 1.0)
+
+	// Act
+	timestamps := make([]int64, 2)
+	s.dataStorage.QueryFirstTimestamps([]uint32{1, 0}, timestamps, -1)
+
+	// Assert
+	s.Equal([]int64{5, -1}, timestamps)
+}
+
 type DataStorageSerializedDataMultiSeriesIteratorSuite struct {
 	suite.Suite
 	lss *cppbridge.LabelSetStorage

pp/go/cppbridge/lss_snapshot.go

@@ -77,12 +77,13 @@ func (lss *LabelSetSnapshot) Query(selector uintptr) *LSSQueryResult {
 	return result
 }
 
+// SeriesGroups group series by label names.
 type SeriesGroups struct {
 	Groups [][]uint32
 }
 
 // GroupSeriesByLabelNames group series by label names
-func (lss *LabelSetSnapshot) GroupSeriesByLabelNames(seriesIDs []uint32, labelNameIDs []uint32) *SeriesGroups {
+func (lss *LabelSetSnapshot) GroupSeriesByLabelNames(seriesIDs, labelNameIDs []uint32) *SeriesGroups {
 	result := &SeriesGroups{
 		Groups: primitivesGroupSeriesByLabelNames(lss.pointer, seriesIDs, labelNameIDs),
 	}