hanahmily commented on code in PR #1190:
URL:
https://github.com/apache/skywalking-banyandb/pull/1190#discussion_r3464098813
##########
pkg/query/logical/trace/trace_plan_distributed.go:
##########
@@ -228,72 +245,80 @@ func (p *distributedPlan) Limit(maxVal int) {
p.maxTraceSize = uint32(maxVal)
}
-var _ sort.Comparable = (*comparableTrace)(nil)
-
-type comparableTrace struct {
- *tracev1.InternalTrace
- sortField []byte
-}
-
-func newComparableTrace(t *tracev1.InternalTrace, sortByTraceID bool)
(*comparableTrace, error) {
- var sortField []byte
- if sortByTraceID {
- sortField = []byte(t.TraceId)
- } else {
- sortField = convert.Int64ToBytes(t.Key)
+// internalTraceToResult converts a proto *tracev1.InternalTrace (row-oriented:
+// per-span tag lists) into a columnar model.TraceResult (per-tag value column
+// aligned across spans, NullTagValue for spans missing a tag). It is the
inverse
+// of the data-node frame build and matches the legacy iterator's
column-building.
+func internalTraceToResult(trace *tracev1.InternalTrace) model.TraceResult {
+ result := model.TraceResult{
+ TID: trace.TraceId,
+ Key: trace.Key,
+ Spans: make([][]byte, 0, len(trace.Spans)),
+ SpanIDs: make([]string, 0, len(trace.Spans)),
}
-
- return &comparableTrace{
- InternalTrace: t,
- sortField: sortField,
- }, nil
-}
-
-func (t *comparableTrace) SortedField() []byte {
- return t.sortField
-}
-
-var _ sort.Iterator[*comparableTrace] = (*sortableTraces)(nil)
-
-type sortableTraces struct {
- cur *comparableTrace
- traces []*tracev1.InternalTrace
- index int
- isSortByTraceID bool
-}
-
-func newSortableTraces(traces []*tracev1.InternalTrace, isSortByTraceID bool)
*sortableTraces {
- return &sortableTraces{
- traces: traces,
- isSortByTraceID: isSortByTraceID,
+ tagValuesByName := make(map[string][]*modelv1.TagValue)
+ var tagOrder []string
+ for spanIdx, span := range trace.Spans {
+ result.Spans = append(result.Spans, span.Span)
+ result.SpanIDs = append(result.SpanIDs, span.SpanId)
+ spanTags := make(map[string]*modelv1.TagValue, len(span.Tags))
+ for _, tag := range span.Tags {
+ if _, exists := tagValuesByName[tag.Key]; !exists {
+ tagOrder = append(tagOrder, tag.Key)
+ tagValuesByName[tag.Key] =
make([]*modelv1.TagValue, spanIdx)
+ for fillIdx := range tagValuesByName[tag.Key] {
+ tagValuesByName[tag.Key][fillIdx] =
pbv1.NullTagValue
+ }
+ }
+ spanTags[tag.Key] = tag.Value
+ }
+ for _, tagName := range tagOrder {
+ tagValue, exists := spanTags[tagName]
+ if !exists {
+ tagValue = pbv1.NullTagValue
+ }
+ tagValuesByName[tagName] =
append(tagValuesByName[tagName], tagValue)
+ }
}
+ for _, tagName := range tagOrder {
+ result.Tags = append(result.Tags, model.Tag{Name: tagName,
Values: tagValuesByName[tagName]})
+ }
+ return result
}
-func (*sortableTraces) Close() error {
- return nil
-}
-
-func (t *sortableTraces) Next() bool {
- return t.iter(func(it *tracev1.InternalTrace) (*comparableTrace, error)
{
- return newComparableTrace(it, t.isSortByTraceID)
- })
-}
-
-func (t *sortableTraces) Val() *comparableTrace {
- return t.cur
+// mergeTraceResultSpans merges spans from src into dst (same TID): a span is
+// appended only when its SpanID is not already present in dst, and the
matching
+// per-tag value column is extended in lockstep. This mirrors the legacy
+// InternalTrace span-dedup but on the columnar model.TraceResult.
+func mergeTraceResultSpans(dst, src *model.TraceResult) {
+ existing := make(map[string]struct{}, len(dst.SpanIDs))
+ for _, spanID := range dst.SpanIDs {
+ existing[spanID] = struct{}{}
+ }
+ for srcIdx, srcSpanID := range src.SpanIDs {
+ if _, ok := existing[srcSpanID]; ok {
+ continue
+ }
+ existing[srcSpanID] = struct{}{}
+ dst.SpanIDs = append(dst.SpanIDs, srcSpanID)
+ dst.Spans = append(dst.Spans, src.Spans[srcIdx])
+ for tagIdx := range dst.Tags {
+ value := pbv1.NullTagValue
+ if srcTag, ok := findTag(src.Tags,
dst.Tags[tagIdx].Name); ok && srcIdx < len(srcTag.Values) {
+ value = srcTag.Values[srcIdx]
+ }
+ dst.Tags[tagIdx].Values =
append(dst.Tags[tagIdx].Values, value)
+ }
+ }
}
Review Comment:
Good catch — fixed in 51976fd6. `mergeTraceResultSpans` now unions src-only
tag columns into `dst` first, `NullTagValue`-backfilling the spans `dst`
already holds so every column stays aligned with `dst.SpanIDs`, before the
existing append loop populates them. Covered by the new
`TestMergeTraceResultSpansUnionsSrcOnlyTagColumns`.
##########
pkg/query/logical/trace/trace_frame.go:
##########
@@ -0,0 +1,325 @@
+// 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 (
+ "encoding/binary"
+ "fmt"
+
+ "google.golang.org/protobuf/proto"
+
+ "github.com/apache/skywalking-banyandb/api/data"
+ modelv1
"github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
+ "github.com/apache/skywalking-banyandb/pkg/query/model"
+)
+
+// traceFrameVersion is the on-wire version of the columnar trace result frame.
+const traceFrameVersion byte = 1
+
+// Tag-value type markers inside the columnar trace frame. Common scalar
variants
+// are encoded directly (no protobuf reflection / oneof dispatch); anything
else
+// (arrays, future variants) falls back to a per-cell proto.Marshal under
+// traceTagProto.
+const (
+ traceTagNull byte = 0
+ traceTagStr byte = 1
+ traceTagInt byte = 2
+ traceTagBinary byte = 3
+ traceTagProto byte = 4
+)
+
+// EncodeTraceResultFrame serializes a columnar slice of model.TraceResult
into a
+// magic+version-prefixed frame: length-delimited traces, each with its spans
and
+// columnar tags (one TagValue per span). It avoids the protobuf
+// message-slice + oneof machinery that dominates the data↔liaison hop.
+func EncodeTraceResultFrame(results []model.TraceResult) []byte {
+ // Pre-size to the exact encoded length so the buffer is allocated once
with no
+ // doubling/realloc churn (the realloc churn was the source of the
native path's
+ // excess bytes/RSS vs the proto path).
+ buf := make([]byte, 0, traceFrameSize(results))
+ buf = append(buf, data.RawFrameMagicLeadingByte, traceFrameVersion)
+ buf = binary.AppendUvarint(buf, uint64(len(results)))
+ for resultIdx := range results {
+ result := &results[resultIdx]
+ buf = appendLenString(buf, result.TID)
+ buf = appendFixed64(buf, uint64(result.Key))
+ buf = binary.AppendUvarint(buf, uint64(len(result.Spans)))
+ for spanIdx, spanBytes := range result.Spans {
+ buf = appendLenBytes(buf, spanBytes)
+ spanID := ""
+ if spanIdx < len(result.SpanIDs) {
+ spanID = result.SpanIDs[spanIdx]
+ }
+ buf = appendLenString(buf, spanID)
+ }
+ buf = binary.AppendUvarint(buf, uint64(len(result.Tags)))
+ for tagIdx := range result.Tags {
+ tag := &result.Tags[tagIdx]
+ buf = appendLenString(buf, tag.Name)
+ buf = binary.AppendUvarint(buf, uint64(len(tag.Values)))
+ for _, tagValue := range tag.Values {
+ buf = appendTraceTagValue(buf, tagValue)
+ }
+ }
+ }
+ return buf
+}
+
+// traceFrameSize computes the exact encoded byte length of results, mirroring
+// EncodeTraceResultFrame field-for-field so the encode buffer is pre-sized
exactly.
+func traceFrameSize(results []model.TraceResult) int {
+ size := 2 + uvarintLen(uint64(len(results)))
+ for resultIdx := range results {
+ result := &results[resultIdx]
+ size += lenStringSize(result.TID) + 8 +
uvarintLen(uint64(len(result.Spans)))
+ for spanIdx, spanBytes := range result.Spans {
+ size += uvarintLen(uint64(len(spanBytes))) +
len(spanBytes)
+ spanIDLen := 0
+ if spanIdx < len(result.SpanIDs) {
+ spanIDLen = len(result.SpanIDs[spanIdx])
+ }
+ size += uvarintLen(uint64(spanIDLen)) + spanIDLen
+ }
+ size += uvarintLen(uint64(len(result.Tags)))
+ for tagIdx := range result.Tags {
+ tag := &result.Tags[tagIdx]
+ size += lenStringSize(tag.Name) +
uvarintLen(uint64(len(tag.Values)))
+ for _, tagValue := range tag.Values {
+ size += 1 + traceTagValueSize(tagValue)
+ }
+ }
+ }
+ return size
+}
+
+// traceTagValueSize returns the encoded byte length of one tag value, matching
+// appendTraceTagValue (1 type byte + payload).
+func traceTagValueSize(tv *modelv1.TagValue) int {
+ switch v := tv.GetValue().(type) {
+ case nil, *modelv1.TagValue_Null:
+ return 0
+ case *modelv1.TagValue_Str:
+ return lenStringSize(v.Str.GetValue())
+ case *modelv1.TagValue_Int:
+ return 8
+ case *modelv1.TagValue_BinaryData:
+ return uvarintLen(uint64(len(v.BinaryData))) + len(v.BinaryData)
+ default:
+ marshaledSize := proto.Size(tv)
+ return uvarintLen(uint64(marshaledSize)) + marshaledSize
+ }
+}
Review Comment:
These paths are already nil-safe via the generated proto getters, so the
noted cases cannot panic: `tv.GetValue()` on a nil `*TagValue` returns `nil`
(the getter guards `if x != nil`), which falls into the `case nil` branch and
returns 0; and `v.Str.GetValue()` on a nil `*Str` returns `""` through the same
nil-receiver guard. The `*TagValue_Int` branch returns a constant 8 without
dereferencing `v.Int`. So both a nil `tv` and a nil nested message are handled.
Leaving as-is.
##########
pkg/query/logical/trace/trace_frame.go:
##########
@@ -0,0 +1,325 @@
+// 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 (
+ "encoding/binary"
+ "fmt"
+
+ "google.golang.org/protobuf/proto"
+
+ "github.com/apache/skywalking-banyandb/api/data"
+ modelv1
"github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
+ "github.com/apache/skywalking-banyandb/pkg/query/model"
+)
+
+// traceFrameVersion is the on-wire version of the columnar trace result frame.
+const traceFrameVersion byte = 1
+
+// Tag-value type markers inside the columnar trace frame. Common scalar
variants
+// are encoded directly (no protobuf reflection / oneof dispatch); anything
else
+// (arrays, future variants) falls back to a per-cell proto.Marshal under
+// traceTagProto.
+const (
+ traceTagNull byte = 0
+ traceTagStr byte = 1
+ traceTagInt byte = 2
+ traceTagBinary byte = 3
+ traceTagProto byte = 4
+)
+
+// EncodeTraceResultFrame serializes a columnar slice of model.TraceResult
into a
+// magic+version-prefixed frame: length-delimited traces, each with its spans
and
+// columnar tags (one TagValue per span). It avoids the protobuf
+// message-slice + oneof machinery that dominates the data↔liaison hop.
+func EncodeTraceResultFrame(results []model.TraceResult) []byte {
+ // Pre-size to the exact encoded length so the buffer is allocated once
with no
+ // doubling/realloc churn (the realloc churn was the source of the
native path's
+ // excess bytes/RSS vs the proto path).
+ buf := make([]byte, 0, traceFrameSize(results))
+ buf = append(buf, data.RawFrameMagicLeadingByte, traceFrameVersion)
+ buf = binary.AppendUvarint(buf, uint64(len(results)))
+ for resultIdx := range results {
+ result := &results[resultIdx]
+ buf = appendLenString(buf, result.TID)
+ buf = appendFixed64(buf, uint64(result.Key))
+ buf = binary.AppendUvarint(buf, uint64(len(result.Spans)))
+ for spanIdx, spanBytes := range result.Spans {
+ buf = appendLenBytes(buf, spanBytes)
+ spanID := ""
+ if spanIdx < len(result.SpanIDs) {
+ spanID = result.SpanIDs[spanIdx]
+ }
+ buf = appendLenString(buf, spanID)
+ }
+ buf = binary.AppendUvarint(buf, uint64(len(result.Tags)))
+ for tagIdx := range result.Tags {
+ tag := &result.Tags[tagIdx]
+ buf = appendLenString(buf, tag.Name)
+ buf = binary.AppendUvarint(buf, uint64(len(tag.Values)))
+ for _, tagValue := range tag.Values {
+ buf = appendTraceTagValue(buf, tagValue)
+ }
+ }
+ }
+ return buf
+}
+
+// traceFrameSize computes the exact encoded byte length of results, mirroring
+// EncodeTraceResultFrame field-for-field so the encode buffer is pre-sized
exactly.
+func traceFrameSize(results []model.TraceResult) int {
+ size := 2 + uvarintLen(uint64(len(results)))
+ for resultIdx := range results {
+ result := &results[resultIdx]
+ size += lenStringSize(result.TID) + 8 +
uvarintLen(uint64(len(result.Spans)))
+ for spanIdx, spanBytes := range result.Spans {
+ size += uvarintLen(uint64(len(spanBytes))) +
len(spanBytes)
+ spanIDLen := 0
+ if spanIdx < len(result.SpanIDs) {
+ spanIDLen = len(result.SpanIDs[spanIdx])
+ }
+ size += uvarintLen(uint64(spanIDLen)) + spanIDLen
+ }
+ size += uvarintLen(uint64(len(result.Tags)))
+ for tagIdx := range result.Tags {
+ tag := &result.Tags[tagIdx]
+ size += lenStringSize(tag.Name) +
uvarintLen(uint64(len(tag.Values)))
+ for _, tagValue := range tag.Values {
+ size += 1 + traceTagValueSize(tagValue)
+ }
+ }
+ }
+ return size
+}
+
+// traceTagValueSize returns the encoded byte length of one tag value, matching
+// appendTraceTagValue (1 type byte + payload).
+func traceTagValueSize(tv *modelv1.TagValue) int {
+ switch v := tv.GetValue().(type) {
+ case nil, *modelv1.TagValue_Null:
+ return 0
+ case *modelv1.TagValue_Str:
+ return lenStringSize(v.Str.GetValue())
+ case *modelv1.TagValue_Int:
+ return 8
+ case *modelv1.TagValue_BinaryData:
+ return uvarintLen(uint64(len(v.BinaryData))) + len(v.BinaryData)
+ default:
+ marshaledSize := proto.Size(tv)
+ return uvarintLen(uint64(marshaledSize)) + marshaledSize
+ }
+}
+
+func lenStringSize(s string) int {
+ return uvarintLen(uint64(len(s))) + len(s)
+}
+
+func uvarintLen(x uint64) int {
+ n := 1
+ for x >= 0x80 {
+ x >>= 7
+ n++
+ }
+ return n
+}
+
+// DecodeTraceResultFrame reconstructs the columnar slice of model.TraceResult
+// from a frame body. All bytes/strings are copied because the body buffer is
+// reused after this call returns.
+func DecodeTraceResultFrame(body []byte) ([]model.TraceResult, error) {
+ if len(body) < 2 || body[0] != data.RawFrameMagicLeadingByte {
+ return nil, fmt.Errorf("trace result frame: invalid magic")
+ }
+ if body[1] != traceFrameVersion {
+ return nil, fmt.Errorf("trace result frame: unsupported version
%d", body[1])
+ }
+ cur := &frameCursor{b: body, pos: 2}
+ nTraces := cur.uvarint()
+ results := make([]model.TraceResult, 0, nTraces)
+ for traceIdx := 0; traceIdx < nTraces && cur.err == nil; traceIdx++ {
+ tid := cur.str()
+ key := int64(cur.fixed64())
+ nSpans := cur.uvarint()
+ spans := make([][]byte, 0, nSpans)
+ spanIDs := make([]string, 0, nSpans)
+ for spanIdx := 0; spanIdx < nSpans && cur.err == nil; spanIdx++
{
+ spans = append(spans, cur.bytesCopy())
+ spanIDs = append(spanIDs, cur.str())
+ }
+ nTags := cur.uvarint()
+ var tags []model.Tag
+ if nTags > 0 {
+ tags = make([]model.Tag, 0, nTags)
+ }
+ for tagIdx := 0; tagIdx < nTags && cur.err == nil; tagIdx++ {
+ tagName := cur.str()
+ nValues := cur.uvarint()
+ values := make([]*modelv1.TagValue, 0, nValues)
+ for valueIdx := 0; valueIdx < nValues && cur.err ==
nil; valueIdx++ {
+ values = append(values, cur.tagValue())
+ }
+ tags = append(tags, model.Tag{Name: tagName, Values:
values})
+ }
+ results = append(results, model.TraceResult{TID: tid, Key: key,
Spans: spans, SpanIDs: spanIDs, Tags: tags})
+ }
+ if cur.err != nil {
+ return nil, cur.err
+ }
+ return results, nil
+}
+
+func appendLenString(buf []byte, s string) []byte {
+ buf = binary.AppendUvarint(buf, uint64(len(s)))
+ return append(buf, s...)
+}
+
+func appendLenBytes(buf, b []byte) []byte {
+ buf = binary.AppendUvarint(buf, uint64(len(b)))
+ return append(buf, b...)
+}
+
+func appendFixed64(buf []byte, v uint64) []byte {
+ var tmp [8]byte
+ binary.LittleEndian.PutUint64(tmp[:], v)
+ return append(buf, tmp[:]...)
+}
+
+func appendTraceTagValue(buf []byte, tv *modelv1.TagValue) []byte {
+ switch v := tv.GetValue().(type) {
+ case nil, *modelv1.TagValue_Null:
+ return append(buf, traceTagNull)
+ case *modelv1.TagValue_Str:
+ buf = append(buf, traceTagStr)
+ return appendLenString(buf, v.Str.GetValue())
+ case *modelv1.TagValue_Int:
+ buf = append(buf, traceTagInt)
+ return appendFixed64(buf, uint64(v.Int.GetValue()))
+ case *modelv1.TagValue_BinaryData:
+ buf = append(buf, traceTagBinary)
+ return appendLenBytes(buf, v.BinaryData)
+ default:
+ marshaled, err := proto.Marshal(tv)
+ if err != nil {
+ // proto.Marshal of a well-formed TagValue cannot fail;
encode an empty
+ // proto-fallback payload so decode reconstructs a zero
TagValue rather
+ // than corrupting the frame.
+ marshaled = nil
+ }
+ buf = append(buf, traceTagProto)
+ return appendLenBytes(buf, marshaled)
+ }
+}
Review Comment:
Same as `traceTagValueSize`: this goes through the nil-safe getters.
`tv.GetValue()` returns `nil` for a nil `tv` (encoded as `traceTagNull`), and
`v.Str.GetValue()` / `v.Int.GetValue()` return the zero value for nil nested
messages rather than panicking. The encode path mirrors `traceTagValueSize`
byte-for-byte (locked by `TestTraceResultFrame_ExactPreSize`), so a nil/zero
value still round-trips consistently. Leaving as-is.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
