chore(metastore): Update metastores after Flushing a dataobj

go.mod

@@ -408,3 +408,5 @@ replace github.com/grafana/loki/pkg/push => ./pkg/push
 
 // leodido fork his project to continue support
 replace github.com/influxdata/go-syslog/v3 => github.com/leodido/go-syslog/v4 v4.2.0
+
+replace github.com/thanos-io/objstore => github.com/benclive/objstore v0.0.0-20250122170312-3f71d73c03dd

go.sum

@@ -228,6 +228,8 @@ github.com/baidubce/bce-sdk-go v0.9.214 h1:bsVfwMh/emI6vreEveUEq9xAr6xtHLycTAGy2
 github.com/baidubce/bce-sdk-go v0.9.214/go.mod h1:zbYJMQwE4IZuyrJiFO8tO8NbtYiKTFTbwh4eIsqjVdg=
 github.com/bboreham/go-loser v0.0.0-20230920113527-fcc2c21820a3 h1:6df1vn4bBlDDo4tARvBm7l6KA9iVMnE3NWizDeWSrps=
 github.com/bboreham/go-loser v0.0.0-20230920113527-fcc2c21820a3/go.mod h1:CIWtjkly68+yqLPbvwwR/fjNJA/idrtULjZWh2v1ys0=
+github.com/benclive/objstore v0.0.0-20250122170312-3f71d73c03dd h1:EtsJOLTb14nQwPSfVDj6b101VhSj25EyUEW9mwJ4X9M=
+github.com/benclive/objstore v0.0.0-20250122170312-3f71d73c03dd/go.mod h1:Quz9HUDjGidU0RQpoytzK4KqJ7kwzP+DMAm4K57/usM=
 github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973/go.mod h1:Dwedo/Wpr24TaqPxmxbtue+5NUziq4I4S80YR8gNf3Q=
 github.com/beorn7/perks v1.0.0/go.mod h1:KWe93zE9D1o94FZ5RNwFwVgaQK1VOXiVxmqh+CedLV8=
 github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
@@ -1124,8 +1126,6 @@ github.com/stvp/tempredis v0.0.0-20181119212430-b82af8480203 h1:QVqDTf3h2WHt08Yu
 github.com/stvp/tempredis v0.0.0-20181119212430-b82af8480203/go.mod h1:oqN97ltKNihBbwlX8dLpwxCl3+HnXKV/R0e+sRLd9C8=
 github.com/tencentyun/cos-go-sdk-v5 v0.7.40 h1:W6vDGKCHe4wBACI1d2UgE6+50sJFhRWU4O8IB2ozzxM=
 github.com/tencentyun/cos-go-sdk-v5 v0.7.40/go.mod h1:4dCEtLHGh8QPxHEkgq+nFaky7yZxQuYwgSJM87icDaw=
-github.com/thanos-io/objstore v0.0.0-20250115091151-a54d0f04b42a h1:wFBHAmtq1tOLPFaiC4LozyG/BzkRa3ZTmVv1KujUNqk=
-github.com/thanos-io/objstore v0.0.0-20250115091151-a54d0f04b42a/go.mod h1:Quz9HUDjGidU0RQpoytzK4KqJ7kwzP+DMAm4K57/usM=
 github.com/tinylib/msgp v1.1.5/go.mod h1:eQsjooMTnV42mHu917E26IogZ2930nFyBQdofk10Udg=
 github.com/tklauser/go-sysconf v0.3.13 h1:GBUpcahXSpR2xN01jhkNAbTLRk2Yzgggk8IM08lq3r4=
 github.com/tklauser/go-sysconf v0.3.13/go.mod h1:zwleP4Q4OehZHGn4CYZDipCgg9usW5IJePewFCGVEa0=

pkg/dataobj/consumer/partition_processor.go

@@ -15,6 +15,7 @@
 	"github.com/twmb/franz-go/pkg/kgo"
 
 	"github.com/grafana/loki/v3/pkg/dataobj"
+	"github.com/grafana/loki/v3/pkg/dataobj/metastore"
 	"github.com/grafana/loki/v3/pkg/kafka"
 )
 
@@ -30,10 +31,10 @@
 	decoder *kafka.Decoder
 
 	// Builder initialization
-	builderOnce sync.Once
-	builderCfg  dataobj.BuilderConfig
-	bucket      objstore.Bucket
-
+	builderOnce      sync.Once
+	builderCfg       dataobj.BuilderConfig
+	bucket           objstore.Bucket
+	metastoreManager *metastore.MetastoreManager
 	// Metrics
 	metrics *partitionOffsetMetrics
 
@@ -60,20 +61,27 @@
 		level.Error(logger).Log("msg", "failed to register partition metrics", "err", err)
 	}
 
