Optimize query performance of series index

CHANGES.md

@@ -18,6 +18,7 @@ Release Notes.
 - Add the stream query trace.
 - Add the topN query trace.
 - Introduce the round-robin selector to Liaison Node.
+- Optimize query performance of series index.
 
 ### Bugs
 

banyand/internal/storage/index.go

@@ -28,18 +28,17 @@ import (
 	"sync"
 	"time"
 
+	"github.com/blugelabs/bluge"
 	"github.com/pkg/errors"
 	"go.uber.org/multierr"
 
 	"github.com/apache/skywalking-banyandb/api/common"
-	databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
 	"github.com/apache/skywalking-banyandb/pkg/index"
 	"github.com/apache/skywalking-banyandb/pkg/index/inverted"
 	"github.com/apache/skywalking-banyandb/pkg/index/posting"
 	"github.com/apache/skywalking-banyandb/pkg/logger"
 	pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
 	"github.com/apache/skywalking-banyandb/pkg/query"
-	"github.com/apache/skywalking-banyandb/pkg/query/logical"
 	"github.com/apache/skywalking-banyandb/pkg/timestamp"
 )
 
@@ -183,7 +182,7 @@ func convertIndexSeriesToSeriesList(indexSeries []index.Series) (pbv1.SeriesList
 	return seriesList, nil
 }
 
