You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by ha...@apache.org on 2023/09/10 00:57:55 UTC
[skywalking-banyandb] 03/03: Add distributed query procedure
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch distributed-query
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit 4844a16d8525e17115e92134ec970c8477586189
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Sun Sep 10 00:40:39 2023 +0000
Add distributed query procedure
Signed-off-by: Gao Hongtao <ha...@gmail.com>
---
CHANGES.md | 1 +
api/proto/banyandb/database/v1/database.proto | 3 +-
banyand/dquery/dquery.go | 68 ++++
banyand/dquery/measure.go | 100 ++++++
banyand/dquery/stream.go | 84 +++++
banyand/dquery/topn.go | 122 +++++++
banyand/internal/cmd/liaison.go | 6 +
banyand/measure/measure.go | 38 +--
banyand/measure/measure_write.go | 2 +-
banyand/measure/metadata.go | 63 +++-
banyand/measure/metadata_test.go | 10 +-
banyand/measure/service.go | 6 +-
banyand/query/processor.go | 5 -
banyand/queue/local.go | 8 +
banyand/queue/pub/pub.go | 18 ++
banyand/queue/queue.go | 1 +
banyand/stream/metadata.go | 65 +++-
banyand/stream/metadata_test.go | 10 +-
banyand/stream/service.go | 12 +-
banyand/stream/stream.go | 34 +-
docs/api-reference.md | 3 +-
pkg/bus/bus.go | 5 +
pkg/iter/sort/sort.go | 133 ++++++++
pkg/iter/sort/sort_test.go | 102 ++++++
pkg/query/executor/interface.go | 22 ++
pkg/query/logical/common.go | 26 +-
pkg/query/logical/iter.go | 134 --------
pkg/query/logical/measure/measure_analyzer.go | 60 ++++
.../logical/measure/measure_plan_distributed.go | 234 ++++++++++++++
.../measure/measure_plan_indexscan_local.go | 9 +-
pkg/query/logical/stream/stream_analyzer.go | 16 +
.../logical/stream/stream_plan_distributed.go | 214 +++++++++++++
.../logical/stream/stream_plan_indexscan_local.go | 4 +-
pkg/run/channel_closer.go | 2 +-
pkg/schema/metadata.go | 350 +++++++++++----------
pkg/schema/schema.go | 96 ++++++
36 files changed, 1651 insertions(+), 415 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index 54f68ee0..0d98483c 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -14,6 +14,7 @@ Release Notes.
- Register the node role to the metadata registry.
- Implement the remote queue to spreading data to data nodes.
- Fix parse environment variables error
+- Implement the distributed query engine.
### Bugs
diff --git a/api/proto/banyandb/database/v1/database.proto b/api/proto/banyandb/database/v1/database.proto
index 23309d0a..8ae7ffc3 100644
--- a/api/proto/banyandb/database/v1/database.proto
+++ b/api/proto/banyandb/database/v1/database.proto
@@ -29,8 +29,7 @@ enum Role {
ROLE_UNSPECIFIED = 0;
ROLE_META = 1;
ROLE_DATA = 2;
- ROLE_QUERY = 3;
- ROLE_LIAISON = 4;
+ ROLE_LIAISON = 3;
}
message Node {
diff --git a/banyand/dquery/dquery.go b/banyand/dquery/dquery.go
index 4bb27520..61f490c8 100644
--- a/banyand/dquery/dquery.go
+++ b/banyand/dquery/dquery.go
@@ -17,3 +17,71 @@
// Package dquery implement the distributed query.
package dquery
+
+import (
+ "context"
+
+ modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
+ "github.com/apache/skywalking-banyandb/banyand/measure"
+ "github.com/apache/skywalking-banyandb/banyand/metadata"
+ "github.com/apache/skywalking-banyandb/banyand/stream"
+ "github.com/apache/skywalking-banyandb/pkg/bus"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ "github.com/apache/skywalking-banyandb/pkg/query/executor"
+ "github.com/apache/skywalking-banyandb/pkg/run"
+)
+
+const (
+ moduleName = "distributed-query"
+)
+
+type queryService struct {
+ log *logger.Logger
+ metaService metadata.Repo
+ sqp *streamQueryProcessor
+ mqp *measureQueryProcessor
+ tqp *topNQueryProcessor
+}
+
+// NewService return a new query service.
+func NewService(metaService metadata.Repo, broadcaster bus.Broadcaster,
+) (run.Unit, error) {
+ svc := &queryService{
+ metaService: metaService,
+ }
+ svc.sqp = &streamQueryProcessor{
+ queryService: svc,
+ broadcaster: broadcaster,
+ }
+ svc.mqp = &measureQueryProcessor{
+ queryService: svc,
+ broadcaster: broadcaster,
+ }
+ svc.tqp = &topNQueryProcessor{
+ queryService: svc,
+ broadcaster: broadcaster,
+ }
+ return svc, nil
+}
+
+func (q *queryService) Name() string {
+ return moduleName
+}
+
+func (q *queryService) PreRun(_ context.Context) error {
+ q.log = logger.GetLogger(moduleName)
+ q.sqp.streamService = stream.NewPortableRepository(q.metaService, q.log)
+ q.mqp.measureService = measure.NewPortableRepository(q.metaService, q.log)
+ return nil
+}
+
+var _ executor.DistributedExecutionContext = (*distributedContext)(nil)
+
+type distributedContext struct {
+ bus.Broadcaster
+ timeRange *modelv1.TimeRange
+}
+
+func (dc *distributedContext) TimeRange() *modelv1.TimeRange {
+ return dc.timeRange
+}
diff --git a/banyand/dquery/measure.go b/banyand/dquery/measure.go
new file mode 100644
index 00000000..522a9cda
--- /dev/null
+++ b/banyand/dquery/measure.go
@@ -0,0 +1,100 @@
+// 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 dquery
+
+import (
+ "context"
+ "time"
+
+ "github.com/apache/skywalking-banyandb/api/common"
+ measurev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/measure/v1"
+ "github.com/apache/skywalking-banyandb/banyand/measure"
+ "github.com/apache/skywalking-banyandb/pkg/bus"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ "github.com/apache/skywalking-banyandb/pkg/query/executor"
+ logical_measure "github.com/apache/skywalking-banyandb/pkg/query/logical/measure"
+)
+
+type measureQueryProcessor struct {
+ measureService measure.Query
+ broadcaster bus.Broadcaster
+ *queryService
+}
+
+func (p *measureQueryProcessor) Rev(message bus.Message) (resp bus.Message) {
+ queryCriteria, ok := message.Data().(*measurev1.QueryRequest)
+ now := time.Now().UnixNano()
+ if !ok {
+ resp = bus.NewMessage(bus.MessageID(now), common.NewError("invalid event data type"))
+ return
+ }
+ ml := p.log.Named("measure", queryCriteria.Metadata.Group, queryCriteria.Metadata.Name)
+ if e := ml.Debug(); e.Enabled() {
+ e.RawJSON("req", logger.Proto(queryCriteria)).Msg("received a query event")
+ }
+
+ meta := queryCriteria.GetMetadata()
+ ec, err := p.measureService.Measure(meta)
+ if err != nil {
+ resp = bus.NewMessage(bus.MessageID(now), common.NewError("fail to get execution context for measure %s: %v", meta.GetName(), err))
+ return
+ }
+
+ s, err := logical_measure.BuildSchema(ec.GetSchema(), ec.GetIndexRules())
+ if err != nil {
+ resp = bus.NewMessage(bus.MessageID(now), common.NewError("fail to build schema for measure %s: %v", meta.GetName(), err))
+ return
+ }
+
+ plan, err := logical_measure.DistributedAnalyze(queryCriteria, s)
+ if err != nil {
+ resp = bus.NewMessage(bus.MessageID(now), common.NewError("fail to analyze the query request for measure %s: %v", meta.GetName(), err))
+ return
+ }
+
+ if e := ml.Debug(); e.Enabled() {
+ e.Str("plan", plan.String()).Msg("query plan")
+ }
+
+ mIterator, err := plan.(executor.MeasureExecutable).Execute(executor.WithDistributedExecutionContext(context.Background(), &distributedContext{
+ Broadcaster: p.broadcaster,
+ timeRange: queryCriteria.TimeRange,
+ }))
+ if err != nil {
+ ml.Error().Err(err).RawJSON("req", logger.Proto(queryCriteria)).Msg("fail to close the query plan")
+ resp = bus.NewMessage(bus.MessageID(now), common.NewError("fail to execute the query plan for measure %s: %v", meta.GetName(), err))
+ return
+ }
+ defer func() {
+ if err = mIterator.Close(); err != nil {
+ ml.Error().Err(err).RawJSON("req", logger.Proto(queryCriteria)).Msg("fail to close the query plan")
+ }
+ }()
+ result := make([]*measurev1.DataPoint, 0)
+ for mIterator.Next() {
+ current := mIterator.Current()
+ if len(current) > 0 {
+ result = append(result, current[0])
+ }
+ }
+ if e := ml.Debug(); e.Enabled() {
+ e.RawJSON("ret", logger.Proto(&measurev1.QueryResponse{DataPoints: result})).Msg("got a measure")
+ }
+ resp = bus.NewMessage(bus.MessageID(now), result)
+ return
+}
diff --git a/banyand/dquery/stream.go b/banyand/dquery/stream.go
new file mode 100644
index 00000000..3a92aaeb
--- /dev/null
+++ b/banyand/dquery/stream.go
@@ -0,0 +1,84 @@
+// 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 dquery
+
+import (
+ "context"
+ "time"
+
+ "github.com/apache/skywalking-banyandb/api/common"
+ streamv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/stream/v1"
+ "github.com/apache/skywalking-banyandb/banyand/stream"
+ "github.com/apache/skywalking-banyandb/pkg/bus"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ "github.com/apache/skywalking-banyandb/pkg/query/executor"
+ logical_stream "github.com/apache/skywalking-banyandb/pkg/query/logical/stream"
+)
+
+type streamQueryProcessor struct {
+ streamService stream.Query
+ broadcaster bus.Broadcaster
+ *queryService
+}
+
+func (p *streamQueryProcessor) Rev(message bus.Message) (resp bus.Message) {
+ now := time.Now().UnixNano()
+ queryCriteria, ok := message.Data().(*streamv1.QueryRequest)
+ if !ok {
+ resp = bus.NewMessage(bus.MessageID(now), common.NewError("invalid event data type"))
+ return
+ }
+ if p.log.Debug().Enabled() {
+ p.log.Debug().RawJSON("criteria", logger.Proto(queryCriteria)).Msg("received a query request")
+ }
+
+ meta := queryCriteria.GetMetadata()
+ ec, err := p.streamService.Stream(meta)
+ if err != nil {
+ resp = bus.NewMessage(bus.MessageID(now), common.NewError("fail to get execution context for stream %s: %v", meta.GetName(), err))
+ return
+ }
+ s, err := logical_stream.BuildSchema(ec.GetSchema(), ec.GetIndexRules())
+ if err != nil {
+ resp = bus.NewMessage(bus.MessageID(now), common.NewError("fail to build schema for stream %s: %v", meta.GetName(), err))
+ return
+ }
+
+ plan, err := logical_stream.DistributedAnalyze(queryCriteria, s)
+ if err != nil {
+ resp = bus.NewMessage(bus.MessageID(now), common.NewError("fail to analyze the query request for stream %s: %v", meta.GetName(), err))
+ return
+ }
+
+ if p.log.Debug().Enabled() {
+ p.log.Debug().Str("plan", plan.String()).Msg("query plan")
+ }
+ entities, err := plan.(executor.StreamExecutable).Execute(executor.WithDistributedExecutionContext(context.Background(), &distributedContext{
+ Broadcaster: p.broadcaster,
+ timeRange: queryCriteria.TimeRange,
+ }))
+ if err != nil {
+ p.log.Error().Err(err).RawJSON("req", logger.Proto(queryCriteria)).Msg("fail to execute the query plan")
+ resp = bus.NewMessage(bus.MessageID(now), common.NewError("execute the query plan for stream %s: %v", meta.GetName(), err))
+ return
+ }
+
+ resp = bus.NewMessage(bus.MessageID(now), entities)
+
+ return
+}
diff --git a/banyand/dquery/topn.go b/banyand/dquery/topn.go
new file mode 100644
index 00000000..df34d337
--- /dev/null
+++ b/banyand/dquery/topn.go
@@ -0,0 +1,122 @@
+// 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 dquery
+
+import (
+ "go.uber.org/multierr"
+ "google.golang.org/protobuf/types/known/timestamppb"
+
+ "github.com/apache/skywalking-banyandb/api/common"
+ "github.com/apache/skywalking-banyandb/api/data"
+ measurev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/measure/v1"
+ modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
+ "github.com/apache/skywalking-banyandb/pkg/bus"
+ "github.com/apache/skywalking-banyandb/pkg/convert"
+ "github.com/apache/skywalking-banyandb/pkg/iter/sort"
+)
+
+type topNQueryProcessor struct {
+ broadcaster bus.Broadcaster
+ *queryService
+}
+
+func (t *topNQueryProcessor) Rev(message bus.Message) (resp bus.Message) {
+ request, ok := message.Data().(*measurev1.TopNRequest)
+ if !ok {
+ t.log.Warn().Msg("invalid event data type")
+ return
+ }
+ if request.GetFieldValueSort() == modelv1.Sort_SORT_UNSPECIFIED {
+ t.log.Warn().Msg("invalid requested sort direction")
+ return
+ }
+ if e := t.log.Debug(); e.Enabled() {
+ e.Stringer("req", request).Msg("received a topN query event")
+ }
+ now := bus.MessageID(request.TimeRange.Begin.Nanos)
+ ff, err := t.broadcaster.Broadcast(data.TopicTopNQuery, bus.NewMessage(now, request))
+ if err != nil {
+ resp = bus.NewMessage(now, common.NewError("execute the query %s: %v", request.Metadata.GetName(), err))
+ return
+ }
+ var allErr error
+ var sii []sort.Iterator[*comparableTopNItem]
+ var latestTimestamp *timestamppb.Timestamp
+ for _, f := range ff {
+ if m, getErr := f.Get(); getErr != nil {
+ allErr = multierr.Append(allErr, getErr)
+ } else {
+ tl := m.Data().(*measurev1.TopNList)
+ un := tl.Timestamp.AsTime().UnixNano()
+ if un > latestTimestamp.AsTime().UnixNano() {
+ latestTimestamp = tl.Timestamp
+ }
+ sii = append(sii, &sortedTopNList{TopNList: tl})
+ }
+ }
+ var desc bool
+ if request.GetFieldValueSort() == modelv1.Sort_SORT_DESC {
+ desc = true
+ }
+ iter := sort.NewItemIter[*comparableTopNItem](sii, desc)
+ defer func() {
+ _ = iter.Close()
+ }()
+ var items []*measurev1.TopNList_Item
+ for iter.Next() && len(items) < int(request.TopN) {
+ items = append(items, iter.Val().TopNList_Item)
+ }
+ resp = bus.NewMessage(now, &measurev1.TopNList{
+ Items: items,
+ Timestamp: latestTimestamp,
+ })
+ return
+}
+
+var _ sort.Comparable = (*comparableTopNItem)(nil)
+
+type comparableTopNItem struct {
+ *measurev1.TopNList_Item
+}
+
+func (c *comparableTopNItem) SortedField() []byte {
+ return convert.Int64ToBytes(c.Value.GetInt().Value)
+}
+
+var _ sort.Iterator[*comparableTopNItem] = (*sortedTopNList)(nil)
+
+type sortedTopNList struct {
+ *measurev1.TopNList
+ index int
+}
+
+func (*sortedTopNList) Close() error {
+ return nil
+}
+
+func (s *sortedTopNList) Next() bool {
+ if s.index >= len(s.Items) {
+ return false
+ }
+ s.index++
+ return s.index < len(s.Items)
+}
+
+func (s *sortedTopNList) Val() *comparableTopNItem {
+ return &comparableTopNItem{s.Items[s.index-1]}
+}
diff --git a/banyand/internal/cmd/liaison.go b/banyand/internal/cmd/liaison.go
index 0b842201..d09428f0 100644
--- a/banyand/internal/cmd/liaison.go
+++ b/banyand/internal/cmd/liaison.go
@@ -24,6 +24,7 @@ import (
"github.com/spf13/cobra"
"github.com/apache/skywalking-banyandb/api/common"
+ "github.com/apache/skywalking-banyandb/banyand/dquery"
"github.com/apache/skywalking-banyandb/banyand/liaison/grpc"
"github.com/apache/skywalking-banyandb/banyand/liaison/http"
"github.com/apache/skywalking-banyandb/banyand/metadata"
@@ -49,10 +50,15 @@ func newLiaisonCmd() *cobra.Command {
profSvc := observability.NewProfService()
metricSvc := observability.NewMetricService()
httpServer := http.NewServer()
+ dQuery, err := dquery.NewService(metaSvc, pipeline)
+ if err != nil {
+ l.Fatal().Err(err).Msg("failed to initiate distributed query service")
+ }
units := []run.Unit{
new(signal.Handler),
pipeline,
+ dQuery,
grpcServer,
httpServer,
profSvc,
diff --git a/banyand/measure/measure.go b/banyand/measure/measure.go
index 9fbfe653..b8a55140 100644
--- a/banyand/measure/measure.go
+++ b/banyand/measure/measure.go
@@ -22,7 +22,6 @@ package measure
import (
"context"
- "math"
"time"
commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
@@ -32,7 +31,6 @@ import (
"github.com/apache/skywalking-banyandb/banyand/tsdb/index"
"github.com/apache/skywalking-banyandb/pkg/logger"
"github.com/apache/skywalking-banyandb/pkg/query/logical"
- resourceSchema "github.com/apache/skywalking-banyandb/pkg/schema"
"github.com/apache/skywalking-banyandb/pkg/timestamp"
)
@@ -43,18 +41,17 @@ const (
)
type measure struct {
- databaseSupplier tsdb.Supplier
- l *logger.Logger
- schema *databasev1.Measure
- indexWriter *index.Writer
- processorManager *topNProcessorManager
- name string
- group string
- indexRules []*databasev1.IndexRule
- topNAggregations []*databasev1.TopNAggregation
- maxObservedModRevision int64
- interval time.Duration
- shardNum uint32
+ databaseSupplier tsdb.Supplier
+ l *logger.Logger
+ schema *databasev1.Measure
+ indexWriter *index.Writer
+ processorManager *topNProcessorManager
+ name string
+ group string
+ indexRules []*databasev1.IndexRule
+ topNAggregations []*databasev1.TopNAggregation
+ interval time.Duration
+ shardNum uint32
}
func (s *measure) startSteamingManager(pipeline queue.Queue) error {
@@ -80,22 +77,10 @@ func (s *measure) GetSchema() *databasev1.Measure {
return s.schema
}
-func (s *measure) GetMetadata() *commonv1.Metadata {
- return s.schema.Metadata
-}
-
func (s *measure) GetIndexRules() []*databasev1.IndexRule {
return s.indexRules
}
-func (s *measure) GetTopN() []*databasev1.TopNAggregation {
- return s.topNAggregations
-}
-
-func (s *measure) MaxObservedModRevision() int64 {
- return s.maxObservedModRevision
-}
-
func (s *measure) Close() error {
if s.processorManager == nil {
return nil
@@ -105,7 +90,6 @@ func (s *measure) Close() error {
func (s *measure) parseSpec() (err error) {
s.name, s.group = s.schema.GetMetadata().GetName(), s.schema.GetMetadata().GetGroup()
- s.maxObservedModRevision = int64(math.Max(float64(resourceSchema.ParseMaxModRevision(s.indexRules)), float64(resourceSchema.ParseMaxModRevision(s.topNAggregations))))
if s.schema.Interval != "" {
s.interval, err = timestamp.ParseDuration(s.schema.Interval)
}
diff --git a/banyand/measure/measure_write.go b/banyand/measure/measure_write.go
index 2b5f561a..3235b5c6 100644
--- a/banyand/measure/measure_write.go
+++ b/banyand/measure/measure_write.go
@@ -134,7 +134,7 @@ func (s *measure) write(shardID common.ShardID, entity []byte, entityValues tsdb
if s.processorManager != nil {
s.processorManager.onMeasureWrite(&measurev1.InternalWriteRequest{
Request: &measurev1.WriteRequest{
- Metadata: s.GetMetadata(),
+ Metadata: s.GetSchema().Metadata,
DataPoint: value,
},
EntityValues: entityValues[1:],
diff --git a/banyand/measure/metadata.go b/banyand/measure/metadata.go
index 9df34dfa..1185442b 100644
--- a/banyand/measure/metadata.go
+++ b/banyand/measure/metadata.go
@@ -20,6 +20,7 @@ package measure
import (
"context"
"fmt"
+ "io"
"path"
"time"
@@ -50,7 +51,7 @@ type schemaRepo struct {
func newSchemaRepo(path string, metadata metadata.Repo,
dbOpts tsdb.DatabaseOpts, l *logger.Logger, pipeline queue.Queue, encoderBufferSize, bufferSize int64,
) schemaRepo {
- return schemaRepo{
+ sr := schemaRepo{
l: l,
metadata: metadata,
Repository: resourceSchema.NewRepository(
@@ -59,6 +60,38 @@ func newSchemaRepo(path string, metadata metadata.Repo,
newSupplier(path, metadata, dbOpts, l, pipeline, encoderBufferSize, bufferSize),
),
}
+ sr.start()
+ return sr
+}
+
+// NewPortableRepository creates a new portable repository.
+func NewPortableRepository(metadata metadata.Repo, l *logger.Logger) Query {
+ r := &schemaRepo{
+ l: l,
+ metadata: metadata,
+ Repository: resourceSchema.NewPortableRepository(
+ metadata,
+ l,
+ newPortableSupplier(metadata, l),
+ ),
+ }
+ r.start()
+ return r
+}
+
+func (sr *schemaRepo) start() {
+ sr.Watcher()
+ sr.metadata.
+ RegisterHandler("measure", schema.KindGroup|schema.KindMeasure|schema.KindIndexRuleBinding|schema.KindIndexRule|schema.KindTopNAggregation,
+ sr)
+}
+
+func (sr *schemaRepo) Measure(metadata *commonv1.Metadata) (Measure, error) {
+ sm, ok := sr.loadMeasure(metadata)
+ if !ok {
+ return nil, errors.WithStack(ErrMeasureNotExist)
+ }
+ return sm, nil
}
func (sr *schemaRepo) OnAddOrUpdate(metadata schema.Metadata) {
@@ -263,7 +296,7 @@ func (sr *schemaRepo) loadMeasure(metadata *commonv1.Metadata) (*measure, bool)
if !ok {
return nil, false
}
- s, ok := r.(*measure)
+ s, ok := r.Delegated().(*measure)
return s, ok
}
@@ -292,12 +325,12 @@ func newSupplier(path string, metadata metadata.Repo, dbOpts tsdb.DatabaseOpts,
}
}
-func (s *supplier) OpenResource(shardNum uint32, db tsdb.Supplier, spec resourceSchema.ResourceSpec) (resourceSchema.Resource, error) {
- measureSchema := spec.Schema.(*databasev1.Measure)
+func (s *supplier) OpenResource(shardNum uint32, db tsdb.Supplier, spec resourceSchema.Resource) (io.Closer, error) {
+ measureSchema := spec.Schema().(*databasev1.Measure)
return openMeasure(shardNum, db, measureSpec{
schema: measureSchema,
- indexRules: spec.IndexRules,
- topNAggregations: spec.Aggregations,
+ indexRules: spec.IndexRules(),
+ topNAggregations: spec.TopN(),
}, s.l, s.pipeline)
}
@@ -349,3 +382,21 @@ func intervalFn(key []byte) time.Duration {
}
return interval
}
+
+type portableSupplier struct {
+ metadata metadata.Repo
+ l *logger.Logger
+}
+
+func newPortableSupplier(metadata metadata.Repo, l *logger.Logger) *portableSupplier {
+ return &portableSupplier{
+ metadata: metadata,
+ l: l,
+ }
+}
+
+func (s *portableSupplier) ResourceSchema(md *commonv1.Metadata) (resourceSchema.ResourceSchema, error) {
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+ defer cancel()
+ return s.metadata.MeasureRegistry().GetMeasure(ctx, md)
+}
diff --git a/banyand/measure/metadata_test.go b/banyand/measure/metadata_test.go
index 9504c1f9..f6cda0f1 100644
--- a/banyand/measure/metadata_test.go
+++ b/banyand/measure/metadata_test.go
@@ -37,6 +37,10 @@ var _ = Describe("Metadata", func() {
BeforeEach(func() {
svcs, deferFn = setUp()
goods = gleak.Goroutines()
+ Eventually(func() bool {
+ _, ok := svcs.measure.LoadGroup("sw_metric")
+ return ok
+ }).WithTimeout(flags.EventuallyTimeout).Should(BeTrue())
})
AfterEach(func() {
@@ -45,12 +49,6 @@ var _ = Describe("Metadata", func() {
})
Context("Manage group", func() {
- It("should pass smoke test", func() {
- Eventually(func() bool {
- _, ok := svcs.measure.LoadGroup("sw_metric")
- return ok
- }).WithTimeout(flags.EventuallyTimeout).Should(BeTrue())
- })
It("should close the group", func() {
deleted, err := svcs.metadataService.GroupRegistry().DeleteGroup(context.TODO(), "sw_metric")
Expect(err).ShouldNot(HaveOccurred())
diff --git a/banyand/measure/service.go b/banyand/measure/service.go
index 47c93684..43940d8f 100644
--- a/banyand/measure/service.go
+++ b/banyand/measure/service.go
@@ -27,7 +27,6 @@ import (
commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
"github.com/apache/skywalking-banyandb/banyand/metadata"
- "github.com/apache/skywalking-banyandb/banyand/metadata/schema"
"github.com/apache/skywalking-banyandb/banyand/observability"
"github.com/apache/skywalking-banyandb/banyand/queue"
"github.com/apache/skywalking-banyandb/banyand/tsdb"
@@ -114,10 +113,7 @@ func (s *service) PreRun(_ context.Context) error {
s.schemaRepo = newSchemaRepo(path, s.metadata, s.dbOpts,
s.l, s.localPipeline, int64(s.BlockEncoderBufferSize), int64(s.BlockBufferSize))
// run a serial watcher
- go s.schemaRepo.Watcher()
- s.metadata.
- RegisterHandler("measure", schema.KindGroup|schema.KindMeasure|schema.KindIndexRuleBinding|schema.KindIndexRule|schema.KindTopNAggregation,
- &s.schemaRepo)
+
s.writeListener = setUpWriteCallback(s.l, &s.schemaRepo)
err := s.pipeline.Subscribe(data.TopicMeasureWrite, s.writeListener)
if err != nil {
diff --git a/banyand/query/processor.go b/banyand/query/processor.go
index ee8a6bf5..a47e6df8 100644
--- a/banyand/query/processor.go
+++ b/banyand/query/processor.go
@@ -25,7 +25,6 @@ import (
"github.com/apache/skywalking-banyandb/api/common"
"github.com/apache/skywalking-banyandb/api/data"
- databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
measurev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/measure/v1"
streamv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/stream/v1"
"github.com/apache/skywalking-banyandb/banyand/measure"
@@ -179,10 +178,6 @@ func (q *queryService) Name() string {
return moduleName
}
-func (q *queryService) Role() databasev1.Role {
- return databasev1.Role_ROLE_QUERY
-}
-
func (q *queryService) PreRun(_ context.Context) error {
q.log = logger.GetLogger(moduleName)
return multierr.Combine(
diff --git a/banyand/queue/local.go b/banyand/queue/local.go
index 659e908e..09ca07aa 100644
--- a/banyand/queue/local.go
+++ b/banyand/queue/local.go
@@ -62,6 +62,14 @@ func (l *local) Publish(topic bus.Topic, message ...bus.Message) (bus.Future, er
return l.local.Publish(topic, message...)
}
+func (l *local) Broadcast(topic bus.Topic, message bus.Message) ([]bus.Future, error) {
+ f, err := l.Publish(topic, message)
+ if err != nil {
+ return nil, err
+ }
+ return []bus.Future{f}, nil
+}
+
func (l local) Name() string {
return "local-pipeline"
}
diff --git a/banyand/queue/pub/pub.go b/banyand/queue/pub/pub.go
index 1a953857..b72e9a07 100644
--- a/banyand/queue/pub/pub.go
+++ b/banyand/queue/pub/pub.go
@@ -68,6 +68,24 @@ func (p *pub) Serve() run.StopNotify {
return p.closer.CloseNotify()
}
+func (p *pub) Broadcast(topic bus.Topic, messages bus.Message) ([]bus.Future, error) {
+ var names []string
+ p.mu.RLock()
+ for k := range p.clients {
+ names = append(names, k)
+ }
+ p.mu.RUnlock()
+ var futures []bus.Future
+ for _, n := range names {
+ f, err := p.Publish(topic, bus.NewMessageWithNode(messages.ID(), n, messages.Data()))
+ if err != nil {
+ return nil, err
+ }
+ futures = append(futures, f)
+ }
+ return futures, nil
+}
+
func (p *pub) Publish(topic bus.Topic, messages ...bus.Message) (bus.Future, error) {
var err error
f := &future{}
diff --git a/banyand/queue/queue.go b/banyand/queue/queue.go
index d899162e..bada6711 100644
--- a/banyand/queue/queue.go
+++ b/banyand/queue/queue.go
@@ -37,6 +37,7 @@ type Queue interface {
type Client interface {
run.Unit
bus.Publisher
+ bus.Broadcaster
NewBatchPublisher() BatchPublisher
}
diff --git a/banyand/stream/metadata.go b/banyand/stream/metadata.go
index e349872e..7a9b34df 100644
--- a/banyand/stream/metadata.go
+++ b/banyand/stream/metadata.go
@@ -19,9 +19,12 @@ package stream
import (
"context"
+ "io"
"path"
"time"
+ "github.com/pkg/errors"
+
"github.com/apache/skywalking-banyandb/api/common"
commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
@@ -33,6 +36,8 @@ import (
resourceSchema "github.com/apache/skywalking-banyandb/pkg/schema"
)
+var _ Query = (*schemaRepo)(nil)
+
type schemaRepo struct {
resourceSchema.Repository
l *logger.Logger
@@ -42,7 +47,7 @@ type schemaRepo struct {
func newSchemaRepo(path string, metadata metadata.Repo,
bufferSize int64, dbOpts tsdb.DatabaseOpts, l *logger.Logger,
) schemaRepo {
- return schemaRepo{
+ r := schemaRepo{
l: l,
metadata: metadata,
Repository: resourceSchema.NewRepository(
@@ -51,6 +56,38 @@ func newSchemaRepo(path string, metadata metadata.Repo,
newSupplier(path, metadata, bufferSize, dbOpts, l),
),
}
+ r.start()
+ return r
+}
+
+// NewPortableRepository creates a new portable repository.
+func NewPortableRepository(metadata metadata.Repo, l *logger.Logger) Query {
+ r := &schemaRepo{
+ l: l,
+ metadata: metadata,
+ Repository: resourceSchema.NewPortableRepository(
+ metadata,
+ l,
+ newPortableSupplier(metadata, l),
+ ),
+ }
+ r.start()
+ return r
+}
+
+func (sr *schemaRepo) start() {
+ sr.Watcher()
+ sr.metadata.RegisterHandler("stream",
+ schema.KindGroup|schema.KindStream|schema.KindIndexRuleBinding|schema.KindIndexRule,
+ sr)
+}
+
+func (sr *schemaRepo) Stream(metadata *commonv1.Metadata) (Stream, error) {
+ sm, ok := sr.loadStream(metadata)
+ if !ok {
+ return nil, errors.WithStack(errStreamNotExist)
+ }
+ return sm, nil
}
func (sr *schemaRepo) OnAddOrUpdate(m schema.Metadata) {
@@ -156,7 +193,7 @@ func (sr *schemaRepo) loadStream(metadata *commonv1.Metadata) (*stream, bool) {
if !ok {
return nil, false
}
- s, ok := r.(*stream)
+ s, ok := r.Delegated().(*stream)
return s, ok
}
@@ -180,11 +217,11 @@ func newSupplier(path string, metadata metadata.Repo, bufferSize int64, dbOpts t
}
}
-func (s *supplier) OpenResource(shardNum uint32, db tsdb.Supplier, spec resourceSchema.ResourceSpec) (resourceSchema.Resource, error) {
- streamSchema := spec.Schema.(*databasev1.Stream)
+func (s *supplier) OpenResource(shardNum uint32, db tsdb.Supplier, resource resourceSchema.Resource) (io.Closer, error) {
+ streamSchema := resource.Schema().(*databasev1.Stream)
return openStream(shardNum, db, streamSpec{
schema: streamSchema,
- indexRules: spec.IndexRules,
+ indexRules: resource.IndexRules(),
}, s.l), nil
}
@@ -222,3 +259,21 @@ func (s *supplier) OpenDB(groupSchema *commonv1.Group) (tsdb.Database, error) {
}),
opts)
}
+
+type portableSupplier struct {
+ metadata metadata.Repo
+ l *logger.Logger
+}
+
+func newPortableSupplier(metadata metadata.Repo, l *logger.Logger) *portableSupplier {
+ return &portableSupplier{
+ metadata: metadata,
+ l: l,
+ }
+}
+
+func (s *portableSupplier) ResourceSchema(md *commonv1.Metadata) (resourceSchema.ResourceSchema, error) {
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+ defer cancel()
+ return s.metadata.StreamRegistry().GetStream(ctx, md)
+}
diff --git a/banyand/stream/metadata_test.go b/banyand/stream/metadata_test.go
index dc184d40..8c4f3bfb 100644
--- a/banyand/stream/metadata_test.go
+++ b/banyand/stream/metadata_test.go
@@ -37,6 +37,10 @@ var _ = Describe("Metadata", func() {
BeforeEach(func() {
goods = gleak.Goroutines()
svcs, deferFn = setUp()
+ Eventually(func() bool {
+ _, ok := svcs.stream.schemaRepo.LoadGroup("default")
+ return ok
+ }).WithTimeout(flags.EventuallyTimeout).Should(BeTrue())
})
AfterEach(func() {
@@ -45,12 +49,6 @@ var _ = Describe("Metadata", func() {
})
Context("Manage group", func() {
- It("should pass smoke test", func() {
- Eventually(func() bool {
- _, ok := svcs.stream.schemaRepo.LoadGroup("default")
- return ok
- }).WithTimeout(flags.EventuallyTimeout).Should(BeTrue())
- })
It("should close the group", func() {
deleted, err := svcs.metadataService.GroupRegistry().DeleteGroup(context.TODO(), "default")
Expect(err).ShouldNot(HaveOccurred())
diff --git a/banyand/stream/service.go b/banyand/stream/service.go
index 3af678d0..4d888cfb 100644
--- a/banyand/stream/service.go
+++ b/banyand/stream/service.go
@@ -27,7 +27,6 @@ import (
commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
"github.com/apache/skywalking-banyandb/banyand/metadata"
- "github.com/apache/skywalking-banyandb/banyand/metadata/schema"
"github.com/apache/skywalking-banyandb/banyand/observability"
"github.com/apache/skywalking-banyandb/banyand/queue"
"github.com/apache/skywalking-banyandb/banyand/tsdb"
@@ -62,11 +61,7 @@ type service struct {
}
func (s *service) Stream(metadata *commonv1.Metadata) (Stream, error) {
- sm, ok := s.schemaRepo.loadStream(metadata)
- if !ok {
- return nil, errors.WithStack(errStreamNotExist)
- }
- return sm, nil
+ return s.schemaRepo.Stream(metadata)
}
func (s *service) FlagSet() *run.FlagSet {
@@ -102,11 +97,6 @@ func (s *service) PreRun(_ context.Context) error {
path := path.Join(s.root, s.Name())
observability.UpdatePath(path)
s.schemaRepo = newSchemaRepo(path, s.metadata, int64(s.blockBufferSize), s.dbOpts, s.l)
- // run a serial watcher
- s.schemaRepo.Watcher()
- s.metadata.RegisterHandler("stream", schema.KindGroup|schema.KindStream|schema.KindIndexRuleBinding|schema.KindIndexRule,
- &s.schemaRepo)
-
s.writeListener = setUpWriteCallback(s.l, &s.schemaRepo)
errWrite := s.pipeline.Subscribe(data.TopicStreamWrite, s.writeListener)
diff --git a/banyand/stream/stream.go b/banyand/stream/stream.go
index fb60bd64..a989ebb2 100644
--- a/banyand/stream/stream.go
+++ b/banyand/stream/stream.go
@@ -22,33 +22,24 @@ package stream
import (
"context"
- commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
"github.com/apache/skywalking-banyandb/banyand/tsdb"
"github.com/apache/skywalking-banyandb/banyand/tsdb/index"
"github.com/apache/skywalking-banyandb/pkg/logger"
- "github.com/apache/skywalking-banyandb/pkg/schema"
)
// a chunk is 1MB.
const chunkSize = 1 << 20
-var _ schema.Resource = (*stream)(nil)
-
type stream struct {
- db tsdb.Supplier
- l *logger.Logger
- schema *databasev1.Stream
- indexWriter *index.Writer
- name string
- group string
- indexRules []*databasev1.IndexRule
- maxObservedModRevision int64
- shardNum uint32
-}
-
-func (s *stream) GetMetadata() *commonv1.Metadata {
- return s.schema.Metadata
+ db tsdb.Supplier
+ l *logger.Logger
+ schema *databasev1.Stream
+ indexWriter *index.Writer
+ name string
+ group string
+ indexRules []*databasev1.IndexRule
+ shardNum uint32
}
func (s *stream) GetSchema() *databasev1.Stream {
@@ -59,21 +50,12 @@ func (s *stream) GetIndexRules() []*databasev1.IndexRule {
return s.indexRules
}
-func (s *stream) MaxObservedModRevision() int64 {
- return s.maxObservedModRevision
-}
-
-func (s *stream) GetTopN() []*databasev1.TopNAggregation {
- return nil
-}
-
func (s *stream) Close() error {
return nil
}
func (s *stream) parseSpec() {
s.name, s.group = s.schema.GetMetadata().GetName(), s.schema.GetMetadata().GetGroup()
- s.maxObservedModRevision = schema.ParseMaxModRevision(s.indexRules)
}
type streamSpec struct {
diff --git a/docs/api-reference.md b/docs/api-reference.md
index 27a53eec..96379185 100644
--- a/docs/api-reference.md
+++ b/docs/api-reference.md
@@ -445,8 +445,7 @@ Metadata is for multi-tenant, multi-model use
| ROLE_UNSPECIFIED | 0 | |
| ROLE_META | 1 | |
| ROLE_DATA | 2 | |
-| ROLE_QUERY | 3 | |
-| ROLE_LIAISON | 4 | |
+| ROLE_LIAISON | 3 | |
diff --git a/pkg/bus/bus.go b/pkg/bus/bus.go
index c7a8247b..6ad65ada 100644
--- a/pkg/bus/bus.go
+++ b/pkg/bus/bus.go
@@ -94,6 +94,11 @@ type Publisher interface {
Publish(topic Topic, message ...Message) (Future, error)
}
+// Broadcaster allow sending Messages to a Topic and receiving the responses.
+type Broadcaster interface {
+ Broadcast(topic Topic, message Message) ([]Future, error)
+}
+
type channel chan event
type chType int
diff --git a/pkg/iter/sort/sort.go b/pkg/iter/sort/sort.go
new file mode 100644
index 00000000..b0d67eb5
--- /dev/null
+++ b/pkg/iter/sort/sort.go
@@ -0,0 +1,133 @@
+// 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 sort provides a generic iterator that merges multiple sorted iterators.
+package sort
+
+import (
+ "bytes"
+ "container/heap"
+ "fmt"
+)
+
+// Comparable is an interface that allows sorting of items.
+type Comparable interface {
+ SortedField() []byte
+}
+
+// Iterator is a stream of items of Comparable type.
+type Iterator[T Comparable] interface {
+ Next() bool
+ Val() T
+ Close() error
+}
+
+type container[T Comparable] struct {
+ item T
+ iter Iterator[T]
+}
+
+type containerHeap[T Comparable] struct {
+ items []*container[T]
+ desc bool
+}
+
+func (h containerHeap[T]) Len() int {
+ return len(h.items)
+}
+
+func (h containerHeap[T]) Less(i, j int) bool {
+ if h.desc {
+ return bytes.Compare(h.items[i].item.SortedField(), h.items[j].item.SortedField()) > 0
+ }
+ return bytes.Compare(h.items[i].item.SortedField(), h.items[j].item.SortedField()) < 0
+}
+
+func (h containerHeap[T]) Swap(i, j int) {
+ h.items[i], h.items[j] = h.items[j], h.items[i]
+}
+
+func (h *containerHeap[T]) Push(x interface{}) {
+ item := x.(*container[T])
+ h.items = append(h.items, item)
+}
+
+func (h *containerHeap[T]) Pop() interface{} {
+ old := h.items
+ n := len(old)
+ item := old[n-1]
+ h.items = old[0 : n-1]
+ return item
+}
+
+type itemIter[T Comparable] struct {
+ curr T
+ h *containerHeap[T]
+ iters []Iterator[T]
+}
+
+// NewItemIter returns a new iterator that merges multiple sorted iterators.
+func NewItemIter[T Comparable](iters []Iterator[T], desc bool) Iterator[T] {
+ var def T
+ it := &itemIter[T]{
+ iters: iters,
+ h: &containerHeap[T]{items: make([]*container[T], 0), desc: desc},
+ curr: def,
+ }
+ it.initialize()
+ return it
+}
+
+func (it *itemIter[T]) initialize() {
+ heap.Init(it.h)
+ for _, iter := range it.iters {
+ if iter.Next() {
+ heap.Push(it.h, &container[T]{item: iter.Val(), iter: iter})
+ }
+ }
+}
+
+func (it *itemIter[T]) pushIterator(iter Iterator[T]) {
+ if iter.Next() {
+ heap.Push(it.h, &container[T]{item: iter.Val(), iter: iter})
+ }
+}
+
+func (it *itemIter[T]) Next() bool {
+ if it.h.Len() == 0 {
+ return false
+ }
+ top := heap.Pop(it.h).(*container[T])
+ it.curr = top.item
+ it.pushIterator(top.iter)
+ return true
+}
+
+func (it *itemIter[T]) Val() T {
+ return it.curr
+}
+
+func (it *itemIter[T]) Close() error {
+ var err error
+ for _, iter := range it.iters {
+ if e := iter.Close(); e != nil {
+ fmt.Println("Error closing iterator:", e)
+ err = e
+ }
+ }
+ return err
+}
diff --git a/pkg/iter/sort/sort_test.go b/pkg/iter/sort/sort_test.go
new file mode 100644
index 00000000..e91a0f96
--- /dev/null
+++ b/pkg/iter/sort/sort_test.go
@@ -0,0 +1,102 @@
+// 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 sort_test
+
+import (
+ "encoding/binary"
+ "testing"
+
+ "github.com/apache/skywalking-banyandb/pkg/iter/sort"
+)
+
+type Int int
+
+func (i Int) SortedField() []byte {
+ b := make([]byte, 8)
+ binary.BigEndian.PutUint64(b, uint64(i))
+ return b
+}
+
+// MockIterator is a mock for the Iterator interface.
+type MockIterator struct {
+ items []Int
+ index int
+}
+
+func NewMockIterator(items []Int) *MockIterator {
+ return &MockIterator{items: items, index: -1}
+}
+
+func (mi *MockIterator) Next() bool {
+ mi.index++
+ return mi.index < len(mi.items)
+}
+
+func (mi *MockIterator) Val() Int {
+ return mi.items[mi.index]
+}
+
+func (mi *MockIterator) Close() error {
+ return nil
+}
+
+func TestItemIter_Ascending(t *testing.T) {
+ iters := []sort.Iterator[Int]{
+ NewMockIterator([]Int{1, 3, 5}),
+ NewMockIterator([]Int{2, 4, 6}),
+ }
+
+ iter := sort.NewItemIter(iters, false) // Ascending order
+ for i := Int(1); i <= 6; i++ {
+ if !iter.Next() {
+ t.Fatalf("expected Next() to be true, got false")
+ }
+ if val := iter.Val(); val != i {
+ t.Errorf("expected Val() to be %d, got %d", i, val)
+ }
+ }
+ if iter.Next() {
+ t.Errorf("expected Next() to be false, got true")
+ }
+ if err := iter.Close(); err != nil {
+ t.Errorf("expected Close() to return nil, got error: %v", err)
+ }
+}
+
+func TestItemIter_Descending(t *testing.T) {
+ iters := []sort.Iterator[Int]{
+ NewMockIterator([]Int{5, 3, 1}),
+ NewMockIterator([]Int{6, 4, 2}),
+ }
+
+ iter := sort.NewItemIter(iters, true) // Descending order
+ for i := Int(6); i >= 1; i-- {
+ if !iter.Next() {
+ t.Fatalf("expected Next() to be true, got false")
+ }
+ if val := iter.Val(); val != i {
+ t.Errorf("expected Val() to be %d, got %d", i, val)
+ }
+ }
+ if iter.Next() {
+ t.Errorf("expected Next() to be false, got true")
+ }
+ if err := iter.Close(); err != nil {
+ t.Errorf("expected Close() to return nil, got error: %v", err)
+ }
+}
diff --git a/pkg/query/executor/interface.go b/pkg/query/executor/interface.go
index 8f4d972b..a206293b 100644
--- a/pkg/query/executor/interface.go
+++ b/pkg/query/executor/interface.go
@@ -26,6 +26,7 @@ import (
modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
streamv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/stream/v1"
"github.com/apache/skywalking-banyandb/banyand/tsdb"
+ "github.com/apache/skywalking-banyandb/pkg/bus"
)
// ExecutionContext allows retrieving data from tsdb.
@@ -95,3 +96,24 @@ type MIterator interface {
type MeasureExecutable interface {
Execute(context.Context) (MIterator, error)
}
+
+// DistributedExecutionContext allows retrieving data through the distributed module.
+type DistributedExecutionContext interface {
+ bus.Broadcaster
+ TimeRange() *modelv1.TimeRange
+}
+
+// DistributedExecutionContextKey is the key of distributed execution context in context.Context.
+type DistributedExecutionContextKey struct{}
+
+var distributedExecutionContextKeyInstance = DistributedExecutionContextKey{}
+
+// WithDistributedExecutionContext returns a new context with distributed execution context.
+func WithDistributedExecutionContext(ctx context.Context, ec DistributedExecutionContext) context.Context {
+ return context.WithValue(ctx, distributedExecutionContextKeyInstance, ec)
+}
+
+// FromDistributedExecutionContext returns the distributed execution context from context.Context.
+func FromDistributedExecutionContext(ctx context.Context) DistributedExecutionContext {
+ return ctx.Value(distributedExecutionContextKeyInstance).(DistributedExecutionContext)
+}
diff --git a/pkg/query/logical/common.go b/pkg/query/logical/common.go
index d1de11ff..4297c296 100644
--- a/pkg/query/logical/common.go
+++ b/pkg/query/logical/common.go
@@ -18,7 +18,6 @@
package logical
import (
- "bytes"
"context"
"io"
"time"
@@ -28,6 +27,7 @@ import (
modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
"github.com/apache/skywalking-banyandb/banyand/tsdb"
+ "github.com/apache/skywalking-banyandb/pkg/iter/sort"
"github.com/apache/skywalking-banyandb/pkg/logger"
"github.com/apache/skywalking-banyandb/pkg/query/executor"
"github.com/apache/skywalking-banyandb/pkg/timestamp"
@@ -48,20 +48,8 @@ type (
// SeekerBuilder wraps the execution of tsdb.SeekerBuilder.
// TODO:// we could have a chance to remove this wrapper.
SeekerBuilder func(builder tsdb.SeekerBuilder)
-
- comparator func(a, b tsdb.Item) bool
)
-func createComparator(sortDirection modelv1.Sort) comparator {
- return func(a, b tsdb.Item) bool {
- comp := bytes.Compare(a.SortedField(), b.SortedField())
- if sortDirection == modelv1.Sort_SORT_DESC {
- return comp == 1
- }
- return comp == -1
- }
-}
-
// ProjectItem parses the item within the StreamExecutionContext.
// projectionFieldRefs must be prepared before calling this method, projectionFieldRefs should be a list of
// tag list where the inner list must exist in the same tag family.
@@ -222,3 +210,15 @@ func StringSlicesEqual(a, b []string) bool {
}
return true
}
+
+// NewItemIter returns a ItemIterator which mergers several tsdb.Iterator by input sorting order.
+func NewItemIter(iters []tsdb.Iterator, s modelv1.Sort) sort.Iterator[tsdb.Item] {
+ var ii []sort.Iterator[tsdb.Item]
+ for _, iter := range iters {
+ ii = append(ii, iter)
+ }
+ if s == modelv1.Sort_SORT_DESC {
+ return sort.NewItemIter[tsdb.Item](ii, true)
+ }
+ return sort.NewItemIter[tsdb.Item](ii, false)
+}
diff --git a/pkg/query/logical/iter.go b/pkg/query/logical/iter.go
deleted file mode 100644
index e522969a..00000000
--- a/pkg/query/logical/iter.go
+++ /dev/null
@@ -1,134 +0,0 @@
-// 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 logical
-
-import (
- "container/heap"
- "io"
-
- "go.uber.org/multierr"
-
- modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
- "github.com/apache/skywalking-banyandb/banyand/tsdb"
-)
-
-var _ ItemIterator = (*itemIter)(nil)
-
-// ItemIterator allow iterating over a tsdb's series.
-type ItemIterator interface {
- HasNext() bool
- Next() tsdb.Item
- io.Closer
-}
-
-var _ heap.Interface = (*containerHeap)(nil)
-
-// container contains both iter and its current item.
-type container struct {
- c comparator
- item tsdb.Item
- iter tsdb.Iterator
-}
-
-type containerHeap []*container
-
-func (h containerHeap) Len() int { return len(h) }
-func (h containerHeap) Less(i, j int) bool { return h[i].c(h[i].item, h[j].item) }
-func (h containerHeap) Swap(i, j int) { h[i], h[j] = h[j], h[i] }
-
-func (h *containerHeap) Push(x interface{}) {
- *h = append(*h, x.(*container))
-}
-
-func (h *containerHeap) Pop() interface{} {
- old := *h
- n := len(old)
- x := old[n-1]
- *h = old[0 : n-1]
- return x
-}
-
-type itemIter struct {
- c comparator
- h *containerHeap
- iters []tsdb.Iterator
-}
-
-// NewItemIter returns a ItemIterator which mergers several tsdb.Iterator by input sorting order.
-func NewItemIter(iters []tsdb.Iterator, sort modelv1.Sort) ItemIterator {
- it := &itemIter{
- c: createComparator(sort),
- iters: iters,
- h: &containerHeap{},
- }
- it.init()
- return it
-}
-
-// init function MUST be called while initialization.
-// 1. Move all iterator to the first item by invoking their Next.
-// 2. Load all first items into a slice.
-func (it *itemIter) init() {
- for _, iter := range it.iters {
- it.pushIterator(iter)
- }
- // heap initialization
- heap.Init(it.h)
-}
-
-// pushIterator pushes the given iterator into the underlying deque.
-// Status will be immediately checked if the Iterator has a next value.
-// 1 - If not, it will be close at once and will not be added to the slice,
-//
-// which means inactive iterator does not exist in the deq.
-//
-// 2 - If so, it will be wrapped into a container and push to the deq.
-//
-// Then we call SliceStable sort to sort the deq.
-func (it *itemIter) pushIterator(iter tsdb.Iterator) {
- if !iter.Next() {
- return
- }
- heap.Push(it.h, &container{
- item: iter.Val(),
- iter: iter,
- c: it.c,
- })
-}
-
-func (it *itemIter) HasNext() bool {
- return it.h.Len() > 0
-}
-
-func (it *itemIter) Next() tsdb.Item {
- // 3. Pop up the minimal item through the order value
- c := heap.Pop(it.h).(*container)
-
- // 4. Move the iterator whose value is popped in step 3, push the next value of this iterator into the slice.
- it.pushIterator(c.iter)
-
- return c.item
-}
-
-// Close closes all underlying iterators.
-func (it *itemIter) Close() (err error) {
- for _, iter := range it.iters {
- err = multierr.Append(err, iter.Close())
- }
- return err
-}
diff --git a/pkg/query/logical/measure/measure_analyzer.go b/pkg/query/logical/measure/measure_analyzer.go
index f9409ee3..0e7f0fff 100644
--- a/pkg/query/logical/measure/measure_analyzer.go
+++ b/pkg/query/logical/measure/measure_analyzer.go
@@ -112,6 +112,66 @@ func Analyze(_ context.Context, criteria *measurev1.QueryRequest, metadata *comm
return p, nil
}
+// DistributedAnalyze converts logical expressions to executable operation tree represented by Plan.
+func DistributedAnalyze(criteria *measurev1.QueryRequest, s logical.Schema) (logical.Plan, error) {
+ groupByEntity := false
+ var groupByTags [][]*logical.Tag
+ if criteria.GetGroupBy() != nil {
+ groupByProjectionTags := criteria.GetGroupBy().GetTagProjection()
+ groupByTags = make([][]*logical.Tag, len(groupByProjectionTags.GetTagFamilies()))
+ tags := make([]string, 0)
+ for i, tagFamily := range groupByProjectionTags.GetTagFamilies() {
+ groupByTags[i] = logical.NewTags(tagFamily.GetName(), tagFamily.GetTags()...)
+ tags = append(tags, tagFamily.GetTags()...)
+ }
+ if logical.StringSlicesEqual(s.EntityList(), tags) {
+ groupByEntity = true
+ }
+ }
+
+ // parse fields
+ plan := newUnresolvedDistributed(criteria)
+
+ // parse limit and offset
+ limitParameter := criteria.GetLimit()
+ if limitParameter == 0 {
+ limitParameter = defaultLimit
+ }
+ pushedLimit := int(limitParameter + criteria.GetOffset())
+
+ if criteria.GetGroupBy() != nil {
+ plan = newUnresolvedGroupBy(plan, groupByTags, groupByEntity)
+ pushedLimit = math.MaxInt
+ }
+
+ if criteria.GetAgg() != nil {
+ plan = newUnresolvedAggregation(plan,
+ logical.NewField(criteria.GetAgg().GetFieldName()),
+ criteria.GetAgg().GetFunction(),
+ criteria.GetGroupBy() != nil,
+ )
+ pushedLimit = math.MaxInt
+ }
+
+ if criteria.GetTop() != nil {
+ plan = top(plan, criteria.GetTop())
+ }
+
+ plan = limit(plan, criteria.GetOffset(), limitParameter)
+ p, err := plan.Analyze(s)
+ if err != nil {
+ return nil, err
+ }
+ rules := []logical.OptimizeRule{
+ logical.NewPushDownOrder(criteria.OrderBy),
+ logical.NewPushDownMaxSize(pushedLimit),
+ }
+ if err := logical.ApplyRules(p, rules...); err != nil {
+ return nil, err
+ }
+ return p, nil
+}
+
// parseFields parses the query request to decide which kind of plan should be generated
// Basically,
// 1 - If no criteria is given, we can only scan all shards
diff --git a/pkg/query/logical/measure/measure_plan_distributed.go b/pkg/query/logical/measure/measure_plan_distributed.go
new file mode 100644
index 00000000..45ee4c66
--- /dev/null
+++ b/pkg/query/logical/measure/measure_plan_distributed.go
@@ -0,0 +1,234 @@
+// 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 measure
+
+import (
+ "context"
+ "fmt"
+
+ "go.uber.org/multierr"
+ "google.golang.org/protobuf/proto"
+
+ "github.com/apache/skywalking-banyandb/api/data"
+ measurev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/measure/v1"
+ modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
+ "github.com/apache/skywalking-banyandb/pkg/bus"
+ "github.com/apache/skywalking-banyandb/pkg/convert"
+ "github.com/apache/skywalking-banyandb/pkg/iter/sort"
+ pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
+ "github.com/apache/skywalking-banyandb/pkg/query/executor"
+ "github.com/apache/skywalking-banyandb/pkg/query/logical"
+)
+
+var _ logical.UnresolvedPlan = (*unresolvedDistributed)(nil)
+
+type unresolvedDistributed struct {
+ originalQuery *measurev1.QueryRequest
+ order *logical.OrderBy
+ maxDataPointsSize int
+}
+
+func newUnresolvedDistributed(query *measurev1.QueryRequest) logical.UnresolvedPlan {
+ return &unresolvedDistributed{
+ originalQuery: query,
+ }
+}
+
+func (ud *unresolvedDistributed) Limit(max int) {
+ ud.maxDataPointsSize = max
+}
+
+func (ud *unresolvedDistributed) Sort(order *logical.OrderBy) {
+ ud.order = order
+}
+
+func (ud *unresolvedDistributed) Analyze(s logical.Schema) (logical.Plan, error) {
+ if ud.originalQuery.TagProjection == nil {
+ return nil, fmt.Errorf("tag projection is required")
+ }
+ if ud.originalQuery.FieldProjection == nil {
+ return nil, fmt.Errorf("filed projection is required")
+ }
+ temp := &measurev1.QueryRequest{
+ TagProjection: ud.originalQuery.TagProjection,
+ FieldProjection: ud.originalQuery.FieldProjection,
+ Metadata: ud.originalQuery.Metadata,
+ Criteria: ud.originalQuery.Criteria,
+ Limit: uint32(ud.maxDataPointsSize),
+ OrderBy: &modelv1.QueryOrder{
+ IndexRuleName: ud.order.Index.Metadata.Name,
+ Sort: ud.order.Sort,
+ },
+ }
+ if ud.order == nil {
+ return &distributedPlan{
+ queryTemplate: temp,
+ s: s,
+ sortByTime: true,
+ }, nil
+ }
+ ok, indexRule := s.IndexRuleDefined(ud.originalQuery.OrderBy.IndexRuleName)
+ if !ok {
+ return nil, fmt.Errorf("index rule %s not found", ud.originalQuery.OrderBy.IndexRuleName)
+ }
+ if len(indexRule.Tags) != 1 {
+ return nil, fmt.Errorf("index rule %s should have only one tag", ud.originalQuery.OrderBy.IndexRuleName)
+ }
+ sortTagSpec := s.FindTagSpecByName(indexRule.Tags[0])
+ if sortTagSpec == nil {
+ return nil, fmt.Errorf("tag %s not found", indexRule.Tags[0])
+ }
+ result := &distributedPlan{
+ queryTemplate: temp,
+ s: s,
+ sortByTime: false,
+ sortTagSpec: *sortTagSpec,
+ }
+ if ud.originalQuery.OrderBy.Sort == modelv1.Sort_SORT_DESC {
+ result.desc = true
+ }
+ return result, nil
+}
+
+type distributedPlan struct {
+ s logical.Schema
+ queryTemplate *measurev1.QueryRequest
+ sortTagSpec logical.TagSpec
+ sortByTime bool
+ desc bool
+}
+
+func (t *distributedPlan) Execute(ctx context.Context) (executor.MIterator, error) {
+ dctx := executor.FromDistributedExecutionContext(ctx)
+ query := proto.Clone(t.queryTemplate).(*measurev1.QueryRequest)
+ query.TimeRange = dctx.TimeRange()
+ ff, err := dctx.Broadcast(data.TopicMeasureQuery, bus.NewMessage(bus.MessageID(dctx.TimeRange().Begin.Nanos), query))
+ if err != nil {
+ return nil, err
+ }
+ var allErr error
+ var see []sort.Iterator[*comparableDataPoint]
+ for _, f := range ff {
+ if m, getErr := f.Get(); getErr != nil {
+ allErr = multierr.Append(allErr, getErr)
+ } else {
+ see = append(see,
+ newSortableElements(m.Data().(*measurev1.QueryResponse).DataPoints,
+ t.sortByTime, t.sortTagSpec))
+ }
+ }
+ return &sortedMIterator{
+ Iterator: sort.NewItemIter[*comparableDataPoint](see, t.desc),
+ }, allErr
+}
+
+func (t *distributedPlan) String() string {
+ return fmt.Sprintf("distributed:%s", t.queryTemplate.String())
+}
+
+func (t *distributedPlan) Children() []logical.Plan {
+ return []logical.Plan{}
+}
+
+func (t *distributedPlan) Schema() logical.Schema {
+ return t.s
+}
+
+var _ sort.Comparable = (*comparableDataPoint)(nil)
+
+type comparableDataPoint struct {
+ *measurev1.DataPoint
+ sortField []byte
+}
+
+func newComparableElement(e *measurev1.DataPoint, sortByTime bool, sortTagSpec logical.TagSpec) (*comparableDataPoint, error) {
+ var sortField []byte
+ if sortByTime {
+ sortField = convert.Uint64ToBytes(uint64(e.Timestamp.AsTime().UnixNano()))
+ } else {
+ var err error
+ sortField, err = pbv1.MarshalTagValue(e.TagFamilies[sortTagSpec.TagFamilyIdx].Tags[sortTagSpec.TagIdx].Value)
+ if err != nil {
+ return nil, err
+ }
+ }
+
+ return &comparableDataPoint{
+ DataPoint: e,
+ sortField: sortField,
+ }, nil
+}
+
+func (e *comparableDataPoint) SortedField() []byte {
+ return e.sortField
+}
+
+var _ sort.Iterator[*comparableDataPoint] = (*sortableElements)(nil)
+
+type sortableElements struct {
+ cur *comparableDataPoint
+ dataPoints []*measurev1.DataPoint
+ sortTagSpec logical.TagSpec
+ index int
+ isSortByTime bool
+}
+
+func newSortableElements(dataPoints []*measurev1.DataPoint, isSortByTime bool, sortTagSpec logical.TagSpec) *sortableElements {
+ return &sortableElements{
+ dataPoints: dataPoints,
+ isSortByTime: isSortByTime,
+ sortTagSpec: sortTagSpec,
+ }
+}
+
+func (*sortableElements) Close() error {
+ return nil
+}
+
+func (s *sortableElements) Next() bool {
+ return s.iter(func(e *measurev1.DataPoint) (*comparableDataPoint, error) {
+ return newComparableElement(e, s.isSortByTime, s.sortTagSpec)
+ })
+}
+
+func (s *sortableElements) Val() *comparableDataPoint {
+ return s.cur
+}
+
+func (s *sortableElements) iter(fn func(*measurev1.DataPoint) (*comparableDataPoint, error)) bool {
+ if s.index >= len(s.dataPoints) {
+ return false
+ }
+ cur, err := fn(s.dataPoints[s.index])
+ s.index++
+ if err != nil {
+ return s.iter(fn)
+ }
+ s.cur = cur
+ return s.index < len(s.dataPoints)
+}
+
+var _ executor.MIterator = (*sortedMIterator)(nil)
+
+type sortedMIterator struct {
+ sort.Iterator[*comparableDataPoint]
+}
+
+func (s *sortedMIterator) Current() []*measurev1.DataPoint {
+ return []*measurev1.DataPoint{s.Val().DataPoint}
+}
diff --git a/pkg/query/logical/measure/measure_plan_indexscan_local.go b/pkg/query/logical/measure/measure_plan_indexscan_local.go
index 678baaea..b140b587 100644
--- a/pkg/query/logical/measure/measure_plan_indexscan_local.go
+++ b/pkg/query/logical/measure/measure_plan_indexscan_local.go
@@ -31,6 +31,7 @@ import (
modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
"github.com/apache/skywalking-banyandb/banyand/tsdb"
"github.com/apache/skywalking-banyandb/pkg/index"
+ "github.com/apache/skywalking-banyandb/pkg/iter/sort"
"github.com/apache/skywalking-banyandb/pkg/logger"
"github.com/apache/skywalking-banyandb/pkg/query/executor"
"github.com/apache/skywalking-banyandb/pkg/query/logical"
@@ -225,7 +226,7 @@ func indexScan(startTime, endTime time.Time, metadata *commonv1.Metadata, projec
var _ executor.MIterator = (*indexScanIterator)(nil)
type indexScanIterator struct {
- inner logical.ItemIterator
+ inner sort.Iterator[tsdb.Item]
err error
current *measurev1.DataPoint
context transformContext
@@ -233,7 +234,7 @@ type indexScanIterator struct {
num int
}
-func newIndexScanIterator(inner logical.ItemIterator, context transformContext, max int) executor.MIterator {
+func newIndexScanIterator(inner sort.Iterator[tsdb.Item], context transformContext, max int) executor.MIterator {
return &indexScanIterator{
inner: inner,
context: context,
@@ -242,10 +243,10 @@ func newIndexScanIterator(inner logical.ItemIterator, context transformContext,
}
func (ism *indexScanIterator) Next() bool {
- if !ism.inner.HasNext() || ism.err != nil || ism.num > ism.max {
+ if !ism.inner.Next() || ism.err != nil || ism.num > ism.max {
return false
}
- nextItem := ism.inner.Next()
+ nextItem := ism.inner.Val()
var err error
if ism.current, err = transform(nextItem, ism.context); err != nil {
ism.err = multierr.Append(ism.err, err)
diff --git a/pkg/query/logical/stream/stream_analyzer.go b/pkg/query/logical/stream/stream_analyzer.go
index 4a429312..919ac9ea 100644
--- a/pkg/query/logical/stream/stream_analyzer.go
+++ b/pkg/query/logical/stream/stream_analyzer.go
@@ -75,6 +75,22 @@ func Analyze(_ context.Context, criteria *streamv1.QueryRequest, metadata *commo
return p, nil
}
+// DistributedAnalyze converts logical expressions to executable operation tree represented by Plan.
+func DistributedAnalyze(criteria *streamv1.QueryRequest, s logical.Schema) (logical.Plan, error) {
+ // parse fields
+ plan := newUnresolvedDistributed(criteria)
+ // parse offset
+ plan = newOffset(plan, criteria.GetOffset())
+
+ // parse limit
+ limitParameter := criteria.GetLimit()
+ if limitParameter == 0 {
+ limitParameter = defaultLimit
+ }
+ plan = newLimit(plan, limitParameter)
+ return plan.Analyze(s)
+}
+
var (
_ logical.Plan = (*limit)(nil)
_ logical.UnresolvedPlan = (*limit)(nil)
diff --git a/pkg/query/logical/stream/stream_plan_distributed.go b/pkg/query/logical/stream/stream_plan_distributed.go
new file mode 100644
index 00000000..7fce94cc
--- /dev/null
+++ b/pkg/query/logical/stream/stream_plan_distributed.go
@@ -0,0 +1,214 @@
+// 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 stream
+
+import (
+ "context"
+ "fmt"
+
+ "go.uber.org/multierr"
+ "google.golang.org/protobuf/proto"
+
+ "github.com/apache/skywalking-banyandb/api/data"
+ modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
+ streamv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/stream/v1"
+ "github.com/apache/skywalking-banyandb/pkg/bus"
+ "github.com/apache/skywalking-banyandb/pkg/convert"
+ "github.com/apache/skywalking-banyandb/pkg/iter/sort"
+ pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
+ "github.com/apache/skywalking-banyandb/pkg/query/executor"
+ "github.com/apache/skywalking-banyandb/pkg/query/logical"
+)
+
+var _ logical.UnresolvedPlan = (*unresolvedDistributed)(nil)
+
+type unresolvedDistributed struct {
+ originalQuery *streamv1.QueryRequest
+}
+
+func newUnresolvedDistributed(query *streamv1.QueryRequest) logical.UnresolvedPlan {
+ return &unresolvedDistributed{
+ originalQuery: query,
+ }
+}
+
+func (ud *unresolvedDistributed) Analyze(s logical.Schema) (logical.Plan, error) {
+ if ud.originalQuery.Projection == nil {
+ return nil, fmt.Errorf("projection is required")
+ }
+ limit := ud.originalQuery.GetLimit()
+ if limit == 0 {
+ limit = defaultLimit
+ }
+ temp := &streamv1.QueryRequest{
+ Projection: ud.originalQuery.Projection,
+ Metadata: ud.originalQuery.Metadata,
+ Criteria: ud.originalQuery.Criteria,
+ Limit: limit,
+ OrderBy: ud.originalQuery.OrderBy,
+ }
+ if ud.originalQuery.OrderBy == nil {
+ return &distributedPlan{
+ queryTemplate: temp,
+ s: s,
+ sortByTime: true,
+ }, nil
+ }
+ ok, indexRule := s.IndexRuleDefined(ud.originalQuery.OrderBy.IndexRuleName)
+ if !ok {
+ return nil, fmt.Errorf("index rule %s not found", ud.originalQuery.OrderBy.IndexRuleName)
+ }
+ if len(indexRule.Tags) != 1 {
+ return nil, fmt.Errorf("index rule %s should have only one tag", ud.originalQuery.OrderBy.IndexRuleName)
+ }
+ sortTagSpec := s.FindTagSpecByName(indexRule.Tags[0])
+ if sortTagSpec == nil {
+ return nil, fmt.Errorf("tag %s not found", indexRule.Tags[0])
+ }
+ result := &distributedPlan{
+ queryTemplate: temp,
+ s: s,
+ sortByTime: false,
+ sortTagSpec: *sortTagSpec,
+ }
+ if ud.originalQuery.OrderBy.Sort == modelv1.Sort_SORT_DESC {
+ result.desc = true
+ }
+ return result, nil
+}
+
+type distributedPlan struct {
+ s logical.Schema
+ queryTemplate *streamv1.QueryRequest
+ sortTagSpec logical.TagSpec
+ sortByTime bool
+ desc bool
+}
+
+func (t *distributedPlan) Execute(ctx context.Context) ([]*streamv1.Element, error) {
+ dctx := executor.FromDistributedExecutionContext(ctx)
+ query := proto.Clone(t.queryTemplate).(*streamv1.QueryRequest)
+ query.TimeRange = dctx.TimeRange()
+ ff, err := dctx.Broadcast(data.TopicMeasureQuery, bus.NewMessage(bus.MessageID(dctx.TimeRange().Begin.Nanos), query))
+ if err != nil {
+ return nil, err
+ }
+ var allErr error
+ var see []sort.Iterator[*comparableElement]
+ for _, f := range ff {
+ if m, getErr := f.Get(); getErr != nil {
+ allErr = multierr.Append(allErr, getErr)
+ } else {
+ see = append(see,
+ newSortableElements(m.Data().(*streamv1.QueryResponse).Elements,
+ t.sortByTime, t.sortTagSpec))
+ }
+ }
+ iter := sort.NewItemIter[*comparableElement](see, t.desc)
+ var result []*streamv1.Element
+ for iter.Next() {
+ result = append(result, iter.Val().Element)
+ }
+ return result, nil
+}
+
+func (t *distributedPlan) String() string {
+ return fmt.Sprintf("distributed:%s", t.queryTemplate.String())
+}
+
+func (t *distributedPlan) Children() []logical.Plan {
+ return []logical.Plan{}
+}
+
+func (t *distributedPlan) Schema() logical.Schema {
+ return t.s
+}
+
+var _ sort.Comparable = (*comparableElement)(nil)
+
+type comparableElement struct {
+ *streamv1.Element
+ sortField []byte
+}
+
+func newComparableElement(e *streamv1.Element, sortByTime bool, sortTagSpec logical.TagSpec) (*comparableElement, error) {
+ var sortField []byte
+ if sortByTime {
+ sortField = convert.Uint64ToBytes(uint64(e.Timestamp.AsTime().UnixNano()))
+ } else {
+ var err error
+ sortField, err = pbv1.MarshalTagValue(e.TagFamilies[sortTagSpec.TagFamilyIdx].Tags[sortTagSpec.TagIdx].Value)
+ if err != nil {
+ return nil, err
+ }
+ }
+
+ return &comparableElement{
+ Element: e,
+ sortField: sortField,
+ }, nil
+}
+
+func (e *comparableElement) SortedField() []byte {
+ return e.sortField
+}
+
+var _ sort.Iterator[*comparableElement] = (*sortableElements)(nil)
+
+type sortableElements struct {
+ cur *comparableElement
+ elements []*streamv1.Element
+ sortTagSpec logical.TagSpec
+ index int
+ isSortByTime bool
+}
+
+func newSortableElements(elements []*streamv1.Element, isSortByTime bool, sortTagSpec logical.TagSpec) *sortableElements {
+ return &sortableElements{
+ elements: elements,
+ isSortByTime: isSortByTime,
+ sortTagSpec: sortTagSpec,
+ }
+}
+
+func (*sortableElements) Close() error {
+ return nil
+}
+
+func (s *sortableElements) Next() bool {
+ return s.iter(func(e *streamv1.Element) (*comparableElement, error) {
+ return newComparableElement(e, s.isSortByTime, s.sortTagSpec)
+ })
+}
+
+func (s *sortableElements) Val() *comparableElement {
+ return s.cur
+}
+
+func (s *sortableElements) iter(fn func(*streamv1.Element) (*comparableElement, error)) bool {
+ if s.index >= len(s.elements) {
+ return false
+ }
+ cur, err := fn(s.elements[s.index])
+ s.index++
+ if err != nil {
+ return s.iter(fn)
+ }
+ s.cur = cur
+ return s.index < len(s.elements)
+}
diff --git a/pkg/query/logical/stream/stream_plan_indexscan_local.go b/pkg/query/logical/stream/stream_plan_indexscan_local.go
index 54ef565e..c011332d 100644
--- a/pkg/query/logical/stream/stream_plan_indexscan_local.go
+++ b/pkg/query/logical/stream/stream_plan_indexscan_local.go
@@ -122,8 +122,8 @@ func (i *localIndexScan) Execute(ctx context.Context) (elements []*streamv1.Elem
defer func() {
err = multierr.Append(err, it.Close())
}()
- for it.HasNext() {
- nextItem := it.Next()
+ for it.Next() {
+ nextItem := it.Val()
tagFamilies, innerErr := logical.ProjectItem(ec, nextItem, i.projectionTagRefs)
if innerErr != nil {
return nil, innerErr
diff --git a/pkg/run/channel_closer.go b/pkg/run/channel_closer.go
index 54c62707..fa1af90d 100644
--- a/pkg/run/channel_closer.go
+++ b/pkg/run/channel_closer.go
@@ -101,6 +101,7 @@ func (c *ChannelCloser) CloseThenWait() {
return
}
+ c.cancel()
c.lock.Lock()
c.closed = true
c.lock.Unlock()
@@ -108,7 +109,6 @@ func (c *ChannelCloser) CloseThenWait() {
c.sender.Done()
c.sender.Wait()
- c.cancel()
c.receiver.Done()
c.receiver.Wait()
}
diff --git a/pkg/schema/metadata.go b/pkg/schema/metadata.go
index 0b4cc94a..3e934873 100644
--- a/pkg/schema/metadata.go
+++ b/pkg/schema/metadata.go
@@ -15,13 +15,11 @@
// specific language governing permissions and limitations
// under the License.
-// Package schema implements a framework to sync schema info from the metadata repository.
package schema
import (
"context"
"io"
- "math"
"sync"
"sync/atomic"
"time"
@@ -38,78 +36,58 @@ import (
"github.com/apache/skywalking-banyandb/pkg/run"
)
-// EventType defines actions of events.
-type EventType uint8
+var _ Resource = (*resourceSpec)(nil)
-// EventType support Add/Update and Delete.
-// All events are idempotent.
-const (
- EventAddOrUpdate EventType = iota
- EventDelete
-)
-
-// EventKind defines category of events.
-type EventKind uint8
-
-// This framework groups events to a hierarchy. A group is the root node.
-const (
- EventKindGroup EventKind = iota
- EventKindResource
-)
+type resourceSpec struct {
+ schema ResourceSchema
+ delegated io.Closer
+ indexRules []*databasev1.IndexRule
+ aggregations []*databasev1.TopNAggregation
+}
-// Group is the root node, allowing get resources from its sub nodes.
-type Group interface {
- GetSchema() *commonv1.Group
- StoreResource(ctx context.Context, resourceSchema ResourceSchema) (Resource, error)
- LoadResource(name string) (Resource, bool)
+func (rs *resourceSpec) Delegated() io.Closer {
+ return rs.delegated
}
-// MetadataEvent is the syncing message between metadata repo and this framework.
-type MetadataEvent struct {
- Metadata *commonv1.Metadata
- Typ EventType
- Kind EventKind
+func (rs *resourceSpec) Close() error {
+ return rs.delegated.Close()
}
-// ResourceSchema allows get the metadata.
-type ResourceSchema interface {
- GetMetadata() *commonv1.Metadata
+func (rs *resourceSpec) Schema() ResourceSchema {
+ return rs.schema
}
-// ResourceSpec wraps required fields to open a resource.
-type ResourceSpec struct {
- Schema ResourceSchema
- // IndexRules are index rules bound to the Schema
- IndexRules []*databasev1.IndexRule
- // Aggregations are topN aggregation bound to the Schema, e.g. TopNAggregation
- Aggregations []*databasev1.TopNAggregation
+func (rs *resourceSpec) IndexRules() []*databasev1.IndexRule {
+ return rs.indexRules
}
-// Resource allows access metadata from a local cache.
-type Resource interface {
- GetIndexRules() []*databasev1.IndexRule
- GetTopN() []*databasev1.TopNAggregation
- MaxObservedModRevision() int64
- ResourceSchema
- io.Closer
+func (rs *resourceSpec) TopN() []*databasev1.TopNAggregation {
+ return rs.aggregations
}
-// ResourceSupplier allows open a resource and its embedded tsdb.
-type ResourceSupplier interface {
- OpenResource(shardNum uint32, db tsdb.Supplier, spec ResourceSpec) (Resource, error)
- ResourceSchema(metdata *commonv1.Metadata) (ResourceSchema, error)
- OpenDB(groupSchema *commonv1.Group) (tsdb.Database, error)
+func (rs *resourceSpec) maxRevision() int64 {
+ return rs.schema.GetMetadata().GetModRevision()
}
-// Repository is the collection of several hierarchies groups by a "Group".
-type Repository interface {
- Watcher()
- SendMetadataEvent(MetadataEvent)
- StoreGroup(groupMeta *commonv1.Metadata) (*group, error)
- LoadGroup(name string) (Group, bool)
- LoadResource(metadata *commonv1.Metadata) (Resource, bool)
- Close()
- StopCh() <-chan struct{}
+func (rs *resourceSpec) isNewThan(other *resourceSpec) bool {
+ if rs.maxRevision() > other.maxRevision() {
+ return true
+ } else if rs.maxRevision() < other.maxRevision() {
+ return false
+ }
+ if len(rs.indexRules) != len(other.indexRules) {
+ return true
+ }
+ if len(rs.aggregations) != len(other.aggregations) {
+ return true
+ }
+ if parseMaxModRevision(rs.indexRules) > parseMaxModRevision(other.indexRules) {
+ return true
+ }
+ if parseMaxModRevision(rs.aggregations) > parseMaxModRevision(other.aggregations) {
+ return true
+ }
+ return false
}
const defaultWorkerNum = 10
@@ -117,17 +95,28 @@ const defaultWorkerNum = 10
var _ Repository = (*schemaRepo)(nil)
type schemaRepo struct {
- metadata metadata.Repo
- resourceSupplier ResourceSupplier
- l *logger.Logger
- data map[string]*group
- workerCloser *run.Closer
- closer *run.Closer
- eventCh chan MetadataEvent
- workerNum int
+ metadata metadata.Repo
+ resourceSupplier ResourceSupplier
+ resourceSchemaSupplier ResourceSchemaSupplier
+ l *logger.Logger
+ data map[string]*group
+ closer *run.ChannelCloser
+ eventCh chan MetadataEvent
+ workerNum int
sync.RWMutex
}
+func (sr *schemaRepo) SendMetadataEvent(event MetadataEvent) {
+ if !sr.closer.AddSender() {
+ return
+ }
+ defer sr.closer.SenderDone()
+ select {
+ case sr.eventCh <- event:
+ case <-sr.closer.CloseNotify():
+ }
+}
+
// StopCh implements Repository.
func (sr *schemaRepo) StopCh() <-chan struct{} {
return sr.closer.CloseNotify()
@@ -140,26 +129,42 @@ func NewRepository(
resourceSupplier ResourceSupplier,
) Repository {
return &schemaRepo{
- metadata: metadata,
- l: l,
- resourceSupplier: resourceSupplier,
- data: make(map[string]*group),
- eventCh: make(chan MetadataEvent, defaultWorkerNum),
- workerNum: defaultWorkerNum,
- workerCloser: run.NewCloser(defaultWorkerNum),
- closer: run.NewCloser(1),
+ metadata: metadata,
+ l: l,
+ resourceSupplier: resourceSupplier,
+ resourceSchemaSupplier: resourceSupplier,
+ data: make(map[string]*group),
+ eventCh: make(chan MetadataEvent, defaultWorkerNum),
+ workerNum: defaultWorkerNum,
+ closer: run.NewChannelCloser(),
}
}
-func (sr *schemaRepo) SendMetadataEvent(event MetadataEvent) {
- sr.eventCh <- event
+// NewPortableRepository return a new Repository without tsdb.
+func NewPortableRepository(
+ metadata metadata.Repo,
+ l *logger.Logger,
+ supplier ResourceSchemaSupplier,
+) Repository {
+ return &schemaRepo{
+ metadata: metadata,
+ l: l,
+ resourceSchemaSupplier: supplier,
+ data: make(map[string]*group),
+ eventCh: make(chan MetadataEvent, defaultWorkerNum),
+ workerNum: defaultWorkerNum,
+ closer: run.NewChannelCloser(),
+ }
}
func (sr *schemaRepo) Watcher() {
for i := 0; i < sr.workerNum; i++ {
go func() {
+ if !sr.closer.AddReceiver() {
+ return
+ }
defer func() {
- sr.workerCloser.Done()
+ sr.closer.ReceiverDone()
if err := recover(); err != nil {
sr.l.Warn().Interface("err", err).Msg("watching the events")
}
@@ -192,13 +197,9 @@ func (sr *schemaRepo) Watcher() {
}
if err != nil && !errors.Is(err, schema.ErrClosed) {
sr.l.Err(err).Interface("event", evt).Msg("fail to handle the metadata event. retry...")
- select {
- case sr.eventCh <- evt:
- case <-sr.workerCloser.CloseNotify():
- return
- }
+ sr.SendMetadataEvent(evt)
}
- case <-sr.workerCloser.CloseNotify():
+ case <-sr.closer.CloseNotify():
return
}
}
@@ -207,27 +208,30 @@ func (sr *schemaRepo) Watcher() {
}
func (sr *schemaRepo) StoreGroup(groupMeta *commonv1.Metadata) (*group, error) {
- ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
- defer cancel()
- groupSchema, err := sr.metadata.GroupRegistry().GetGroup(ctx, groupMeta.GetName())
- if err != nil {
- return nil, err
- }
- name := groupSchema.GetMetadata().GetName()
+ name := groupMeta.GetName()
sr.Lock()
defer sr.Unlock()
g, ok := sr.getGroup(name)
if !ok {
sr.l.Info().Str("group", name).Msg("creating a tsdb")
- var db tsdb.Database
- db, err = sr.resourceSupplier.OpenDB(groupSchema)
- if err != nil {
+ g = sr.createGroup(name)
+ if err := g.init(name); err != nil {
return nil, err
}
- g = newGroup(groupSchema, sr.metadata, db, sr.l, sr.resourceSupplier)
- sr.data[name] = g
return g, nil
}
+ if !g.isInit() {
+ if err := g.init(name); err != nil {
+ return nil, err
+ }
+ return g, nil
+ }
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+ defer cancel()
+ groupSchema, err := g.metadata.GroupRegistry().GetGroup(ctx, name)
+ if err != nil {
+ return nil, err
+ }
prevGroupSchema := g.GetSchema()
if groupSchema.GetMetadata().GetModRevision() <= prevGroupSchema.Metadata.ModRevision {
return g, nil
@@ -236,15 +240,22 @@ func (sr *schemaRepo) StoreGroup(groupMeta *commonv1.Metadata) (*group, error) {
db := g.SupplyTSDB()
db.Close()
sr.l.Info().Str("group", name).Msg("creating a new tsdb")
- newDB, err := sr.resourceSupplier.OpenDB(groupSchema)
- if err != nil {
+ if err := g.init(name); err != nil {
return nil, err
}
- g.setDB(newDB)
- g.groupSchema.Store(groupSchema)
return g, nil
}
+func (sr *schemaRepo) createGroup(name string) (g *group) {
+ if sr.resourceSupplier != nil {
+ g = newGroup(sr.metadata, sr.l, sr.resourceSupplier)
+ } else {
+ g = newPortableGroup(sr.metadata, sr.l)
+ }
+ sr.data[name] = g
+ return
+}
+
func (sr *schemaRepo) deleteGroup(groupMeta *commonv1.Metadata) error {
name := groupMeta.GetName()
sr.Lock()
@@ -273,7 +284,11 @@ func (sr *schemaRepo) getGroup(name string) (*group, bool) {
func (sr *schemaRepo) LoadGroup(name string) (Group, bool) {
sr.RLock()
defer sr.RUnlock()
- return sr.getGroup(name)
+ g, ok := sr.getGroup(name)
+ if !ok {
+ return nil, false
+ }
+ return g, g.isInit()
}
func (sr *schemaRepo) LoadResource(metadata *commonv1.Metadata) (Resource, bool) {
@@ -285,18 +300,18 @@ func (sr *schemaRepo) LoadResource(metadata *commonv1.Metadata) (Resource, bool)
}
func (sr *schemaRepo) storeResource(metadata *commonv1.Metadata) (Resource, error) {
- group, ok := sr.LoadGroup(metadata.Group)
+ g, ok := sr.LoadGroup(metadata.Group)
if !ok {
var err error
- if group, err = sr.StoreGroup(&commonv1.Metadata{Name: metadata.Group}); err != nil {
+ if g, err = sr.StoreGroup(&commonv1.Metadata{Name: metadata.Group}); err != nil {
return nil, errors.WithMessagef(err, "create unknown group:%s", metadata.Group)
}
}
- stm, err := sr.resourceSupplier.ResourceSchema(metadata)
+ stm, err := sr.resourceSchemaSupplier.ResourceSchema(metadata)
if err != nil {
return nil, errors.WithMessage(err, "fails to get the resource")
}
- return group.StoreResource(context.Background(), stm)
+ return g.(*group).storeResource(context.Background(), stm)
}
func (sr *schemaRepo) deleteResource(metadata *commonv1.Metadata) error {
@@ -313,7 +328,8 @@ func (sr *schemaRepo) Close() {
sr.l.Warn().Interface("err", err).Msg("closing resource")
}
}()
- sr.workerCloser.CloseThenWait()
+ sr.closer.CloseThenWait()
+ close(sr.eventCh)
sr.RLock()
defer sr.RUnlock()
@@ -323,8 +339,6 @@ func (sr *schemaRepo) Close() {
sr.l.Err(err).RawJSON("group", logger.Proto(g.GetSchema().Metadata)).Msg("closing")
}
}
- sr.closer.Done()
- sr.closer.CloseThenWait()
}
var _ Group = (*group)(nil)
@@ -335,14 +349,12 @@ type group struct {
db atomic.Value
groupSchema atomic.Pointer[commonv1.Group]
l *logger.Logger
- schemaMap map[string]Resource
+ schemaMap map[string]*resourceSpec
mapMutex sync.RWMutex
}
func newGroup(
- groupSchema *commonv1.Group,
metadata metadata.Repo,
- db tsdb.Database,
l *logger.Logger,
resourceSupplier ResourceSupplier,
) *group {
@@ -350,12 +362,49 @@ func newGroup(
groupSchema: atomic.Pointer[commonv1.Group]{},
metadata: metadata,
l: l,
- schemaMap: make(map[string]Resource),
+ schemaMap: make(map[string]*resourceSpec),
resourceSupplier: resourceSupplier,
}
+ return g
+}
+
+func newPortableGroup(
+ metadata metadata.Repo,
+ l *logger.Logger,
+) *group {
+ g := &group{
+ groupSchema: atomic.Pointer[commonv1.Group]{},
+ metadata: metadata,
+ l: l,
+ schemaMap: make(map[string]*resourceSpec),
+ }
+ return g
+}
+
+func (g *group) init(name string) error {
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+ defer cancel()
+ groupSchema, err := g.metadata.GroupRegistry().GetGroup(ctx, name)
+ if errors.As(err, schema.ErrGRPCResourceNotFound) {
+ return nil
+ }
+ if err != nil {
+ return err
+ }
g.groupSchema.Store(groupSchema)
+ if g.isPortable() {
+ return nil
+ }
+ db, err := g.resourceSupplier.OpenDB(groupSchema)
+ if err != nil {
+ return err
+ }
g.db.Store(db)
- return g
+ return nil
+}
+
+func (g *group) isInit() bool {
+ return g.GetSchema() != nil
}
func (g *group) GetSchema() *commonv1.Group {
@@ -366,46 +415,15 @@ func (g *group) SupplyTSDB() tsdb.Database {
return g.db.Load().(tsdb.Database)
}
-func (g *group) setDB(db tsdb.Database) {
- g.db.Store(db)
-}
-
-func (g *group) StoreResource(ctx context.Context, resourceSchema ResourceSchema) (Resource, error) {
+func (g *group) storeResource(ctx context.Context, resourceSchema ResourceSchema) (Resource, error) {
g.mapMutex.Lock()
defer g.mapMutex.Unlock()
- key := resourceSchema.GetMetadata().GetName()
- preResource := g.schemaMap[key]
- var localCtx context.Context
- var cancel context.CancelFunc
- if preResource != nil &&
- resourceSchema.GetMetadata().GetModRevision() <= preResource.GetMetadata().GetModRevision() {
- // we only need to check the max modifications revision observed for index rules
- localCtx, cancel = context.WithTimeout(ctx, 5*time.Second)
- idxRules, errIndexRules := g.metadata.IndexRules(localCtx, resourceSchema.GetMetadata())
- cancel()
- if errIndexRules != nil {
- return nil, errIndexRules
- }
- localCtx, cancel = context.WithTimeout(ctx, 5*time.Second)
- topNAggrs, errTopN := g.metadata.MeasureRegistry().TopNAggregations(localCtx, resourceSchema.GetMetadata())
- cancel()
- if errTopN != nil {
- return nil, errTopN
- }
- if len(idxRules) == len(preResource.GetIndexRules()) && len(topNAggrs) == len(preResource.GetTopN()) {
- maxModRevision := int64(math.Max(float64(ParseMaxModRevision(idxRules)), float64(ParseMaxModRevision(topNAggrs))))
- if preResource.MaxObservedModRevision() >= maxModRevision {
- return preResource, nil
- }
- }
- }
- localCtx, cancel = context.WithTimeout(ctx, 5*time.Second)
+ localCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
defer cancel()
idxRules, err := g.metadata.IndexRules(localCtx, resourceSchema.GetMetadata())
if err != nil {
return nil, err
}
-
var topNAggrs []*databasev1.TopNAggregation
if _, ok := resourceSchema.(*databasev1.Measure); ok {
localCtx, cancel = context.WithTimeout(ctx, 5*time.Second)
@@ -416,20 +434,28 @@ func (g *group) StoreResource(ctx context.Context, resourceSchema ResourceSchema
return nil, innerErr
}
}
-
- sm, errTS := g.resourceSupplier.OpenResource(g.GetSchema().GetResourceOpts().ShardNum, g, ResourceSpec{
- Schema: resourceSchema,
- IndexRules: idxRules,
- Aggregations: topNAggrs,
- })
- if errTS != nil {
- return nil, errTS
+ resource := &resourceSpec{
+ schema: resourceSchema,
+ indexRules: idxRules,
+ aggregations: topNAggrs,
+ }
+ key := resourceSchema.GetMetadata().GetName()
+ preResource := g.schemaMap[key]
+ if preResource != nil && preResource.isNewThan(resource) {
+ return preResource, nil
}
- g.schemaMap[key] = sm
+ if !g.isPortable() {
+ sm, errTS := g.resourceSupplier.OpenResource(g.GetSchema().GetResourceOpts().ShardNum, g, resource)
+ if errTS != nil {
+ return nil, errTS
+ }
+ resource.delegated = sm
+ }
+ g.schemaMap[key] = resource
if preResource != nil {
_ = preResource.Close()
}
- return sm, nil
+ return resource, nil
}
func (g *group) deleteResource(metadata *commonv1.Metadata) error {
@@ -445,6 +471,10 @@ func (g *group) deleteResource(metadata *commonv1.Metadata) error {
return nil
}
+func (g *group) isPortable() bool {
+ return g.resourceSupplier == nil
+}
+
func (g *group) LoadResource(name string) (Resource, bool) {
g.mapMutex.RLock()
s := g.schemaMap[name]
@@ -461,11 +491,13 @@ func (g *group) close() (err error) {
err = multierr.Append(err, s.Close())
}
g.mapMutex.RUnlock()
+ if !g.isInit() {
+ return nil
+ }
return multierr.Append(err, g.SupplyTSDB().Close())
}
-// ParseMaxModRevision gives the max revision from resources' metadata.
-func ParseMaxModRevision[T ResourceSchema](indexRules []T) (maxRevisionForIdxRules int64) {
+func parseMaxModRevision[T ResourceSchema](indexRules []T) (maxRevisionForIdxRules int64) {
maxRevisionForIdxRules = int64(0)
for _, idxRule := range indexRules {
if idxRule.GetMetadata().GetModRevision() > maxRevisionForIdxRules {
diff --git a/pkg/schema/schema.go b/pkg/schema/schema.go
new file mode 100644
index 00000000..f19a4230
--- /dev/null
+++ b/pkg/schema/schema.go
@@ -0,0 +1,96 @@
+// 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 schema implements a framework to sync schema info from the metadata repository.
+package schema
+
+import (
+ "io"
+
+ commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
+ databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+ "github.com/apache/skywalking-banyandb/banyand/tsdb"
+)
+
+// EventType defines actions of events.
+type EventType uint8
+
+// EventType support Add/Update and Delete.
+// All events are idempotent.
+const (
+ EventAddOrUpdate EventType = iota
+ EventDelete
+)
+
+// EventKind defines category of events.
+type EventKind uint8
+
+// This framework groups events to a hierarchy. A group is the root node.
+const (
+ EventKindGroup EventKind = iota
+ EventKindResource
+)
+
+// Group is the root node, allowing get resources from its sub nodes.
+type Group interface {
+ GetSchema() *commonv1.Group
+ LoadResource(name string) (Resource, bool)
+}
+
+// MetadataEvent is the syncing message between metadata repo and this framework.
+type MetadataEvent struct {
+ Metadata *commonv1.Metadata
+ Typ EventType
+ Kind EventKind
+}
+
+// ResourceSchema allows get the metadata.
+type ResourceSchema interface {
+ GetMetadata() *commonv1.Metadata
+}
+
+// Resource allows access metadata from a local cache.
+type Resource interface {
+ io.Closer
+ IndexRules() []*databasev1.IndexRule
+ TopN() []*databasev1.TopNAggregation
+ Schema() ResourceSchema
+ Delegated() io.Closer
+}
+
+// ResourceSchemaSupplier allows get a ResourceSchema from the metadata.
+type ResourceSchemaSupplier interface {
+ ResourceSchema(metadata *commonv1.Metadata) (ResourceSchema, error)
+}
+
+// ResourceSupplier allows open a resource and its embedded tsdb.
+type ResourceSupplier interface {
+ ResourceSchemaSupplier
+ OpenResource(shardNum uint32, db tsdb.Supplier, spec Resource) (io.Closer, error)
+ OpenDB(groupSchema *commonv1.Group) (tsdb.Database, error)
+}
+
+// Repository is the collection of several hierarchies groups by a "Group".
+type Repository interface {
+ Watcher()
+ SendMetadataEvent(MetadataEvent)
+ StoreGroup(groupMeta *commonv1.Metadata) (*group, error)
+ LoadGroup(name string) (Group, bool)
+ LoadResource(metadata *commonv1.Metadata) (Resource, bool)
+ Close()
+ StopCh() <-chan struct{}
+}