+	metastoreManager, err := metastore.NewMetastoreManager(bucket, tenantID, logger, reg)
+	if err != nil {
+		level.Error(logger).Log("msg", "failed to create metastore manager", "err", err)
+		return nil
+	}
+
 	return &partitionProcessor{
-		client:     client,
-		logger:     log.With(logger, "topic", topic, "partition", partition),
-		topic:      topic,
-		partition:  partition,
-		records:    make(chan *kgo.Record, 1000),
-		ctx:        ctx,
-		cancel:     cancel,
-		decoder:    decoder,
-		reg:        reg,
-		builderCfg: builderCfg,
-		bucket:     bucket,
-		tenantID:   []byte(tenantID),
-		metrics:    metrics,
+		client:           client,
+		logger:           log.With(logger, "topic", topic, "partition", partition),
+		topic:            topic,
+		partition:        partition,
+		records:          make(chan *kgo.Record, 1000),
+		ctx:              ctx,
+		cancel:           cancel,
+		decoder:          decoder,
+		reg:              reg,
+		builderCfg:       builderCfg,
+		bucket:           bucket,
+		tenantID:         []byte(tenantID),
+		metrics:          metrics,
+		metastoreManager: metastoreManager,
 	}
 }
 
@@ -157,8 +165,9 @@
 			MaxBackoff: 10 * time.Second,
 		})
 
+		var flushResult dataobj.FlushResult
 		for backoff.Ongoing() {
-			err = p.builder.Flush(p.ctx)
+			flushResult, err = p.builder.Flush(p.ctx)
 			if err == nil {
 				break
 			}
@@ -167,6 +176,11 @@
 			backoff.Wait()
 		}
 
