Improve granularity of index in measure by leveling up from data point to series

CHANGES.md

@@ -23,6 +23,7 @@ Release Notes.
 - Add the Istio scenario stress test based on the data generated by the integration access log.
 - Generalize the index's docID to uint64.
 - Remove redundant ID tag type.
+- Improve granularity of index in `measure` by leveling up from data point to series.
 
 ### Bugs
 

banyand/measure/measure_write.go

@@ -56,7 +56,8 @@ func (s *measure) write(shardID common.ShardID, entity []byte, entityValues tsdb
 	if err != nil {
 		return err
 	}
-	series, err := shard.Series().Get(entity, entityValues)
+	seriesDB := shard.Series()
+	series, err := seriesDB.Get(entity, entityValues)
 	if err != nil {
 		return err
 	}
@@ -69,18 +70,18 @@ func (s *measure) write(shardID common.ShardID, entity []byte, entityValues tsdb
 		}
 		return err
 	}
-	writeFn := func() (tsdb.Writer, error) {
+	writeFn := func() error {
 		builder := wp.WriterBuilder().Time(t)
 		for fi, family := range value.GetTagFamilies() {
 			spec := s.schema.GetTagFamilies()[fi]
 			bb, errMarshal := pbv1.EncodeFamily(spec, family)
 			if errMarshal != nil {
-				return nil, errMarshal
+				return errMarshal
 			}
 			builder.Family(familyIdentity(spec.GetName(), pbv1.TagFlag), bb)
 		}
 		if len(value.GetFields()) > len(s.schema.GetFields()) {
-			return nil, errors.Wrap(errMalformedElement, "fields number is more than expected")
+			return errors.Wrap(errMalformedElement, "fields number is more than expected")
 		}
 		for fi, fieldValue := range value.GetFields() {
 			fieldSpec := s.schema.GetFields()[fi]
@@ -90,7 +91,7 @@ func (s *measure) write(shardID common.ShardID, entity []byte, entityValues tsdb
 				continue
 			}
 			if fType != fieldSpec.GetFieldType() {
-				return nil, errors.Wrapf(errMalformedElement, "field %s type is unexpected", fieldSpec.GetName())
+				return errors.Wrapf(errMalformedElement, "field %s type is unexpected", fieldSpec.GetName())
 			}
 			data := encodeFieldValue(fieldValue)
 			if data == nil {
@@ -101,7 +102,7 @@ func (s *measure) write(shardID common.ShardID, entity []byte, entityValues tsdb
 		}
 		writer, errWrite := builder.Build()
 		if errWrite != nil {
-			return nil, errWrite
+			return errWrite
 		}
 		_, errWrite = writer.Write()
 		if s.l.Debug().Enabled() {
@@ -114,15 +115,15 @@ func (s *measure) write(shardID common.ShardID, entity []byte, entityValues tsdb
 				Int("shard_id", int(shardID)).
 				Msg("write measure")
 		}
-		return writer, errWrite
+		return errWrite
 	}
-	writer, err := writeFn()
-	if err != nil {
+
+	if err = writeFn(); err != nil {
 		_ = wp.Close()
 		return err
 	}
 	m := index.Message{
-		LocalWriter: writer,
+		IndexWriter: tsdb.NewSeriesIndexWriter(series.ID(), seriesDB),
 		Value: index.Value{
 			TagFamilies: value.GetTagFamilies(),
 			Timestamp:   t,

banyand/measure/service.go

@@ -202,5 +202,8 @@ func NewService(_ context.Context, metadata metadata.Repo, repo discovery.Servic
 		repo:     repo,
 		pipeline: pipeline,
 		stopCh:   make(chan struct{}),
+		dbOpts: tsdb.DatabaseOpts{
+			IndexGranularity: tsdb.IndexGranularitySeries,
+		},
 	}, nil
 }

banyand/observability/meter_prom.go

@@ -21,6 +21,8 @@
 package observability
 
 import (
+	"sync"
+
 	grpcprom "github.com/grpc-ecosystem/go-grpc-middleware/providers/prometheus"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
@@ -30,7 +32,12 @@ import (
 	"google.golang.org/grpc"
 )
 
-var reg = prometheus.NewRegistry()
+var (
+	reg = prometheus.NewRegistry()
+
+	once       = sync.Once{}
+	srvMetrics *grpcprom.ServerMetrics
+)
 
 func init() {
 	reg.MustRegister(prometheus.NewGoCollector())
@@ -48,11 +55,13 @@ func NewMeterProvider(scope meter.Scope) meter.Provider {
 
 // MetricsServerInterceptor returns a server interceptor for metrics.
 func MetricsServerInterceptor() (grpc.UnaryServerInterceptor, grpc.StreamServerInterceptor) {
-	srvMetrics := grpcprom.NewServerMetrics(
-		grpcprom.WithServerHandlingTimeHistogram(
-			grpcprom.WithHistogramBuckets([]float64{0.001, 0.01, 0.1, 0.3, 0.6, 1, 3, 6, 9, 20, 30, 60, 90, 120}),
-		),
-	)
-	reg.MustRegister(srvMetrics)
+	once.Do(func() {
+		srvMetrics = grpcprom.NewServerMetrics(
+			grpcprom.WithServerHandlingTimeHistogram(
+				grpcprom.WithHistogramBuckets([]float64{0.001, 0.01, 0.1, 0.3, 0.6, 1, 3, 6, 9, 20, 30, 60, 90, 120}),
+			),
+		)
+		reg.MustRegister(srvMetrics)
+	})
 	return srvMetrics.UnaryServerInterceptor(), srvMetrics.StreamServerInterceptor()
 }

banyand/stream/service.go

@@ -162,6 +162,7 @@ func NewService(_ context.Context, metadata metadata.Repo, repo discovery.Servic
 		pipeline: pipeline,
 		dbOpts: tsdb.DatabaseOpts{
 			EnableGlobalIndex: true,
+			IndexGranularity:  tsdb.IndexGranularityBlock,
 		},
 		stopCh: make(chan struct{}),
 	}, nil

banyand/stream/stream_write.go

@@ -120,8 +120,9 @@ func (s *stream) write(shardID common.ShardID, entity []byte, entityValues tsdb.
 		return err
 	}
 	m := index.Message{
-		Scope:       tsdb.Entry(s.name),
-		LocalWriter: writer,
+		Scope:        tsdb.Entry(s.name),
+		IndexWriter:  writer,
+		GlobalItemID: writer.ItemID(),
 		Value: index.Value{
 			TagFamilies: value.GetTagFamilies(),
 			Timestamp:   t,

banyand/tsdb/block.go

@@ -108,7 +108,7 @@ type block struct {
 
 type openOpts struct {
 	tsTableFactory TSTableFactory
-	inverted       inverted.StoreOpts
+	inverted       *inverted.StoreOpts
 	lsm            lsm.StoreOpts
 }
 
@@ -163,10 +163,12 @@ func options(ctx context.Context, root string, l *logger.Logger) (openOpts, erro
 	}
 	options = o.(DatabaseOpts)
 	var opts openOpts
-	opts.inverted = inverted.StoreOpts{
-		Path:         path.Join(root, componentSecondInvertedIdx),
-		Logger:       l.Named(componentSecondInvertedIdx),
-		BatchWaitSec: options.BlockInvertedIndex.BatchWaitSec,
+	if options.IndexGranularity == IndexGranularityBlock {
+		opts.inverted = &inverted.StoreOpts{
+			Path:         path.Join(root, componentSecondInvertedIdx),
+			Logger:       l.Named(componentSecondInvertedIdx),
+			BatchWaitSec: options.BlockInvertedIndex.BatchWaitSec,
+		}
 	}
 	opts.lsm = lsm.StoreOpts{
 		Path:         path.Join(root, componentSecondLSMIdx),
@@ -198,13 +200,16 @@ func (b *block) open() (err error) {
 		return err
 	}
 	b.closableLst = append(b.closableLst, b.tsTable)
-	if b.invertedIndex, err = inverted.NewStore(b.openOpts.inverted); err != nil {
-		return err
+	if b.openOpts.inverted != nil {
+		if b.invertedIndex, err = inverted.NewStore(*b.openOpts.inverted); err != nil {
+			return err
+		}
+		b.closableLst = append(b.closableLst, b.invertedIndex)
 	}
 	if b.lsmIndex, err = lsm.NewStore(b.openOpts.lsm); err != nil {
 		return err
 	}
-	b.closableLst = append(b.closableLst, b.invertedIndex, b.lsmIndex)
+	b.closableLst = append(b.closableLst, b.lsmIndex)
 	b.ref.Store(0)
 	b.closed.Store(false)
 	blockOpenedTimeSecondsGauge.Set(float64(time.Now().Unix()), b.position.LabelValues()...)
@@ -431,6 +436,9 @@ func (d *bDelegate) writeLSMIndex(fields []index.Field, id common.ItemID) error
 }
 
 func (d *bDelegate) writeInvertedIndex(fields []index.Field, id common.ItemID) error {
+	if d.delegate.invertedIndex == nil {
+		return errors.New("inverted index is not enabled")
+	}
 	total := 0
 	for _, f := range fields {
 		total += len(f.Marshal())

banyand/tsdb/buffer.go

@@ -161,6 +161,9 @@ func (b *Buffer) Read(key []byte, ts time.Time) ([]byte, bool) {
 
 // Close gracefully closes the Buffer and ensures that all pending operations are completed.
 func (b *Buffer) Close() error {
+	if b == nil {
+		return nil
+	}
 	b.closerOnce.Do(func() {
 		b.entryCloser.Done()
 		b.entryCloser.CloseThenWait()

banyand/tsdb/index/writer.go

@@ -38,10 +38,11 @@ import (
 
 // Message wraps value and other info to generate relative indices.
 type Message struct {
-	Value       Value
-	LocalWriter tsdb.Writer
-	BlockCloser io.Closer
-	Scope       tsdb.Entry
+	IndexWriter  tsdb.IndexWriter
+	BlockCloser  io.Closer
+	Value        Value
+	Scope        tsdb.Entry
+	GlobalItemID tsdb.GlobalItemID
 }
 
 // Value represents the input data for generating indices.
@@ -122,8 +123,8 @@ func NewWriter(ctx context.Context, options WriterOptions) *Writer {
 
 func (s *Writer) Write(m Message) {
 	err := multierr.Combine(
-		s.writeLocalIndex(m.LocalWriter, m.Value),
-		s.writeGlobalIndex(m.Scope, m.LocalWriter.ItemID(), m.Value),
+		s.writeLocalIndex(m.IndexWriter, m.Value),
+		s.writeGlobalIndex(m.Scope, m.GlobalItemID, m.Value),
 		m.BlockCloser.Close(),
 	)
 	if err != nil {
@@ -191,7 +192,7 @@ func (s *Writer) writeGlobalIndex(scope tsdb.Entry, ref tsdb.GlobalItemID, value
 	)
 }
 
-func (s *Writer) writeLocalIndex(writer tsdb.Writer, value Value) (err error) {
+func (s *Writer) writeLocalIndex(writer tsdb.IndexWriter, value Value) (err error) {
 	collect := func(ruleIndexes []*partition.IndexRuleLocator, fn func(fields []index.Field) error) error {
 		fields := make([]index.Field, 0)
 		for _, ruleIndex := range ruleIndexes {

banyand/tsdb/scope.go

@@ -21,6 +21,7 @@ import (
 	"context"
 
 	"github.com/apache/skywalking-banyandb/api/common"
+	"github.com/apache/skywalking-banyandb/pkg/index"
 )
 
 var _ Shard = (*scopedShard)(nil)
@@ -73,6 +74,18 @@ type scopedSeriesDatabase struct {
 	scope     Entry
 }
 
+func (sdd *scopedSeriesDatabase) writeInvertedIndex(fields []index.Field, seriesID common.SeriesID) error {
+	return sdd.delegated.writeInvertedIndex(fields, seriesID)
+}
+
+func (sdd *scopedSeriesDatabase) writeLSMIndex(fields []index.Field, seriesID common.SeriesID) error {
+	return sdd.delegated.writeLSMIndex(fields, seriesID)
+}
+
+func (sdd *scopedSeriesDatabase) Search(ctx context.Context, path Path, filter index.Filter, order *OrderBy) (SeriesList, error) {
+	return sdd.delegated.Search(ctx, path.prepend(sdd.scope), filter, order)
+}
+
 func (sdd *scopedSeriesDatabase) Close() error {
 	return nil
 }

banyand/tsdb/series_seek.go

@@ -62,7 +62,6 @@ type seekerBuilder struct {
 	seriesSpan          *seriesSpan
 	indexRuleForSorting *databasev1.IndexRule
 	l                   *logger.Logger
-	rangeOptsForSorting index.RangeOpts
 	order               modelv1.Sort
 }
 

banyand/tsdb/series_seek_sort.go

@@ -37,6 +37,7 @@ import (
 var (
 	errUnspecifiedIndexType = errors.New("Unspecified index type")
 	emptyFilters            = make([]filterFn, 0)
+	rangeOpts               = index.RangeOpts{}
 )
 
 func (s *seekerBuilder) OrderByIndex(indexRule *databasev1.IndexRule, order modelv1.Sort) SeekerBuilder {
@@ -84,9 +85,9 @@ func (s *seekerBuilder) buildSeriesByIndex() (series []Iterator, err error) {
 		}
 		switch s.indexRuleForSorting.GetType() {
 		case databasev1.IndexRule_TYPE_TREE:
-			inner, err = b.lsmIndexReader().Iterator(fieldKey, s.rangeOptsForSorting, s.order)
+			inner, err = b.lsmIndexReader().Iterator(fieldKey, rangeOpts, s.order)
 		case databasev1.IndexRule_TYPE_INVERTED:
-			inner, err = b.invertedIndexReader().Iterator(fieldKey, s.rangeOptsForSorting, s.order)
+			inner, err = b.invertedIndexReader().Iterator(fieldKey, rangeOpts, s.order)
 		case databasev1.IndexRule_TYPE_UNSPECIFIED:
 			return nil, errors.WithMessagef(errUnspecifiedIndexType, "index rule:%v", s.indexRuleForSorting)
 		}

banyand/tsdb/series_write.go

@@ -207,3 +207,26 @@ func (w *writer) Write() (GlobalItemID, error) {
 		Term: convert.Int64ToBytes(w.ts.UnixNano()),
 	}, id.ID)
 }
+
+var _ IndexWriter = (*seriesIndexWriter)(nil)
+
+type seriesIndexWriter struct {
+	seriesDB SeriesDatabase
+	seriesID common.SeriesID
+}
+
+func (s *seriesIndexWriter) WriteInvertedIndex(fields []index.Field) error {
+	return s.seriesDB.writeInvertedIndex(fields, s.seriesID)
+}
+
+func (s *seriesIndexWriter) WriteLSMIndex(fields []index.Field) error {
+	return s.seriesDB.writeLSMIndex(fields, s.seriesID)
+}
+
+// NewSeriesIndexWriter returns a new series index writer.
+func NewSeriesIndexWriter(seriesID common.SeriesID, seriesDB SeriesDatabase) IndexWriter {
+	return &seriesIndexWriter{
+		seriesID: seriesID,
+		seriesDB: seriesDB,
+	}
+}