Head keeper

pp-pkg/storage/adapter.go

@@ -169,7 +169,7 @@ func (ar *Adapter) ChunkQuerier(mint, maxt int64) (storage.ChunkQuerier, error)
 	ahead := ar.proxy.Get()
 	queriers = append(
 		queriers,
-		querier.NewQuerier(ahead, querier.NewNoOpShardedDeduplicator, mint, maxt, nil, ar.activeQuerierMetrics),
+		querier.NewChunkQuerier(ahead, querier.NewNoOpShardedDeduplicator, mint, maxt, nil),
 	)
 
 	for head := range ar.proxy.RangeQueriableHeads(mint, maxt) {
@@ -179,7 +179,7 @@ func (ar *Adapter) ChunkQuerier(mint, maxt int64) (storage.ChunkQuerier, error)
 
 		queriers = append(
 			queriers,
-			querier.NewQuerier(head, querier.NewNoOpShardedDeduplicator, mint, maxt, nil, ar.storageQuerierMetrics),
+			querier.NewChunkQuerier(head, querier.NewNoOpShardedDeduplicator, mint, maxt, nil),
 		)
 	}
 
@@ -198,7 +198,7 @@ func (ar *Adapter) Close() error {
 }
 
 // HeadQuerier returns [storage.Querier] from active head.
