Optimize query performance of series index

CHANGES.md

@@ -20,6 +20,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

@@ -25,14 +25,12 @@ import (
 	"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/query/model"
 )
 
@@ -172,7 +170,9 @@ func convertIndexSeriesToSeriesList(indexSeries []index.Series) (pbv1.SeriesList
 	return seriesList, nil
 }
 
-func (s *seriesIndex) Search(ctx context.Context, series []*pbv1.Series, filter index.Filter, order *model.OrderBy, preloadSize int) (sl pbv1.SeriesList, err error) {
+func (s *seriesIndex) Search(ctx context.Context, series []*pbv1.Series, blugeQuery *inverted.BlugeQuery,
+	filter index.Filter, order *model.OrderBy, preloadSize int,
+) (sl pbv1.SeriesList, err error) {
 	tracer := query.GetTracer(ctx)
 	if tracer != nil {
 		var span *query.Span
@@ -190,7 +190,7 @@ 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 {
@@ -206,9 +206,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 {

banyand/internal/storage/storage.go

@@ -34,6 +34,7 @@ import (
 	commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
 	"github.com/apache/skywalking-banyandb/pkg/fs"
 	"github.com/apache/skywalking-banyandb/pkg/index"
+	"github.com/apache/skywalking-banyandb/pkg/index/inverted"
 	"github.com/apache/skywalking-banyandb/pkg/logger"
 	pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
 	"github.com/apache/skywalking-banyandb/pkg/query/model"
@@ -67,7 +68,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 *model.OrderBy, preloadSize int) (pbv1.SeriesList, error)
+	Search(ctx context.Context, series []*pbv1.Series, query *inverted.BlugeQuery, filter index.Filter, order *model.OrderBy, preloadSize int) (pbv1.SeriesList, error)
 }
 
 // TSDB allows listing and getting shard details.

banyand/measure/query.go

@@ -151,7 +151,7 @@ func (s *measure) searchSeriesList(ctx context.Context, series []*pbv1.Series, m
 ) (sl pbv1.SeriesList, tables []*tsTable, err error) {
 	for i := range segments {
 		tables = append(tables, segments[i].Tables()...)
-		sll, err := segments[i].IndexDB().Search(ctx, series, mqo.Filter, mqo.Order, preloadSize)
+		sll, err := segments[i].IndexDB().Search(ctx, series, mqo.Query, mqo.Filter, mqo.Order, preloadSize)
 		if err != nil {
 			return nil, nil, err
 		}

banyand/stream/query.go

@@ -30,14 +30,14 @@ import (
 	"github.com/apache/skywalking-banyandb/banyand/internal/storage"
 	"github.com/apache/skywalking-banyandb/pkg/convert"
 	"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/index/posting/roaring"
 	itersort "github.com/apache/skywalking-banyandb/pkg/iter/sort"
 	"github.com/apache/skywalking-banyandb/pkg/logger"
 	"github.com/apache/skywalking-banyandb/pkg/partition"
 	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/query/model"
 )
 
@@ -519,7 +519,7 @@ func (qr *queryResult) mergeByTimestamp() *model.StreamResult {
 func indexSearch(sqo model.StreamQueryOptions,
 	tabs []*tsTable, seriesList pbv1.SeriesList,
 ) (posting.List, error) {
-	if sqo.Filter == nil || sqo.Filter == logical.ENode {
+	if sqo.Filter == nil || sqo.Filter == inverted.ENode {
 		return nil, nil
 	}
 	result := roaring.NewPostingList()

pkg/convert/json.go

@@ -0,0 +1,29 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package convert
+
+import "encoding/json"
+
+// JSONToString converts a JSON marshaler to its JSON string representation.
+func JSONToString(marshaler json.Marshaler) string {
+	bb, err := marshaler.MarshalJSON()
+	if err != nil {
+		return err.Error()
+	}
+	return string(bb)
+}

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.