Simplify interface, just fallback to noop

Signed-off-by: Oleg Zaytsev <[email protected]>

Author: colega

storage/interface.go

@@ -289,13 +289,14 @@ type SeriesWithRef struct {
 	Ref    SeriesRef
 }
 
-type BatchAppender interface {
-	Appender
-	// BatchSeriesRefs will create series and return their SeriesRef and existing labels, reusing buf when possible.
-	// Series that could not be created because of invalid labelsets will have their SeriesRef set to 0.
-	// Series that have to be created will copy (calling Labels.Copy()) the labels before storing them,
-	// so no references to series will be kept.
-	// Series that already exist and didn't have to be created will return the labels stored in the TSDB.
+type BatchSeriesReferencer interface {
+	// BatchSeriesRefs will attempt to, for each one of the series provided:
+	// - If it does not exist, clone the labels and create a new series, returning the reference and the cloned labels.
+	// - If it does exist, return the reference and the labels stored in the TSDB.
+	// The result will contain the references and labels for each series in the same order as the input.
+	// If the reference is 0, it means that series was not created,
+	// this could happen because the input was not valid (e.g. empty labels),
+	// or because the implementation does not support this feature (for example, initAppender when head is not initialized yet).
 	BatchSeriesRefs(series []labels.Labels, buf []SeriesWithRef) []SeriesWithRef
 }
 

tsdb/db.go

@@ -1203,27 +1203,6 @@ func (db *DB) run(ctx context.Context) {
 	}
 }
 