+		if err := p.metastoreManager.UpdateMetastore(p.ctx, flushResult); err != nil {
+			level.Error(p.logger).Log("msg", "failed to update metastore", "err", err)
+			return
+		}
+
 		backoff.Reset()
 		for backoff.Ongoing() {
 			err = p.client.CommitRecords(p.ctx, record)

pkg/dataobj/dataobj.go

@@ -9,6 +9,9 @@ import (
 	"errors"
 	"flag"
 	"fmt"
+	"io"
+	"sort"
+	"time"
 
 	"github.com/grafana/dskit/flagext"
 	lru "github.com/hashicorp/golang-lru/v2"
@@ -138,6 +141,11 @@ const (
 	builderStateFlush
 )
 
+type FlushResult struct {
+	Path                       string
+	MinTimestamp, MaxTimestamp time.Time
+}
+
 // NewBuilder creates a new Builder which stores data objects for the specified
 // tenant in a bucket.
 //
@@ -180,6 +188,39 @@ func NewBuilder(cfg BuilderConfig, bucket objstore.Bucket, tenantID string) (*Bu
 	}, nil
 }
 
+// FromExisting updates this builder with content from an existing data object, replicating all the state like stream IDs and logs.
+func (b *Builder) FromExisting(f io.ReadSeeker) error {
+	if b.currentSizeEstimate > 0 {
+		return fmt.Errorf("builder already has data, cannot use FromExisting")
+	}
+
+	dec := encoding.ReadSeekerDecoder(f)
+
+	var streamIDs = make(map[int64]*labels.Labels, 32)
+	for result := range streams.Iter(context.Background(), dec) {
+		stream, err := result.Value()
+		if err != nil {
+			return err
+		}
+		sort.Sort(stream.Labels)
+		streamIDs[stream.ID] = &stream.Labels
+	}
+
+	for result := range logs.Iter(context.Background(), dec) {
+		record, err := result.Value()
+		if err != nil {
+			return err
+		}
+		streamLabels := streamIDs[record.StreamID]
+
+		b.streams.Record(*streamLabels, record.Timestamp)
+		b.logs.Append(record)
+	}
+
+	b.state = builderStateDirty
+	return nil
+}
+
 // Append buffers a stream to be written to a data object. Append returns an
 // error if the stream labels cannot be parsed or [ErrBufferFull] if the
 // builder is full.
@@ -286,15 +327,10 @@ func streamSizeEstimate(stream logproto.Stream) int {
 // If Flush builds an object but fails to upload it to object storage, the
 // built object is cached and can be retried. [Builder.Reset] can be called to
 // discard any pending data and allow new data to be appended.
-func (b *Builder) Flush(ctx context.Context) error {
-	switch b.state {
-	case builderStateEmpty:
-		return nil // Nothing to flush
-	case builderStateDirty:
-		if err := b.buildObject(); err != nil {
-			return fmt.Errorf("building object: %w", err)
-		}
-		b.state = builderStateFlush
+func (b *Builder) Flush(ctx context.Context) (FlushResult, error) {
+	_, err := b.FlushToBuffer()
+	if err != nil {
+		return FlushResult{}, err
 	}
 
 	timer := prometheus.NewTimer(b.metrics.flushTime)
@@ -305,11 +341,32 @@ func (b *Builder) Flush(ctx context.Context) error {
 
 	objectPath := fmt.Sprintf("tenant-%s/objects/%s/%s", b.tenantID, sumStr[:b.cfg.SHAPrefixSize], sumStr[b.cfg.SHAPrefixSize:])
 	if err := b.bucket.Upload(ctx, objectPath, bytes.NewReader(b.flushBuffer.Bytes())); err != nil {
-		return err
+		return FlushResult{}, fmt.Errorf("uploading object: %w", err)
 	}
 
+	minTimestamp, maxTimestamp := b.streams.GetBounds()
+
 	b.Reset()
-	return nil
+
+	return FlushResult{
+		Path:         objectPath,
+		MinTimestamp: minTimestamp,
+		MaxTimestamp: maxTimestamp,
+	}, nil
+}
+
+func (b *Builder) FlushToBuffer() (*bytes.Buffer, error) {
+	switch b.state {
+	case builderStateEmpty:
+		return nil, nil // Nothing to flush
+	case builderStateDirty:
+		if err := b.buildObject(); err != nil {
+			return nil, fmt.Errorf("building object: %w", err)
+		}
+		b.state = builderStateFlush
+	}
+
+	return b.flushBuffer, nil
 }
 
 func (b *Builder) buildObject() error {
@@ -345,6 +402,7 @@ func (b *Builder) Reset() {
 	b.state = builderStateEmpty
 	b.flushBuffer.Reset()
 	b.metrics.sizeEstimate.Set(0)
+	b.currentSizeEstimate = 0
 }
 
 // RegisterMetrics registers metrics about builder to report to reg. All

pkg/dataobj/dataobj_test.go

@@ -83,7 +83,8 @@ func Test(t *testing.T) {
 		for _, entry := range streams {
 			require.NoError(t, builder.Append(entry))
 		}
-		require.NoError(t, builder.Flush(context.Background()))
+		_, err = builder.Flush(context.Background())
+		require.NoError(t, err)
 	})
 
 	t.Run("Read", func(t *testing.T) {

pkg/dataobj/internal/sections/streams/streams.go

@@ -3,9 +3,11 @@
 package streams
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"sort"
+	"sync"
 	"time"
 
 	"github.com/prometheus/client_golang/prometheus"
@@ -16,6 +18,7 @@
 	"github.com/grafana/loki/v3/pkg/dataobj/internal/encoding"
 	"github.com/grafana/loki/v3/pkg/dataobj/internal/metadata/datasetmd"
 	"github.com/grafana/loki/v3/pkg/dataobj/internal/metadata/streamsmd"
+	"github.com/grafana/loki/v3/pkg/dataobj/internal/result"
 	"github.com/grafana/loki/v3/pkg/dataobj/internal/streamio"
 	"github.com/grafana/loki/v3/pkg/dataobj/internal/util/sliceclear"
 )
@@ -33,6 +36,20 @@
 	Rows         int           // Number of rows in the stream.
 }
 
+func (s *Stream) Reset() {
+	s.ID = 0
+	s.Labels = nil
+	s.MinTimestamp = time.Time{}
+	s.MaxTimestamp = time.Time{}
+	s.Rows = 0
+}
+
+var streamPool = sync.Pool{
+	New: func() interface{} {
+		return &Stream{}
+	},
+}
+
 // Streams tracks information about streams in a data object.
 type Streams struct {
 	metrics  *Metrics
@@ -61,10 +78,26 @@
 	return &Streams{
 		metrics:  metrics,
 		pageSize: pageSize,
-		lookup:   make(map[uint64][]*Stream),
+		lookup:   make(map[uint64][]*Stream, 1024),
+		ordered:  make([]*Stream, 0, 1024),
 	}
 }
 
+func (s *Streams) Iter(ctx context.Context) result.Seq[Stream] {
+	return result.Iter(func(yield func(Stream) bool) error {
+		for _, stream := range s.ordered {
+			if !yield(*stream) {
+				return nil
+			}
+		}
+		return nil
+	})
+}
+
+func (s *Streams) GetBounds() (time.Time, time.Time) {
+	return s.globalMinTimestamp, s.globalMaxTimestamp
+}
+
 // Record a stream record within the Streams section. The provided timestamp is
 // used to track the minimum and maximum timestamp of a stream. The number of
 // calls to Record is used to track the number of rows for a stream.
@@ -153,7 +186,11 @@
 		s.currentLabelsSize += len(lbl.Value)
 	}
 
-	newStream := &Stream{ID: s.lastID.Add(1), Labels: streamLabels}
+	newStream := streamPool.Get().(*Stream)
+	newStream.Reset()
+	newStream.ID = s.lastID.Add(1)
+	newStream.Labels = streamLabels
+
 	s.lookup[hash] = append(s.lookup[hash], newStream)
 	s.ordered = append(s.ordered, newStream)
 	s.metrics.streamCount.Inc()
@@ -187,7 +224,6 @@
 func (s *Streams) EncodeTo(enc *encoding.Encoder) error {
 	timer := prometheus.NewTimer(s.metrics.encodeSeconds)
 	defer timer.ObserveDuration()
-	defer s.Reset()
 
 	// TODO(rfratto): handle one section becoming too large. This can happen when
 	// the number of columns is very wide. There are two approaches to handle
@@ -333,6 +369,9 @@
 // Reset resets all state, allowing Streams to be reused.
 func (s *Streams) Reset() {
 	s.lastID.Store(0)
+	for _, stream := range s.ordered {
+		streamPool.Put(stream)
+	}
 	clear(s.lookup)
 	s.ordered = sliceclear.Clear(s.ordered)
 	s.currentLabelsSize = 0