You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by lu...@apache.org on 2023/10/29 02:58:44 UTC
(skywalking-banyandb) branch main updated: Unify topn&query plan (#343)
This is an automated email from the ASF dual-hosted git repository.
lujiajing pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
The following commit(s) were added to refs/heads/main by this push:
new dbb75ae9 Unify topn&query plan (#343)
dbb75ae9 is described below
commit dbb75ae9c930739b8cdfac33ad54dbf3b25ed75b
Author: A3bz <59...@users.noreply.github.com>
AuthorDate: Sun Oct 29 10:58:38 2023 +0800
Unify topn&query plan (#343)
* unify query & topn planer
* get schema from measure registry rather than topn measure
* add entity to schema registry
* assign tags to variable with groupByTags array index zero
---------
Co-authored-by: Jiajing LU <lu...@gmail.com>
---
banyand/measure/measure_query.go | 5 +
banyand/measure/metadata.go | 1 +
banyand/query/processor_topn.go | 187 ++++-----------
pkg/query/logical/measure/topn_analyzer.go | 94 ++++++++
pkg/query/logical/measure/topn_plan_localscan.go | 254 +++++++++++++++++++++
test/cases/topn/data/input/asc.yaml | 22 --
test/cases/topn/data/input/desc.yaml | 22 --
test/cases/topn/data/want/aggr_desc.yaml | 3 +-
test/cases/topn/data/want/asc.yaml | 174 --------------
test/cases/topn/data/want/condition_aggr_desc.yaml | 3 +-
test/cases/topn/data/want/desc.yaml | 174 --------------
test/cases/topn/data/want/null_group.yaml | 3 +-
test/cases/topn/topn.go | 2 -
13 files changed, 406 insertions(+), 538 deletions(-)
diff --git a/banyand/measure/measure_query.go b/banyand/measure/measure_query.go
index bd4b9e38..a34c0f7b 100644
--- a/banyand/measure/measure_query.go
+++ b/banyand/measure/measure_query.go
@@ -54,10 +54,15 @@ type Measure interface {
GetSchema() *databasev1.Measure
GetIndexRules() []*databasev1.IndexRule
GetInterval() time.Duration
+ SetSchema(schema *databasev1.Measure)
}
var _ Measure = (*measure)(nil)
+func (s *measure) SetSchema(schema *databasev1.Measure) {
+ s.schema = schema
+}
+
func (s *measure) GetInterval() time.Duration {
return s.interval
}
diff --git a/banyand/measure/metadata.go b/banyand/measure/metadata.go
index 27dfa090..a63acad2 100644
--- a/banyand/measure/metadata.go
+++ b/banyand/measure/metadata.go
@@ -217,6 +217,7 @@ func createOrUpdateTopNMeasure(ctx context.Context, measureSchemaRegistry schema
},
},
Fields: []*databasev1.FieldSpec{TopNValueFieldSpec},
+ Entity: sourceMeasureSchema.GetEntity(),
}
if oldTopNSchema == nil {
if _, innerErr := measureSchemaRegistry.CreateMeasure(ctx, newTopNMeasure); innerErr != nil {
diff --git a/banyand/query/processor_topn.go b/banyand/query/processor_topn.go
index 1f266c7e..48743d9d 100644
--- a/banyand/query/processor_topn.go
+++ b/banyand/query/processor_topn.go
@@ -18,29 +18,24 @@
package query
import (
- "bytes"
"container/heap"
"context"
"time"
- "github.com/pkg/errors"
"golang.org/x/exp/slices"
- "google.golang.org/protobuf/proto"
"google.golang.org/protobuf/types/known/timestamppb"
- databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+ "github.com/apache/skywalking-banyandb/api/common"
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/banyand/measure"
"github.com/apache/skywalking-banyandb/banyand/tsdb"
"github.com/apache/skywalking-banyandb/pkg/bus"
- "github.com/apache/skywalking-banyandb/pkg/convert"
"github.com/apache/skywalking-banyandb/pkg/flow"
- "github.com/apache/skywalking-banyandb/pkg/flow/streaming"
"github.com/apache/skywalking-banyandb/pkg/logger"
- pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
"github.com/apache/skywalking-banyandb/pkg/query/aggregation"
- "github.com/apache/skywalking-banyandb/pkg/timestamp"
+ "github.com/apache/skywalking-banyandb/pkg/query/executor"
+ logical_measure "github.com/apache/skywalking-banyandb/pkg/query/logical/measure"
)
type topNQueryProcessor struct {
@@ -50,10 +45,15 @@ type topNQueryProcessor struct {
func (t *topNQueryProcessor) Rev(message bus.Message) (resp bus.Message) {
request, ok := message.Data().(*measurev1.TopNRequest)
+ now := time.Now().UnixNano()
if !ok {
t.log.Warn().Msg("invalid event data type")
return
}
+ ml := t.log.Named("topn", request.Metadata.Group, request.Metadata.Name)
+ if e := ml.Debug(); e.Enabled() {
+ e.RawJSON("req", logger.Proto(request)).Msg("received a topn event")
+ }
if request.GetFieldValueSort() == modelv1.Sort_SORT_UNSPECIFIED {
t.log.Warn().Msg("invalid requested sort direction")
return
@@ -81,157 +81,68 @@ func (t *topNQueryProcessor) Rev(message bus.Message) (resp bus.Message) {
Msg("fail to find source measure")
return
}
- shards, err := sourceMeasure.CompanionShards(topNMetadata)
+
+ schema, err := t.metaService.MeasureRegistry().GetMeasure(context.TODO(), topNMetadata)
if err != nil {
t.log.Error().Err(err).
Str("topN", topNMetadata.GetName()).
- Msg("fail to list shards")
+ Msg("fail to find topN measure")
return
}
- aggregator := CreateTopNPostAggregator(request.GetTopN(),
- request.GetAgg(), request.GetFieldValueSort())
- entity, err := locateEntity(topNSchema, request.GetFieldValueSort(), request.GetConditions())
+
+ sourceMeasure.SetSchema(schema)
+ s, err := logical_measure.BuildTopNSchema(schema, topNSchema.GetGroupByTagNames())
if err != nil {
t.log.Error().Err(err).
Str("topN", topNMetadata.GetName()).
- Msg("fail to parse entity")
- return
+ Msg("fail to build schema")
}
- for _, shard := range shards {
- // TODO: support condition
- sl, innerErr := shard.Series().List(context.WithValue(
- context.Background(),
- logger.ContextKey,
- t.log,
- ), tsdb.NewPath(entity))
- if innerErr != nil {
- t.log.Error().Err(innerErr).
- Str("topN", topNMetadata.GetName()).
- Msg("fail to list series")
- return
- }
- for _, series := range sl {
- iters, scanErr := t.scanSeries(series, request)
- if scanErr != nil {
- t.log.Error().Err(innerErr).
- Str("topN", topNMetadata.GetName()).
- Msg("fail to scan series")
- return
- }
- if len(iters) < 1 {
- continue
- }
- for _, iter := range iters {
- for iter.Next() {
- tuple, parseErr := parseTopNFamily(iter.Val(), sourceMeasure.GetInterval())
- if parseErr != nil {
- t.log.Error().Err(parseErr).
- Str("topN", topNMetadata.GetName()).
- Msg("fail to parse topN family")
- return
- }
- _ = aggregator.Put(tuple.V1.([]*modelv1.TagValue), tuple.V2.(int64), iter.Val().Time())
- }
- _ = iter.Close()
- }
- }
+ plan, err := logical_measure.TopNAnalyze(context.TODO(), request, schema, topNSchema, s)
+ if err != nil {
+ resp = bus.NewMessage(bus.MessageID(now), common.NewError("fail to analyze the query request for topn %s: %v", topNMetadata.GetName(), err))
+ return
}
- now := time.Now().UnixNano()
- resp = bus.NewMessage(bus.MessageID(now), &measurev1.TopNResponse{Lists: aggregator.Val(sourceMeasure.GetSchema().GetEntity().GetTagNames())})
-
- return
-}
-
-func locateEntity(topNSchema *databasev1.TopNAggregation, sortDirection modelv1.Sort,
- conditions []*modelv1.Condition,
-) (tsdb.Entity, error) {
- entityMap := make(map[string]int)
- entity := make([]tsdb.Entry, 1+1+len(topNSchema.GetGroupByTagNames()))
- // sortDirection
- entity[0] = convert.Int64ToBytes(int64(sortDirection.Number()))
- // rankNumber
- entity[1] = tsdb.AnyEntry
- for idx, tagName := range topNSchema.GetGroupByTagNames() {
- entityMap[tagName] = idx + 2
- // allow to make fuzzy search with partial conditions
- entity[idx+2] = tsdb.AnyEntry
- }
- for _, pairQuery := range conditions {
- if pairQuery.GetOp() != modelv1.Condition_BINARY_OP_EQ {
- return nil, errors.New("op other than EQ is not supported")
- }
- if entityIdx, ok := entityMap[pairQuery.GetName()]; ok {
- switch v := pairQuery.GetValue().GetValue().(type) {
- case *modelv1.TagValue_Str:
- entity[entityIdx] = []byte(v.Str.GetValue())
- case *modelv1.TagValue_Int:
- entity[entityIdx] = convert.Int64ToBytes(v.Int.GetValue())
- default:
- return nil, errors.New("unsupported condition tag type for entity")
- }
- continue
- }
- return nil, errors.New("only groupBy tag name is supported")
+ if e := ml.Debug(); e.Enabled() {
+ e.Str("plan", plan.String()).Msg("topn plan")
}
- return entity, nil
-}
-
-func parseTopNFamily(item tsdb.Item, interval time.Duration) (*streaming.Tuple2, error) {
- familyRawBytes, err := item.Family(familyIdentity(measure.TopNTagFamily, pbv1.TagFlag))
- if err != nil {
- return nil, err
- }
- tagFamily := &modelv1.TagFamilyForWrite{}
- err = proto.Unmarshal(familyRawBytes, tagFamily)
- if err != nil {
- return nil, err
- }
- fieldBytes, err := item.Family(familyIdentity(measure.TopNValueFieldSpec.GetName(),
- pbv1.EncoderFieldFlag(measure.TopNValueFieldSpec, interval)))
+ mIterator, err := plan.(executor.MeasureExecutable).Execute(executor.WithMeasureExecutionContext(context.Background(), sourceMeasure))
if err != nil {
- return nil, err
+ ml.Error().Err(err).RawJSON("req", logger.Proto(request)).Msg("fail to close the topn plan")
+ resp = bus.NewMessage(bus.MessageID(now), common.NewError("fail to execute the topn plan for measure %s: %v", topNMetadata.GetName(), err))
}
- fieldValue, err := pbv1.DecodeFieldValue(fieldBytes, measure.TopNValueFieldSpec)
- if err != nil {
- return nil, err
+ defer func() {
+ if err = mIterator.Close(); err != nil {
+ ml.Error().Err(err).RawJSON("req", logger.Proto(request))
+ }
+ }()
+
+ result := make([]*measurev1.DataPoint, 0)
+ for mIterator.Next() {
+ current := mIterator.Current()
+ if len(current) > 0 {
+ result = append(result, current[0])
+ }
}
- return &streaming.Tuple2{
- // GroupValues
- V1: tagFamily.GetTags()[1:],
- // FieldValue
- V2: fieldValue.GetInt().GetValue(),
- }, nil
-}
-func familyIdentity(name string, flag []byte) []byte {
- return bytes.Join([][]byte{tsdb.Hash([]byte(name)), flag}, nil)
+ resp = bus.NewMessage(bus.MessageID(now), toTopNResponse(result))
+ return
}
-func (t *topNQueryProcessor) scanSeries(series tsdb.Series, request *measurev1.TopNRequest) ([]tsdb.Iterator, error) {
- ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
- defer cancel()
- seriesSpan, err := series.Span(ctx, timestamp.NewInclusiveTimeRange(
- request.GetTimeRange().GetBegin().AsTime(),
- request.GetTimeRange().GetEnd().AsTime()),
- )
- if err != nil {
- if errors.Is(err, tsdb.ErrEmptySeriesSpan) {
- return nil, nil
+func toTopNResponse(dps []*measurev1.DataPoint) *measurev1.TopNResponse {
+ topNList := make([]*measurev1.TopNList, 0)
+ topNItems := make([]*measurev1.TopNList_Item, len(dps))
+ for i, dp := range dps {
+ topNItems[i] = &measurev1.TopNList_Item{
+ Entity: dp.GetTagFamilies()[0].GetTags(),
+ Value: dp.GetFields()[0].GetValue(),
}
- return nil, err
}
- defer func(seriesSpan tsdb.SeriesSpan) {
- if seriesSpan != nil {
- _ = seriesSpan.Close()
- }
- }(seriesSpan)
- seeker, err := seriesSpan.SeekerBuilder().OrderByTime(modelv1.Sort_SORT_ASC).Build()
- if err != nil {
- return nil, err
- }
- return seeker.Seek()
+ topNList = append(topNList, &measurev1.TopNList{
+ Items: topNItems,
+ })
+ return &measurev1.TopNResponse{Lists: topNList}
}
var _ heap.Interface = (*postAggregationProcessor)(nil)
diff --git a/pkg/query/logical/measure/topn_analyzer.go b/pkg/query/logical/measure/topn_analyzer.go
new file mode 100644
index 00000000..2def6f73
--- /dev/null
+++ b/pkg/query/logical/measure/topn_analyzer.go
@@ -0,0 +1,94 @@
+// 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 implements execution operations for querying measure data.
+package measure
+
+import (
+ "context"
+
+ commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
+ databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+ measurev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/measure/v1"
+ "github.com/apache/skywalking-banyandb/pkg/query/logical"
+)
+
+// BuildTopNSchema returns Schema loaded from the metadata repository.
+func BuildTopNSchema(md *databasev1.Measure, entityList []string) (logical.Schema, error) {
+ md.GetEntity()
+
+ ms := &schema{
+ common: &logical.CommonSchema{
+ TagSpecMap: make(map[string]*logical.TagSpec),
+ EntityList: entityList,
+ },
+ measure: md,
+ fieldMap: make(map[string]*logical.FieldSpec),
+ }
+
+ ms.common.RegisterTagFamilies(md.GetTagFamilies())
+
+ for fieldIdx, spec := range md.GetFields() {
+ ms.registerField(fieldIdx, spec)
+ }
+
+ return ms, nil
+}
+
+// TopNAnalyze converts logical expressions to executable operation tree represented by Plan.
+func TopNAnalyze(_ context.Context, criteria *measurev1.TopNRequest, schema *databasev1.Measure,
+ topNAggregation *databasev1.TopNAggregation, s logical.Schema,
+) (logical.Plan, error) {
+ groupByProjectionTags := schema.GetEntity().GetTagNames()
+ groupByTags := make([][]*logical.Tag, len(schema.GetTagFamilies()))
+ tagFamily := schema.GetTagFamilies()[0]
+ groupByTags[0] = logical.NewTags(tagFamily.GetName(), groupByProjectionTags...)
+
+ projectionFields := make([]*logical.Field, len(schema.GetFields()))
+ for i, fieldSpecProj := range schema.GetFields() {
+ projectionFields[i] = logical.NewField(fieldSpecProj.GetName())
+ }
+ // parse fields
+ plan := parse(criteria, schema.GetMetadata(), projectionFields, groupByTags)
+
+ if criteria.GetAgg() != 0 {
+ plan = newUnresolvedGroupBy(plan, groupByTags, false)
+ plan = newUnresolvedAggregation(plan,
+ logical.NewField(topNAggregation.GetFieldName()),
+ criteria.GetAgg(),
+ true)
+ }
+
+ plan = top(plan, &measurev1.QueryRequest_Top{
+ Number: criteria.GetTopN(),
+ FieldName: topNAggregation.GetFieldName(),
+ FieldValueSort: criteria.GetFieldValueSort(),
+ })
+ p, err := plan.Analyze(s)
+ if err != nil {
+ return nil, err
+ }
+ return p, nil
+}
+
+func parse(criteria *measurev1.TopNRequest, metadata *commonv1.Metadata,
+ projFields []*logical.Field, projTags [][]*logical.Tag,
+) logical.UnresolvedPlan {
+ timeRange := criteria.GetTimeRange()
+ return local(timeRange.GetBegin().AsTime(), timeRange.GetEnd().AsTime(),
+ metadata, projTags, projFields, criteria.GetConditions(), criteria.GetFieldValueSort())
+}
diff --git a/pkg/query/logical/measure/topn_plan_localscan.go b/pkg/query/logical/measure/topn_plan_localscan.go
new file mode 100644
index 00000000..702dc57b
--- /dev/null
+++ b/pkg/query/logical/measure/topn_plan_localscan.go
@@ -0,0 +1,254 @@
+// 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 implements execution operations for querying measure data.
+package measure
+
+import (
+ "context"
+ "fmt"
+ "io"
+ "time"
+
+ "github.com/pkg/errors"
+ "go.uber.org/multierr"
+
+ commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
+ 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/banyand/measure"
+ "github.com/apache/skywalking-banyandb/banyand/tsdb"
+ "github.com/apache/skywalking-banyandb/pkg/convert"
+ "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"
+ "github.com/apache/skywalking-banyandb/pkg/timestamp"
+)
+
+var _ logical.UnresolvedPlan = (*unresolvedLocalScan)(nil)
+
+type unresolvedLocalScan struct {
+ startTime time.Time
+ endTime time.Time
+ metadata *commonv1.Metadata
+ conditions []*modelv1.Condition
+ projectionTags [][]*logical.Tag
+ projectionFields []*logical.Field
+ sort modelv1.Sort
+}
+
+func (uls *unresolvedLocalScan) Analyze(s logical.Schema) (logical.Plan, error) {
+ var projTagsRefs [][]*logical.TagRef
+ if len(uls.projectionTags) > 0 {
+ var err error
+ projTagsRefs, err = s.CreateTagRef(uls.projectionTags...)
+ if err != nil {
+ return nil, err
+ }
+ }
+
+ var projFieldRefs []*logical.FieldRef
+ if len(uls.projectionFields) > 0 {
+ var err error
+ projFieldRefs, err = s.CreateFieldRef(uls.projectionFields...)
+ if err != nil {
+ return nil, err
+ }
+ }
+
+ entity, err := uls.locateEntity(s.EntityList())
+ if err != nil {
+ return nil, err
+ }
+
+ return &localScan{
+ timeRange: timestamp.NewInclusiveTimeRange(uls.startTime, uls.endTime),
+ schema: s,
+ projectionTagsRefs: projTagsRefs,
+ projectionFieldsRefs: projFieldRefs,
+ metadata: uls.metadata,
+ entity: entity,
+ l: logger.GetLogger("topn", "measure", uls.metadata.Group, uls.metadata.Name, "local-index"),
+ }, nil
+}
+
+func (uls *unresolvedLocalScan) locateEntity(entityList []string) (tsdb.Entity, error) {
+ entityMap := make(map[string]int)
+ entity := make([]tsdb.Entry, 1+1+len(entityList))
+ // sortDirection
+ entity[0] = convert.Int64ToBytes(int64(uls.sort.Number()))
+ // rankNumber
+ entity[1] = tsdb.AnyEntry
+ for idx, tagName := range entityList {
+ entityMap[tagName] = idx + 2
+ // allow to make fuzzy search with partial conditions
+ entity[idx+2] = tsdb.AnyEntry
+ }
+ for _, pairQuery := range uls.conditions {
+ if pairQuery.GetOp() != modelv1.Condition_BINARY_OP_EQ {
+ return nil, errors.Errorf("tag belongs to the entity only supports EQ operation in condition(%v)", pairQuery)
+ }
+ if entityIdx, ok := entityMap[pairQuery.GetName()]; ok {
+ switch v := pairQuery.GetValue().GetValue().(type) {
+ case *modelv1.TagValue_Str:
+ entity[entityIdx] = []byte(v.Str.GetValue())
+ case *modelv1.TagValue_Int:
+ entity[entityIdx] = convert.Int64ToBytes(v.Int.GetValue())
+ default:
+ return nil, errors.New("unsupported condition tag type for entity")
+ }
+ continue
+ }
+ return nil, errors.New("only groupBy tag name is supported")
+ }
+
+ return entity, nil
+}
+
+func local(startTime, endTime time.Time, metadata *commonv1.Metadata, projectionTags [][]*logical.Tag,
+ projectionFields []*logical.Field, conditions []*modelv1.Condition, sort modelv1.Sort,
+) logical.UnresolvedPlan {
+ return &unresolvedLocalScan{
+ startTime: startTime,
+ endTime: endTime,
+ metadata: metadata,
+ projectionTags: projectionTags,
+ projectionFields: projectionFields,
+ conditions: conditions,
+ sort: sort,
+ }
+}
+
+var _ logical.Plan = (*localScan)(nil)
+
+type localScan struct {
+ schema logical.Schema
+ metadata *commonv1.Metadata
+ l *logger.Logger
+ timeRange timestamp.TimeRange
+ projectionTagsRefs [][]*logical.TagRef
+ projectionFieldsRefs []*logical.FieldRef
+ entity tsdb.Entity
+ sort modelv1.Sort
+}
+
+func (i *localScan) Execute(ctx context.Context) (mit executor.MIterator, err error) {
+ var seriesList tsdb.SeriesList
+ ec := executor.FromMeasureExecutionContext(ctx)
+ shards, err := ec.(measure.Measure).CompanionShards(i.metadata)
+ if err != nil {
+ return nil, err
+ }
+ for _, shard := range shards {
+ sl, errInternal := shard.Series().List(context.WithValue(
+ ctx,
+ logger.ContextKey,
+ i.l,
+ ), tsdb.NewPath(i.entity))
+ if errInternal != nil {
+ return nil, errInternal
+ }
+ seriesList = seriesList.Merge(sl)
+ }
+ if len(seriesList) == 0 {
+ return dummyIter, nil
+ }
+ var builders []logical.SeekerBuilder
+ builders = append(builders, func(builder tsdb.SeekerBuilder) {
+ builder.OrderByTime(i.sort)
+ })
+ iters, closers, err := logical.ExecuteForShard(ctx, i.l, seriesList, i.timeRange, builders...)
+ if err != nil {
+ return nil, err
+ }
+ if len(closers) > 0 {
+ defer func(closers []io.Closer) {
+ for _, c := range closers {
+ err = multierr.Append(err, c.Close())
+ }
+ }(closers)
+ }
+
+ if len(iters) == 0 {
+ return dummyIter, nil
+ }
+ tc := transformContext{
+ ec: ec,
+ projectionTagsRefs: i.projectionTagsRefs,
+ projectionFieldsRefs: i.projectionFieldsRefs,
+ }
+ it := logical.NewItemIter(iters, i.sort)
+
+ return newLocalScanIterator(it, tc), nil
+}
+
+func (i *localScan) String() string {
+ return fmt.Sprintf("IndexScan: startTime=%d,endTime=%d,Metadata{group=%s,name=%s}; projection=%s; sort=%s;",
+ i.timeRange.Start.Unix(), i.timeRange.End.Unix(), i.metadata.GetGroup(), i.metadata.GetName(),
+ logical.FormatTagRefs(", ", i.projectionTagsRefs...), i.sort)
+}
+
+func (i *localScan) Children() []logical.Plan {
+ return []logical.Plan{}
+}
+
+func (i *localScan) Schema() logical.Schema {
+ if len(i.projectionTagsRefs) == 0 {
+ return i.schema
+ }
+ return i.schema.ProjTags(i.projectionTagsRefs...).ProjFields(i.projectionFieldsRefs...)
+}
+
+type localScanIterator struct {
+ inner sort.Iterator[tsdb.Item]
+ err error
+ current *measurev1.DataPoint
+ context transformContext
+ num int
+}
+
+func (lst *localScanIterator) Next() bool {
+ if !lst.inner.Next() || lst.err != nil {
+ return false
+ }
+ nextItem := lst.inner.Val()
+ var err error
+ if lst.current, err = transform(nextItem, lst.context); err != nil {
+ lst.err = multierr.Append(lst.err, err)
+ }
+ lst.num++
+ return true
+}
+
+func (lst *localScanIterator) Current() []*measurev1.DataPoint {
+ if lst.current == nil {
+ return nil
+ }
+ return []*measurev1.DataPoint{lst.current}
+}
+
+func (lst *localScanIterator) Close() error {
+ return multierr.Combine(lst.err, lst.inner.Close())
+}
+
+func newLocalScanIterator(inner sort.Iterator[tsdb.Item], context transformContext) executor.MIterator {
+ return &localScanIterator{
+ inner: inner,
+ context: context,
+ }
+}
diff --git a/test/cases/topn/data/input/asc.yaml b/test/cases/topn/data/input/asc.yaml
deleted file mode 100644
index 9a8732df..00000000
--- a/test/cases/topn/data/input/asc.yaml
+++ /dev/null
@@ -1,22 +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.
-
-metadata:
- name: "service_instance_cpm_minute_top_bottom_100"
- group: "sw_metric"
-topN: 2
-fieldValueSort: 2
diff --git a/test/cases/topn/data/input/desc.yaml b/test/cases/topn/data/input/desc.yaml
deleted file mode 100644
index 63a4d565..00000000
--- a/test/cases/topn/data/input/desc.yaml
+++ /dev/null
@@ -1,22 +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.
-
-metadata:
- name: "service_instance_cpm_minute_top_bottom_100"
- group: "sw_metric"
-topN: 2
-fieldValueSort: 1
diff --git a/test/cases/topn/data/want/aggr_desc.yaml b/test/cases/topn/data/want/aggr_desc.yaml
index 96cc77b2..b39f77e7 100644
--- a/test/cases/topn/data/want/aggr_desc.yaml
+++ b/test/cases/topn/data/want/aggr_desc.yaml
@@ -52,5 +52,4 @@ lists:
value: entity_2
value:
int:
- value: "10"
- timestamp: "2022-10-17T12:54:00Z"
\ No newline at end of file
+ value: "10"
\ No newline at end of file
diff --git a/test/cases/topn/data/want/asc.yaml b/test/cases/topn/data/want/asc.yaml
deleted file mode 100644
index 2731344d..00000000
--- a/test/cases/topn/data/want/asc.yaml
+++ /dev/null
@@ -1,174 +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.
-
-lists:
-- items:
- - entity:
- - key: service_id
- value:
- str:
- value: svc_1
- - key: entity_id
- value:
- str:
- value: entity_1
- value:
- int:
- value: "1"
- - entity:
- - key: service_id
- value:
- str:
- value: ""
- - key: entity_id
- value:
- str:
- value: entity_2
- value:
- int:
- value: "10"
- timestamp: "2022-10-17T12:49:00Z"
-- items:
- - entity:
- - key: service_id
- value:
- str:
- value: svc_2
- - key: entity_id
- value:
- str:
- value: entity_2
- value:
- int:
- value: "2"
- - entity:
- - key: service_id
- value:
- str:
- value: svc_1
- - key: entity_id
- value:
- str:
- value: entity_2
- value:
- int:
- value: "9"
- timestamp: "2022-10-17T12:50:00Z"
-- items:
- - entity:
- - key: service_id
- value:
- str:
- value: svc_1
- - key: entity_id
- value:
- str:
- value: entity_2
- value:
- int:
- value: "3"
- - entity:
- - key: service_id
- value:
- str:
- value: svc_4
- - key: entity_id
- value:
- str:
- value: entity_2
- value:
- int:
- value: "8"
- timestamp: "2022-10-17T12:51:00Z"
-- items:
- - entity:
- - key: service_id
- value:
- str:
- value: svc_2
- - key: entity_id
- value:
- str:
- value: entity_3
- value:
- int:
- value: "5"
- - entity:
- - key: service_id
- value:
- str:
- value: svc_1
- - key: entity_id
- value:
- str:
- value: entity_3
- value:
- int:
- value: "11"
- timestamp: "2022-10-17T12:52:00Z"
-- items:
- - entity:
- - key: service_id
- value:
- str:
- value: ""
- - key: entity_id
- value:
- str:
- value: entity_1
- value:
- int:
- value: "4"
- - entity:
- - key: service_id
- value:
- str:
- value: svc_2
- - key: entity_id
- value:
- str:
- value: entity_1
- value:
- int:
- value: "12"
- timestamp: "2022-10-17T12:53:00Z"
-- items:
- - entity:
- - key: service_id
- value:
- str:
- value: svc_1
- - key: entity_id
- value:
- str:
- value: entity_1
- value:
- int:
- value: "6"
- - entity:
- - key: service_id
- value:
- str:
- value: ""
- - key: entity_id
- value:
- str:
- value: entity_1
- value:
- int:
- value: "7"
- timestamp: "2022-10-17T12:54:00Z"
\ No newline at end of file
diff --git a/test/cases/topn/data/want/condition_aggr_desc.yaml b/test/cases/topn/data/want/condition_aggr_desc.yaml
index ebbb77f3..fdd710a8 100644
--- a/test/cases/topn/data/want/condition_aggr_desc.yaml
+++ b/test/cases/topn/data/want/condition_aggr_desc.yaml
@@ -28,5 +28,4 @@ lists:
value: entity_3
value:
int:
- value: "11"
- timestamp: "2022-10-17T12:54:00Z"
\ No newline at end of file
+ value: "11"
\ No newline at end of file
diff --git a/test/cases/topn/data/want/desc.yaml b/test/cases/topn/data/want/desc.yaml
deleted file mode 100644
index a937323c..00000000
--- a/test/cases/topn/data/want/desc.yaml
+++ /dev/null
@@ -1,174 +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.
-
-lists:
-- items:
- - entity:
- - key: service_id
- value:
- str:
- value: ""
- - key: entity_id
- value:
- str:
- value: entity_2
- value:
- int:
- value: "10"
- - entity:
- - key: service_id
- value:
- str:
- value: svc_1
- - key: entity_id
- value:
- str:
- value: entity_1
- value:
- int:
- value: "1"
- timestamp: "2022-10-17T12:49:00Z"
-- items:
- - entity:
- - key: service_id
- value:
- str:
- value: svc_1
- - key: entity_id
- value:
- str:
- value: entity_2
- value:
- int:
- value: "9"
- - entity:
- - key: service_id
- value:
- str:
- value: svc_2
- - key: entity_id
- value:
- str:
- value: entity_2
- value:
- int:
- value: "2"
- timestamp: "2022-10-17T12:50:00Z"
-- items:
- - entity:
- - key: service_id
- value:
- str:
- value: svc_4
- - key: entity_id
- value:
- str:
- value: entity_2
- value:
- int:
- value: "8"
- - entity:
- - key: service_id
- value:
- str:
- value: svc_1
- - key: entity_id
- value:
- str:
- value: entity_2
- value:
- int:
- value: "3"
- timestamp: "2022-10-17T12:51:00Z"
-- items:
- - entity:
- - key: service_id
- value:
- str:
- value: svc_1
- - key: entity_id
- value:
- str:
- value: entity_3
- value:
- int:
- value: "11"
- - entity:
- - key: service_id
- value:
- str:
- value: svc_2
- - key: entity_id
- value:
- str:
- value: entity_3
- value:
- int:
- value: "5"
- timestamp: "2022-10-17T12:52:00Z"
-- items:
- - entity:
- - key: service_id
- value:
- str:
- value: svc_2
- - key: entity_id
- value:
- str:
- value: entity_1
- value:
- int:
- value: "12"
- - entity:
- - key: service_id
- value:
- str:
- value: ""
- - key: entity_id
- value:
- str:
- value: entity_1
- value:
- int:
- value: "4"
- timestamp: "2022-10-17T12:53:00Z"
-- items:
- - entity:
- - key: service_id
- value:
- str:
- value: ""
- - key: entity_id
- value:
- str:
- value: entity_1
- value:
- int:
- value: "7"
- - entity:
- - key: service_id
- value:
- str:
- value: svc_1
- - key: entity_id
- value:
- str:
- value: entity_1
- value:
- int:
- value: "6"
- timestamp: "2022-10-17T12:54:00Z"
\ No newline at end of file
diff --git a/test/cases/topn/data/want/null_group.yaml b/test/cases/topn/data/want/null_group.yaml
index c9841a7d..b39f77e7 100644
--- a/test/cases/topn/data/want/null_group.yaml
+++ b/test/cases/topn/data/want/null_group.yaml
@@ -52,5 +52,4 @@ lists:
value: entity_2
value:
int:
- value: "10"
- timestamp: "2023-03-18T13:19:00Z"
\ No newline at end of file
+ value: "10"
\ No newline at end of file
diff --git a/test/cases/topn/topn.go b/test/cases/topn/topn.go
index 48f6d54e..e1fc8c39 100644
--- a/test/cases/topn/topn.go
+++ b/test/cases/topn/topn.go
@@ -40,8 +40,6 @@ var (
)
var _ = g.DescribeTable("TopN Tests", verify,
- g.Entry("desc", helpers.Args{Input: "desc", Duration: 25 * time.Minute, Offset: -20 * time.Minute}),
- g.Entry("asc", helpers.Args{Input: "asc", Duration: 25 * time.Minute, Offset: -20 * time.Minute}),
g.Entry("max top3 order by desc", helpers.Args{Input: "aggr_desc", Duration: 25 * time.Minute, Offset: -20 * time.Minute}),
g.Entry("max top3 with condition order by desc", helpers.Args{Input: "condition_aggr_desc", Duration: 25 * time.Minute, Offset: -20 * time.Minute}),
g.Entry("max top3 for null group order by desc", helpers.Args{Input: "null_group", Duration: 25 * time.Minute, Offset: -20 * time.Minute}),