-// HeadInitialized returns true if the Head has been initialized with a starting time.
-func (db *DB) HeadInitialized() bool {
-	return db.head.initialized()
-}
-
-// InitializeHead the Head with a starting time.
-// After Initialize() is called for the first time, Initialized() is always true.
-func (db *DB) InitializeHead(t int64) {
-	db.head.initTime(t)
-}
-
-// BatchAppender returns a storage.BatchAppender, it requires the head to be initialized.
-// If the Head is not initialized, it returns nil and false.
-func (db *DB) BatchAppender(ctx context.Context) (storage.BatchAppender, bool) {
-	app, ok := db.head.batchAppender(ctx)
-	if ok {
-		return batchDBAppender{BatchAppender: app, dba: dbAppender{Appender: app, db: db}}, true
-	}
-	return nil, false
-}
-
 // Appender opens a new appender against the database.
 func (db *DB) Appender(ctx context.Context) storage.Appender {
 	return dbAppender{db: db, Appender: db.head.Appender(ctx)}
@@ -1302,6 +1281,7 @@ type dbAppender struct {
 }
 
 var _ storage.GetRef = dbAppender{}
+var _ storage.BatchSeriesReferencer = dbAppender{}
 
 func (a dbAppender) GetRef(lset labels.Labels, hash uint64) (storage.SeriesRef, labels.Labels) {
 	if g, ok := a.Appender.(storage.GetRef); ok {
@@ -1310,6 +1290,13 @@ func (a dbAppender) GetRef(lset labels.Labels, hash uint64) (storage.SeriesRef,
 	return 0, labels.EmptyLabels()
 }
 
+func (a dbAppender) BatchSeriesRefs(series []labels.Labels, buf []storage.SeriesWithRef) []storage.SeriesWithRef {
+	if r, ok := a.Appender.(storage.BatchSeriesReferencer); ok {
+		return r.BatchSeriesRefs(series, buf)
+	}
+	return unknownBatchSeriesRefs(series, buf)
+}
+
 func (a dbAppender) Commit() error {
 	err := a.Appender.Commit()
 
@@ -1324,21 +1311,6 @@ func (a dbAppender) Commit() error {
 	return err
 }
 
-var _ storage.BatchAppender = batchDBAppender{}
-
-type batchDBAppender struct {
-	storage.BatchAppender
-	dba dbAppender
-}
-
-func (a batchDBAppender) GetRef(lset labels.Labels, hash uint64) (storage.SeriesRef, labels.Labels) {
-	return a.dba.GetRef(lset, hash)
-}
-
-func (a batchDBAppender) Commit() error {
-	return a.dba.Commit()
-}
-
 // waitingForCompactionDelay returns true if the DB is waiting for the Head compaction delay.
 // This doesn't guarantee that the Head is really compactable.
 func (db *DB) waitingForCompactionDelay() bool {

tsdb/db_test.go

@@ -335,12 +335,13 @@ func TestBatchDBAppenderAddRef(t *testing.T) {
 	}()
 
 	ctx := context.Background()
-	db.InitializeHead(123)
-	app1, ok := db.BatchAppender(ctx)
+	db.Head().Init(123)
+	app1 := db.Appender(ctx)
+	batchApp1, ok := app1.(storage.BatchSeriesReferencer)
 	require.True(t, ok)
 
 	lset1 := labels.FromStrings("a", "b")
-	entries1 := app1.BatchSeriesRefs([]labels.Labels{lset1}, nil)
+	entries1 := batchApp1.BatchSeriesRefs([]labels.Labels{lset1}, nil)
 	require.NotZero(t, entries1[0])
 	require.True(t, labels.Equal(entries1[0].Labels, lset1))
 
@@ -353,12 +354,13 @@ func TestBatchDBAppenderAddRef(t *testing.T) {
 	err = app1.Commit()
 	require.NoError(t, err)
 
-	app2, ok := db.BatchAppender(ctx)
+	app2 := db.Appender(ctx)
+	batchApp2, ok := app2.(storage.BatchSeriesReferencer)
 	require.True(t, ok)
 
 	lset2 := labels.FromStrings("a", "c")
 	lset3 := labels.FromStrings("x", "y")
-	entries2 := app1.BatchSeriesRefs([]labels.Labels{lset1, lset2, lset3, labels.EmptyLabels()}, nil)
+	entries2 := batchApp2.BatchSeriesRefs([]labels.Labels{lset1, lset2, lset3, labels.EmptyLabels()}, nil)
 	require.Equal(t, entries1[0], entries2[0])
 	require.NotEqual(t, entries1[0], entries2[2])
 	require.Zero(t, entries2[3], "Empty labels should not create a new series")
@@ -398,15 +400,16 @@ func TestBatchDBAppenderBatchSeriesRefsCopiesLabels(t *testing.T) {
 	}()
 
 	ctx := context.Background()
-	db.InitializeHead(123)
-	app1, ok := db.BatchAppender(ctx)
+	db.Head().Init(123)
+	app1 := db.Appender(ctx)
+	batchApp1, ok := app1.(storage.BatchSeriesReferencer)
 	require.True(t, ok)
 
 	b := labels.NewScratchBuilder(0)
 	b.Add("a", "b")
 	unsafeMutableLabels := b.Labels()
 
-	entries1 := app1.BatchSeriesRefs([]labels.Labels{unsafeMutableLabels}, nil)
+	entries1 := batchApp1.BatchSeriesRefs([]labels.Labels{unsafeMutableLabels}, nil)
 	require.NotZero(t, entries1[0])
 	require.True(t, labels.Equal(entries1[0].Labels, unsafeMutableLabels))
 
@@ -426,9 +429,10 @@ func TestBatchDBAppenderBatchSeriesRefsCopiesLabels(t *testing.T) {
 	require.NoError(t, err)
 
 	// Check that overwritten labels create new series
-	app2, ok := db.BatchAppender(ctx)
+	app2 := db.Appender(ctx)
+	batchApp2, ok := app2.(storage.BatchSeriesReferencer)
 	require.True(t, ok)
-	entries2 := app1.BatchSeriesRefs([]labels.Labels{unsafeMutableLabels}, nil)
+	entries2 := batchApp2.BatchSeriesRefs([]labels.Labels{unsafeMutableLabels}, nil)
 	require.NotEqual(t, entries2[0], entries1[0])
 	require.True(t, labels.Equal(entries2[0].Labels, unsafeMutableLabels))
 
@@ -442,24 +446,26 @@ func TestBatchDBAppenderBatchSeriesRefsReturnsLabelsFromStorage(t *testing.T) {
 	}()
 
 	ctx := context.Background()
-	db.InitializeHead(123)
-	app1, ok := db.BatchAppender(ctx)
+	db.Head().Init(123)
+	app1 := db.Appender(ctx)
+	batchApp1, ok := app1.(storage.BatchSeriesReferencer)
 	require.True(t, ok)
 
 	b := labels.NewScratchBuilder(0)
 	b.Add("a", "b")
 	unsafeMutableLabels := b.Labels()
 
-	entries1 := app1.BatchSeriesRefs([]labels.Labels{unsafeMutableLabels}, nil)
+	entries1 := batchApp1.BatchSeriesRefs([]labels.Labels{unsafeMutableLabels}, nil)
 	require.NotZero(t, entries1[0])
 	require.True(t, labels.Equal(entries1[0].Labels, unsafeMutableLabels))
 
 	require.NoError(t, app1.Commit())
 
 	// Retrieve same labels again, they should exist now.
-	app2, ok := db.BatchAppender(ctx)
+	app2 := db.Appender(ctx)
+	batchApp2, ok := app2.(storage.BatchSeriesReferencer)
 	require.True(t, ok)
-	entries2 := app1.BatchSeriesRefs([]labels.Labels{unsafeMutableLabels}, nil)
+	entries2 := batchApp2.BatchSeriesRefs([]labels.Labels{unsafeMutableLabels}, nil)
 	require.Equal(t, entries2[0], entries1[0])
 	require.True(t, labels.Equal(entries2[0].Labels, unsafeMutableLabels))
 
@@ -481,11 +487,12 @@ func TestBatchDBAppenderTracksCreatedSeriesInWAL(t *testing.T) {
 
 	// Append to that DB and close it.
 	{
-		db.InitializeHead(123)
-		app, ok := db.BatchAppender(ctx)
+		db.Head().Init(123)
+		app := db.Appender(ctx)
+		batchApp, ok := app.(storage.BatchSeriesReferencer)
 		require.True(t, ok)
 
-		refs := app.BatchSeriesRefs([]labels.Labels{lbls}, nil)
+		refs := batchApp.BatchSeriesRefs([]labels.Labels{lbls}, nil)
 		require.NotZero(t, refs[0])
 		_, err := app.Append(refs[0].Ref, labels.EmptyLabels(), 0, 1)
 		require.NoError(t, err)
@@ -595,14 +602,15 @@ func TestBatchSeriesRefEmptyLabelsIgnored(t *testing.T) {
 	}()
 
 	ctx := context.Background()
-	db.InitializeHead(123)
-	app1, ok := db.BatchAppender(ctx)
+	db.Head().Init(123)
+	app1 := db.Appender(ctx)
+	batchApp1, ok := app1.(storage.BatchSeriesReferencer)
 	require.True(t, ok)
 
 	lset1 := labels.FromStrings("a", "b")
 	lset2 := labels.FromStrings("a", "b", "c", "")
 
-	entries := app1.BatchSeriesRefs([]labels.Labels{lset1, lset2}, nil)
+	entries := batchApp1.BatchSeriesRefs([]labels.Labels{lset1, lset2}, nil)
 	require.Len(t, entries, 2)
 	require.Equal(t, entries[0], entries[1])
 

tsdb/head_append.go

@@ -40,6 +40,7 @@ type initAppender struct {
 }
 
 var _ storage.GetRef = &initAppender{}
+var _ storage.BatchSeriesReferencer = &initAppender{}
 
 func (a *initAppender) SetOptions(opts *storage.AppendOptions) {
 	if a.app != nil {
@@ -132,6 +133,13 @@ func (a *initAppender) GetRef(lset labels.Labels, hash uint64) (storage.SeriesRe
 	return 0, labels.EmptyLabels()
 }
 
+func (a *initAppender) BatchSeriesRefs(series []labels.Labels, buf []storage.SeriesWithRef) []storage.SeriesWithRef {
+	if r, ok := a.app.(storage.BatchSeriesReferencer); ok {
+		return r.BatchSeriesRefs(series, buf)
+	}
+	return unknownBatchSeriesRefs(series, buf)
+}
+
 func (a *initAppender) Commit() error {
 	if a.app == nil {
 		a.head.metrics.activeAppenders.Dec()
@@ -148,15 +156,6 @@ func (a *initAppender) Rollback() error {
 	return a.app.Rollback()
 }
 
-// batchAppender returns a storage.BatchAppender, it requires the head to be initialized.
-// If the Head is not initialized, it returns nil and false.
-func (h *Head) batchAppender(_ context.Context) (app storage.BatchAppender, ok bool) {
-	if h.initialized() {
-		return h.appender(), true
-	}
-	return nil, false
-}
-
 // Appender returns a new Appender on the database.
 func (h *Head) Appender(_ context.Context) storage.Appender {
 	h.metrics.activeAppenders.Inc()
@@ -2096,6 +2095,14 @@ func (a *headAppender) Rollback() (err error) {
 	return a.log()
 }
 
+// unknownBatchSeriesRefs is a default non-implementation for storage.BatchSeriesReferencer
+// that will return zero refs and labels for each series provided.
+func unknownBatchSeriesRefs(series []labels.Labels, buf []storage.SeriesWithRef) []storage.SeriesWithRef {
+	result := reuseOrMake(&buf, len(series), len(series))
+	clear(result)
+	return result
+}
+
 // reuseOrMake will try to reuse the reusable slice if it has enough capacity, setting it to the desired len.
 // If it can't be reused, it will make a new slice with the desired length and capacity.
 // If a new slice is allocated, it will update the reusable slice to point to the new slice.