-func (ar *Adapter) HeadQuerier(ctx context.Context, mint, maxt int64) (storage.Querier, error) {
+func (ar *Adapter) HeadQuerier(mint, maxt int64) (storage.Querier, error) {
 	return querier.NewQuerier(
 		ar.proxy.Get(),
 		querier.NewNoOpShardedDeduplicator,
@@ -210,7 +210,7 @@ func (ar *Adapter) HeadQuerier(ctx context.Context, mint, maxt int64) (storage.Q
 }
 
 // HeadStatus returns stats of Head.
-func (ar *Adapter) HeadStatus(ctx context.Context, limit int) (querier.HeadStatus, error) {
+func (ar *Adapter) HeadStatus(ctx context.Context, limit int) (*querier.HeadStatus, error) {
 	return querier.QueryHeadStatus(ctx, ar.proxy.Get(), limit)
 }
 

pp/go/cppbridge/head.go

@@ -30,6 +30,17 @@ type TimeInterval struct {
 	MaxT int64
 }
 
+func NewInvalidTimeInterval() TimeInterval {
+	return TimeInterval{
+		MinT: math.MaxInt64,
+		MaxT: math.MinInt64,
+	}
+}
+
+func (t *TimeInterval) IsInvalid() bool {
+	return t.MinT == math.MaxInt64 && t.MaxT == math.MinInt64
+}
+
 type Sample struct {
 	Timestamp int64
 	Value     float64
@@ -39,13 +50,15 @@ type Sample struct {
 type HeadDataStorage struct {
 	dataStorage       uintptr
 	gcDestroyDetector *uint64
+	timeInterval      TimeInterval
 }
 
 // NewHeadDataStorage - constructor.
 func NewHeadDataStorage() *HeadDataStorage {
 	ds := &HeadDataStorage{
 		dataStorage:       seriesDataDataStorageCtor(),
 		gcDestroyDetector: &gcDestroyDetector,
+		timeInterval:      NewInvalidTimeInterval(),
 	}
 
 	runtime.SetFinalizer(ds, func(ds *HeadDataStorage) {
@@ -58,12 +71,16 @@ func NewHeadDataStorage() *HeadDataStorage {
 // Reset - resets data storage.
 func (ds *HeadDataStorage) Reset() {
 	seriesDataDataStorageReset(ds.dataStorage)
+	ds.timeInterval = NewInvalidTimeInterval()
 }
 
-func (ds *HeadDataStorage) TimeInterval() TimeInterval {
-	res := seriesDataDataStorageTimeInterval(ds.dataStorage)
-	runtime.KeepAlive(ds)
-	return res
+func (ds *HeadDataStorage) TimeInterval(invalidateCache bool) TimeInterval {
+	if invalidateCache || ds.timeInterval.IsInvalid() {
+		ds.timeInterval = seriesDataDataStorageTimeInterval(ds.dataStorage)
+		runtime.KeepAlive(ds)
+	}
+
+	return ds.timeInterval
 }
 
 func (ds *HeadDataStorage) GetQueriedSeriesBitset() []byte {
@@ -237,7 +254,7 @@ type HeadDataStorageSerializedChunks struct {
 
 type HeadDataStorageSerializedChunkMetadata [SerializedChunkMetadataSize]byte
 
-func (cm *HeadDataStorageSerializedChunkMetadata) SeriesID() uint32 {
+func (cm HeadDataStorageSerializedChunkMetadata) SeriesID() uint32 {
 	return *(*uint32)(unsafe.Pointer(&cm[0]))
 }
 
@@ -257,6 +274,11 @@ func (r *HeadDataStorageSerializedChunks) Data() []byte {
 	return r.data
 }
 
+func (r *HeadDataStorageSerializedChunks) Metadata(chunkIndex int) HeadDataStorageSerializedChunkMetadata {
+	offset := Uint32Size + chunkIndex*SerializedChunkMetadataSize
+	return HeadDataStorageSerializedChunkMetadata(r.data[offset : offset+SerializedChunkMetadataSize])
+}
+
 type HeadDataStorageSerializedChunkIndex struct {
 	m map[uint32][]int
 }

pp/go/cppbridge/head_test.go

@@ -164,10 +164,15 @@ func (s *HeadSuite) TestTimeInterval() {
 	encoder.Encode(1, 3, 1.0)
 
 	// Act
-	timeInterval := dataStorage.TimeInterval()
+	timeInterval := dataStorage.TimeInterval(false)
+	encoder.Encode(1, 4, 1.0)
+	cachedTimeInterval := dataStorage.TimeInterval(false)
+	actualTimeInterval := dataStorage.TimeInterval(true)
 
 	// Assert
 	s.Equal(cppbridge.TimeInterval{MinT: 1, MaxT: 3}, timeInterval)
+	s.Equal(cppbridge.TimeInterval{MinT: 1, MaxT: 3}, cachedTimeInterval)
+	s.Equal(cppbridge.TimeInterval{MinT: 1, MaxT: 4}, actualTimeInterval)
 }
 
 func (s *HeadSuite) TestInstantQuery() {

pp/go/cppbridge/head_wal_test.go

@@ -30,5 +30,5 @@ func TestHeadWalDecoder_DecodeToDataStorage(t *testing.T) {
 
 	// Assert
 	require.NoError(t, err)
-	require.Equal(t, cppbridge.TimeInterval{MinT: 1660828401000, MaxT: 1660828410000}, dataStorage.TimeInterval())
+	require.Equal(t, cppbridge.TimeInterval{MinT: 1660828401000, MaxT: 1660828410000}, dataStorage.TimeInterval(false))
 }

pp/go/relabeler/block/writer.go

@@ -301,7 +301,7 @@ func (w *Writer) createWriters(shard relabeler.Shard) (blockWriters, error) {
 	var writers blockWriters
 
 	shard.DataStorageRLock()
-	timeInterval := shard.DataStorage().TimeInterval()
+	timeInterval := shard.DataStorage().TimeInterval(false)
 	shard.DataStorageRUnlock()
 
 	quantStart := (timeInterval.MinT / w.blockDurationMs) * w.blockDurationMs

pp/go/relabeler/head/shard.go

@@ -133,8 +133,8 @@ func (ds *DataStorage) CreateRevertableLoader(lss *cppbridge.LabelSetStorage, ls
 	return ds.dataStorage.CreateRevertableLoader(lss, lsIdBatchSize)
 }
 
-func (ds *DataStorage) TimeInterval() cppbridge.TimeInterval {
-	return ds.dataStorage.TimeInterval()
+func (ds *DataStorage) TimeInterval(invalidateCache bool) cppbridge.TimeInterval {
+	return ds.dataStorage.TimeInterval(invalidateCache)
 }
 
 func (ds *DataStorage) GetQueriedSeriesBitset() []byte {

pp/go/relabeler/interface.go

@@ -23,7 +23,7 @@ type DataStorage interface {
 	CreateUnusedSeriesDataUnloader() *cppbridge.UnusedSeriesDataUnloader
 	CreateLoader(queriers []uintptr) *cppbridge.UnloadedDataLoader
 	CreateRevertableLoader(lss *cppbridge.LabelSetStorage, lsIdBatchSize uint32) *cppbridge.UnloadedDataRevertableLoader
-	TimeInterval() cppbridge.TimeInterval
+	TimeInterval(invalidateCache bool) cppbridge.TimeInterval
 	GetQueriedSeriesBitset() []byte
 }
 

pp/go/storage/appender/appender_test.go

@@ -14,7 +14,7 @@ func TestXxx(t *testing.T) {
 	lss := &shard.LSS{}
 	ds := shard.NewDataStorage()
 	wl := &testWal{}
-	sd := shard.NewShard(lss, ds, wl, 0)
+	sd := shard.NewShard(lss, ds, nil, nil, wl, 0)
 	id := "test-head-id"
 	generation := uint64(0)
 

pp/go/storage/block/block.go

@@ -0,0 +1,142 @@
+package block
+
+import (
+	"fmt"
+	"io"
+	"math"
+	"unsafe"
+
+	"github.com/prometheus/prometheus/pp/go/cppbridge"
+	"github.com/prometheus/prometheus/tsdb/chunkenc"
+)
+
+type Chunk struct {
+	rc *cppbridge.RecodedChunk
+}
+
+func (c *Chunk) MinT() int64 {
+	return c.rc.MinT
+}
+
+func (c *Chunk) MaxT() int64 {
+	return c.rc.MaxT
+}
+
+func (c *Chunk) SeriesID() uint32 {
+	return c.rc.SeriesId
+}
+
+func (c *Chunk) Encoding() chunkenc.Encoding {
+	return chunkenc.EncXOR
+}
+
+func (c *Chunk) SampleCount() uint8 {
+	return c.rc.SamplesCount
+}
+
+func (c *Chunk) Bytes() []byte {
+	return c.rc.ChunkData
+}
+
+type ChunkIterator struct {
+	r  *cppbridge.ChunkRecoder
+	rc *cppbridge.RecodedChunk
+}
+
+func NewChunkIterator(lss *cppbridge.LabelSetStorage, lsIdBatchSize uint32, ds *cppbridge.HeadDataStorage, minT, maxT int64) ChunkIterator {
+	return ChunkIterator{r: cppbridge.NewChunkRecoder(lss, lsIdBatchSize, ds, cppbridge.TimeInterval{MinT: minT, MaxT: maxT})}
+}
+
+func (i *ChunkIterator) Next() bool {
+	if i.rc != nil && !i.rc.HasMoreData {
+		return false
+	}
+
+	rc := i.r.RecodeNextChunk()
+	i.rc = &rc
+	return rc.SeriesId != math.MaxUint32
+}
+
+func (i *ChunkIterator) NextBatch() bool {
+	i.rc.HasMoreData = i.r.NextBatch()
+	return i.rc.HasMoreData
+}
+
+func (i *ChunkIterator) At() Chunk {
+	return Chunk{rc: i.rc}
+}
+
+type IndexWriter struct {
+	cppIndexWriter  *cppbridge.IndexWriter
+	isPrefixWritten bool
+}
+
+func (iw *IndexWriter) WriteSeriesTo(id uint32, chunks []ChunkMetadata, w io.Writer) (n int64, err error) {
+	if !iw.isPrefixWritten {
+		var bytesWritten int
+		bytesWritten, err = w.Write(iw.cppIndexWriter.WriteHeader())
+		n += int64(bytesWritten)
+		if err != nil {
+			return n, fmt.Errorf("failed to write header: %w", err)
+		}
+
+		bytesWritten, err = w.Write(iw.cppIndexWriter.WriteSymbols())
+		n += int64(bytesWritten)
+		if err != nil {
+			return n, fmt.Errorf("failed to write symbols: %w", err)
+		}
+		iw.isPrefixWritten = true
+	}
+
+	bytesWritten, err := w.Write(iw.cppIndexWriter.WriteSeries(id, *(*[]cppbridge.ChunkMetadata)(unsafe.Pointer(&chunks))))
+	n += int64(bytesWritten)
+	if err != nil {
+		return n, fmt.Errorf("failed to write series: %w", err)
+	}
+
+	return n, nil
+}
+
+func (iw *IndexWriter) WriteRestTo(w io.Writer) (n int64, err error) {
+	bytesWritten, err := w.Write(iw.cppIndexWriter.WriteLabelIndices())
+	n += int64(bytesWritten)
+	if err != nil {
+		return n, fmt.Errorf("failed to write label indicies: %w", err)
+	}
+
+	for {
+		data, hasMoreData := iw.cppIndexWriter.WriteNextPostingsBatch(1 << 20)
+		bytesWritten, err = w.Write(data)
+		if err != nil {
+			return n, fmt.Errorf("failed to write postings: %w", err)
+		}
+		n += int64(bytesWritten)
+		if !hasMoreData {
+			break
+		}
+	}
+
+	bytesWritten, err = w.Write(iw.cppIndexWriter.WriteLabelIndicesTable())
+	if err != nil {
+		return n, fmt.Errorf("failed to write label indicies table: %w", err)
+	}
+	n += int64(bytesWritten)
+
+	bytesWritten, err = w.Write(iw.cppIndexWriter.WritePostingsTableOffsets())
+	if err != nil {
+		return n, fmt.Errorf("failed to write posting table offsets: %w", err)
+	}
+	n += int64(bytesWritten)
+
+	bytesWritten, err = w.Write(iw.cppIndexWriter.WriteTableOfContents())
+	if err != nil {
+		return n, fmt.Errorf("failed to write table of content: %w", err)
+	}
+	n += int64(bytesWritten)
+
+	return n, nil
+}
+
+func NewIndexWriter(lss *cppbridge.LabelSetStorage) IndexWriter {
+	return IndexWriter{cppIndexWriter: cppbridge.NewIndexWriter(lss)}
+}