Agg series set

cmd/prometheus/main.go

@@ -2315,7 +2315,7 @@ func readPromPPFeatures(logger log.Logger) {
 			_ = level.Info(logger).Log("msg", "[FEATURE] Shrink shard copier is enabled.")
 
 		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,
@@ -2330,3 +2330,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

@@ -21,6 +21,9 @@ import (
 	"github.com/prometheus/prometheus/storage"
 )
 
+// defaultCacheCheckIntervalMs is the default cache check interval in milliseconds.
+const defaultCacheCheckIntervalMs = int64(5*time.Minute) / 1e6 // ns to ms
+
 //
 // Adapter
 //
@@ -36,6 +39,7 @@ type Adapter struct {
 	hashdexLimits         atomic.Value // stores cppbridge.WALHashdexLimits
 	transparentState      *cppbridge.StateV2
 	mergeOutOfOrderChunks func()
+	scrapeInterval        atomic.Int64
 
 	// stat
 	activeQuerierMetrics  *querier.Metrics
@@ -132,7 +136,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 +254,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 / 1e6)) // ns to ms
 	return nil
 }
 
@@ -264,12 +272,15 @@ func (ar *Adapter) Close() error {
 
 // HeadQuerier returns [storage.Querier] from active head.
 func (ar *Adapter) HeadQuerier(mint, maxt int64) (storage.Querier, error) {
+	ahead := ar.proxy.Get()
+	aTimeInterval := headTimeIntervalWithValidateCache(ahead, defaultCacheCheckIntervalMs)
 	return querier.NewQuerier(
-		ar.proxy.Get(),
+		ahead,
 		querier.NewNoOpShardedDeduplicator,
 		mint,
 		maxt,
-		nil,
+		ar.scrapeInterval.Load(),
+		aTimeInterval.MinT,
 		ar.activeQuerierMetrics,
 	), nil
 }
@@ -294,9 +305,18 @@ func (ar *Adapter) MergeOutOfOrderChunks() {
 func (ar *Adapter) Querier(mint, maxt int64) (storage.Querier, error) {
 	queriers := make([]storage.Querier, 0, 1) //revive:disable-line:add-constant // the best way
 	ahead := ar.proxy.Get()
+	aTimeInterval := headTimeIntervalWithValidateCache(ahead, defaultCacheCheckIntervalMs)
 	queriers = append(
 		queriers,
-		querier.NewQuerier(ahead, querier.NewNoOpShardedDeduplicator, mint, maxt, nil, ar.activeQuerierMetrics),
+		querier.NewQuerier(
+			ahead,
+			querier.NewNoOpShardedDeduplicator,
+			mint,
+			maxt,
+			ar.scrapeInterval.Load(),
+			aTimeInterval.MinT,
+			ar.activeQuerierMetrics,
+		),
 	)
 
 	for _, head := range ar.proxy.Heads() {
@@ -316,7 +336,8 @@ func (ar *Adapter) Querier(mint, maxt int64) (storage.Querier, error) {
 				querier.NewNoOpShardedDeduplicator,
 				mint,
 				maxt,
-				nil,
+				ar.scrapeInterval.Load(),
+				aTimeInterval.MinT,
 				ar.storageQuerierMetrics,
 			),
 		)
@@ -342,3 +363,15 @@ func headTimeInterval(head *pp_storage.Head) cppbridge.TimeInterval {
 
 	return timeInterval
 }
+
+// headTimeIntervalWithValidateCache returns [cppbridge.TimeInterval] from [pp_storage.Head] with validate cache.
+func headTimeIntervalWithValidateCache(head *pp_storage.Head, cacheCheckIntervalMs int64) cppbridge.TimeInterval {
+	timeInterval := cppbridge.NewInvalidTimeInterval()
+	for _, shard := range head.Shards() {
+		interval := shard.TimeIntervalWithValidateCache(cacheCheckIntervalMs)
+		timeInterval.MinT = min(interval.MinT, timeInterval.MinT)
+		timeInterval.MaxT = max(interval.MaxT, timeInterval.MaxT)
+	}
+
+	return timeInterval
+}

pp-pkg/storage/batch_storage.go

@@ -84,11 +84,31 @@ 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
+	aTimeInterval := cppbridge.NewInvalidTimeInterval()
+	for _, shard := range bs.transactionHead.Shards() {
+		interval := shard.TimeIntervalWithValidateCache(defaultCacheCheckIntervalMs)
+		aTimeInterval.MinT = min(interval.MinT, aTimeInterval.MinT)
+		aTimeInterval.MaxT = max(interval.MaxT, aTimeInterval.MaxT)
+	}
+
+	return querier.NewQuerier(
+		bs.transactionHead,
+		querier.NewNoOpShardedDeduplicator,
+		mint,
+		maxt,
+		bs.adapter.scrapeInterval.Load(),
+		aTimeInterval.MinT,
+		nil,
+	), nil
 }

pp/entrypoint/series_data_data_storage.cpp

@@ -240,27 +240,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) {
@@ -518,4 +522,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

@@ -168,14 +168,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

@@ -2,23 +2,25 @@ package cppbridge
 
 import (
 	"runtime"
+	"sync"
 	"sync/atomic"
+	"time"
 	"unsafe"
 )
 
 // DataStorage is Go wrapper around series_data::Data_storage.
 type DataStorage struct {
-	dataStorage       uintptr
-	gcDestroyDetector *uint64
-	timeInterval      atomic.Pointer[TimeInterval]
+	dataStorage    uintptr
+	timeInterval   atomic.Pointer[TimeInterval]
+	lastUpdateTime atomic.Int64
+	m              sync.Mutex
 }
 
 // NewDataStorage - constructor.
 func NewDataStorage() *DataStorage {
 	ds := &DataStorage{
-		dataStorage:       seriesDataDataStorageCtor(),
-		gcDestroyDetector: &gcDestroyDetector,
-		timeInterval:      atomic.Pointer[TimeInterval]{},
+		dataStorage:  seriesDataDataStorageCtor(),
+		timeInterval: atomic.Pointer[TimeInterval]{},
 	}
 	ds.timeInterval.Store(newInvalidTimeIntervalPtr())
 
@@ -45,6 +47,25 @@ func (ds *DataStorage) TimeInterval(invalidateCache bool) TimeInterval {
 	return *ds.timeInterval.Load()
 }
 
+// TimeIntervalWithValidateCache gets time interval from [DataStorage] with validate cache.
+func (ds *DataStorage) TimeIntervalWithValidateCache(cacheCheckIntervalMs int64) TimeInterval {
+	now := time.Now().UnixMilli()
+	if now-ds.lastUpdateTime.Load() > cacheCheckIntervalMs {
+		// slow path
+		ds.m.Lock()
+		if now-ds.lastUpdateTime.Load() > cacheCheckIntervalMs {
+			timeInterval := seriesDataDataStorageTimeInterval(ds.dataStorage)
+			ds.timeInterval.Store(&timeInterval)
+			ds.lastUpdateTime.Store(now)
+		}
+		ds.m.Unlock()
+
+		runtime.KeepAlive(ds)
+	}
+
+	return *ds.timeInterval.Load()
+}
+
 func (ds *DataStorage) GetQueriedSeriesBitset() []byte {
 	size := seriesDataDataStorageQueriedSeriesBitsetSize(ds.dataStorage)
 	bitset := seriesDataDataStorageQueriedSeriesBitset(ds.dataStorage, make([]byte, 0, size))
@@ -121,8 +142,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 {
@@ -2182,20 +2180,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

@@ -1707,14 +1707,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.