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 2025/05/13 23:31:55 UTC
(skywalking-banyandb) 01/02: Add multi-groups
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch multi-group
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit 7fbdf47d4ca57fbf26ff18d7a36f74c16eff5043
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Tue May 13 22:02:39 2025 +0800
Add multi-groups
Signed-off-by: Gao Hongtao <ha...@gmail.com>
---
banyand/dquery/measure.go | 42 ++--
banyand/dquery/stream.go | 42 ++--
banyand/query/processor.go | 95 ++++-----
banyand/query/processor_topn.go | 84 ++++----
banyand/stream/query_by_ts.go | 3 +
docs/interacting/bydbctl/property.md | 6 +-
pkg/index/inverted/query.go | 4 +
pkg/query/executor/interface.go | 30 ---
pkg/query/logical/measure/measure_analyzer.go | 55 ++++--
.../measure/measure_plan_indexscan_local.go | 14 +-
pkg/query/logical/measure/measure_plan_merge.go | 193 +++++++++++++++++++
pkg/query/logical/measure/schema.go | 66 ++++++-
pkg/query/logical/measure/topn_analyzer.go | 79 +++++++-
pkg/query/logical/measure/topn_plan_localscan.go | 6 +-
pkg/query/logical/measure/topn_plan_merge.go | 141 ++++++++++++++
pkg/query/logical/schema.go | 102 +++++++++-
pkg/query/logical/stream/schema.go | 50 +++--
pkg/query/logical/stream/stream_analyzer.go | 48 +++--
.../logical/stream/stream_plan_indexscan_local.go | 4 +-
pkg/query/logical/stream/stream_plan_merge.go | 169 ++++++++++++++++
pkg/query/logical/stream/stream_plan_tag_filter.go | 10 +-
.../testdata/group_stages/updated.json} | 8 +-
.../testdata/groups/updated.json} | 6 +-
.../measures/service_cpm_minute_updated.json | 42 ++++
.../service_instance_cpm_minute_updated.json | 52 +++++
...service_instance_cpm_minute_top_bottom_100.json | 17 ++
pkg/test/stream/etcd.go | 50 +++--
pkg/test/stream/testdata/group.json | 21 +-
pkg/test/stream/testdata/group_with_stages.json | 95 ++++++---
.../testdata/index_rule_bindings/updated.json | 16 ++
.../testdata/index_rules/duration_updated.json | 12 ++
pkg/test/stream/testdata/streams/sw_updated.json | 90 +++++++++
test/cases/init.go | 2 +
.../data/input/multi_group_new_tag_field.yaml | 25 +++
.../measure/data/input/multi_group_unchanged.yaml | 25 +++
.../testdata/service_cpm_minute_updated_data.json | 92 +++++++++
.../data/want/multi_group_new_tag_field.yaml | 213 +++++++++++++++++++++
.../measure/data/want/multi_group_unchanged.yaml | 153 +++++++++++++++
test/cases/measure/measure.go | 2 +
test/cases/stream/data/data.go | 9 +-
.../stream/data/input/multi_group_new_tag.yaml | 23 +++
.../data/input/multi_group_sort_duration.yaml | 26 +++
.../stream/data/input/multi_group_tag_type.yaml | 23 +++
.../stream/data/input/multi_group_unchanged.yaml | 23 +++
test/cases/stream/data/testdata/sw_updated.json | 59 ++++++
.../stream/data/want/multi_group_new_tag.yaml | 149 ++++++++++++++
.../data/want/multi_group_sort_duration.yaml | 122 ++++++++++++
.../stream/data/want/multi_group_tag_type.yaml | 152 +++++++++++++++
.../stream/data/want/multi_group_unchanged.yaml | 141 ++++++++++++++
test/cases/stream/stream.go | 4 +
.../topn/data/input/multi_group_aggr_desc.yaml | 22 +++
test/cases/topn/topn.go | 1 +
52 files changed, 2635 insertions(+), 283 deletions(-)
diff --git a/banyand/dquery/measure.go b/banyand/dquery/measure.go
index 1b1a62e0..63ce9464 100644
--- a/banyand/dquery/measure.go
+++ b/banyand/dquery/measure.go
@@ -31,6 +31,7 @@ import (
"github.com/apache/skywalking-banyandb/pkg/logger"
"github.com/apache/skywalking-banyandb/pkg/query"
"github.com/apache/skywalking-banyandb/pkg/query/executor"
+ "github.com/apache/skywalking-banyandb/pkg/query/logical"
logical_measure "github.com/apache/skywalking-banyandb/pkg/query/logical/measure"
)
@@ -49,34 +50,33 @@ func (p *measureQueryProcessor) Rev(ctx context.Context, message bus.Message) (r
resp = bus.NewMessage(bus.MessageID(now), common.NewError("invalid event data type"))
return
}
- if len(queryCriteria.Groups) > 1 {
- resp = bus.NewMessage(bus.MessageID(now), common.NewError("only support one group in the query request"))
- return
- }
ml := p.log.Named("measure", queryCriteria.Groups[0], queryCriteria.Name)
if e := ml.Debug(); e.Enabled() {
e.RawJSON("req", logger.Proto(queryCriteria)).Msg("received a query event")
}
- meta := &commonv1.Metadata{
- Name: queryCriteria.Name,
- Group: queryCriteria.Groups[0],
- }
- 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
+ var schemas []logical.Schema
+ for _, g := range queryCriteria.Groups {
+ meta := &commonv1.Metadata{
+ Name: queryCriteria.Name,
+ Group: g,
+ }
+ 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
+ }
+ schemas = append(schemas, s)
}
- plan, err := logical_measure.DistributedAnalyze(queryCriteria, s)
+ plan, err := logical_measure.DistributedAnalyze(queryCriteria, schemas)
if err != nil {
- resp = bus.NewMessage(bus.MessageID(now), common.NewError("fail to analyze the query request for measure %s: %v", meta.GetName(), err))
+ resp = bus.NewMessage(bus.MessageID(now), common.NewError("fail to analyze the query request for measure %s: %v", queryCriteria.Name, err))
return
}
@@ -133,7 +133,7 @@ func (p *measureQueryProcessor) Rev(ctx context.Context, message bus.Message) (r
}))
if err != nil {
ml.Error().Err(err).Dur("latency", time.Since(n)).RawJSON("req", logger.Proto(queryCriteria)).Msg("fail to query")
- resp = bus.NewMessage(bus.MessageID(now), common.NewError("fail to execute the query plan for measure %s: %v", meta.GetName(), err))
+ resp = bus.NewMessage(bus.MessageID(now), common.NewError("fail to execute the query plan for measure %s: %v", queryCriteria.Name, err))
return
}
defer func() {
diff --git a/banyand/dquery/stream.go b/banyand/dquery/stream.go
index 184e3438..98feb553 100644
--- a/banyand/dquery/stream.go
+++ b/banyand/dquery/stream.go
@@ -30,6 +30,7 @@ import (
"github.com/apache/skywalking-banyandb/pkg/logger"
"github.com/apache/skywalking-banyandb/pkg/query"
"github.com/apache/skywalking-banyandb/pkg/query/executor"
+ "github.com/apache/skywalking-banyandb/pkg/query/logical"
logical_stream "github.com/apache/skywalking-banyandb/pkg/query/logical/stream"
)
@@ -51,28 +52,29 @@ func (p *streamQueryProcessor) Rev(ctx context.Context, message bus.Message) (re
if p.log.Debug().Enabled() {
p.log.Debug().RawJSON("criteria", logger.Proto(queryCriteria)).Msg("received a query request")
}
- if len(queryCriteria.Groups) > 1 {
- resp = bus.NewMessage(bus.MessageID(now), common.NewError("only support one group in the query request"))
- return
- }
- meta := &commonv1.Metadata{
- Name: queryCriteria.Name,
- Group: queryCriteria.Groups[0],
- }
- 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
+
+ var schemas []logical.Schema
+ for _, group := range queryCriteria.Groups {
+ meta := &commonv1.Metadata{
+ Name: queryCriteria.Name,
+ Group: group,
+ }
+ 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
+ }
+ schemas = append(schemas, s)
}
- plan, err := logical_stream.DistributedAnalyze(queryCriteria, s)
+ plan, err := logical_stream.DistributedAnalyze(queryCriteria, schemas)
if err != nil {
- resp = bus.NewMessage(bus.MessageID(now), common.NewError("fail to analyze the query request for stream %s: %v", meta.GetName(), err))
+ resp = bus.NewMessage(bus.MessageID(now), common.NewError("fail to analyze the query request for stream %s: %v", queryCriteria.Name, err))
return
}
@@ -130,7 +132,7 @@ func (p *streamQueryProcessor) Rev(ctx context.Context, message bus.Message) (re
}))
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))
+ resp = bus.NewMessage(bus.MessageID(now), common.NewError("execute the query plan for stream %s: %v", queryCriteria.Name, err))
return
}
diff --git a/banyand/query/processor.go b/banyand/query/processor.go
index e1b6864a..7610997a 100644
--- a/banyand/query/processor.go
+++ b/banyand/query/processor.go
@@ -34,6 +34,7 @@ import (
"github.com/apache/skywalking-banyandb/pkg/logger"
"github.com/apache/skywalking-banyandb/pkg/query"
"github.com/apache/skywalking-banyandb/pkg/query/executor"
+ "github.com/apache/skywalking-banyandb/pkg/query/logical"
logical_measure "github.com/apache/skywalking-banyandb/pkg/query/logical/measure"
logical_stream "github.com/apache/skywalking-banyandb/pkg/query/logical/stream"
"github.com/apache/skywalking-banyandb/pkg/run"
@@ -73,28 +74,32 @@ func (p *streamQueryProcessor) Rev(ctx context.Context, message bus.Message) (re
resp = bus.NewMessage(bus.MessageID(time.Now().UnixNano()), common.NewError("panic"))
}
}()
- if len(queryCriteria.Groups) > 1 {
- resp = bus.NewMessage(bus.MessageID(now), common.NewError("only support one group in the query request"))
- return
- }
- meta := &commonv1.Metadata{
- Name: queryCriteria.Name,
- Group: queryCriteria.Groups[0],
- }
- 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
+ var metadata []*commonv1.Metadata
+ var schemas []logical.Schema
+ var ecc []executor.StreamExecutionContext
+ for i := range queryCriteria.Groups {
+ meta := &commonv1.Metadata{
+ Name: queryCriteria.Name,
+ Group: queryCriteria.Groups[i],
+ }
+ 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
+ }
+ ecc = append(ecc, ec)
+ 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
+ }
+ schemas = append(schemas, s)
+ metadata = append(metadata, meta)
}
- plan, err := logical_stream.Analyze(ctx, queryCriteria, meta, s)
+ plan, err := logical_stream.Analyze(queryCriteria, metadata, schemas, ecc)
if err != nil {
- resp = bus.NewMessage(bus.MessageID(now), common.NewError("fail to analyze the query request for stream %s: %v", meta.GetName(), err))
+ resp = bus.NewMessage(bus.MessageID(now), common.NewError("fail to analyze the query request for stream %s: %v", queryCriteria.GetName(), err))
return
}
@@ -123,10 +128,10 @@ func (p *streamQueryProcessor) Rev(ctx context.Context, message bus.Message) (re
}
se := plan.(executor.StreamExecutable)
defer se.Close()
- entities, err := se.Execute(executor.WithStreamExecutionContext(ctx, ec))
+ entities, err := se.Execute(ctx)
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))
+ resp = bus.NewMessage(bus.MessageID(now), common.NewError("execute the query plan for stream %s: %v", queryCriteria.GetName(), err))
return
}
@@ -161,34 +166,36 @@ func (p *measureQueryProcessor) Rev(ctx context.Context, message bus.Message) (r
resp = bus.NewMessage(bus.MessageID(time.Now().UnixNano()), common.NewError("panic"))
}
}()
- if len(queryCriteria.Groups) > 1 {
- resp = bus.NewMessage(bus.MessageID(now), common.NewError("only support one group in the query request"))
- return
+ var metadata []*commonv1.Metadata
+ var schemas []logical.Schema
+ var ecc []executor.MeasureExecutionContext
+ for i := range queryCriteria.Groups {
+ meta := &commonv1.Metadata{
+ Name: queryCriteria.Name,
+ Group: queryCriteria.Groups[i],
+ }
+ 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
+ }
+ ecc = append(ecc, ec)
+ schemas = append(schemas, s)
+ metadata = append(metadata, meta)
}
ml := p.log.Named("measure", queryCriteria.Groups[0], queryCriteria.Name)
if e := ml.Debug(); e.Enabled() {
e.RawJSON("req", logger.Proto(queryCriteria)).Msg("received a query event")
}
- meta := &commonv1.Metadata{
- Name: queryCriteria.Name,
- Group: queryCriteria.Groups[0],
- }
- 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.Analyze(ctx, queryCriteria, meta, s)
+ plan, err := logical_measure.Analyze(queryCriteria, metadata, schemas, ecc)
if err != nil {
- resp = bus.NewMessage(bus.MessageID(now), common.NewError("fail to analyze the query request for measure %s: %v", meta.GetName(), err))
+ resp = bus.NewMessage(bus.MessageID(now), common.NewError("fail to analyze the query request for measure %s: %v", queryCriteria.GetName(), err))
return
}
var tracer *query.Tracer
@@ -216,10 +223,10 @@ func (p *measureQueryProcessor) Rev(ctx context.Context, message bus.Message) (r
e.Str("plan", plan.String()).Msg("query plan")
}
- mIterator, err := plan.(executor.MeasureExecutable).Execute(executor.WithMeasureExecutionContext(ctx, ec))
+ mIterator, err := plan.(executor.MeasureExecutable).Execute(ctx)
if err != nil {
ml.Error().Err(err).RawJSON("req", logger.Proto(queryCriteria)).Msg("fail to query")
- resp = bus.NewMessage(bus.MessageID(now), common.NewError("fail to execute the query plan for measure %s: %v", meta.GetName(), err))
+ resp = bus.NewMessage(bus.MessageID(now), common.NewError("fail to execute the query plan for measure %s: %v", queryCriteria.GetName(), err))
return
}
defer func() {
diff --git a/banyand/query/processor_topn.go b/banyand/query/processor_topn.go
index d765391a..f3246c1f 100644
--- a/banyand/query/processor_topn.go
+++ b/banyand/query/processor_topn.go
@@ -23,12 +23,14 @@ import (
"errors"
"fmt"
"slices"
+ "strings"
"time"
"google.golang.org/protobuf/types/known/timestamppb"
"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"
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"
@@ -56,13 +58,9 @@ func (t *topNQueryProcessor) Rev(ctx context.Context, message bus.Message) (resp
t.log.Warn().Msg("invalid event data type")
return
}
- if len(request.Groups) > 1 {
- resp = bus.NewMessage(bus.MessageID(now), common.NewError("only support one group in the query request"))
- return
- }
- ml := t.log.Named("topn", request.Groups[0], request.Name)
+ ml := t.log.Named("topn", strings.Join(request.Groups, ","), request.Name)
if e := ml.Debug(); e.Enabled() {
- e.RawJSON("req", logger.Proto(request)).Msg("received a topn event")
+ e.RawJSON("req", logger.Proto(request)).Msg("received a topn event for groups: " + strings.Join(request.Groups, ","))
}
if request.GetFieldValueSort() == modelv1.Sort_SORT_UNSPECIFIED {
t.log.Warn().Msg("invalid requested sort direction")
@@ -71,44 +69,56 @@ func (t *topNQueryProcessor) Rev(ctx context.Context, message bus.Message) (resp
if e := t.log.Debug(); e.Enabled() {
e.Stringer("req", request).Msg("received a topN query event")
}
- topNMetadata := &commonv1.Metadata{
- Name: request.Name,
- Group: request.Groups[0],
- }
- topNSchema, err := t.metaService.TopNAggregationRegistry().GetTopNAggregation(ctx, topNMetadata)
- if err != nil {
- t.log.Error().Err(err).
- Str("topN", topNMetadata.GetName()).
- Msg("fail to get execution context")
- return
- }
- if topNSchema.GetFieldValueSort() != modelv1.Sort_SORT_UNSPECIFIED &&
- topNSchema.GetFieldValueSort() != request.GetFieldValueSort() {
- t.log.Warn().Msg("unmatched sort direction")
- return
- }
- sourceMeasure, err := t.measureService.Measure(topNSchema.GetSourceMeasure())
- if err != nil {
- t.log.Error().Err(err).
- Str("topN", topNMetadata.GetName()).
- Msg("fail to find source measure")
- return
+ // Process all groups
+ var sourceMeasureSchemas []*databasev1.Measure
+ var topNSchemas []*databasev1.TopNAggregation
+ var ecc []executor.MeasureExecutionContext
+
+ for _, group := range request.Groups {
+ topNMetadata := &commonv1.Metadata{
+ Name: request.Name,
+ Group: group,
+ }
+ topNSchema, err := t.metaService.TopNAggregationRegistry().GetTopNAggregation(ctx, topNMetadata)
+ if err != nil {
+ t.log.Error().Err(err).
+ Str("group", group).
+ Msg("fail to get execution context")
+ return
+ }
+ if topNSchema.GetFieldValueSort() != modelv1.Sort_SORT_UNSPECIFIED &&
+ topNSchema.GetFieldValueSort() != request.GetFieldValueSort() {
+ t.log.Warn().Str("group", group).Msg("unmatched sort direction")
+ return
+ }
+ sourceMeasure, err := t.measureService.Measure(topNSchema.GetSourceMeasure())
+ if err != nil {
+ t.log.Error().Err(err).
+ Str("group", group).
+ Msg("fail to find source measure")
+ return
+ }
+ topNResultMeasure, err := t.measureService.Measure(measure.GetTopNSchemaMetadata(group))
+ if err != nil {
+ ml.Error().Err(err).Str("group", group).Msg("fail to find topn result measure")
+ return
+ }
+
+ sourceMeasureSchemas = append(sourceMeasureSchemas, sourceMeasure.GetSchema())
+ topNSchemas = append(topNSchemas, topNSchema)
+ ecc = append(ecc, topNResultMeasure)
}
- plan, err := logical_measure.TopNAnalyze(request, sourceMeasure.GetSchema(), topNSchema)
+ plan, err := logical_measure.TopNAnalyze(request, sourceMeasureSchemas, topNSchemas, ecc)
if err != nil {
- resp = bus.NewMessage(bus.MessageID(now), common.NewError("fail to analyze the query request for topn %s: %v", topNMetadata.GetName(), err))
+ resp = bus.NewMessage(bus.MessageID(now), common.NewError("fail to analyze the query request for topn %s: %v", request.Name, err))
return
}
if e := ml.Debug(); e.Enabled() {
e.Str("plan", plan.String()).Msg("topn plan")
}
- topNResultMeasure, err := t.measureService.Measure(measure.GetTopNSchemaMetadata(topNMetadata.Group))
- if err != nil {
- ml.Error().Err(err).Str("topN", topNMetadata.GetName()).Msg("fail to find topn result measure")
- return
- }
+
var tracer *query.Tracer
var span *query.Span
if request.Trace {
@@ -129,10 +139,10 @@ func (t *topNQueryProcessor) Rev(ctx context.Context, message bus.Message) (resp
span.Stop()
}()
}
- mIterator, err := plan.(executor.MeasureExecutable).Execute(executor.WithMeasureExecutionContext(ctx, topNResultMeasure))
+ mIterator, err := plan.(executor.MeasureExecutable).Execute(ctx)
if err != nil {
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))
+ resp = bus.NewMessage(bus.MessageID(now), common.NewError("fail to execute the topn plan for measure %s: %v", request.Name, err))
return
}
defer func() {
diff --git a/banyand/stream/query_by_ts.go b/banyand/stream/query_by_ts.go
index f1dfa7e1..5d5289d8 100644
--- a/banyand/stream/query_by_ts.go
+++ b/banyand/stream/query_by_ts.go
@@ -172,6 +172,9 @@ func loadBlockCursor(bc *blockCursor, tmpBlock *block, qo queryOptions, sm *stre
for _, tagFamilyProj := range bc.tagProjection {
for j, tagProj := range tagFamilyProj.Names {
tagSpec := is.tagMap[tagProj]
+ if tagSpec == nil {
+ continue
+ }
if tagSpec.IndexedOnly {
continue
}
diff --git a/docs/interacting/bydbctl/property.md b/docs/interacting/bydbctl/property.md
index a6e5a2de..f0fd729a 100644
--- a/docs/interacting/bydbctl/property.md
+++ b/docs/interacting/bydbctl/property.md
@@ -31,7 +31,7 @@ Then, below command will create a new property:
```shell
bydbctl property apply -f - <<EOF
metadata:
- container:
+ metadata:
group: sw
name: temp_data
id: General-Service
@@ -52,7 +52,7 @@ The operation supports updating partial tags.
```shell
bydbctl property apply -f - <<EOF
metadata:
- container:
+ metadata:
group: sw
name: temp_data
id: General-Service
@@ -93,7 +93,7 @@ Query operation queries all properties in a group with a container name.
```shell
bydbctl property query -f - <<EOF
groups: ["sw"]
-container: temp_data
+name: temp_data
EOF
```
diff --git a/pkg/index/inverted/query.go b/pkg/index/inverted/query.go
index 9ff1fc56..9812d253 100644
--- a/pkg/index/inverted/query.go
+++ b/pkg/index/inverted/query.go
@@ -729,3 +729,7 @@ func (p *schema) ProjFields(...*logical.FieldRef) logical.Schema {
func (p *schema) ProjTags(...[]*logical.TagRef) logical.Schema {
panic("unimplemented")
}
+
+func (p *schema) Children() []logical.Schema {
+ return nil
+}
diff --git a/pkg/query/executor/interface.go b/pkg/query/executor/interface.go
index d48675f6..14323577 100644
--- a/pkg/query/executor/interface.go
+++ b/pkg/query/executor/interface.go
@@ -33,21 +33,6 @@ type StreamExecutionContext interface {
Query(ctx context.Context, opts model.StreamQueryOptions) (model.StreamQueryResult, error)
}
-// StreamExecutionContextKey is the key of stream execution context in context.Context.
-type StreamExecutionContextKey struct{}
-
-var streamExecutionContextKeyInstance = StreamExecutionContextKey{}
-
-// WithStreamExecutionContext returns a new context with stream execution context.
-func WithStreamExecutionContext(ctx context.Context, ec StreamExecutionContext) context.Context {
- return context.WithValue(ctx, streamExecutionContextKeyInstance, ec)
-}
-
-// FromStreamExecutionContext returns the stream execution context from context.Context.
-func FromStreamExecutionContext(ctx context.Context) StreamExecutionContext {
- return ctx.Value(streamExecutionContextKeyInstance).(StreamExecutionContext)
-}
-
// StreamExecutable allows querying in the stream schema.
type StreamExecutable interface {
Execute(context.Context) ([]*streamv1.Element, error)
@@ -59,21 +44,6 @@ type MeasureExecutionContext interface {
Query(ctx context.Context, opts model.MeasureQueryOptions) (model.MeasureQueryResult, error)
}
-// MeasureExecutionContextKey is the key of measure execution context in context.Context.
-type MeasureExecutionContextKey struct{}
-
-var measureExecutionContextKeyInstance = MeasureExecutionContextKey{}
-
-// WithMeasureExecutionContext returns a new context with measure execution context.
-func WithMeasureExecutionContext(ctx context.Context, ec MeasureExecutionContext) context.Context {
- return context.WithValue(ctx, measureExecutionContextKeyInstance, ec)
-}
-
-// FromMeasureExecutionContext returns the measure execution context from context.Context.
-func FromMeasureExecutionContext(ctx context.Context) MeasureExecutionContext {
- return ctx.Value(measureExecutionContextKeyInstance).(MeasureExecutionContext)
-}
-
// MIterator allows iterating in a measure data set.
type MIterator interface {
Next() bool
diff --git a/pkg/query/logical/measure/measure_analyzer.go b/pkg/query/logical/measure/measure_analyzer.go
index 453db063..5eb69674 100644
--- a/pkg/query/logical/measure/measure_analyzer.go
+++ b/pkg/query/logical/measure/measure_analyzer.go
@@ -18,12 +18,13 @@
package measure
import (
- "context"
+ "fmt"
"math"
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/executor"
"github.com/apache/skywalking-banyandb/pkg/query/logical"
)
@@ -53,7 +54,10 @@ func BuildSchema(md *databasev1.Measure, indexRules []*databasev1.IndexRule) (lo
}
// Analyze converts logical expressions to executable operation tree represented by Plan.
-func Analyze(_ context.Context, criteria *measurev1.QueryRequest, metadata *commonv1.Metadata, s logical.Schema) (logical.Plan, error) {
+func Analyze(criteria *measurev1.QueryRequest, metadata []*commonv1.Metadata, ss []logical.Schema, ecc []executor.MeasureExecutionContext) (logical.Plan, error) {
+ if len(metadata) != len(ss) {
+ return nil, fmt.Errorf("number of schemas %d not equal to metadata count %d", len(ss), len(metadata))
+ }
groupByEntity := false
var groupByTags [][]*logical.Tag
if criteria.GetGroupBy() != nil {
@@ -64,13 +68,29 @@ func Analyze(_ context.Context, criteria *measurev1.QueryRequest, metadata *comm
groupByTags[i] = logical.NewTags(tagFamily.GetName(), tagFamily.GetTags()...)
tags = append(tags, tagFamily.GetTags()...)
}
- if logical.StringSlicesEqual(s.EntityList(), tags) {
+ if logical.StringSlicesEqual(ss[0].EntityList(), tags) {
groupByEntity = true
}
}
-
- // parse fields
- plan := parseFields(criteria, metadata, groupByEntity)
+ var plan logical.UnresolvedPlan
+ var s logical.Schema
+ tagProjection := logical.ToTags(criteria.GetTagProjection())
+ if len(metadata) == 1 {
+ plan = parseFields(criteria, metadata[0], ecc[0], groupByEntity, tagProjection)
+ s = ss[0]
+ } else {
+ var err error
+ if s, err = mergeSchema(ss); err != nil {
+ return nil, err
+ }
+ plan = &unresolvedMerger{
+ criteria: criteria,
+ metadata: metadata,
+ ecc: ecc,
+ tagProjection: tagProjection,
+ groupByEntity: groupByEntity,
+ }
+ }
// parse limit and offset
limitParameter := criteria.GetLimit()
@@ -113,7 +133,7 @@ func Analyze(_ context.Context, criteria *measurev1.QueryRequest, metadata *comm
}
// DistributedAnalyze converts logical expressions to executable operation tree represented by Plan.
-func DistributedAnalyze(criteria *measurev1.QueryRequest, s logical.Schema) (logical.Plan, error) {
+func DistributedAnalyze(criteria *measurev1.QueryRequest, ss []logical.Schema) (logical.Plan, error) {
var groupByTags [][]*logical.Tag
if criteria.GetGroupBy() != nil {
groupByProjectionTags := criteria.GetGroupBy().GetTagProjection()
@@ -152,6 +172,17 @@ func DistributedAnalyze(criteria *measurev1.QueryRequest, s logical.Schema) (log
}
plan = limit(plan, criteria.GetOffset(), limitParameter)
+
+ var s logical.Schema
+ var err error
+ if len(ss) == 1 {
+ s = ss[0]
+ } else {
+ if s, err = mergeSchema(ss); err != nil {
+ return nil, err
+ }
+ }
+
p, err := plan.Analyze(s)
if err != nil {
return nil, err
@@ -166,16 +197,14 @@ func DistributedAnalyze(criteria *measurev1.QueryRequest, s logical.Schema) (log
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
-// 2 - If criteria is given, but all of those fields exist in the "entity" definition.
-func parseFields(criteria *measurev1.QueryRequest, metadata *commonv1.Metadata, groupByEntity bool) logical.UnresolvedPlan {
+func parseFields(criteria *measurev1.QueryRequest, metadata *commonv1.Metadata, ec executor.MeasureExecutionContext,
+ groupByEntity bool, tagProjection [][]*logical.Tag,
+) logical.UnresolvedPlan {
projFields := make([]*logical.Field, len(criteria.GetFieldProjection().GetNames()))
for i, fieldNameProj := range criteria.GetFieldProjection().GetNames() {
projFields[i] = logical.NewField(fieldNameProj)
}
timeRange := criteria.GetTimeRange()
return indexScan(timeRange.GetBegin().AsTime(), timeRange.GetEnd().AsTime(), metadata,
- logical.ToTags(criteria.GetTagProjection()), projFields, groupByEntity, criteria.GetCriteria())
+ tagProjection, projFields, groupByEntity, criteria.GetCriteria(), ec)
}
diff --git a/pkg/query/logical/measure/measure_plan_indexscan_local.go b/pkg/query/logical/measure/measure_plan_indexscan_local.go
index bb4b4d9b..c5fc25ac 100644
--- a/pkg/query/logical/measure/measure_plan_indexscan_local.go
+++ b/pkg/query/logical/measure/measure_plan_indexscan_local.go
@@ -43,6 +43,7 @@ var _ logical.UnresolvedPlan = (*unresolvedIndexScan)(nil)
type unresolvedIndexScan struct {
startTime time.Time
endTime time.Time
+ ec executor.MeasureExecutionContext
metadata *commonv1.Metadata
criteria *modelv1.Criteria
projectionTags [][]*logical.Tag
@@ -99,6 +100,7 @@ func (uis *unresolvedIndexScan) Analyze(s logical.Schema) (logical.Plan, error)
groupByEntity: uis.groupByEntity,
uis: uis,
l: logger.GetLogger("query", "measure", uis.metadata.Group, uis.metadata.Name, "local-index"),
+ ec: uis.ec,
}, nil
}
@@ -128,6 +130,7 @@ func (uis *unresolvedIndexScan) Analyze(s logical.Schema) (logical.Plan, error)
groupByEntity: uis.groupByEntity,
uis: uis,
l: logger.GetLogger("query", "measure", uis.metadata.Group, uis.metadata.Name, "local-index"),
+ ec: uis.ec,
}, nil
}
@@ -137,18 +140,19 @@ var (
)
type localIndexScan struct {
- query index.Query
+ ec executor.MeasureExecutionContext
schema logical.Schema
+ query index.Query
uis *unresolvedIndexScan
order *logical.OrderBy
metadata *commonv1.Metadata
l *logger.Logger
timeRange timestamp.TimeRange
- projectionTags []model.TagProjection
projectionTagsRefs [][]*logical.TagRef
projectionFieldsRefs []*logical.FieldRef
entities [][]*modelv1.TagValue
projectionFields []string
+ projectionTags []model.TagProjection
groupByEntity bool
}
@@ -176,10 +180,9 @@ func (i *localIndexScan) Execute(ctx context.Context) (mit executor.MIterator, e
}
orderBy.Type = index.OrderByTypeSeries
}
- ec := executor.FromMeasureExecutionContext(ctx)
ctx, stop := i.startSpan(ctx, query.GetTracer(ctx), orderBy)
defer stop(err)
- result, err := ec.Query(ctx, model.MeasureQueryOptions{
+ result, err := i.ec.Query(ctx, model.MeasureQueryOptions{
Name: i.metadata.GetName(),
TimeRange: &i.timeRange,
Entities: i.entities,
@@ -214,7 +217,7 @@ func (i *localIndexScan) Schema() logical.Schema {
}
func indexScan(startTime, endTime time.Time, metadata *commonv1.Metadata, projectionTags [][]*logical.Tag,
- projectionFields []*logical.Field, groupByEntity bool, criteria *modelv1.Criteria,
+ projectionFields []*logical.Field, groupByEntity bool, criteria *modelv1.Criteria, ec executor.MeasureExecutionContext,
) logical.UnresolvedPlan {
return &unresolvedIndexScan{
startTime: startTime,
@@ -224,6 +227,7 @@ func indexScan(startTime, endTime time.Time, metadata *commonv1.Metadata, projec
projectionFields: projectionFields,
groupByEntity: groupByEntity,
criteria: criteria,
+ ec: ec,
}
}
diff --git a/pkg/query/logical/measure/measure_plan_merge.go b/pkg/query/logical/measure/measure_plan_merge.go
new file mode 100644
index 00000000..71b8b480
--- /dev/null
+++ b/pkg/query/logical/measure/measure_plan_merge.go
@@ -0,0 +1,193 @@
+// 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"
+
+ 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/pkg/iter/sort"
+ "github.com/apache/skywalking-banyandb/pkg/query/executor"
+ "github.com/apache/skywalking-banyandb/pkg/query/logical"
+ "github.com/apache/skywalking-banyandb/pkg/query/model"
+)
+
+var _ logical.UnresolvedPlan = (*unresolvedMerger)(nil)
+
+type unresolvedMerger struct {
+ criteria *measurev1.QueryRequest
+ metadata []*commonv1.Metadata
+ ecc []executor.MeasureExecutionContext
+ tagProjection [][]*logical.Tag
+ groupByEntity bool
+}
+
+func (u *unresolvedMerger) Analyze(s logical.Schema) (logical.Plan, error) {
+ ss := s.Children()
+ if len(ss) != len(u.metadata) {
+ return nil, fmt.Errorf("number of schemas %d not equal to metadata count %d", len(ss), len(u.metadata))
+ }
+
+ projTags := make([]model.TagProjection, len(u.tagProjection))
+ if len(u.tagProjection) > 0 {
+ for i := range u.tagProjection {
+ for _, tag := range u.tagProjection[i] {
+ projTags[i].Family = tag.GetFamilyName()
+ projTags[i].Names = append(projTags[i].Names, tag.GetTagName())
+ }
+ }
+ projectionTagRefs, err := s.CreateTagRef(u.tagProjection...)
+ if err != nil {
+ return nil, err
+ }
+ s = s.ProjTags(projectionTagRefs...)
+ }
+
+ mp := &mergePlan{
+ s: s,
+ }
+
+ for i := range u.metadata {
+ subPlan := parseFields(u.criteria, u.metadata[i], u.ecc[i], u.groupByEntity, u.tagProjection)
+ sp, err := subPlan.Analyze(ss[i])
+ if err != nil {
+ return nil, err
+ }
+ mp.subPlans = append(mp.subPlans, sp)
+ }
+
+ if u.criteria.OrderBy == nil {
+ mp.sortByTime = true
+ return mp, nil
+ }
+
+ if u.criteria.OrderBy.IndexRuleName == "" {
+ mp.sortByTime = true
+ if u.criteria.OrderBy.Sort == modelv1.Sort_SORT_DESC {
+ mp.desc = true
+ }
+ return mp, nil
+ }
+
+ ok, indexRule := s.IndexRuleDefined(u.criteria.OrderBy.IndexRuleName)
+ if !ok {
+ return nil, fmt.Errorf("index rule %s not found", u.criteria.OrderBy.IndexRuleName)
+ }
+
+ if len(indexRule.Tags) != 1 {
+ return nil, fmt.Errorf("index rule %s should have one tag", u.criteria.OrderBy.IndexRuleName)
+ }
+
+ sortTagSpec := s.FindTagSpecByName(indexRule.Tags[0])
+ if sortTagSpec == nil {
+ return nil, fmt.Errorf("tag %s not found", indexRule.Tags[0])
+ }
+
+ mp.sortTagSpec = *sortTagSpec
+ if u.criteria.OrderBy.Sort == modelv1.Sort_SORT_DESC {
+ mp.desc = true
+ }
+
+ return mp, nil
+}
+
+var (
+ _ logical.Plan = (*mergePlan)(nil)
+ _ executor.MeasureExecutable = (*mergePlan)(nil)
+)
+
+type mergePlan struct {
+ s logical.Schema
+ subPlans []logical.Plan
+ sortTagSpec logical.TagSpec
+ sortByTime bool
+ desc bool
+}
+
+func (m *mergePlan) Execute(ctx context.Context) (executor.MIterator, error) {
+ var allErr error
+ var iters []sort.Iterator[*comparableDataPoint]
+
+ for _, sp := range m.subPlans {
+ iter, err := sp.(executor.MeasureExecutable).Execute(ctx)
+ if err != nil {
+ allErr = multierr.Append(allErr, err)
+ continue
+ }
+
+ see := &sortableDataPoints{
+ iter: iter,
+ sortTagSpec: m.sortTagSpec,
+ sortByTime: m.sortByTime,
+ }
+ iters = append(iters, see)
+ }
+ if allErr != nil {
+ return nil, allErr
+ }
+
+ iter := &sortedMIterator{
+ Iterator: sort.NewItemIter(iters, m.desc),
+ }
+ iter.init()
+ return iter, nil
+}
+
+func (m *mergePlan) Children() []logical.Plan {
+ return m.subPlans
+}
+
+func (m *mergePlan) Schema() logical.Schema {
+ return m.s
+}
+
+func (m *mergePlan) String() string {
+ return fmt.Sprintf("MergePlan: subPlans=%d, sortByTime=%t, desc=%t, sortTag=%s",
+ len(m.subPlans), m.sortByTime, m.desc, m.sortTagSpec.Spec.GetName())
+}
+
+type sortableDataPoints struct {
+ iter executor.MIterator
+ current *comparableDataPoint
+ sortTagSpec logical.TagSpec
+ sortByTime bool
+}
+
+func (s *sortableDataPoints) Next() bool {
+ if !s.iter.Next() {
+ return false
+ }
+
+ dp := s.iter.Current()[0]
+ var err error
+ s.current, err = newComparableElement(dp, s.sortByTime, s.sortTagSpec)
+ return err == nil
+}
+
+func (s *sortableDataPoints) Val() *comparableDataPoint {
+ return s.current
+}
+
+func (s *sortableDataPoints) Close() error {
+ return s.iter.Close()
+}
diff --git a/pkg/query/logical/measure/schema.go b/pkg/query/logical/measure/schema.go
index 5ac0b009..800d7ced 100644
--- a/pkg/query/logical/measure/schema.go
+++ b/pkg/query/logical/measure/schema.go
@@ -33,6 +33,7 @@ type schema struct {
measure *databasev1.Measure
fieldMap map[string]*logical.FieldSpec
common *logical.CommonSchema
+ children []logical.Schema
}
func (m *schema) EntityList() []string {
@@ -56,12 +57,10 @@ func (m *schema) CreateTagRef(tags ...[]*logical.Tag) ([][]*logical.TagRef, erro
}
func (m *schema) CreateFieldRef(fields ...*logical.Field) ([]*logical.FieldRef, error) {
- fieldRefs := make([]*logical.FieldRef, len(fields))
- for idx, field := range fields {
+ fieldRefs := make([]*logical.FieldRef, 0, len(fields))
+ for _, field := range fields {
if fs, ok := m.fieldMap[field.Name]; ok {
- fieldRefs[idx] = &logical.FieldRef{Field: field, Spec: fs}
- } else {
- return nil, errors.Wrap(errFieldNotDefined, field.Name)
+ fieldRefs = append(fieldRefs, &logical.FieldRef{Field: field, Spec: fs})
}
}
return fieldRefs, nil
@@ -112,3 +111,60 @@ func (m *schema) registerField(fieldIdx int, spec *databasev1.FieldSpec) {
Spec: spec,
}
}
+
+func (m *schema) Children() []logical.Schema {
+ return m.children
+}
+
+func mergeSchema(schemas []logical.Schema) (logical.Schema, error) {
+ if len(schemas) == 0 {
+ return nil, nil
+ }
+ if len(schemas) == 1 {
+ return schemas[0], nil
+ }
+
+ var commonSchemas []*logical.CommonSchema
+ var tagFamilies []*databasev1.TagFamilySpec
+ fieldMap := make(map[string]*logical.FieldSpec)
+
+ for _, sm := range schemas {
+ if sm == nil {
+ continue
+ }
+ mSchema, ok := sm.(*schema)
+ if !ok {
+ continue
+ }
+
+ tagFamilies = logical.MergeTagFamilySpecs(tagFamilies, mSchema.measure.GetTagFamilies())
+
+ commonSchemas = append(commonSchemas, mSchema.common)
+
+ for name, spec := range mSchema.fieldMap {
+ if existing, exists := fieldMap[name]; exists {
+ if existing.Spec.FieldType != spec.Spec.FieldType {
+ existing.Spec.FieldType = databasev1.FieldType_FIELD_TYPE_UNSPECIFIED
+ }
+ } else {
+ fieldMap[name] = &logical.FieldSpec{
+ FieldIdx: spec.FieldIdx,
+ Spec: spec.Spec,
+ }
+ }
+ }
+ }
+
+ mergedCommon, err := logical.MergeSchemas(commonSchemas)
+ if err != nil {
+ return nil, err
+ }
+
+ ret := &schema{
+ common: mergedCommon,
+ children: schemas,
+ fieldMap: fieldMap,
+ }
+ ret.common.RegisterTagFamilies(tagFamilies)
+ return ret, nil
+}
diff --git a/pkg/query/logical/measure/topn_analyzer.go b/pkg/query/logical/measure/topn_analyzer.go
index 1a52f4dd..4ac59297 100644
--- a/pkg/query/logical/measure/topn_analyzer.go
+++ b/pkg/query/logical/measure/topn_analyzer.go
@@ -20,27 +20,88 @@ package measure
import (
"errors"
+ "strconv"
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/banyand/measure"
+ "github.com/apache/skywalking-banyandb/pkg/query/executor"
"github.com/apache/skywalking-banyandb/pkg/query/logical"
)
// TopNAnalyze converts logical expressions to executable operation tree represented by Plan.
-func TopNAnalyze(criteria *measurev1.TopNRequest, sourceMeasureSchema *databasev1.Measure, topNAggSchema *databasev1.TopNAggregation,
+func TopNAnalyze(criteria *measurev1.TopNRequest, sourceMeasureSchemaList []*databasev1.Measure, topNAggSchemaList []*databasev1.TopNAggregation,
+ ecc []executor.MeasureExecutionContext,
) (logical.Plan, error) {
- // parse fields
+ // Ensure input slices have identical lengths
+ if len(sourceMeasureSchemaList) != len(topNAggSchemaList) || len(sourceMeasureSchemaList) != len(ecc) {
+ return nil, errors.New("sourceMeasureSchemaList, topNAggSchemaList, and ecc must have identical lengths")
+ }
+ if len(sourceMeasureSchemaList) == 0 {
+ return nil, errors.New("sourceMeasureSchemaList is empty")
+ }
+
timeRange := criteria.GetTimeRange()
var plan logical.UnresolvedPlan
- plan = &unresolvedLocalScan{
- name: criteria.Name,
- startTime: timeRange.GetBegin().AsTime(),
- endTime: timeRange.GetEnd().AsTime(),
- conditions: criteria.Conditions,
- sort: criteria.FieldValueSort,
- groupByTags: topNAggSchema.GroupByTagNames,
+ var sourceMeasureSchema *databasev1.Measure
+ var topNAggSchema *databasev1.TopNAggregation
+ var ec executor.MeasureExecutionContext
+ if len(sourceMeasureSchemaList) == 1 {
+ sourceMeasureSchema = sourceMeasureSchemaList[0]
+ topNAggSchema = topNAggSchemaList[0]
+ ec = ecc[0]
+ // parse fields
+ plan = &unresolvedLocalScan{
+ name: criteria.Name,
+ startTime: timeRange.GetBegin().AsTime(),
+ endTime: timeRange.GetEnd().AsTime(),
+ conditions: criteria.Conditions,
+ sort: criteria.FieldValueSort,
+ groupByTags: topNAggSchema.GetGroupByTagNames(),
+ ec: ec,
+ }
+ } else {
+ subPlans := make([]*unresolvedLocalScan, 0, len(sourceMeasureSchemaList))
+ for i := range sourceMeasureSchemaList {
+ subPlans = append(subPlans, &unresolvedLocalScan{
+ name: criteria.Name,
+ startTime: timeRange.GetBegin().AsTime(),
+ endTime: timeRange.GetEnd().AsTime(),
+ conditions: criteria.Conditions,
+ sort: criteria.FieldValueSort,
+ groupByTags: topNAggSchemaList[i].GetGroupByTagNames(),
+ ec: ecc[i],
+ })
+ }
+ plan = &unresolvedTopNMerger{
+ subPlans: subPlans,
+ }
+ baseEntity := sourceMeasureSchemaList[0].GetEntity()
+ baseTagNames := baseEntity.GetTagNames()
+ for i, m := range sourceMeasureSchemaList[1:] {
+ entity := m.GetEntity()
+ tagNames := entity.GetTagNames()
+ if len(tagNames) != len(baseTagNames) {
+ return nil, errors.New("all source measures must have the same entity: tag name count mismatch at index " + strconv.Itoa(i+1))
+ }
+ for j := range tagNames {
+ if tagNames[j] != baseTagNames[j] {
+ return nil, errors.New("all source measures must have the same entity: tag name mismatch at index " + strconv.Itoa(i+1))
+ }
+ }
+ }
+
+ topNAggSchema = topNAggSchemaList[0]
+ baseFieldName := topNAggSchema.FieldName
+ for i, agg := range topNAggSchemaList[1:] {
+ if agg.GetFieldName() != baseFieldName {
+ return nil, errors.New("all TopNAggregation must have the same FieldName: mismatch at index " + strconv.Itoa(i+1))
+ }
+ }
+
+ sourceMeasureSchema = sourceMeasureSchemaList[0]
}
+
s, err := buildVirtualSchema(sourceMeasureSchema, topNAggSchema.FieldName)
if err != nil {
return nil, err
diff --git a/pkg/query/logical/measure/topn_plan_localscan.go b/pkg/query/logical/measure/topn_plan_localscan.go
index 92421573..f7c85323 100644
--- a/pkg/query/logical/measure/topn_plan_localscan.go
+++ b/pkg/query/logical/measure/topn_plan_localscan.go
@@ -48,6 +48,7 @@ var (
type unresolvedLocalScan struct {
startTime time.Time
endTime time.Time
+ ec executor.MeasureExecutionContext
name string
conditions []*modelv1.Condition
groupByTags []string
@@ -102,6 +103,7 @@ func (uls *unresolvedLocalScan) Analyze(s logical.Schema) (logical.Plan, error)
},
FieldProjection: fieldProjection,
},
+ ec: uls.ec,
}, nil
}
@@ -142,12 +144,12 @@ var _ logical.Plan = (*localScan)(nil)
type localScan struct {
s logical.Schema
+ ec executor.MeasureExecutionContext
options model.MeasureQueryOptions
}
func (i *localScan) Execute(ctx context.Context) (mit executor.MIterator, err error) {
- ec := executor.FromMeasureExecutionContext(ctx)
- result, err := ec.Query(ctx, i.options)
+ result, err := i.ec.Query(ctx, i.options)
if err != nil {
return nil, fmt.Errorf("failed to query measure: %w", err)
}
diff --git a/pkg/query/logical/measure/topn_plan_merge.go b/pkg/query/logical/measure/topn_plan_merge.go
new file mode 100644
index 00000000..62bce81d
--- /dev/null
+++ b/pkg/query/logical/measure/topn_plan_merge.go
@@ -0,0 +1,141 @@
+// 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"
+
+ "github.com/pkg/errors"
+ "go.uber.org/multierr"
+
+ measurev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/measure/v1"
+ "github.com/apache/skywalking-banyandb/pkg/query/executor"
+ "github.com/apache/skywalking-banyandb/pkg/query/logical"
+)
+
+var _ logical.UnresolvedPlan = (*unresolvedTopNMerger)(nil)
+
+type unresolvedTopNMerger struct {
+ subPlans []*unresolvedLocalScan
+}
+
+// Analyze implements logical.UnresolvedPlan.
+func (u *unresolvedTopNMerger) Analyze(s logical.Schema) (logical.Plan, error) {
+ subPlans := make([]*localScan, 0, len(u.subPlans))
+ for _, subPlan := range u.subPlans {
+ analyzed, err := subPlan.Analyze(s)
+ if err != nil {
+ return nil, errors.Wrap(err, "failed to analyze sub plan")
+ }
+ ls, ok := analyzed.(*localScan)
+ if !ok {
+ return nil, errors.New("sub plan must be localScan")
+ }
+ subPlans = append(subPlans, ls)
+ }
+
+ return &topNMerger{
+ subPlans: subPlans,
+ schema: s,
+ }, nil
+}
+
+var _ logical.Plan = (*topNMerger)(nil)
+
+type topNMerger struct {
+ schema logical.Schema
+ subPlans []*localScan
+}
+
+func (t *topNMerger) Execute(ctx context.Context) (executor.MIterator, error) {
+ iters := make([]executor.MIterator, 0, len(t.subPlans))
+ for _, subPlan := range t.subPlans {
+ iter, err := subPlan.Execute(ctx)
+ if err != nil {
+ return nil, errors.Wrap(err, "failed to execute sub plan")
+ }
+ iters = append(iters, iter)
+ }
+
+ return &topNMergingIterator{
+ iters: iters,
+ }, nil
+}
+
+func (t *topNMerger) String() string {
+ return fmt.Sprintf("TopNMerge: %d sub plans", len(t.subPlans))
+}
+
+func (t *topNMerger) Children() []logical.Plan {
+ children := make([]logical.Plan, 0, len(t.subPlans))
+ for _, sp := range t.subPlans {
+ children = append(children, sp)
+ }
+ return children
+}
+
+func (t *topNMerger) Schema() logical.Schema {
+ return t.schema
+}
+
+type topNMergingIterator struct {
+ iters []executor.MIterator
+ currentDps []*measurev1.DataPoint
+ currentIt int
+}
+
+func (t *topNMergingIterator) Next() bool {
+ // Try to get next batch from current iterator
+ if t.currentIt < len(t.iters) && len(t.currentDps) > 0 {
+ t.currentDps = t.currentDps[1:]
+ if len(t.currentDps) > 0 {
+ return true
+ }
+ }
+
+ // Move to next iterator
+ for t.currentIt < len(t.iters) {
+ if t.iters[t.currentIt].Next() {
+ t.currentDps = t.iters[t.currentIt].Current()
+ if len(t.currentDps) > 0 {
+ return true
+ }
+ } else {
+ t.currentIt++
+ }
+ }
+
+ return false
+}
+
+func (t *topNMergingIterator) Current() []*measurev1.DataPoint {
+ if len(t.currentDps) == 0 {
+ return nil
+ }
+ return []*measurev1.DataPoint{t.currentDps[0]}
+}
+
+func (t *topNMergingIterator) Close() error {
+ var err error
+ for _, iter := range t.iters {
+ err = multierr.Append(err, iter.Close())
+ }
+ return err
+}
diff --git a/pkg/query/logical/schema.go b/pkg/query/logical/schema.go
index 0f9b0ffb..cdbb7d80 100644
--- a/pkg/query/logical/schema.go
+++ b/pkg/query/logical/schema.go
@@ -53,7 +53,7 @@ type Schema interface {
CreateFieldRef(fields ...*Field) ([]*FieldRef, error)
ProjTags(refs ...[]*TagRef) Schema
ProjFields(refs ...*FieldRef) Schema
- Equal(Schema) bool
+ Children() []Schema
}
// TagSpec wraps offsets to access a tag in the raw data swiftly.
@@ -154,17 +154,107 @@ func (cs *CommonSchema) IndexRuleDefined(indexRuleName string) (bool, *databasev
// The family name of the tag is actually not used
// since the uniqueness of the tag names can be guaranteed across families.
func (cs *CommonSchema) CreateRef(tags ...[]*Tag) ([][]*TagRef, error) {
- tagRefs := make([][]*TagRef, len(tags))
- for i, tagInFamily := range tags {
+ tagRefs := make([][]*TagRef, 0, len(tags))
+ for _, tagInFamily := range tags {
var tagRefsInFamily []*TagRef
for _, tag := range tagInFamily {
if ts, ok := cs.TagSpecMap[tag.GetTagName()]; ok {
tagRefsInFamily = append(tagRefsInFamily, &TagRef{tag, ts})
- } else {
- return nil, errors.Wrap(errTagNotDefined, tag.GetCompoundName())
}
}
- tagRefs[i] = tagRefsInFamily
+ tagRefs = append(tagRefs, tagRefsInFamily)
}
return tagRefs, nil
}
+
+func stringSliceEqual(a, b []string) bool {
+ if len(a) != len(b) {
+ return false
+ }
+ for i := range a {
+ if a[i] != b[i] {
+ return false
+ }
+ }
+ return true
+}
+
+// MergeSchemas merges multiple schemas into one.
+func MergeSchemas(schemas []*CommonSchema) (*CommonSchema, error) {
+ if len(schemas) == 0 {
+ return nil, errors.New("no schemas to merge")
+ }
+
+ baseEntity := schemas[0].EntityList
+ for _, s := range schemas[1:] {
+ if !stringSliceEqual(baseEntity, s.EntityList) {
+ return nil, errors.New("schemas have different entity lists")
+ }
+ }
+
+ merged := &CommonSchema{
+ TagSpecMap: make(map[string]*TagSpec),
+ IndexRules: make([]*databasev1.IndexRule, 0),
+ EntityList: baseEntity,
+ }
+
+ indexRuleMap := make(map[string]*databasev1.IndexRule)
+
+ for _, s := range schemas {
+ for _, rule := range s.IndexRules {
+ if existedRule := indexRuleMap[rule.Metadata.Name]; existedRule == nil {
+ indexRuleMap[rule.Metadata.Name] = rule
+ } else if !stringSliceEqual(existedRule.GetTags(), rule.GetTags()) {
+ return nil, errors.Errorf("index rule %s has different tags", rule.Metadata.Name)
+ }
+ }
+ }
+ for _, rule := range indexRuleMap {
+ merged.IndexRules = append(merged.IndexRules, rule)
+ }
+
+ return merged, nil
+}
+
+func mergeTagSpecs(dst, src []*databasev1.TagSpec) []*databasev1.TagSpec {
+ res := make([]*databasev1.TagSpec, 0, len(dst)+len(src))
+ res = append(res, dst...)
+ for _, s := range src {
+ found := false
+ for i, d := range dst {
+ if d.Name == s.Name {
+ if d.Type != s.Type {
+ // If the type is different, the tag spec is not compatible.
+ // We need to set the type to unspecifed.
+ res[i].Type = databasev1.TagType_TAG_TYPE_UNSPECIFIED
+ }
+ found = true
+ break
+ }
+ }
+ if !found {
+ res = append(res, s)
+ }
+ }
+ return res
+}
+
+// MergeTagFamilySpecs merges two slices of TagFamilySpec.
+func MergeTagFamilySpecs(dst []*databasev1.TagFamilySpec, src []*databasev1.TagFamilySpec) []*databasev1.TagFamilySpec {
+ res := make([]*databasev1.TagFamilySpec, 0, len(dst)+len(src))
+ res = append(res, dst...)
+ for _, s := range src {
+ found := false
+ for i, d := range dst {
+ if d.Name == s.Name {
+ res[i].Tags = mergeTagSpecs(d.Tags, s.Tags)
+ found = true
+ break
+ }
+ }
+ if !found {
+ res = append(res, s)
+ }
+ }
+ return res
+}
diff --git a/pkg/query/logical/stream/schema.go b/pkg/query/logical/stream/schema.go
index a48b789e..77297a0b 100644
--- a/pkg/query/logical/stream/schema.go
+++ b/pkg/query/logical/stream/schema.go
@@ -19,8 +19,6 @@
package stream
import (
- "github.com/google/go-cmp/cmp"
-
databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
"github.com/apache/skywalking-banyandb/pkg/query/logical"
)
@@ -28,8 +26,9 @@ import (
var _ logical.Schema = (*schema)(nil)
type schema struct {
- stream *databasev1.Stream
- common *logical.CommonSchema
+ stream *databasev1.Stream
+ common *logical.CommonSchema
+ children []logical.Schema
}
func (s *schema) FindTagSpecByName(name string) *logical.TagSpec {
@@ -53,13 +52,6 @@ func (s *schema) IndexDefined(tagName string) (bool, *databasev1.IndexRule) {
return s.common.IndexDefined(tagName)
}
-func (s *schema) Equal(s2 logical.Schema) bool {
- if other, ok := s2.(*schema); ok {
- return cmp.Equal(other.common.TagSpecMap, s.common.TagSpecMap)
- }
- return false
-}
-
// CreateTagRef create TagRef to the given tags.
// The family name of the tag is actually not used
// since the uniqueness of the tag names can be guaranteed across families.
@@ -83,3 +75,39 @@ func (s *schema) ProjTags(refs ...[]*logical.TagRef) logical.Schema {
func (s *schema) ProjFields(...*logical.FieldRef) logical.Schema {
panic("stream does not support field")
}
+
+func (s *schema) Children() []logical.Schema {
+ return s.children
+}
+
+func mergeSchema(schemas []logical.Schema) (logical.Schema, error) {
+ if len(schemas) == 0 {
+ return nil, nil
+ }
+ if len(schemas) == 1 {
+ return schemas[0], nil
+ }
+ var commonSchemas []*logical.CommonSchema
+ var tagFamilies []*databasev1.TagFamilySpec
+ for _, sm := range schemas {
+ if sm == nil {
+ continue
+ }
+ s := sm.(*schema)
+ if s == nil {
+ continue
+ }
+ tagFamilies = logical.MergeTagFamilySpecs(tagFamilies, s.stream.GetTagFamilies())
+ commonSchemas = append(commonSchemas, s.common)
+ }
+ merged, err := logical.MergeSchemas(commonSchemas)
+ if err != nil {
+ return nil, err
+ }
+ ret := &schema{
+ common: merged,
+ children: schemas,
+ }
+ ret.common.RegisterTagFamilies(tagFamilies)
+ return ret, nil
+}
diff --git a/pkg/query/logical/stream/stream_analyzer.go b/pkg/query/logical/stream/stream_analyzer.go
index 30537627..e2eae758 100644
--- a/pkg/query/logical/stream/stream_analyzer.go
+++ b/pkg/query/logical/stream/stream_analyzer.go
@@ -47,9 +47,29 @@ func BuildSchema(sm *databasev1.Stream, indexRules []*databasev1.IndexRule) (log
}
// Analyze converts logical expressions to executable operation tree represented by Plan.
-func Analyze(_ context.Context, criteria *streamv1.QueryRequest, metadata *commonv1.Metadata, s logical.Schema) (logical.Plan, error) {
+func Analyze(criteria *streamv1.QueryRequest, metadata []*commonv1.Metadata, ss []logical.Schema, ecc []executor.StreamExecutionContext) (logical.Plan, error) {
// parse fields
- plan := parseTags(criteria, metadata)
+ if len(metadata) != len(ss) {
+ return nil, fmt.Errorf("number of schemas %d not equal to number of metadata %d", len(ss), len(metadata))
+ }
+ var plan logical.UnresolvedPlan
+ var s logical.Schema
+ tagProjection := logical.ToTags(criteria.GetProjection())
+ if len(metadata) == 1 {
+ plan = parseTags(criteria, metadata[0], ecc[0], tagProjection)
+ s = ss[0]
+ } else {
+ var err error
+ if s, err = mergeSchema(ss); err != nil {
+ return nil, err
+ }
+ plan = &unresolvedMerger{
+ criteria: criteria,
+ metadata: metadata,
+ ecc: ecc,
+ tagProjection: tagProjection,
+ }
+ }
// parse limit
limitParameter := criteria.GetLimit()
@@ -73,8 +93,17 @@ func Analyze(_ context.Context, criteria *streamv1.QueryRequest, metadata *commo
}
// DistributedAnalyze converts logical expressions to executable operation tree represented by Plan.
-func DistributedAnalyze(criteria *streamv1.QueryRequest, s logical.Schema) (logical.Plan, error) {
+func DistributedAnalyze(criteria *streamv1.QueryRequest, ss []logical.Schema) (logical.Plan, error) {
// parse fields
+ var s logical.Schema
+ if len(ss) == 1 {
+ s = ss[0]
+ } else {
+ var err error
+ if s, err = mergeSchema(ss); err != nil {
+ return nil, err
+ }
+ }
plan := newUnresolvedDistributed(criteria)
// parse limit
@@ -173,15 +202,10 @@ func newLimit(input logical.UnresolvedPlan, offset, num uint32) logical.Unresolv
}
}
-// parseTags 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
-// 2 - If criteria is given, but all of those fields exist in the "entity" definition,
-//
-// i.e. they are top-level sharding keys. For example, for the current skywalking's streamSchema,
-// we use service_id + service_instance_id + state as the compound sharding keys.
-func parseTags(criteria *streamv1.QueryRequest, metadata *commonv1.Metadata) logical.UnresolvedPlan {
+func parseTags(criteria *streamv1.QueryRequest, metadata *commonv1.Metadata,
+ ec executor.StreamExecutionContext, tagProjection [][]*logical.Tag,
+) logical.UnresolvedPlan {
timeRange := criteria.GetTimeRange()
return tagFilter(timeRange.GetBegin().AsTime(), timeRange.GetEnd().AsTime(), metadata,
- criteria.Criteria, logical.ToTags(criteria.GetProjection()))
+ criteria.Criteria, tagProjection, ec)
}
diff --git a/pkg/query/logical/stream/stream_plan_indexscan_local.go b/pkg/query/logical/stream/stream_plan_indexscan_local.go
index d485c6ff..3374d8fe 100644
--- a/pkg/query/logical/stream/stream_plan_indexscan_local.go
+++ b/pkg/query/logical/stream/stream_plan_indexscan_local.go
@@ -48,6 +48,7 @@ type localIndexScan struct {
schema logical.Schema
filter index.Filter
result model.StreamQueryResult
+ ec executor.StreamExecutionContext
order *logical.OrderBy
metadata *commonv1.Metadata
l *logger.Logger
@@ -88,9 +89,8 @@ func (i *localIndexScan) Execute(ctx context.Context) ([]*streamv1.Element, erro
Sort: i.order.Sort,
}
}
- ec := executor.FromStreamExecutionContext(ctx)
var err error
- if i.result, err = ec.Query(ctx, model.StreamQueryOptions{
+ if i.result, err = i.ec.Query(ctx, model.StreamQueryOptions{
Name: i.metadata.GetName(),
TimeRange: &i.timeRange,
Entities: i.entities,
diff --git a/pkg/query/logical/stream/stream_plan_merge.go b/pkg/query/logical/stream/stream_plan_merge.go
new file mode 100644
index 00000000..18e7a3f0
--- /dev/null
+++ b/pkg/query/logical/stream/stream_plan_merge.go
@@ -0,0 +1,169 @@
+// 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"
+
+ commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
+ 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/iter/sort"
+ "github.com/apache/skywalking-banyandb/pkg/query/executor"
+ "github.com/apache/skywalking-banyandb/pkg/query/logical"
+ "github.com/apache/skywalking-banyandb/pkg/query/model"
+)
+
+var _ logical.UnresolvedPlan = (*unresolvedMerger)(nil)
+
+type unresolvedMerger struct {
+ criteria *streamv1.QueryRequest
+ metadata []*commonv1.Metadata
+ ecc []executor.StreamExecutionContext
+ tagProjection [][]*logical.Tag
+}
+
+// Analyze implements logical.UnresolvedPlan.
+func (u *unresolvedMerger) Analyze(s logical.Schema) (logical.Plan, error) {
+ ss := s.Children()
+ if len(ss) != len(u.metadata) {
+ return nil, fmt.Errorf("number of schemas %d not equal to number of metadata %d", len(ss), len(u.metadata))
+ }
+ projTags := make([]model.TagProjection, len(u.tagProjection))
+ if len(u.tagProjection) > 0 {
+ for i := range u.tagProjection {
+ for _, tag := range u.tagProjection[i] {
+ projTags[i].Family = tag.GetFamilyName()
+ projTags[i].Names = append(projTags[i].Names, tag.GetTagName())
+ }
+ }
+ projectionTagRefs, errProject := s.CreateTagRef(u.tagProjection...)
+ if errProject != nil {
+ return nil, errProject
+ }
+ s = s.ProjTags(projectionTagRefs...)
+ }
+
+ mp := &mergePlan{
+ s: s,
+ }
+ for i := range u.metadata {
+ subPlan := parseTags(u.criteria, u.metadata[i], u.ecc[i], u.tagProjection)
+ sp, err := subPlan.Analyze(ss[i])
+ if err != nil {
+ return nil, err
+ }
+ mp.subPlans = append(mp.subPlans, sp)
+ }
+ if u.criteria.OrderBy == nil {
+ mp.sortByTime = true
+ return mp, nil
+ }
+ if u.criteria.OrderBy.IndexRuleName == "" {
+ mp.sortByTime = true
+ if u.criteria.OrderBy.Sort == modelv1.Sort_SORT_DESC {
+ mp.desc = true
+ }
+ return mp, nil
+ }
+ ok, indexRule := s.IndexRuleDefined(u.criteria.OrderBy.IndexRuleName)
+ if !ok {
+ return nil, fmt.Errorf("index rule %s not found", u.criteria.OrderBy.IndexRuleName)
+ }
+ if len(indexRule.Tags) != 1 {
+ return nil, fmt.Errorf("index rule %s should have only one tag", u.criteria.OrderBy.IndexRuleName)
+ }
+ sortTagSpec := s.FindTagSpecByName(indexRule.Tags[0])
+ if sortTagSpec == nil {
+ return nil, fmt.Errorf("tag %s not found", indexRule.Tags[0])
+ }
+ mp.sortTagSpec = *sortTagSpec
+ if u.criteria.OrderBy.Sort == modelv1.Sort_SORT_DESC {
+ mp.desc = true
+ }
+
+ return mp, nil
+}
+
+var (
+ _ logical.Plan = (*mergePlan)(nil)
+ _ executor.StreamExecutable = (*mergePlan)(nil)
+)
+
+type mergePlan struct {
+ s logical.Schema
+ subPlans []logical.Plan
+ sortTagSpec logical.TagSpec
+ sortByTime bool
+ desc bool
+}
+
+// Close implements executor.StreamExecutable.
+func (m *mergePlan) Close() {
+ for _, p := range m.subPlans {
+ p.(executor.StreamExecutable).Close()
+ }
+}
+
+// Execute implements executor.StreamExecutable.
+func (m *mergePlan) Execute(ctx context.Context) ([]*streamv1.Element, error) {
+ var allErr error
+ var see []sort.Iterator[*comparableElement]
+
+ for _, sp := range m.subPlans {
+ elements, err := sp.(executor.StreamExecutable).Execute(ctx)
+ if err != nil {
+ allErr = multierr.Append(allErr, err)
+ continue
+ }
+
+ iter := &sortableElements{
+ elements: elements,
+ isSortByTime: m.sortByTime,
+ sortTagSpec: m.sortTagSpec,
+ }
+ see = append(see, iter)
+ }
+
+ iter := sort.NewItemIter(see, m.desc)
+ var result []*streamv1.Element
+ for iter.Next() {
+ result = append(result, iter.Val().Element)
+ }
+
+ return result, allErr
+}
+
+// Children implements logical.Plan.
+func (m *mergePlan) Children() []logical.Plan {
+ return m.subPlans
+}
+
+// Schema implements logical.Plan.
+func (m *mergePlan) Schema() logical.Schema {
+ return m.s
+}
+
+// String implements logical.Plan.
+func (m *mergePlan) String() string {
+ return fmt.Sprintf("MergePlan: subPlans=%d, sortByTime=%t, desc=%t, sortTag=%s",
+ len(m.subPlans), m.sortByTime, m.desc, m.sortTagSpec.Spec.GetName())
+}
diff --git a/pkg/query/logical/stream/stream_plan_tag_filter.go b/pkg/query/logical/stream/stream_plan_tag_filter.go
index e7acf94e..b1a100df 100644
--- a/pkg/query/logical/stream/stream_plan_tag_filter.go
+++ b/pkg/query/logical/stream/stream_plan_tag_filter.go
@@ -39,6 +39,7 @@ var _ logical.UnresolvedPlan = (*unresolvedTagFilter)(nil)
type unresolvedTagFilter struct {
startTime time.Time
endTime time.Time
+ ec executor.StreamExecutionContext
metadata *commonv1.Metadata
criteria *modelv1.Criteria
projectionTags [][]*logical.Tag
@@ -75,7 +76,7 @@ func (uis *unresolvedTagFilter) Analyze(s logical.Schema) (logical.Plan, error)
}
}
ctx.projectionTags = projTags
- plan := uis.selectIndexScanner(ctx)
+ plan := uis.selectIndexScanner(ctx, uis.ec)
if uis.criteria != nil {
tagFilter, errFilter := logical.BuildTagFilter(uis.criteria, entityDict, s, len(ctx.globalConditions) > 1)
if errFilter != nil {
@@ -89,7 +90,7 @@ func (uis *unresolvedTagFilter) Analyze(s logical.Schema) (logical.Plan, error)
return plan, err
}
-func (uis *unresolvedTagFilter) selectIndexScanner(ctx *analyzeContext) logical.Plan {
+func (uis *unresolvedTagFilter) selectIndexScanner(ctx *analyzeContext, ec executor.StreamExecutionContext) logical.Plan {
return &localIndexScan{
timeRange: timestamp.NewInclusiveTimeRange(uis.startTime, uis.endTime),
schema: ctx.s,
@@ -99,10 +100,12 @@ func (uis *unresolvedTagFilter) selectIndexScanner(ctx *analyzeContext) logical.
filter: ctx.filter,
entities: ctx.entities,
l: logger.GetLogger("query", "stream", "local-index"),
+ ec: ec,
}
}
-func tagFilter(startTime, endTime time.Time, metadata *commonv1.Metadata, criteria *modelv1.Criteria, projection [][]*logical.Tag,
+func tagFilter(startTime, endTime time.Time, metadata *commonv1.Metadata, criteria *modelv1.Criteria,
+ projection [][]*logical.Tag, ec executor.StreamExecutionContext,
) logical.UnresolvedPlan {
return &unresolvedTagFilter{
startTime: startTime,
@@ -110,6 +113,7 @@ func tagFilter(startTime, endTime time.Time, metadata *commonv1.Metadata, criter
metadata: metadata,
criteria: criteria,
projectionTags: projection,
+ ec: ec,
}
}
diff --git a/pkg/test/stream/testdata/group_with_stages.json b/pkg/test/measure/testdata/group_stages/updated.json
similarity index 84%
copy from pkg/test/stream/testdata/group_with_stages.json
copy to pkg/test/measure/testdata/group_stages/updated.json
index f38a2436..1b8a1ee3 100644
--- a/pkg/test/stream/testdata/group_with_stages.json
+++ b/pkg/test/measure/testdata/group_stages/updated.json
@@ -1,8 +1,8 @@
{
"metadata": {
- "name": "default"
+ "name": "sw_updated"
},
- "catalog": "CATALOG_STREAM",
+ "catalog": "CATALOG_MEASURE",
"resource_opts": {
"shard_num": 2,
"segment_interval": {
@@ -11,7 +11,7 @@
},
"ttl": {
"unit": "UNIT_DAY",
- "num": 3
+ "num": 7
},
"stages": [
{
@@ -23,7 +23,7 @@
},
"ttl": {
"unit": "UNIT_DAY",
- "num": 7
+ "num": 30
},
"node_selector": "type=warm"
}
diff --git a/pkg/test/stream/testdata/group.json b/pkg/test/measure/testdata/groups/updated.json
similarity index 75%
copy from pkg/test/stream/testdata/group.json
copy to pkg/test/measure/testdata/groups/updated.json
index b9e8dc80..654806c4 100644
--- a/pkg/test/stream/testdata/group.json
+++ b/pkg/test/measure/testdata/groups/updated.json
@@ -1,8 +1,8 @@
{
"metadata": {
- "name": "default"
+ "name": "sw_updated"
},
- "catalog": "CATALOG_STREAM",
+ "catalog": "CATALOG_MEASURE",
"resource_opts": {
"shard_num": 2,
"segment_interval": {
@@ -11,7 +11,7 @@
},
"ttl": {
"unit": "UNIT_DAY",
- "num": 3
+ "num": 7
}
},
"updated_at": "2021-04-15T01:30:15.01Z"
diff --git a/pkg/test/measure/testdata/measures/service_cpm_minute_updated.json b/pkg/test/measure/testdata/measures/service_cpm_minute_updated.json
new file mode 100644
index 00000000..3d0d6cb9
--- /dev/null
+++ b/pkg/test/measure/testdata/measures/service_cpm_minute_updated.json
@@ -0,0 +1,42 @@
+{
+ "metadata": {
+ "name": "service_cpm_minute",
+ "group": "sw_updated"
+ },
+ "tag_families": [
+ {
+ "name": "default",
+ "tags": [
+ {
+ "name": "entity_id",
+ "type": "TAG_TYPE_INT"
+ },
+ {
+ "name": "extra_tag",
+ "type": "TAG_TYPE_STRING"
+ }
+ ]
+ }
+ ],
+ "fields": [
+ {
+ "name": "total",
+ "field_type": "FIELD_TYPE_INT",
+ "encoding_method": "ENCODING_METHOD_GORILLA",
+ "compression_method": "COMPRESSION_METHOD_ZSTD"
+ },
+ {
+ "name": "extra_field",
+ "field_type": "FIELD_TYPE_STRING",
+ "encoding_method": "ENCODING_METHOD_GORILLA",
+ "compression_method": "COMPRESSION_METHOD_ZSTD"
+ }
+ ],
+ "entity": {
+ "tag_names": [
+ "entity_id"
+ ]
+ },
+ "interval": "1m",
+ "updated_at": "2024-06-01T00:00:00Z"
+}
diff --git a/pkg/test/measure/testdata/measures/service_instance_cpm_minute_updated.json b/pkg/test/measure/testdata/measures/service_instance_cpm_minute_updated.json
new file mode 100644
index 00000000..1bc84feb
--- /dev/null
+++ b/pkg/test/measure/testdata/measures/service_instance_cpm_minute_updated.json
@@ -0,0 +1,52 @@
+{
+ "metadata": {
+ "name": "service_instance_cpm_minute",
+ "group": "sw_updated"
+ },
+ "tag_families": [
+ {
+ "name": "default",
+ "tags": [
+ {
+ "name": "id",
+ "type": "TAG_TYPE_STRING"
+ },
+ {
+ "name": "entity_id",
+ "type": "TAG_TYPE_STRING"
+ },
+ {
+ "name": "service_id",
+ "type": "TAG_TYPE_STRING"
+ }
+ ]
+ }
+ ],
+ "fields": [
+ {
+ "name": "total",
+ "field_type": "FIELD_TYPE_INT",
+ "encoding_method": "ENCODING_METHOD_GORILLA",
+ "compression_method": "COMPRESSION_METHOD_ZSTD"
+ },
+ {
+ "name": "value",
+ "field_type": "FIELD_TYPE_INT",
+ "encoding_method": "ENCODING_METHOD_GORILLA",
+ "compression_method": "COMPRESSION_METHOD_ZSTD"
+ }
+ ],
+ "entity": {
+ "tag_names": [
+ "service_id",
+ "entity_id"
+ ]
+ },
+ "sharding_key": {
+ "tag_names": [
+ "service_id"
+ ]
+ },
+ "interval": "1m",
+ "updated_at": "2021-04-15T01:30:15.01Z"
+}
\ No newline at end of file
diff --git a/pkg/test/measure/testdata/topn_aggregations/updated_service_instance_cpm_minute_top_bottom_100.json b/pkg/test/measure/testdata/topn_aggregations/updated_service_instance_cpm_minute_top_bottom_100.json
new file mode 100644
index 00000000..492c828a
--- /dev/null
+++ b/pkg/test/measure/testdata/topn_aggregations/updated_service_instance_cpm_minute_top_bottom_100.json
@@ -0,0 +1,17 @@
+{
+ "metadata": {
+ "name": "service_instance_cpm_minute_top_bottom_100",
+ "group": "sw_updated"
+ },
+ "source_measure": {
+ "name": "service_instance_cpm_minute",
+ "group": "sw_updated"
+ },
+ "field_name": "value",
+ "field_value_sort": 0,
+ "group_by_tag_names": [
+ "service_id"
+ ],
+ "counters_number": 1000,
+ "lru_size": 10
+}
\ No newline at end of file
diff --git a/pkg/test/stream/etcd.go b/pkg/test/stream/etcd.go
index d48c3f4f..fca9c6c5 100644
--- a/pkg/test/stream/etcd.go
+++ b/pkg/test/stream/etcd.go
@@ -21,6 +21,7 @@ package stream
import (
"context"
"embed"
+ "encoding/json"
"errors"
"path"
@@ -118,19 +119,24 @@ func LoadSchemaWithStages(ctx context.Context, e schema.Registry) error {
if e == nil {
return nil
}
- g := &commonv1.Group{}
- if err := protojson.Unmarshal([]byte(groupWithStagesJSON), g); err != nil {
+ var rawGroups []json.RawMessage
+ if err := json.Unmarshal([]byte(groupWithStagesJSON), &rawGroups); err != nil {
return err
}
- _, err := e.GetGroup(ctx, g.Metadata.Name)
- if !errors.Is(err, schema.ErrGRPCResourceNotFound) {
- logger.Infof("group %s already exists", g.Metadata.Name)
- return nil
- }
- if innerErr := e.CreateGroup(ctx, g); innerErr != nil {
- return innerErr
+ for _, raw := range rawGroups {
+ g := &commonv1.Group{}
+ if err := protojson.Unmarshal(raw, g); err != nil {
+ return err
+ }
+ _, err := e.GetGroup(ctx, g.Metadata.Name)
+ if !errors.Is(err, schema.ErrGRPCResourceNotFound) {
+ logger.Infof("group %s already exists", g.Metadata.Name)
+ return nil
+ }
+ if innerErr := e.CreateGroup(ctx, g); innerErr != nil {
+ return innerErr
+ }
}
-
return loadSchemas(ctx, e)
}
@@ -140,17 +146,23 @@ func PreloadSchema(ctx context.Context, e schema.Registry) error {
if e == nil {
return nil
}
- g := &commonv1.Group{}
- if err := protojson.Unmarshal([]byte(groupJSON), g); err != nil {
+ var rawGroups []json.RawMessage
+ if err := json.Unmarshal([]byte(groupJSON), &rawGroups); err != nil {
return err
}
- _, err := e.GetGroup(ctx, g.Metadata.Name)
- if !errors.Is(err, schema.ErrGRPCResourceNotFound) {
- logger.Infof("group %s already exists", g.Metadata.Name)
- return nil
- }
- if innerErr := e.CreateGroup(ctx, g); innerErr != nil {
- return innerErr
+ for _, raw := range rawGroups {
+ g := &commonv1.Group{}
+ if err := protojson.Unmarshal(raw, g); err != nil {
+ return err
+ }
+ _, err := e.GetGroup(ctx, g.Metadata.Name)
+ if !errors.Is(err, schema.ErrGRPCResourceNotFound) {
+ logger.Infof("group %s already exists", g.Metadata.Name)
+ return nil
+ }
+ if innerErr := e.CreateGroup(ctx, g); innerErr != nil {
+ return innerErr
+ }
}
return loadSchemas(ctx, e)
diff --git a/pkg/test/stream/testdata/group.json b/pkg/test/stream/testdata/group.json
index b9e8dc80..352560dd 100644
--- a/pkg/test/stream/testdata/group.json
+++ b/pkg/test/stream/testdata/group.json
@@ -1,4 +1,4 @@
-{
+[{
"metadata": {
"name": "default"
},
@@ -15,4 +15,21 @@
}
},
"updated_at": "2021-04-15T01:30:15.01Z"
-}
\ No newline at end of file
+},{
+ "metadata": {
+ "name": "updated"
+ },
+ "catalog": "CATALOG_STREAM",
+ "resource_opts": {
+ "shard_num": 2,
+ "segment_interval": {
+ "unit": "UNIT_DAY",
+ "num": 1
+ },
+ "ttl": {
+ "unit": "UNIT_DAY",
+ "num": 3
+ }
+ },
+ "updated_at": "2021-04-15T01:30:15.01Z"
+}]
\ No newline at end of file
diff --git a/pkg/test/stream/testdata/group_with_stages.json b/pkg/test/stream/testdata/group_with_stages.json
index f38a2436..b5aad645 100644
--- a/pkg/test/stream/testdata/group_with_stages.json
+++ b/pkg/test/stream/testdata/group_with_stages.json
@@ -1,33 +1,68 @@
-{
- "metadata": {
- "name": "default"
- },
- "catalog": "CATALOG_STREAM",
- "resource_opts": {
- "shard_num": 2,
- "segment_interval": {
- "unit": "UNIT_DAY",
- "num": 1
+[
+ {
+ "metadata": {
+ "name": "default"
},
- "ttl": {
- "unit": "UNIT_DAY",
- "num": 3
+ "catalog": "CATALOG_STREAM",
+ "resource_opts": {
+ "shard_num": 2,
+ "segment_interval": {
+ "unit": "UNIT_DAY",
+ "num": 1
+ },
+ "ttl": {
+ "unit": "UNIT_DAY",
+ "num": 3
+ },
+ "stages": [
+ {
+ "name": "warm",
+ "shard_num": 1,
+ "segment_interval": {
+ "unit": "UNIT_DAY",
+ "num": 3
+ },
+ "ttl": {
+ "unit": "UNIT_DAY",
+ "num": 7
+ },
+ "node_selector": "type=warm"
+ }
+ ]
},
- "stages": [
- {
- "name": "warm",
- "shard_num": 1,
- "segment_interval": {
- "unit": "UNIT_DAY",
- "num": 3
- },
- "ttl": {
- "unit": "UNIT_DAY",
- "num": 7
- },
- "node_selector": "type=warm"
- }
- ]
+ "updated_at": "2021-04-15T01:30:15.01Z"
},
- "updated_at": "2021-04-15T01:30:15.01Z"
-}
\ No newline at end of file
+ {
+ "metadata": {
+ "name": "updated"
+ },
+ "catalog": "CATALOG_STREAM",
+ "resource_opts": {
+ "shard_num": 2,
+ "segment_interval": {
+ "unit": "UNIT_DAY",
+ "num": 1
+ },
+ "ttl": {
+ "unit": "UNIT_DAY",
+ "num": 3
+ },
+ "stages": [
+ {
+ "name": "warm",
+ "shard_num": 1,
+ "segment_interval": {
+ "unit": "UNIT_DAY",
+ "num": 3
+ },
+ "ttl": {
+ "unit": "UNIT_DAY",
+ "num": 7
+ },
+ "node_selector": "type=warm"
+ }
+ ]
+ },
+ "updated_at": "2021-04-15T01:30:15.01Z"
+ }
+]
\ No newline at end of file
diff --git a/pkg/test/stream/testdata/index_rule_bindings/updated.json b/pkg/test/stream/testdata/index_rule_bindings/updated.json
new file mode 100644
index 00000000..9eb12e76
--- /dev/null
+++ b/pkg/test/stream/testdata/index_rule_bindings/updated.json
@@ -0,0 +1,16 @@
+{
+ "metadata": {
+ "name": "sw-index-rule-binding",
+ "group": "updated"
+ },
+ "rules": [
+ "duration"
+ ],
+ "subject":{
+ "catalog": "CATALOG_STREAM",
+ "name": "sw"
+ },
+ "begin_at": "2021-04-15T01:30:15.01Z",
+ "expire_at": "2121-04-15T01:30:15.01Z",
+ "updated_at": "2021-04-15T01:30:15.01Z"
+}
diff --git a/pkg/test/stream/testdata/index_rules/duration_updated.json b/pkg/test/stream/testdata/index_rules/duration_updated.json
new file mode 100644
index 00000000..d9980a3c
--- /dev/null
+++ b/pkg/test/stream/testdata/index_rules/duration_updated.json
@@ -0,0 +1,12 @@
+{
+ "metadata": {
+ "id": 3,
+ "name": "duration",
+ "group": "updated"
+ },
+ "tags": [
+ "duration"
+ ],
+ "type": "TYPE_INVERTED",
+ "updated_at": "2021-04-15T01:30:15.01Z"
+}
diff --git a/pkg/test/stream/testdata/streams/sw_updated.json b/pkg/test/stream/testdata/streams/sw_updated.json
new file mode 100644
index 00000000..46731df9
--- /dev/null
+++ b/pkg/test/stream/testdata/streams/sw_updated.json
@@ -0,0 +1,90 @@
+{
+ "metadata": {
+ "name": "sw",
+ "group": "updated"
+ },
+ "tag_families": [
+ {
+ "name": "data",
+ "tags": [
+ {
+ "name": "data_binary",
+ "type": "TAG_TYPE_DATA_BINARY"
+ }
+ ]
+ },
+ {
+ "name": "searchable",
+ "tags": [
+ {
+ "name": "service_instance_id",
+ "type": "TAG_TYPE_STRING"
+ },
+ {
+ "name": "trace_id",
+ "type": "TAG_TYPE_STRING"
+ },
+ {
+ "name": "duration",
+ "type": "TAG_TYPE_INT"
+ },
+ {
+ "name": "state",
+ "type": "TAG_TYPE_INT"
+ },
+ {
+ "name": "service_id",
+ "type": "TAG_TYPE_STRING"
+ },
+ {
+ "name": "endpoint_id",
+ "type": "TAG_TYPE_STRING"
+ },
+ {
+ "name": "http.method",
+ "type": "TAG_TYPE_STRING"
+ },
+ {
+ "name": "span_id",
+ "type": "TAG_TYPE_STRING"
+ },
+ {
+ "name": "db.type",
+ "type": "TAG_TYPE_STRING"
+ },
+ {
+ "name": "db.instance",
+ "type": "TAG_TYPE_STRING"
+ },
+ {
+ "name": "mq.topic",
+ "type": "TAG_TYPE_STRING"
+ },
+ {
+ "name": "mq.broker",
+ "type": "TAG_TYPE_STRING"
+ },
+ {
+ "name": "extended_tags",
+ "type": "TAG_TYPE_STRING_ARRAY"
+ },
+ {
+ "name": "new_tag",
+ "type": "TAG_TYPE_STRING"
+ },
+ {
+ "name": "status_code",
+ "type": "TAG_TYPE_STRING"
+ }
+ ]
+ }
+ ],
+ "entity": {
+ "tag_names": [
+ "service_id",
+ "service_instance_id",
+ "state"
+ ]
+ },
+ "updated_at": "2024-06-10T12:00:00Z"
+}
diff --git a/test/cases/init.go b/test/cases/init.go
index 741c5b5c..6dd05a41 100644
--- a/test/cases/init.go
+++ b/test/cases/init.go
@@ -39,6 +39,7 @@ func Initialize(addr string, now time.Time) {
// stream
casesstreamdata.Write(conn, "sw", now, interval)
casesstreamdata.Write(conn, "duplicated", now, 0)
+ casesstreamdata.WriteToGroup(conn, "sw", "updated", "sw_updated", now.Add(time.Minute), interval)
// measure
interval = time.Minute
casesmeasuredata.Write(conn, "service_traffic", "index_mode", "service_traffic_data_old.json", now.AddDate(0, 0, -2), interval)
@@ -57,4 +58,5 @@ func Initialize(addr string, now time.Time) {
casesmeasuredata.Write(conn, "service_instance_latency_minute", "sw_metric", "service_instance_latency_minute_data1.json", now.Add(1*time.Minute), interval)
casesmeasuredata.Write(conn, "endpoint_traffic", "sw_metric", "endpoint_traffic.json", now, interval)
casesmeasuredata.Write(conn, "duplicated", "exception", "duplicated.json", now, 0)
+ casesmeasuredata.Write(conn, "service_cpm_minute", "sw_updated", "service_cpm_minute_updated_data.json", now.Add(10*time.Minute), interval)
}
diff --git a/test/cases/measure/data/input/multi_group_new_tag_field.yaml b/test/cases/measure/data/input/multi_group_new_tag_field.yaml
new file mode 100644
index 00000000..d9301d32
--- /dev/null
+++ b/test/cases/measure/data/input/multi_group_new_tag_field.yaml
@@ -0,0 +1,25 @@
+# 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.
+
+name: "service_cpm_minute"
+groups: [ "sw_metric", "sw_updated" ]
+tagProjection:
+ tagFamilies:
+ - name: "default"
+ tags: [ "entity_id", "extra_tag" ]
+fieldProjection:
+ names: [ "total", "extra_field" ]
diff --git a/test/cases/measure/data/input/multi_group_unchanged.yaml b/test/cases/measure/data/input/multi_group_unchanged.yaml
new file mode 100644
index 00000000..13c3236a
--- /dev/null
+++ b/test/cases/measure/data/input/multi_group_unchanged.yaml
@@ -0,0 +1,25 @@
+# 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.
+
+name: "service_cpm_minute"
+groups: [ "sw_metric", "sw_updated" ]
+tagProjection:
+ tagFamilies:
+ - name: "default"
+ tags: [ "entity_id" ]
+fieldProjection:
+ names: [ "total" ]
diff --git a/test/cases/measure/data/testdata/service_cpm_minute_updated_data.json b/test/cases/measure/data/testdata/service_cpm_minute_updated_data.json
new file mode 100644
index 00000000..9c372c7e
--- /dev/null
+++ b/test/cases/measure/data/testdata/service_cpm_minute_updated_data.json
@@ -0,0 +1,92 @@
+[
+ {
+ "tag_families": [
+ {
+ "tags": [
+ {
+ "int": {
+ "value": 1
+ }
+ },
+ {
+ "str": {
+ "value": "extraA"
+ }
+ }
+ ]
+ }
+ ],
+ "fields": [
+ {
+ "int": {
+ "value": 100
+ }
+ },
+ {
+ "str": {
+ "value": "fieldA"
+ }
+ }
+ ]
+ },
+ {
+ "tag_families": [
+ {
+ "tags": [
+ {
+ "int": {
+ "value": 2
+ }
+ },
+ {
+ "str": {
+ "value": "extraB"
+ }
+ }
+ ]
+ }
+ ],
+ "fields": [
+ {
+ "int": {
+ "value": 200
+ }
+ },
+ {
+ "str": {
+ "value": "fieldB"
+ }
+ }
+ ]
+ },
+ {
+ "tag_families": [
+ {
+ "tags": [
+ {
+ "int": {
+ "value": 3
+ }
+ },
+ {
+ "str": {
+ "value": "extraC"
+ }
+ }
+ ]
+ }
+ ],
+ "fields": [
+ {
+ "int": {
+ "value": 300
+ }
+ },
+ {
+ "str": {
+ "value": "fieldC"
+ }
+ }
+ ]
+ }
+]
diff --git a/test/cases/measure/data/want/multi_group_new_tag_field.yaml b/test/cases/measure/data/want/multi_group_new_tag_field.yaml
new file mode 100644
index 00000000..e3372ce4
--- /dev/null
+++ b/test/cases/measure/data/want/multi_group_new_tag_field.yaml
@@ -0,0 +1,213 @@
+# 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.
+
+dataPoints:
+- fields:
+ - name: total
+ value:
+ int:
+ value: "100"
+ - name: extra_field
+ value:
+ "null": null
+ sid: "17296083384258897030"
+ tagFamilies:
+ - name: default
+ tags:
+ - key: entity_id
+ value:
+ str:
+ value: entity_1
+ - key: extra_tag
+ value:
+ "null": null
+ timestamp: "2025-05-11T11:07:00Z"
+ version: "1746961971558055000"
+- fields:
+ - name: total
+ value:
+ int:
+ value: "100"
+ - name: extra_field
+ value:
+ "null": null
+ sid: "13293155473432710288"
+ tagFamilies:
+ - name: default
+ tags:
+ - key: entity_id
+ value:
+ str:
+ value: entity_2
+ - key: extra_tag
+ value:
+ "null": null
+ timestamp: "2025-05-11T11:08:00Z"
+ version: "1746961971558070000"
+- fields:
+ - name: total
+ value:
+ int:
+ value: "100"
+ - name: extra_field
+ value:
+ "null": null
+ sid: "8985479151150409059"
+ tagFamilies:
+ - name: default
+ tags:
+ - key: entity_id
+ value:
+ str:
+ value: entity_3
+ - key: extra_tag
+ value:
+ "null": null
+ timestamp: "2025-05-11T11:09:00Z"
+ version: "1746961971558081000"
+- fields:
+ - name: total
+ value:
+ int:
+ value: "100"
+ - name: extra_field
+ value:
+ "null": null
+ sid: "11425316604636028751"
+ tagFamilies:
+ - name: default
+ tags:
+ - key: entity_id
+ value:
+ str:
+ value: entity_4
+ - key: extra_tag
+ value:
+ "null": null
+ timestamp: "2025-05-11T11:10:00Z"
+ version: "1746961971558093000"
+- fields:
+ - name: total
+ value:
+ int:
+ value: "50"
+ - name: extra_field
+ value:
+ "null": null
+ sid: "13633132181018445499"
+ tagFamilies:
+ - name: default
+ tags:
+ - key: entity_id
+ value:
+ str:
+ value: entity_5
+ - key: extra_tag
+ value:
+ "null": null
+ timestamp: "2025-05-11T11:11:00Z"
+ version: "1746961971558104000"
+- fields:
+ - name: total
+ value:
+ int:
+ value: "300"
+ - name: extra_field
+ value:
+ "null": null
+ sid: "12160318715725963202"
+ tagFamilies:
+ - name: default
+ tags:
+ - key: entity_id
+ value:
+ str:
+ value: entity_6
+ - key: extra_tag
+ value:
+ "null": null
+ timestamp: "2025-05-11T11:12:00Z"
+ version: "1746961971558114000"
+- fields:
+ - name: total
+ value:
+ int:
+ value: "100"
+ - name: extra_field
+ value:
+ str:
+ value: fieldA
+ sid: "335438207074479228"
+ tagFamilies:
+ - name: default
+ tags:
+ - key: entity_id
+ value:
+ int:
+ value: "1"
+ - key: extra_tag
+ value:
+ str:
+ value: extraA
+ timestamp: "2025-05-11T11:20:00Z"
+ version: "1746961972237248000"
+- fields:
+ - name: total
+ value:
+ int:
+ value: "200"
+ - name: extra_field
+ value:
+ str:
+ value: fieldB
+ sid: "16973798048885028135"
+ tagFamilies:
+ - name: default
+ tags:
+ - key: entity_id
+ value:
+ int:
+ value: "2"
+ - key: extra_tag
+ value:
+ str:
+ value: extraB
+ timestamp: "2025-05-11T11:21:00Z"
+ version: "1746961972237281000"
+- fields:
+ - name: total
+ value:
+ int:
+ value: "300"
+ - name: extra_field
+ value:
+ str:
+ value: fieldC
+ sid: "13265636622351444845"
+ tagFamilies:
+ - name: default
+ tags:
+ - key: entity_id
+ value:
+ int:
+ value: "3"
+ - key: extra_tag
+ value:
+ str:
+ value: extraC
+ timestamp: "2025-05-11T11:22:00Z"
+ version: "1746961972237328000"
diff --git a/test/cases/measure/data/want/multi_group_unchanged.yaml b/test/cases/measure/data/want/multi_group_unchanged.yaml
new file mode 100644
index 00000000..aee7c791
--- /dev/null
+++ b/test/cases/measure/data/want/multi_group_unchanged.yaml
@@ -0,0 +1,153 @@
+# 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.
+
+dataPoints:
+ - fields:
+ - name: total
+ value:
+ int:
+ value: "100"
+ sid: "17296083384258897030"
+ tagFamilies:
+ - name: default
+ tags:
+ - key: entity_id
+ value:
+ str:
+ value: entity_1
+ timestamp: "2025-05-11T10:50:00Z"
+ version: "1746960914435901000"
+ - fields:
+ - name: total
+ value:
+ int:
+ value: "100"
+ sid: "13293155473432710288"
+ tagFamilies:
+ - name: default
+ tags:
+ - key: entity_id
+ value:
+ str:
+ value: entity_2
+ timestamp: "2025-05-11T10:51:00Z"
+ version: "1746960914435914000"
+ - fields:
+ - name: total
+ value:
+ int:
+ value: "100"
+ sid: "8985479151150409059"
+ tagFamilies:
+ - name: default
+ tags:
+ - key: entity_id
+ value:
+ str:
+ value: entity_3
+ timestamp: "2025-05-11T10:52:00Z"
+ version: "1746960914435924000"
+ - fields:
+ - name: total
+ value:
+ int:
+ value: "100"
+ sid: "11425316604636028751"
+ tagFamilies:
+ - name: default
+ tags:
+ - key: entity_id
+ value:
+ str:
+ value: entity_4
+ timestamp: "2025-05-11T10:53:00Z"
+ version: "1746960914435933000"
+ - fields:
+ - name: total
+ value:
+ int:
+ value: "50"
+ sid: "13633132181018445499"
+ tagFamilies:
+ - name: default
+ tags:
+ - key: entity_id
+ value:
+ str:
+ value: entity_5
+ timestamp: "2025-05-11T10:54:00Z"
+ version: "1746960914435943000"
+ - fields:
+ - name: total
+ value:
+ int:
+ value: "300"
+ sid: "12160318715725963202"
+ tagFamilies:
+ - name: default
+ tags:
+ - key: entity_id
+ value:
+ str:
+ value: entity_6
+ timestamp: "2025-05-11T10:55:00Z"
+ version: "1746960914435952000"
+ - fields:
+ - name: total
+ value:
+ int:
+ value: "100"
+ sid: "335438207074479228"
+ tagFamilies:
+ - name: default
+ tags:
+ - key: entity_id
+ value:
+ int:
+ value: "1"
+ timestamp: "2025-05-11T11:03:00Z"
+ version: "1746960915119637000"
+ - fields:
+ - name: total
+ value:
+ int:
+ value: "200"
+ sid: "16973798048885028135"
+ tagFamilies:
+ - name: default
+ tags:
+ - key: entity_id
+ value:
+ int:
+ value: "2"
+ timestamp: "2025-05-11T11:04:00Z"
+ version: "1746960915119651000"
+ - fields:
+ - name: total
+ value:
+ int:
+ value: "300"
+ sid: "13265636622351444845"
+ tagFamilies:
+ - name: default
+ tags:
+ - key: entity_id
+ value:
+ int:
+ value: "3"
+ timestamp: "2025-05-11T11:05:00Z"
+ version: "1746960915119661000"
\ No newline at end of file
diff --git a/test/cases/measure/measure.go b/test/cases/measure/measure.go
index e42e0681..a7ec3979 100644
--- a/test/cases/measure/measure.go
+++ b/test/cases/measure/measure.go
@@ -81,4 +81,6 @@ var _ = g.DescribeTable("Scanning Measures", verify,
g.Entry("range of index mode", helpers.Args{Input: "index_mode_range", Duration: 25 * time.Minute, Offset: -20 * time.Minute}),
g.Entry("none of index mode", helpers.Args{Input: "index_mode_none", WantEmpty: true, Duration: 25 * time.Minute, Offset: -20 * time.Minute}),
g.Entry("query by id in index mode", helpers.Args{Input: "index_mode_by_id", Duration: 25 * time.Minute, Offset: -20 * time.Minute}),
+ g.Entry("multi groups: unchanged", helpers.Args{Input: "multi_group_unchanged", Duration: 35 * time.Minute, Offset: -20 * time.Minute}),
+ g.Entry("multi groups: new tag and fields", helpers.Args{Input: "multi_group_new_tag_field", Duration: 35 * time.Minute, Offset: -20 * time.Minute}),
)
diff --git a/test/cases/stream/data/data.go b/test/cases/stream/data/data.go
index b3df58d4..3316179c 100644
--- a/test/cases/stream/data/data.go
+++ b/test/cases/stream/data/data.go
@@ -171,9 +171,14 @@ func loadData(stream streamv1.StreamService_WriteClient, metadata *commonv1.Meta
// Write data into the server.
func Write(conn *grpclib.ClientConn, name string, baseTime time.Time, interval time.Duration) {
+ WriteToGroup(conn, name, "default", name, baseTime, interval)
+}
+
+// WriteToGroup data into the server with a specific group.
+func WriteToGroup(conn *grpclib.ClientConn, name, group, fileName string, baseTime time.Time, interval time.Duration) {
metadata := &commonv1.Metadata{
Name: name,
- Group: "default",
+ Group: group,
}
schema := databasev1.NewStreamRegistryServiceClient(conn)
resp, err := schema.Get(context.Background(), &databasev1.StreamRegistryServiceGetRequest{Metadata: metadata})
@@ -184,7 +189,7 @@ func Write(conn *grpclib.ClientConn, name string, baseTime time.Time, interval t
ctx := context.Background()
writeClient, err := c.Write(ctx)
gm.Expect(err).NotTo(gm.HaveOccurred())
- loadData(writeClient, metadata, fmt.Sprintf("%s.json", name), baseTime, interval)
+ loadData(writeClient, metadata, fmt.Sprintf("%s.json", fileName), baseTime, interval)
gm.Expect(writeClient.CloseSend()).To(gm.Succeed())
gm.Eventually(func() error {
_, err := writeClient.Recv()
diff --git a/test/cases/stream/data/input/multi_group_new_tag.yaml b/test/cases/stream/data/input/multi_group_new_tag.yaml
new file mode 100644
index 00000000..90bb4f8c
--- /dev/null
+++ b/test/cases/stream/data/input/multi_group_new_tag.yaml
@@ -0,0 +1,23 @@
+# 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.
+
+name: "sw"
+groups: [ "default", "updated" ]
+projection:
+ tagFamilies:
+ - name: "searchable"
+ tags: [ "trace_id", "service_id", "new_tag" ]
diff --git a/test/cases/stream/data/input/multi_group_sort_duration.yaml b/test/cases/stream/data/input/multi_group_sort_duration.yaml
new file mode 100644
index 00000000..d30041b8
--- /dev/null
+++ b/test/cases/stream/data/input/multi_group_sort_duration.yaml
@@ -0,0 +1,26 @@
+# 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.
+
+name: "sw"
+groups: [ "default", "updated" ]
+projection:
+ tagFamilies:
+ - name: "searchable"
+ tags: [ "trace_id", "duration" ]
+orderBy:
+ indexRuleName: "duration"
+ sort: "SORT_DESC"
diff --git a/test/cases/stream/data/input/multi_group_tag_type.yaml b/test/cases/stream/data/input/multi_group_tag_type.yaml
new file mode 100644
index 00000000..628ef914
--- /dev/null
+++ b/test/cases/stream/data/input/multi_group_tag_type.yaml
@@ -0,0 +1,23 @@
+# 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.
+
+name: "sw"
+groups: [ "default", "updated" ]
+projection:
+ tagFamilies:
+ - name: "searchable"
+ tags: [ "trace_id", "service_id", "status_code" ]
diff --git a/test/cases/stream/data/input/multi_group_unchanged.yaml b/test/cases/stream/data/input/multi_group_unchanged.yaml
new file mode 100644
index 00000000..9d4f7edb
--- /dev/null
+++ b/test/cases/stream/data/input/multi_group_unchanged.yaml
@@ -0,0 +1,23 @@
+# 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.
+
+name: "sw"
+groups: [ "default", "updated" ]
+projection:
+ tagFamilies:
+ - name: "searchable"
+ tags: [ "trace_id", "service_id", "service_instance_id" ]
diff --git a/test/cases/stream/data/testdata/sw_updated.json b/test/cases/stream/data/testdata/sw_updated.json
new file mode 100644
index 00000000..cac2584b
--- /dev/null
+++ b/test/cases/stream/data/testdata/sw_updated.json
@@ -0,0 +1,59 @@
+[
+ {
+ "tags": [
+ {"str": {"value": "instance_1"}},
+ {"str": {"value": "trace_001"}},
+ {"int": {"value": 1234}},
+ {"int": {"value": 1}},
+ {"str": {"value": "service_1"}},
+ {"str": {"value": "/api/v1"}},
+ {"str": {"value": "GET"}},
+ {"str": {"value": "span_1"}},
+ {"str": {"value": "mysql"}},
+ {"str": {"value": "db1"}},
+ {"str": {"value": "topic1"}},
+ {"str": {"value": "broker1"}},
+ {"str_array": {"value": ["tagA", "tagB"]}},
+ {"str": {"value": "new_value"}},
+ {"str": {"value": "200"}}
+ ]
+ },
+ {
+ "tags": [
+ {"str": {"value": "instance_2"}},
+ {"str": {"value": "trace_002"}},
+ {"int": {"value": 5678}},
+ {"int": {"value": 0}},
+ {"str": {"value": "service_2"}},
+ {"str": {"value": "/api/v2"}},
+ {"str": {"value": "POST"}},
+ {"str": {"value": "span_2"}},
+ {"str": {"value": "postgresql"}},
+ {"str": {"value": "db2"}},
+ {"str": {"value": "topic2"}},
+ {"str": {"value": "broker2"}},
+ {"str_array": {"value": ["tagC"]}},
+ {"str": {"value": "another_value"}},
+ {"str": {"value": "404"}}
+ ]
+ },
+ {
+ "tags": [
+ {"str": {"value": "instance_3"}},
+ {"str": {"value": "trace_003"}},
+ {"int": {"value": 91011}},
+ {"int": {"value": 1}},
+ {"str": {"value": "service_3"}},
+ {"str": {"value": "/api/v3"}},
+ {"str": {"value": "PUT"}},
+ {"str": {"value": "span_3"}},
+ {"str": {"value": "mongodb"}},
+ {"str": {"value": "db3"}},
+ {"str": {"value": "topic3"}},
+ {"str": {"value": "broker3"}},
+ {"str_array": {"value": ["tagD", "tagE", "tagF"]}},
+ {"str": {"value": "test_tag"}},
+ {"str": {"value": "500"}}
+ ]
+ }
+]
diff --git a/test/cases/stream/data/want/multi_group_new_tag.yaml b/test/cases/stream/data/want/multi_group_new_tag.yaml
new file mode 100644
index 00000000..fa2fd472
--- /dev/null
+++ b/test/cases/stream/data/want/multi_group_new_tag.yaml
@@ -0,0 +1,149 @@
+# 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.
+
+elements:
+- elementId: 47318c4050f63c95
+ tagFamilies:
+ - name: searchable
+ tags:
+ - key: trace_id
+ value:
+ str:
+ value: "1"
+ - key: service_id
+ value:
+ str:
+ value: webapp_id
+ - key: new_tag
+ value:
+ "null": null
+ timestamp: "2025-05-11T00:07:00Z"
+- elementId: a1126fa17892e2bd
+ tagFamilies:
+ - name: searchable
+ tags:
+ - key: trace_id
+ value:
+ str:
+ value: "2"
+ - key: service_id
+ value:
+ str:
+ value: webapp_id
+ - key: new_tag
+ value:
+ "null": null
+ timestamp: "2025-05-11T00:07:00.500Z"
+- elementId: cfb608ee34cb7413
+ tagFamilies:
+ - name: searchable
+ tags:
+ - key: trace_id
+ value:
+ str:
+ value: "3"
+ - key: service_id
+ value:
+ str:
+ value: webapp_id
+ - key: new_tag
+ value:
+ "null": null
+ timestamp: "2025-05-11T00:07:01Z"
+- elementId: 7d589b1fdae3557a
+ tagFamilies:
+ - name: searchable
+ tags:
+ - key: trace_id
+ value:
+ str:
+ value: "4"
+ - key: service_id
+ value:
+ str:
+ value: webapp_id
+ - key: new_tag
+ value:
+ "null": null
+ timestamp: "2025-05-11T00:07:01.500Z"
+- elementId: 2dcb300ca20b88c3
+ tagFamilies:
+ - name: searchable
+ tags:
+ - key: trace_id
+ value:
+ str:
+ value: "5"
+ - key: service_id
+ value:
+ str:
+ value: webapp_id
+ - key: new_tag
+ value:
+ "null": null
+ timestamp: "2025-05-11T00:07:02Z"
+- elementId: 47318c4050f63c95
+ tagFamilies:
+ - name: searchable
+ tags:
+ - key: trace_id
+ value:
+ str:
+ value: trace_001
+ - key: service_id
+ value:
+ str:
+ value: service_1
+ - key: new_tag
+ value:
+ str:
+ value: new_value
+ timestamp: "2025-05-11T00:08:00Z"
+- elementId: a1126fa17892e2bd
+ tagFamilies:
+ - name: searchable
+ tags:
+ - key: trace_id
+ value:
+ str:
+ value: trace_002
+ - key: service_id
+ value:
+ str:
+ value: service_2
+ - key: new_tag
+ value:
+ str:
+ value: another_value
+ timestamp: "2025-05-11T00:08:00.500Z"
+- elementId: cfb608ee34cb7413
+ tagFamilies:
+ - name: searchable
+ tags:
+ - key: trace_id
+ value:
+ str:
+ value: trace_003
+ - key: service_id
+ value:
+ str:
+ value: service_3
+ - key: new_tag
+ value:
+ str:
+ value: test_tag
+ timestamp: "2025-05-11T00:08:01Z"
diff --git a/test/cases/stream/data/want/multi_group_sort_duration.yaml b/test/cases/stream/data/want/multi_group_sort_duration.yaml
new file mode 100644
index 00000000..29e4c853
--- /dev/null
+++ b/test/cases/stream/data/want/multi_group_sort_duration.yaml
@@ -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.
+
+elements:
+- elementId: cfb608ee34cb7413
+ tagFamilies:
+ - name: searchable
+ tags:
+ - key: trace_id
+ value:
+ str:
+ value: trace_003
+ - key: duration
+ value:
+ int:
+ value: "91011"
+ timestamp: "2025-05-11T00:44:01Z"
+- elementId: a1126fa17892e2bd
+ tagFamilies:
+ - name: searchable
+ tags:
+ - key: trace_id
+ value:
+ str:
+ value: trace_002
+ - key: duration
+ value:
+ int:
+ value: "5678"
+ timestamp: "2025-05-11T00:44:00.500Z"
+- elementId: 47318c4050f63c95
+ tagFamilies:
+ - name: searchable
+ tags:
+ - key: trace_id
+ value:
+ str:
+ value: trace_001
+ - key: duration
+ value:
+ int:
+ value: "1234"
+ timestamp: "2025-05-11T00:44:00Z"
+- elementId: 47318c4050f63c95
+ tagFamilies:
+ - name: searchable
+ tags:
+ - key: trace_id
+ value:
+ str:
+ value: "1"
+ - key: duration
+ value:
+ int:
+ value: "1000"
+ timestamp: "2025-05-11T00:43:00Z"
+- elementId: a1126fa17892e2bd
+ tagFamilies:
+ - name: searchable
+ tags:
+ - key: trace_id
+ value:
+ str:
+ value: "2"
+ - key: duration
+ value:
+ int:
+ value: "500"
+ timestamp: "2025-05-11T00:43:00.500Z"
+- elementId: 2dcb300ca20b88c3
+ tagFamilies:
+ - name: searchable
+ tags:
+ - key: trace_id
+ value:
+ str:
+ value: "5"
+ - key: duration
+ value:
+ int:
+ value: "300"
+ timestamp: "2025-05-11T00:43:02Z"
+- elementId: 7d589b1fdae3557a
+ tagFamilies:
+ - name: searchable
+ tags:
+ - key: trace_id
+ value:
+ str:
+ value: "4"
+ - key: duration
+ value:
+ int:
+ value: "60"
+ timestamp: "2025-05-11T00:43:01.500Z"
+- elementId: cfb608ee34cb7413
+ tagFamilies:
+ - name: searchable
+ tags:
+ - key: trace_id
+ value:
+ str:
+ value: "3"
+ - key: duration
+ value:
+ int:
+ value: "30"
+ timestamp: "2025-05-11T00:43:01Z"
diff --git a/test/cases/stream/data/want/multi_group_tag_type.yaml b/test/cases/stream/data/want/multi_group_tag_type.yaml
new file mode 100644
index 00000000..71d13b8f
--- /dev/null
+++ b/test/cases/stream/data/want/multi_group_tag_type.yaml
@@ -0,0 +1,152 @@
+# 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.
+
+elements:
+- elementId: 47318c4050f63c95
+ tagFamilies:
+ - name: searchable
+ tags:
+ - key: trace_id
+ value:
+ str:
+ value: "1"
+ - key: service_id
+ value:
+ str:
+ value: webapp_id
+ - key: status_code
+ value:
+ "null": null
+ timestamp: "2025-05-11T00:17:00Z"
+- elementId: a1126fa17892e2bd
+ tagFamilies:
+ - name: searchable
+ tags:
+ - key: trace_id
+ value:
+ str:
+ value: "2"
+ - key: service_id
+ value:
+ str:
+ value: webapp_id
+ - key: status_code
+ value:
+ "null": null
+ timestamp: "2025-05-11T00:17:00.500Z"
+- elementId: cfb608ee34cb7413
+ tagFamilies:
+ - name: searchable
+ tags:
+ - key: trace_id
+ value:
+ str:
+ value: "3"
+ - key: service_id
+ value:
+ str:
+ value: webapp_id
+ - key: status_code
+ value:
+ int:
+ value: "500"
+ timestamp: "2025-05-11T00:17:01Z"
+- elementId: 7d589b1fdae3557a
+ tagFamilies:
+ - name: searchable
+ tags:
+ - key: trace_id
+ value:
+ str:
+ value: "4"
+ - key: service_id
+ value:
+ str:
+ value: webapp_id
+ - key: status_code
+ value:
+ int:
+ value: "400"
+ timestamp: "2025-05-11T00:17:01.500Z"
+- elementId: 2dcb300ca20b88c3
+ tagFamilies:
+ - name: searchable
+ tags:
+ - key: trace_id
+ value:
+ str:
+ value: "5"
+ - key: service_id
+ value:
+ str:
+ value: webapp_id
+ - key: status_code
+ value:
+ int:
+ value: "500"
+ timestamp: "2025-05-11T00:17:02Z"
+- elementId: 47318c4050f63c95
+ tagFamilies:
+ - name: searchable
+ tags:
+ - key: trace_id
+ value:
+ str:
+ value: trace_001
+ - key: service_id
+ value:
+ str:
+ value: service_1
+ - key: status_code
+ value:
+ str:
+ value: "200"
+ timestamp: "2025-05-11T00:18:00Z"
+- elementId: a1126fa17892e2bd
+ tagFamilies:
+ - name: searchable
+ tags:
+ - key: trace_id
+ value:
+ str:
+ value: trace_002
+ - key: service_id
+ value:
+ str:
+ value: service_2
+ - key: status_code
+ value:
+ str:
+ value: "404"
+ timestamp: "2025-05-11T00:18:00.500Z"
+- elementId: cfb608ee34cb7413
+ tagFamilies:
+ - name: searchable
+ tags:
+ - key: trace_id
+ value:
+ str:
+ value: trace_003
+ - key: service_id
+ value:
+ str:
+ value: service_3
+ - key: status_code
+ value:
+ str:
+ value: "500"
+ timestamp: "2025-05-11T00:18:01Z"
diff --git a/test/cases/stream/data/want/multi_group_unchanged.yaml b/test/cases/stream/data/want/multi_group_unchanged.yaml
new file mode 100644
index 00000000..9759193f
--- /dev/null
+++ b/test/cases/stream/data/want/multi_group_unchanged.yaml
@@ -0,0 +1,141 @@
+# 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.
+
+elements:
+- elementId: 47318c4050f63c95
+ tagFamilies:
+ - name: searchable
+ tags:
+ - key: trace_id
+ value:
+ str:
+ value: "1"
+ - key: service_id
+ value:
+ str:
+ value: webapp_id
+ - key: service_instance_id
+ value:
+ "null": null
+- elementId: a1126fa17892e2bd
+ tagFamilies:
+ - name: searchable
+ tags:
+ - key: trace_id
+ value:
+ str:
+ value: "2"
+ - key: service_id
+ value:
+ str:
+ value: webapp_id
+ - key: service_instance_id
+ value:
+ "null": null
+- elementId: cfb608ee34cb7413
+ tagFamilies:
+ - name: searchable
+ tags:
+ - key: trace_id
+ value:
+ str:
+ value: "3"
+ - key: service_id
+ value:
+ str:
+ value: webapp_id
+ - key: service_instance_id
+ value:
+ "null": null
+- elementId: 7d589b1fdae3557a
+ tagFamilies:
+ - name: searchable
+ tags:
+ - key: trace_id
+ value:
+ str:
+ value: "4"
+ - key: service_id
+ value:
+ str:
+ value: webapp_id
+ - key: service_instance_id
+ value:
+ "null": null
+- elementId: 2dcb300ca20b88c3
+ tagFamilies:
+ - name: searchable
+ tags:
+ - key: trace_id
+ value:
+ str:
+ value: "5"
+ - key: service_id
+ value:
+ str:
+ value: webapp_id
+ - key: service_instance_id
+ value:
+ "null": null
+- elementId: 47318c4050f63c95
+ tagFamilies:
+ - name: searchable
+ tags:
+ - key: trace_id
+ value:
+ str:
+ value: trace_001
+ - key: service_id
+ value:
+ str:
+ value: service_1
+ - key: service_instance_id
+ value:
+ str:
+ value: instance_1
+- elementId: a1126fa17892e2bd
+ tagFamilies:
+ - name: searchable
+ tags:
+ - key: trace_id
+ value:
+ str:
+ value: trace_002
+ - key: service_id
+ value:
+ str:
+ value: service_2
+ - key: service_instance_id
+ value:
+ str:
+ value: instance_2
+- elementId: cfb608ee34cb7413
+ tagFamilies:
+ - name: searchable
+ tags:
+ - key: trace_id
+ value:
+ str:
+ value: trace_003
+ - key: service_id
+ value:
+ str:
+ value: service_3
+ - key: service_instance_id
+ value:
+ str:
+ value: instance_3
diff --git a/test/cases/stream/stream.go b/test/cases/stream/stream.go
index fcb29051..69cbfd4f 100644
--- a/test/cases/stream/stream.go
+++ b/test/cases/stream/stream.go
@@ -87,4 +87,8 @@ var _ = g.DescribeTable("Scanning Streams", func(args helpers.Args) {
g.Entry("duplicated index filter", helpers.Args{Input: "duplicated_index_filter", Duration: 1 * time.Hour}),
g.Entry("duplicated order by index", helpers.Args{Input: "duplicated_order_by_index", Duration: 1 * time.Hour}),
g.Entry("duplicated order by index with the index filter", helpers.Args{Input: "duplicated_order_by_filter", Duration: 1 * time.Hour}),
+ g.Entry("multi-groups: unchanged tags", helpers.Args{Input: "multi_group_unchanged", Duration: 1 * time.Hour, IgnoreElementID: true}),
+ g.Entry("multi-groups: new tags", helpers.Args{Input: "multi_group_new_tag", Duration: 1 * time.Hour, IgnoreElementID: true}),
+ g.Entry("multi-groups: update tag type", helpers.Args{Input: "multi_group_tag_type", Duration: 1 * time.Hour, IgnoreElementID: true}),
+ g.Entry("multi-groups: sort duration", helpers.Args{Input: "multi_group_sort_duration", Duration: 1 * time.Hour, IgnoreElementID: true}),
)
diff --git a/test/cases/topn/data/input/multi_group_aggr_desc.yaml b/test/cases/topn/data/input/multi_group_aggr_desc.yaml
new file mode 100644
index 00000000..6f5b4775
--- /dev/null
+++ b/test/cases/topn/data/input/multi_group_aggr_desc.yaml
@@ -0,0 +1,22 @@
+# 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.
+
+name: "service_instance_cpm_minute_top_bottom_100"
+groups: ["sw_metric", "sw_updated"]
+topN: 3
+fieldValueSort: 1
+agg: 2
diff --git a/test/cases/topn/topn.go b/test/cases/topn/topn.go
index e1fc8c39..357eeaeb 100644
--- a/test/cases/topn/topn.go
+++ b/test/cases/topn/topn.go
@@ -43,4 +43,5 @@ var _ = g.DescribeTable("TopN Tests", verify,
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}),
+ g.Entry("multi-group: max top3 order by desc", helpers.Args{Input: "multi_group_aggr_desc", Want: "aggr_desc", Duration: 25 * time.Minute, Offset: -20 * time.Minute}),
)