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/08/29 15:33:43 UTC
(skywalking-banyandb) 04/12: Refactor trace query interfaces and improve documentation for filter implementations
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch trace/query
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit 84605febb2d9ba931938deb85ed481942677e80c
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Fri Aug 29 11:22:15 2025 +0800
Refactor trace query interfaces and improve documentation for filter implementations
---
pkg/query/executor/interface.go | 3 +-
pkg/query/logical/trace/index_filter.go | 23 +-
pkg/query/logical/trace/trace_analyzer.go | 274 +++++++++++++++++++++++
pkg/query/logical/trace/trace_plan_local.go | 116 ++++++++++
pkg/query/logical/trace/trace_plan_tag_filter.go | 217 ++++++++++++++++++
5 files changed, 621 insertions(+), 12 deletions(-)
diff --git a/pkg/query/executor/interface.go b/pkg/query/executor/interface.go
index e1dbc4c9..f9ce9717 100644
--- a/pkg/query/executor/interface.go
+++ b/pkg/query/executor/interface.go
@@ -24,7 +24,6 @@ import (
measurev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/measure/v1"
modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
streamv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/stream/v1"
- tracev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/trace/v1"
"github.com/apache/skywalking-banyandb/pkg/bus"
"github.com/apache/skywalking-banyandb/pkg/query/model"
)
@@ -88,6 +87,6 @@ type TraceExecutionContext interface {
// TraceExecutable allows querying in the trace schema.
type TraceExecutable interface {
- Execute(context.Context) ([]*tracev1.Span, error)
+ Execute(context.Context) (model.TraceResult, error)
Close()
}
diff --git a/pkg/query/logical/trace/index_filter.go b/pkg/query/logical/trace/index_filter.go
index 9b2dc9e1..77d1e55c 100644
--- a/pkg/query/logical/trace/index_filter.go
+++ b/pkg/query/logical/trace/index_filter.go
@@ -124,7 +124,7 @@ func parseConditionToFilter(cond *modelv1.Condition, entity []*modelv1.TagValue,
return nil, nil, errors.Errorf("unsupported condition operation: %v", cond.Op)
}
-// traceFilter implements index.Filter for trace queries
+// traceFilter implements index.Filter for trace queries.
type traceFilter struct {
op string
tagName string
@@ -144,7 +144,7 @@ func (tf *traceFilter) String() string {
return tf.op + ":" + tf.tagName
}
-// traceAndFilter implements index.Filter for AND operations in trace queries
+// traceAndFilter implements index.Filter for AND operations in trace queries.
type traceAndFilter struct {
left index.Filter
right index.Filter
@@ -171,7 +171,7 @@ func (taf *traceAndFilter) String() string {
return "and(" + taf.left.String() + "," + taf.right.String() + ")"
}
-// traceOrFilter implements index.Filter for OR operations in trace queries
+// traceOrFilter implements index.Filter for OR operations in trace queries.
type traceOrFilter struct {
left index.Filter
right index.Filter
@@ -198,12 +198,12 @@ func (tof *traceOrFilter) String() string {
return "or(" + tof.left.String() + "," + tof.right.String() + ")"
}
-// traceEqFilter implements index.Filter for EQ operations in trace queries
+// traceEqFilter implements index.Filter for EQ operations in trace queries.
type traceEqFilter struct {
+ expr logical.LiteralExpr
+ cond *modelv1.Condition
op string
tagName string
- cond *modelv1.Condition
- expr logical.LiteralExpr
}
func (tef *traceEqFilter) Execute(_ index.GetSearcher, _ common.SeriesID, _ *index.RangeOpts) (posting.List, posting.List, error) {
@@ -223,12 +223,12 @@ func (tef *traceEqFilter) String() string {
return tef.op + ":" + tef.tagName
}
-// traceRangeFilter implements index.Filter for range operations in trace queries
+// traceRangeFilter implements index.Filter for range operations in trace queries.
type traceRangeFilter struct {
+ expr logical.LiteralExpr
+ cond *modelv1.Condition
op string
tagName string
- cond *modelv1.Condition
- expr logical.LiteralExpr
}
func (trf *traceRangeFilter) Execute(_ index.GetSearcher, _ common.SeriesID, _ *index.RangeOpts) (posting.List, posting.List, error) {
@@ -248,6 +248,9 @@ func (trf *traceRangeFilter) ShouldSkip(tagFilters index.FilterOp) (bool, error)
opts = trf.expr.RangeOpts(true, false, false)
case modelv1.Condition_BINARY_OP_LE:
opts = trf.expr.RangeOpts(true, false, true)
+ default:
+ // For non-range operations, return false (don't skip)
+ return false, nil
}
return tagFilters.Range(trf.tagName, opts)
}
@@ -258,7 +261,7 @@ func (trf *traceRangeFilter) String() string {
return trf.op + ":" + trf.tagName
}
-// traceMatchFilter implements index.Filter for MATCH operations in trace queries
+// traceMatchFilter implements index.Filter for MATCH operations in trace queries.
type traceMatchFilter struct {
op string
tagName string
diff --git a/pkg/query/logical/trace/trace_analyzer.go b/pkg/query/logical/trace/trace_analyzer.go
new file mode 100644
index 00000000..ee5bff71
--- /dev/null
+++ b/pkg/query/logical/trace/trace_analyzer.go
@@ -0,0 +1,274 @@
+// 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 trace
+
+import (
+ "context"
+ "fmt"
+
+ commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
+ tracev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/trace/v1"
+ "github.com/apache/skywalking-banyandb/pkg/query/executor"
+ "github.com/apache/skywalking-banyandb/pkg/query/logical"
+ "github.com/apache/skywalking-banyandb/pkg/query/model"
+)
+
+const defaultLimit uint32 = 20
+
+// Analyze converts logical expressions to executable operation tree represented by Plan.
+func Analyze(criteria *tracev1.QueryRequest, metadata []*commonv1.Metadata, ss []logical.Schema, ecc []executor.TraceExecutionContext) (logical.Plan, error) {
+ // parse fields
+ 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 := convertStringProjectionToTags(criteria.GetTagProjection())
+ if len(metadata) == 1 {
+ plan = parseTraceTags(criteria, metadata[0], ecc[0], tagProjection)
+ s = ss[0]
+ } else {
+ var err error
+ if s, err = mergeSchema(ss); err != nil {
+ return nil, err
+ }
+ plan = &unresolvedTraceMerger{
+ criteria: criteria,
+ metadata: metadata,
+ ecc: ecc,
+ tagProjection: tagProjection,
+ }
+ }
+
+ // parse limit
+ limitParameter := criteria.GetLimit()
+ if limitParameter == 0 {
+ limitParameter = defaultLimit
+ }
+ plan = newTraceLimit(plan, criteria.GetOffset(), limitParameter)
+
+ p, err := plan.Analyze(s)
+ if err != nil {
+ return nil, err
+ }
+ rules := []logical.OptimizeRule{
+ logical.NewPushDownOrder(criteria.OrderBy),
+ logical.NewPushDownMaxSize(int(limitParameter + criteria.GetOffset())),
+ }
+ if err := logical.ApplyRules(p, rules...); err != nil {
+ return nil, err
+ }
+ return p, nil
+}
+
+// DistributedAnalyze converts logical expressions to executable operation tree represented by Plan.
+func DistributedAnalyze(criteria *tracev1.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 := newUnresolvedTraceDistributed(criteria)
+
+ // parse limit
+ limitParameter := criteria.GetLimit()
+ if limitParameter == 0 {
+ limitParameter = defaultLimit
+ }
+ plan = newDistributedTraceLimit(plan, criteria.Offset, limitParameter)
+ return plan.Analyze(s)
+}
+
+var (
+ _ logical.Plan = (*traceLimit)(nil)
+ _ logical.UnresolvedPlan = (*traceLimit)(nil)
+ _ executor.TraceExecutable = (*traceLimit)(nil)
+)
+
+// Parent refers to a parent node in the execution tree(plan).
+type Parent struct {
+ UnresolvedInput logical.UnresolvedPlan
+ Input logical.Plan
+}
+
+type traceLimit struct {
+ *Parent
+ limitNum uint32
+ offsetNum uint32
+}
+
+func (l *traceLimit) Close() {
+ l.Parent.Input.(executor.TraceExecutable).Close()
+}
+
+func (l *traceLimit) Execute(ctx context.Context) (model.TraceResult, error) {
+ // For trace queries, we typically get one result per execution
+ // The limit and offset are handled at the span level within each trace
+ result, err := l.Parent.Input.(executor.TraceExecutable).Execute(ctx)
+ if err != nil {
+ return model.TraceResult{}, err
+ }
+
+ // Apply offset and limit to spans within the trace
+ if len(result.Spans) == 0 {
+ return result, nil
+ }
+
+ offset := int(l.offsetNum)
+ limit := int(l.limitNum)
+
+ if offset >= len(result.Spans) {
+ return model.TraceResult{
+ Error: result.Error,
+ Spans: [][]byte{},
+ TID: result.TID,
+ Tags: result.Tags,
+ }, nil
+ }
+
+ endIndex := offset + limit
+ if endIndex > len(result.Spans) {
+ endIndex = len(result.Spans)
+ }
+
+ return model.TraceResult{
+ Error: result.Error,
+ Spans: result.Spans[offset:endIndex],
+ TID: result.TID,
+ Tags: result.Tags,
+ }, nil
+}
+
+func (l *traceLimit) Analyze(s logical.Schema) (logical.Plan, error) {
+ var err error
+ l.Input, err = l.UnresolvedInput.Analyze(s)
+ if err != nil {
+ return nil, err
+ }
+ return l, nil
+}
+
+func (l *traceLimit) Schema() logical.Schema {
+ return l.Input.Schema()
+}
+
+func (l *traceLimit) String() string {
+ return fmt.Sprintf("%s TraceLimit: %d", l.Input.String(), l.limitNum)
+}
+
+func (l *traceLimit) Children() []logical.Plan {
+ return []logical.Plan{l.Input}
+}
+
+func newTraceLimit(input logical.UnresolvedPlan, offset, num uint32) logical.UnresolvedPlan {
+ return &traceLimit{
+ Parent: &Parent{
+ UnresolvedInput: input,
+ },
+ offsetNum: offset,
+ limitNum: num,
+ }
+}
+
+func parseTraceTags(criteria *tracev1.QueryRequest, metadata *commonv1.Metadata,
+ ec executor.TraceExecutionContext, tagProjection [][]*logical.Tag,
+) logical.UnresolvedPlan {
+ timeRange := criteria.GetTimeRange()
+ return &unresolvedTraceTagFilter{
+ startTime: timeRange.GetBegin().AsTime(),
+ endTime: timeRange.GetEnd().AsTime(),
+ metadata: metadata,
+ criteria: criteria.Criteria,
+ projectionTags: tagProjection,
+ ec: ec,
+ }
+}
+
+// convertStringProjectionToTags converts a string array projection to tag projection format.
+// For trace, we create a single tag family with empty name since traces don't have families.
+func convertStringProjectionToTags(tagNames []string) [][]*logical.Tag {
+ if len(tagNames) == 0 {
+ return nil
+ }
+
+ // For trace, create a single tag family (empty name) with all tags
+ tags := make([]*logical.Tag, len(tagNames))
+ for i, tagName := range tagNames {
+ tags[i] = logical.NewTag("", tagName) // Empty family name for trace
+ }
+
+ return [][]*logical.Tag{tags}
+}
+
+// Placeholder implementations for distributed query components.
+var _ logical.UnresolvedPlan = (*unresolvedTraceMerger)(nil)
+
+type unresolvedTraceMerger struct {
+ criteria *tracev1.QueryRequest
+ metadata []*commonv1.Metadata
+ ecc []executor.TraceExecutionContext
+ tagProjection [][]*logical.Tag
+}
+
+func (utm *unresolvedTraceMerger) Analyze(s logical.Schema) (logical.Plan, error) {
+ // TODO: Implement trace merger logic
+ return nil, fmt.Errorf("trace merger not implemented yet")
+}
+
+func newUnresolvedTraceDistributed(criteria *tracev1.QueryRequest) logical.UnresolvedPlan {
+ // TODO: Implement distributed trace query
+ return &unresolvedTraceDistributed{criteria: criteria}
+}
+
+var _ logical.UnresolvedPlan = (*unresolvedTraceDistributed)(nil)
+
+type unresolvedTraceDistributed struct {
+ criteria *tracev1.QueryRequest
+}
+
+func (utd *unresolvedTraceDistributed) Analyze(s logical.Schema) (logical.Plan, error) {
+ // TODO: Implement distributed trace analysis
+ return nil, fmt.Errorf("distributed trace query not implemented yet")
+}
+
+func newDistributedTraceLimit(input logical.UnresolvedPlan, offset, num uint32) logical.UnresolvedPlan {
+ // TODO: Implement distributed trace limit
+ return &distributedTraceLimit{
+ input: input,
+ offsetNum: offset,
+ limitNum: num,
+ }
+}
+
+var _ logical.UnresolvedPlan = (*distributedTraceLimit)(nil)
+
+type distributedTraceLimit struct {
+ input logical.UnresolvedPlan
+ offsetNum uint32
+ limitNum uint32
+}
+
+func (dtl *distributedTraceLimit) Analyze(s logical.Schema) (logical.Plan, error) {
+ // TODO: Implement distributed trace limit analysis
+ return nil, fmt.Errorf("distributed trace limit not implemented yet")
+}
diff --git a/pkg/query/logical/trace/trace_plan_local.go b/pkg/query/logical/trace/trace_plan_local.go
new file mode 100644
index 00000000..15552f39
--- /dev/null
+++ b/pkg/query/logical/trace/trace_plan_local.go
@@ -0,0 +1,116 @@
+// 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 trace
+
+import (
+ "context"
+ "fmt"
+
+ commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
+ "github.com/apache/skywalking-banyandb/pkg/index"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ "github.com/apache/skywalking-banyandb/pkg/query/executor"
+ "github.com/apache/skywalking-banyandb/pkg/query/logical"
+ "github.com/apache/skywalking-banyandb/pkg/query/model"
+ "github.com/apache/skywalking-banyandb/pkg/timestamp"
+)
+
+var (
+ _ logical.Plan = (*localScan)(nil)
+ _ logical.Sorter = (*localScan)(nil)
+ _ logical.VolumeLimiter = (*localScan)(nil)
+ _ executor.TraceExecutable = (*localScan)(nil)
+)
+
+type localScan struct {
+ schema logical.Schema
+ skippingFilter index.Filter
+ result model.TraceQueryResult
+ ec executor.TraceExecutionContext
+ order *logical.OrderBy
+ metadata *commonv1.Metadata
+ l *logger.Logger
+ projectionTags *model.TagProjection
+ timeRange timestamp.TimeRange
+ projectionTagRefs [][]*logical.TagRef
+ maxTraceSize int
+}
+
+func (i *localScan) Close() {
+ if i.result != nil {
+ i.result.Release()
+ }
+}
+
+func (i *localScan) Limit(maxVal int) {
+ i.maxTraceSize = maxVal
+}
+
+func (i *localScan) Sort(order *logical.OrderBy) {
+ i.order = order
+}
+
+func (i *localScan) Execute(ctx context.Context) (model.TraceResult, error) {
+ select {
+ case <-ctx.Done():
+ return model.TraceResult{}, ctx.Err()
+ default:
+ }
+ if i.result != nil {
+ return *i.result.Pull(), nil
+ }
+ var orderBy *index.OrderBy
+ if i.order != nil {
+ orderBy = &index.OrderBy{
+ Index: i.order.Index,
+ Sort: i.order.Sort,
+ }
+ }
+ var err error
+ if i.result, err = i.ec.Query(ctx, model.TraceQueryOptions{
+ Name: i.metadata.GetName(),
+ TimeRange: &i.timeRange,
+ SkippingFilter: i.skippingFilter,
+ Order: orderBy,
+ TagProjection: i.projectionTags,
+ MaxTraceSize: i.maxTraceSize,
+ }); err != nil {
+ return model.TraceResult{}, err
+ }
+ if i.result == nil {
+ return model.TraceResult{}, nil
+ }
+ return *i.result.Pull(), nil
+}
+
+func (i *localScan) String() string {
+ return fmt.Sprintf("TraceScan: startTime=%d,endTime=%d,Metadata{group=%s,name=%s},conditions=%s; projection=%s; orderBy=%s; limit=%d",
+ i.timeRange.Start.Unix(), i.timeRange.End.Unix(), i.metadata.GetGroup(), i.metadata.GetName(),
+ i.skippingFilter, logical.FormatTagRefs(", ", i.projectionTagRefs...), i.order, i.maxTraceSize)
+}
+
+func (i *localScan) Children() []logical.Plan {
+ return []logical.Plan{}
+}
+
+func (i *localScan) Schema() logical.Schema {
+ if len(i.projectionTagRefs) == 0 {
+ return i.schema
+ }
+ return i.schema.ProjTags(i.projectionTagRefs...)
+}
diff --git a/pkg/query/logical/trace/trace_plan_tag_filter.go b/pkg/query/logical/trace/trace_plan_tag_filter.go
new file mode 100644
index 00000000..7107cb8e
--- /dev/null
+++ b/pkg/query/logical/trace/trace_plan_tag_filter.go
@@ -0,0 +1,217 @@
+// 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 trace
+
+import (
+ "context"
+ "fmt"
+ "time"
+
+ commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
+ modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
+ "github.com/apache/skywalking-banyandb/pkg/index"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ "github.com/apache/skywalking-banyandb/pkg/query/executor"
+ "github.com/apache/skywalking-banyandb/pkg/query/logical"
+ "github.com/apache/skywalking-banyandb/pkg/query/model"
+ "github.com/apache/skywalking-banyandb/pkg/timestamp"
+)
+
+var _ logical.UnresolvedPlan = (*unresolvedTraceTagFilter)(nil)
+
+type unresolvedTraceTagFilter struct {
+ startTime time.Time
+ endTime time.Time
+ ec executor.TraceExecutionContext
+ metadata *commonv1.Metadata
+ criteria *modelv1.Criteria
+ projectionTags [][]*logical.Tag
+}
+
+func (uis *unresolvedTraceTagFilter) Analyze(s logical.Schema) (logical.Plan, error) {
+ ctx := newTraceAnalyzerContext(s)
+ entityList := s.EntityList()
+ entityDict := make(map[string]int)
+ for idx, e := range entityList {
+ entityDict[e] = idx
+ }
+ var err error
+ // For trace, we only use skipping filter (no inverted filter or entities)
+ ctx.skippingFilter, err = buildTraceFilter(uis.criteria, s, entityDict)
+ if err != nil {
+ return nil, err
+ }
+
+ if len(uis.projectionTags) > 0 {
+ // For trace, we use a single TagProjection since traces don't have tag families
+ ctx.projectionTags = &model.TagProjection{
+ Family: "", // Empty family name for trace
+ Names: make([]string, 0),
+ }
+ for i := range uis.projectionTags {
+ for _, tag := range uis.projectionTags[i] {
+ ctx.projectionTags.Names = append(ctx.projectionTags.Names, tag.GetTagName())
+ }
+ }
+ var errProject error
+ ctx.projTagsRefs, errProject = s.CreateTagRef(uis.projectionTags...)
+ if errProject != nil {
+ return nil, errProject
+ }
+ }
+ plan := uis.selectTraceScanner(ctx, uis.ec)
+ if uis.criteria != nil {
+ tagFilter, errFilter := logical.BuildTagFilter(uis.criteria, entityDict, s, len(ctx.globalConditions) > 1)
+ if errFilter != nil {
+ return nil, errFilter
+ }
+ if tagFilter != logical.DummyFilter {
+ // create tagFilter with a projected view
+ plan = newTraceTagFilter(s.ProjTags(ctx.projTagsRefs...), plan, tagFilter)
+ }
+ }
+ return plan, err
+}
+
+func (uis *unresolvedTraceTagFilter) selectTraceScanner(ctx *traceAnalyzeContext, ec executor.TraceExecutionContext) logical.Plan {
+ return &localScan{
+ timeRange: timestamp.NewInclusiveTimeRange(uis.startTime, uis.endTime),
+ schema: ctx.s,
+ projectionTagRefs: ctx.projTagsRefs,
+ projectionTags: ctx.projectionTags,
+ metadata: uis.metadata,
+ skippingFilter: ctx.skippingFilter,
+ l: logger.GetLogger("query", "trace", "local-scan"),
+ ec: ec,
+ }
+}
+
+func traceTagFilter(startTime, endTime time.Time, metadata *commonv1.Metadata, criteria *modelv1.Criteria,
+ projection [][]*logical.Tag, ec executor.TraceExecutionContext,
+) logical.UnresolvedPlan {
+ return &unresolvedTraceTagFilter{
+ startTime: startTime,
+ endTime: endTime,
+ metadata: metadata,
+ criteria: criteria,
+ projectionTags: projection,
+ ec: ec,
+ }
+}
+
+type traceAnalyzeContext struct {
+ s logical.Schema
+ skippingFilter index.Filter
+ projectionTags *model.TagProjection
+ globalConditions []interface{}
+ projTagsRefs [][]*logical.TagRef
+}
+
+func newTraceAnalyzerContext(s logical.Schema) *traceAnalyzeContext {
+ return &traceAnalyzeContext{
+ globalConditions: make([]interface{}, 0),
+ s: s,
+ }
+}
+
+// buildTraceFilter builds a filter for trace queries.
+// Unlike stream, trace only needs skipping filter.
+func buildTraceFilter(criteria *modelv1.Criteria, s logical.Schema, entityDict map[string]int) (index.Filter, error) {
+ if criteria == nil {
+ return nil, nil
+ }
+ // Create a map of valid tag names from the schema
+ tagNames := make(map[string]bool)
+ // For trace schema, tags are stored directly (no tag families)
+ for tagName := range s.(*schema).common.TagSpecMap {
+ tagNames[tagName] = true
+ }
+
+ filter, _, err := buildFilter(criteria, tagNames, entityDict, nil)
+ return filter, err
+}
+
+var (
+ _ logical.Plan = (*traceTagFilterPlan)(nil)
+ _ executor.TraceExecutable = (*traceTagFilterPlan)(nil)
+)
+
+type traceTagFilterPlan struct {
+ s logical.Schema
+ parent logical.Plan
+ tagFilter logical.TagFilter
+}
+
+func (t *traceTagFilterPlan) Close() {
+ t.parent.(executor.TraceExecutable).Close()
+}
+
+func newTraceTagFilter(s logical.Schema, parent logical.Plan, tagFilter logical.TagFilter) logical.Plan {
+ return &traceTagFilterPlan{
+ s: s,
+ parent: parent,
+ tagFilter: tagFilter,
+ }
+}
+
+func (t *traceTagFilterPlan) Execute(ctx context.Context) (model.TraceResult, error) {
+ result, err := t.parent.(executor.TraceExecutable).Execute(ctx)
+ if err != nil {
+ return model.TraceResult{}, err
+ }
+
+ // For trace, we need to check if the result matches the tag filter
+ // Convert trace tags to TagFamilies format for filter matching
+ var tagFamilies []*modelv1.TagFamily
+ if len(result.Tags) > 0 {
+ // Create a single tag family for trace tags (trace has no families)
+ family := &modelv1.TagFamily{
+ Name: "",
+ Tags: make([]*modelv1.Tag, len(result.Tags)),
+ }
+ for i, tag := range result.Tags {
+ family.Tags[i] = &modelv1.Tag{
+ Key: tag.Name,
+ Value: tag.Values[0], // Use the first value from the tag's Values slice
+ }
+ }
+ tagFamilies = []*modelv1.TagFamily{family}
+ }
+
+ ok, err := t.tagFilter.Match(logical.TagFamilies(tagFamilies), t.s)
+ if err != nil {
+ return model.TraceResult{}, err
+ }
+ if !ok {
+ return model.TraceResult{}, nil
+ }
+
+ return result, nil
+}
+
+func (t *traceTagFilterPlan) String() string {
+ return fmt.Sprintf("%s trace-tag-filter:%s", t.parent, t.tagFilter.String())
+}
+
+func (t *traceTagFilterPlan) Children() []logical.Plan {
+ return []logical.Plan{t.parent}
+}
+
+func (t *traceTagFilterPlan) Schema() logical.Schema {
+ return t.s
+}