-func (s *seriesIndex) Search(ctx context.Context, series []*pbv1.Series, filter index.Filter, order *pbv1.OrderBy, preloadSize int) (sl pbv1.SeriesList, err error) {
+func (s *seriesIndex) Search(ctx context.Context, series []*pbv1.Series, blugeQuery bluge.Query, order *pbv1.OrderBy, preloadSize int) (sl pbv1.SeriesList, err error) {
 	tracer := query.GetTracer(ctx)
 	if tracer != nil {
 		var span *query.Span
@@ -201,12 +200,11 @@ func (s *seriesIndex) Search(ctx context.Context, series []*pbv1.Series, filter
 	}
 
 	pl := seriesList.ToList()
-	if filter != nil && filter != logical.ENode {
+	if blugeQuery != nil {
 		var plFilter posting.List
 		func() {
 			if tracer != nil {
 				span, _ := tracer.StartSpan(ctx, "filter")
-				span.Tag("exp", filter.String())
 				defer func() {
 					if err != nil {
 						span.Error(err)
@@ -217,9 +215,7 @@ func (s *seriesIndex) Search(ctx context.Context, series []*pbv1.Series, filter
 					span.Stop()
 				}()
 			}
-			if plFilter, err = filter.Execute(func(_ databasev1.IndexRule_Type) (index.Searcher, error) {
-				return s.store, nil
-			}, 0); err != nil {
+			if plFilter, err = s.store.Execute(blugeQuery); err != nil {
 				return
 			}
 			if plFilter == nil {
@@ -499,19 +495,19 @@ func (sic *seriesIndexController[T, O]) searchPrimary(ctx context.Context, serie
 }
 
 func (sic *seriesIndexController[T, O]) Search(ctx context.Context, series []*pbv1.Series,
-	filter index.Filter, order *pbv1.OrderBy, preloadSize int,
+	query bluge.Query, order *pbv1.OrderBy, preloadSize int,
 ) (pbv1.SeriesList, error) {
 	sic.RLock()
 	defer sic.RUnlock()
 
-	sl, err := sic.hot.Search(ctx, series, filter, order, preloadSize)
+	sl, err := sic.hot.Search(ctx, series, query, order, preloadSize)
 	if err != nil {
 		return nil, err
 	}
 	if len(sl) > 0 || sic.standby == nil {
 		return sl, nil
 	}
-	return sic.standby.Search(ctx, series, filter, order, preloadSize)
+	return sic.standby.Search(ctx, series, query, order, preloadSize)
 }
 
 func (sic *seriesIndexController[T, O]) Close() error {

banyand/internal/storage/storage.go

@@ -28,6 +28,7 @@ import (
 	"io"
 	"time"
 
+	"github.com/blugelabs/bluge"
 	"github.com/pkg/errors"
 
 	"github.com/apache/skywalking-banyandb/api/common"
@@ -66,7 +67,7 @@ type SupplyTSDB[T TSTable] func() T
 // IndexDB is the interface of index database.
 type IndexDB interface {
 	Write(docs index.Documents) error
-	Search(ctx context.Context, series []*pbv1.Series, filter index.Filter, order *pbv1.OrderBy, preloadSize int) (pbv1.SeriesList, error)
+	Search(ctx context.Context, series []*pbv1.Series, query bluge.Query, order *pbv1.OrderBy, preloadSize int) (pbv1.SeriesList, error)
 }
 
 // TSDB allows listing and getting shard details.

banyand/measure/query.go

@@ -89,7 +89,7 @@ func (s *measure) Query(ctx context.Context, mqo pbv1.MeasureQueryOptions) (mqr
 		}
 	}
 
-	sl, err := tsdb.IndexDB().Search(ctx, series, mqo.Filter, mqo.Order, preloadSize)
+	sl, err := tsdb.IndexDB().Search(ctx, series, mqo.Query, mqo.Order, preloadSize)
 	if err != nil {
 		return nil, err
 	}

pkg/index/index.go

@@ -24,6 +24,8 @@ import (
 	"fmt"
 	"io"
 
+	"github.com/blugelabs/bluge"
+
 	"github.com/apache/skywalking-banyandb/api/common"
 	databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
 	modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
@@ -188,6 +190,8 @@ type SeriesStore interface {
 	Store
 	// Search returns a list of series that match the given matchers.
 	Search(context.Context, []SeriesMatcher) ([]Series, error)
+	// Execute returns a posting list that matches the given query.
+	Execute(bluge.Query) (posting.List, error)
 }
 
 // SeriesMatcherType represents the type of series matcher.

pkg/index/inverted/inverted.go

@@ -61,10 +61,11 @@ var (
 	defaultRangePreloadSize = 1000
 )
 
-var analyzers map[databasev1.IndexRule_Analyzer]*analysis.Analyzer
+// Analyzers is a map that associates each IndexRule_Analyzer type with a corresponding Analyzer.
+var Analyzers map[databasev1.IndexRule_Analyzer]*analysis.Analyzer
 
 func init() {
-	analyzers = map[databasev1.IndexRule_Analyzer]*analysis.Analyzer{
+	Analyzers = map[databasev1.IndexRule_Analyzer]*analysis.Analyzer{
 		databasev1.IndexRule_ANALYZER_KEYWORD:  analyzer.NewKeywordAnalyzer(),
 		databasev1.IndexRule_ANALYZER_SIMPLE:   analyzer.NewSimpleAnalyzer(),
 		databasev1.IndexRule_ANALYZER_STANDARD: analyzer.NewStandardAnalyzer(),
@@ -120,7 +121,7 @@ func (s *store) Batch(batch index.Batch) error {
 				tf.StoreValue().Sortable()
 			}
 			if f.Key.Analyzer != databasev1.IndexRule_ANALYZER_UNSPECIFIED {
-				tf = tf.WithAnalyzer(analyzers[f.Key.Analyzer])
+				tf = tf.WithAnalyzer(Analyzers[f.Key.Analyzer])
 			}
 			doc.AddField(tf)
 		}
@@ -149,7 +150,7 @@ func NewStore(opts StoreOpts) (index.SeriesStore, error) {
 			WithPersisterNapTimeMSec(int(opts.BatchWaitSec * 1000))
 	}
 	config := bluge.DefaultConfigWithIndexConfig(indexConfig)
-	config.DefaultSearchAnalyzer = analyzers[databasev1.IndexRule_ANALYZER_KEYWORD]
+	config.DefaultSearchAnalyzer = Analyzers[databasev1.IndexRule_ANALYZER_KEYWORD]
 	config.Logger = log.New(opts.Logger, opts.Logger.Module(), 0)
 	w, err := bluge.OpenWriter(config)
 	if err != nil {
@@ -266,7 +267,7 @@ func (s *store) Match(fieldKey index.FieldKey, matches []string) (posting.List,
 	if err != nil {
 		return nil, err
 	}
-	analyzer := analyzers[fieldKey.Analyzer]
+	analyzer := Analyzers[fieldKey.Analyzer]
 	fk := fieldKey.Marshal()
 	query := bluge.NewBooleanQuery()
 	if fieldKey.HasSeriesID() {
@@ -304,6 +305,26 @@ func (s *store) Range(fieldKey index.FieldKey, opts index.RangeOpts) (list posti
 	return
 }
 
+func (s *store) Execute(query bluge.Query) (posting.List, error) {
+	reader, err := s.writer.Reader()
+	if err != nil {
+		return nil, err
+	}
+	documentMatchIterator, err := reader.Search(context.Background(), bluge.NewAllMatches(query))
+	if err != nil {
+		return nil, err
+	}
+	iter := newBlugeMatchIterator(documentMatchIterator, reader, nil)
+	defer func() {
+		err = multierr.Append(err, iter.Close())
+	}()
+	list := roaring.NewPostingList()
+	for iter.Next() {
+		list.Insert(iter.Val().docID)
+	}
+	return list, err
+}
+
 func (s *store) SizeOnDisk() int64 {
 	_, bytes := s.writer.DirectoryStats()
 	return int64(bytes)

pkg/pb/v1/metadata.go

@@ -21,6 +21,8 @@ package v1
 import (
 	"context"
 
+	"github.com/blugelabs/bluge"
+
 	"github.com/apache/skywalking-banyandb/api/common"
 	databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
 	modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
@@ -158,7 +160,7 @@ const (
 
 // MeasureQueryOptions is the options of a measure query.
 type MeasureQueryOptions struct {
-	Filter          index.Filter
+	Query           bluge.Query
 	TimeRange       *timestamp.TimeRange
 	Order           *OrderBy
 	Name            string