This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
The following commit(s) were added to refs/heads/main by this push:
new 3c8659ba5 perf(trace): vectorized trace query — limit pushdown,
bounded sidx scan, native columnar distributed wire frame (#1190)
3c8659ba5 is described below
commit 3c8659ba5aed8538f153540e12751f58c436e02a
Author: Gao Hongtao <[email protected]>
AuthorDate: Wed Jun 24 20:33:07 2026 +0800
perf(trace): vectorized trace query — limit pushdown, bounded sidx scan,
native columnar distributed wire frame (#1190)
* perf(trace): push offset+limit into the vectorized SIDX scan
The ordered SIDX (tag-filter) vectorized path built Phase-1 with no trace
cap, so it carried forward and materialized span data for every matching trace
before the outer traceLimit truncated to the limit -- O(matches) work vs the
row path's O(limit). Pass maxRows (= offset+limit) to BuildMergePhase1 so
Phase-1 stops after the top-N distinct trace IDs in sorted-merge order (via the
existing ErrLimitExhausted), matching the row path. Results are identical (the
merge is globally sorted); [...]
---
api/data/codec.go | 17 ++
api/data/codec_test.go | 5 +-
api/data/data.go | 9 +-
api/data/trace_frame.go | 50 ++++
api/data/trace_frame_test.go | 109 +++++++
banyand/internal/migration/fsutil.go | 22 +-
banyand/internal/migration/fsutil_test.go | 100 +++++++
banyand/internal/sidx/query.go | 38 ++-
banyand/internal/sidx/query_test.go | 55 ++++
banyand/internal/sidx/sidx.go | 37 +++
banyand/measure/measure_suite_test.go | 2 +-
banyand/measure/migration_e2e_test.go | 2 +-
banyand/query/processor.go | 119 +++++++-
banyand/query/query.go | 3 +-
banyand/queue/sub/sub.go | 4 +-
banyand/stream/stream_suite_test.go | 2 +-
banyand/trace/query_vectorized.go | 10 +-
banyand/trace/svc_liaison.go | 2 +
banyand/trace/svc_standalone.go | 4 +
banyand/trace/trace_suite_test.go | 2 +-
banyand/trace/vectorized_parity_test.go | 57 ++--
pkg/cmdsetup/data.go | 2 +-
pkg/cmdsetup/standalone.go | 2 +-
pkg/query/logical/trace/trace_frame.go | 325 +++++++++++++++++++++
pkg/query/logical/trace/trace_frame_test.go | 130 +++++++++
pkg/query/logical/trace/trace_plan_distributed.go | 263 +++++++++--------
.../logical/trace/trace_plan_distributed_test.go | 140 +++++----
27 files changed, 1272 insertions(+), 239 deletions(-)
diff --git a/api/data/codec.go b/api/data/codec.go
index eea56df35..9e8bf4a19 100644
--- a/api/data/codec.go
+++ b/api/data/codec.go
@@ -149,6 +149,23 @@ func MeasureWireModeRaw() bool {
return measureWireModeRaw.Load()
}
+// traceWireModeRaw is the per-process wire mode for TopicTraceQuery. Set once
at
+// trace-service startup; raw==true selects the native columnar trace frame
body,
+// raw==false keeps the proto body. Default false so any process that never
sets
+// it keeps the pre-existing proto behavior.
+var traceWireModeRaw atomic.Bool
+
+// SetTraceWireModeRaw publishes the per-process wire mode for TopicTraceQuery.
+func SetTraceWireModeRaw(raw bool) {
+ traceWireModeRaw.Store(raw)
+}
+
+// TraceWireModeRaw reports whether this process emits/decodes native trace
frame
+// bodies for TopicTraceQuery.
+func TraceWireModeRaw() bool {
+ return traceWireModeRaw.Load()
+}
+
// measureQueryResponseCodec dispatches TopicInternalMeasureQuery encode/decode
// by the per-process wire mode: flag-on → RawFrameCodec, flag-off →
ProtoCodec.
// One static supplier serves both flag-on raw bodies and flag-off proto bodies
diff --git a/api/data/codec_test.go b/api/data/codec_test.go
index da49f99fa..b9de83f1d 100644
--- a/api/data/codec_test.go
+++ b/api/data/codec_test.go
@@ -105,8 +105,9 @@ func TestRawFrameCodec_WrongMagic_FailsLoud(t *testing.T) {
func TestTopicResponseMap_ProtoCodec_RoundTripsByteIdentical(t *testing.T) {
var protoTopics int
for topic, codec := range TopicResponseMap {
- if topic == TopicInternalMeasureQuery {
- // Wire-mode-dispatching codec; covered separately
below.
+ if topic == TopicInternalMeasureQuery || topic ==
TopicTraceQuery {
+ // Wire-mode-dispatching codecs; covered separately
(measure below,
+ // trace in trace_frame_test.go including flag-off
byte-identity).
continue
}
pc, ok := codec.(*ProtoCodec)
diff --git a/api/data/data.go b/api/data/data.go
index b11baea5c..ad9ca6420 100644
--- a/api/data/data.go
+++ b/api/data/data.go
@@ -208,9 +208,12 @@ var (
TopicPropertyRepair: NewProtoCodec(func() proto.Message {
return &propertyv1.InternalRepairResponse{}
}),
- TopicTraceQuery: NewProtoCodec(func() proto.Message {
- return &tracev1.InternalQueryResponse{}
- }),
+ TopicTraceQuery: &traceQueryResponseCodec{
+ proto: NewProtoCodec(func() proto.Message {
+ return &tracev1.InternalQueryResponse{}
+ }),
+ raw: NewRawFrameCodec(),
+ },
TopicMeasureCollectDataInfo: NewProtoCodec(func() proto.Message
{
return &databasev1.DataInfo{}
}),
diff --git a/api/data/trace_frame.go b/api/data/trace_frame.go
new file mode 100644
index 000000000..71b716b86
--- /dev/null
+++ b/api/data/trace_frame.go
@@ -0,0 +1,50 @@
+// 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 data
+
+// traceQueryResponseCodec dispatches TopicTraceQuery encode/decode by the
+// per-process wire mode, mirroring measureQueryResponseCodec: flag-on passes
the
+// caller-supplied columnar frame ([]byte) through RawFrameCodec; flag-off (and
+// any proto.Message the caller still emits, e.g. the tracing path) keeps the
+// byte-identical proto body via ProtoCodec.
+type traceQueryResponseCodec struct {
+ proto *ProtoCodec
+ raw *RawFrameCodec
+}
+
+// Marshal encodes the response. In raw wire mode a []byte body (the columnar
+// trace frame) is passed through unchanged; everything else (proto.Message)
goes
+// through the proto path, covering both flag-off and the proto fallback.
+func (c *traceQueryResponseCodec) Marshal(v any) ([]byte, error) {
+ if TraceWireModeRaw() {
+ if b, ok := v.([]byte); ok {
+ return c.raw.Marshal(b)
+ }
+ }
+ return c.proto.Marshal(v)
+}
+
+// Unmarshal decodes the response. In raw wire mode a magic-prefixed body is
+// returned as []byte (the columnar frame, decoded downstream by the liaison);
+// otherwise it is proto-unmarshaled.
+func (c *traceQueryResponseCodec) Unmarshal(body []byte) (any, error) {
+ if TraceWireModeRaw() && len(body) > 0 && body[0] ==
RawFrameMagicLeadingByte {
+ return c.raw.Unmarshal(body)
+ }
+ return c.proto.Unmarshal(body)
+}
diff --git a/api/data/trace_frame_test.go b/api/data/trace_frame_test.go
new file mode 100644
index 000000000..7a8d125c0
--- /dev/null
+++ b/api/data/trace_frame_test.go
@@ -0,0 +1,109 @@
+// 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 data
+
+import (
+ "testing"
+
+ "github.com/stretchr/testify/require"
+ "google.golang.org/protobuf/proto"
+
+ tracev1
"github.com/apache/skywalking-banyandb/api/proto/banyandb/trace/v1"
+)
+
+func newTraceCodec() *traceQueryResponseCodec {
+ return &traceQueryResponseCodec{
+ proto: NewProtoCodec(func() proto.Message { return
&tracev1.InternalQueryResponse{} }),
+ raw: NewRawFrameCodec(),
+ }
+}
+
+// TestTraceCodec_RawPassthrough verifies that flag-on the codec passes a
+// columnar frame ([]byte) through unchanged on both encode and decode.
+func TestTraceCodec_RawPassthrough(t *testing.T) {
+ SetTraceWireModeRaw(true)
+ defer SetTraceWireModeRaw(false)
+ c := newTraceCodec()
+
+ frame := []byte{RawFrameMagicLeadingByte, 0x01, 0x02, 0x03}
+ body, err := c.Marshal(frame)
+ require.NoError(t, err)
+ require.Equal(t, frame, body, "raw frame must pass through Marshal
unchanged")
+
+ got, err := c.Unmarshal(body)
+ require.NoError(t, err)
+ decoded, ok := got.([]byte)
+ require.True(t, ok, "raw-mode Unmarshal of a magic-prefixed body must
return []byte")
+ require.Equal(t, frame, decoded, "raw frame must pass through Unmarshal
unchanged")
+}
+
+// TestTraceCodec_RawEmptyBody verifies that flag-on, a nil/empty body (a
+// legitimate empty distributed result) falls through to the proto path,
yielding
+// an empty *tracev1.InternalQueryResponse rather than being magic-validated.
+func TestTraceCodec_RawEmptyBody(t *testing.T) {
+ SetTraceWireModeRaw(true)
+ defer SetTraceWireModeRaw(false)
+ c := newTraceCodec()
+
+ got, err := c.Unmarshal(nil)
+ require.NoError(t, err)
+ decoded, ok := got.(*tracev1.InternalQueryResponse)
+ require.True(t, ok)
+ require.Empty(t, decoded.GetInternalTraces())
+}
+
+// TestTraceCodec_FlagOffIsProto verifies flag-off encodes byte-identically to
+// plain proto.Marshal and round-trips through the proto path.
+func TestTraceCodec_FlagOffIsProto(t *testing.T) {
+ SetTraceWireModeRaw(false)
+ c := newTraceCodec()
+ orig := &tracev1.InternalQueryResponse{
+ InternalTraces: []*tracev1.InternalTrace{
+ {TraceId: "trace-1", Key: -42, Spans:
[]*tracev1.Span{{SpanId: "span-1", Span: []byte("payload-1")}}},
+ },
+ }
+
+ body, err := c.Marshal(orig)
+ require.NoError(t, err)
+ want, err := proto.Marshal(orig)
+ require.NoError(t, err)
+ require.Equal(t, want, body, "flag-off must encode byte-identically to
proto.Marshal")
+
+ got, err := c.Unmarshal(body)
+ require.NoError(t, err)
+ require.True(t, proto.Equal(orig, got.(*tracev1.InternalQueryResponse)))
+}
+
+// TestTraceCodec_FlagOnProtoFallback verifies that flag-on, a proto.Message
+// value (e.g. the tracing path) still goes through the proto path.
+func TestTraceCodec_FlagOnProtoFallback(t *testing.T) {
+ SetTraceWireModeRaw(true)
+ defer SetTraceWireModeRaw(false)
+ c := newTraceCodec()
+ orig := &tracev1.InternalQueryResponse{}
+
+ body, err := c.Marshal(orig)
+ require.NoError(t, err)
+ want, err := proto.Marshal(orig)
+ require.NoError(t, err)
+ require.Equal(t, want, body)
+
+ got, err := c.Unmarshal(body)
+ require.NoError(t, err)
+ require.True(t, proto.Equal(orig, got.(*tracev1.InternalQueryResponse)))
+}
diff --git a/banyand/internal/migration/fsutil.go
b/banyand/internal/migration/fsutil.go
index f649d3d03..ab8c3c22d 100644
--- a/banyand/internal/migration/fsutil.go
+++ b/banyand/internal/migration/fsutil.go
@@ -55,9 +55,27 @@ func (f NoFsyncFS) Write(buffer []byte, name string,
permission fs.Mode) (int, e
return n, err
}
-// WriteAtomic degrades to a plain (non-atomic, non-synced) write.
+// WriteAtomic writes buffer to name atomically — to a temp sibling, then
+// rename — while skipping the fsync/dir-sync that the durable path performs.
+// NoFsyncFS trades crash durability for throughput, but atomic VISIBILITY must
+// be preserved: callers such as the snapshot-manifest write
+// (tsTable.mustWriteSnapshot) rely on WriteAtomic never exposing a partially
+// written final file to a concurrent reader. A plain O_TRUNC write to the
final
+// path opened a window where a concurrent snapshot read saw a truncated file,
+// surfacing as a "<epoch>.snp: unexpected end of JSON" parse panic in
+// mustReadSnapshot under -race.
func (f NoFsyncFS) WriteAtomic(buffer []byte, name string, permission fs.Mode)
(int, error) {
- return f.Write(buffer, name, permission)
+ tmpName := name + ".tmp"
+ n, err := f.Write(buffer, tmpName, permission)
+ if err != nil {
+ _ = os.Remove(tmpName)
+ return n, err
+ }
+ if renameErr := os.Rename(tmpName, name); renameErr != nil {
+ _ = os.Remove(tmpName)
+ return n, fmt.Errorf("rename %s -> %s: %w", tmpName, name,
renameErr)
+ }
+ return n, nil
}
// SyncPath is a no-op.
diff --git a/banyand/internal/migration/fsutil_test.go
b/banyand/internal/migration/fsutil_test.go
index 3de297885..760ea4f09 100644
--- a/banyand/internal/migration/fsutil_test.go
+++ b/banyand/internal/migration/fsutil_test.go
@@ -18,12 +18,17 @@
package migration
import (
+ "encoding/json"
+ "fmt"
"os"
"path/filepath"
+ "sync"
"testing"
"github.com/stretchr/testify/require"
+ "github.com/apache/skywalking-banyandb/banyand/internal/storage"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
"github.com/apache/skywalking-banyandb/pkg/index/inverted"
)
@@ -72,3 +77,98 @@ func TestCopyDir_MissingSource(t *testing.T) {
_, err := CopyDir(filepath.Join(t.TempDir(), "does-not-exist"), dst)
require.Error(t, err)
}
+
+// TestNoFsyncFS_WriteAtomic_NoTmpLeftover asserts WriteAtomic produces the
exact
+// content and leaves no ".tmp" sibling behind on success.
+func TestNoFsyncFS_WriteAtomic_NoTmpLeftover(t *testing.T) {
+ dir := t.TempDir()
+ nofs := NoFsyncFS{FileSystem: fs.NewLocalFileSystem()}
+ name := filepath.Join(dir, "manifest.json")
+
+ payload := []byte(`["0000000000000001","0000000000000002"]`)
+ n, err := nofs.WriteAtomic(payload, name, storage.FilePerm)
+ require.NoError(t, err)
+ require.Equal(t, len(payload), n)
+
+ got, err := os.ReadFile(name)
+ require.NoError(t, err)
+ require.Equal(t, payload, got)
+ require.NoFileExists(t, name+".tmp", "WriteAtomic must not leave a .tmp
sibling on success")
+}
+
+// TestNoFsyncFS_WriteAtomic_ConcurrentReadsNeverSeePartial locks the atomicity
+// of NoFsyncFS.WriteAtomic: a reader parsing the manifest concurrently with
+// repeated rewrites must never observe a truncated file. Before the fix,
+// WriteAtomic did a plain O_TRUNC write to the final path, so a concurrent
+// ReadSnapshotPartNames caught the truncation window and failed with
+// "unexpected end of JSON" — the root cause of the flaky migration snapshot
+// tests. Run with -race for maximum sensitivity.
+func TestNoFsyncFS_WriteAtomic_ConcurrentReadsNeverSeePartial(t *testing.T) {
+ dir := t.TempDir()
+ lfs := fs.NewLocalFileSystem()
+ nofs := NoFsyncFS{FileSystem: lfs}
+ snpPath := filepath.Join(dir, "0000000000000001.snp")
+
+ // Seed so the manifest always exists before readers start; rename then
keeps
+ // it continuously present, so a reader never sees ENOENT, only (with
the bug)
+ // a partial file.
+ seed, err := json.Marshal([]string{"0000000000000001"})
+ require.NoError(t, err)
+ _, err = nofs.WriteAtomic(seed, snpPath, storage.FilePerm)
+ require.NoError(t, err)
+
+ const writes = 3000
+ var wg sync.WaitGroup
+ stop := make(chan struct{})
+ errCh := make(chan error, 8)
+
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ defer close(stop)
+ for i := 0; i < writes; i++ {
+ // Vary the payload length so the truncation window a
non-atomic write
+ // would expose differs each iteration.
+ names := make([]string, 1+i%17)
+ for j := range names {
+ names[j] = fmt.Sprintf("%016x", i*100+j)
+ }
+ data, mErr := json.Marshal(names)
+ if mErr != nil {
+ errCh <- mErr
+ return
+ }
+ if _, wErr := nofs.WriteAtomic(data, snpPath,
storage.FilePerm); wErr != nil {
+ errCh <- wErr
+ return
+ }
+ }
+ }()
+
+ for r := 0; r < 4; r++ {
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ for {
+ select {
+ case <-stop:
+ return
+ default:
+ }
+ if _, rErr :=
storage.ReadSnapshotPartNames(lfs, snpPath); rErr != nil {
+ select {
+ case errCh <- rErr:
+ default:
+ }
+ return
+ }
+ }
+ }()
+ }
+
+ wg.Wait()
+ close(errCh)
+ for e := range errCh {
+ t.Fatalf("WriteAtomic exposed a partial snapshot to a
concurrent reader: %v", e)
+ }
+}
diff --git a/banyand/internal/sidx/query.go b/banyand/internal/sidx/query.go
index 494f66bf8..3519d662e 100644
--- a/banyand/internal/sidx/query.go
+++ b/banyand/internal/sidx/query.go
@@ -347,12 +347,27 @@ func (s *sidx) prepareSyncResources(
}, true
}
+// errSyncBudgetReached signals that the sync query loop has collected
MaxBatchSize
+// distinct results and the block scan can stop early. It is not a failure.
+var errSyncBudgetReached = errors.New("sidx: sync query result budget reached")
+
func (s *sidx) processSyncLoop(ctx context.Context, req QueryRequest,
resources *syncQueryResources) ([]*QueryResponse, error) {
var results []*QueryResponse
var metrics *batchMetrics
if query.GetTracer(ctx) != nil {
metrics = &batchMetrics{}
}
+ // When MaxBatchSize bounds the result, stop scanning once that many
distinct
+ // elements (= trace IDs; mergeSync dedups by data) have been
collected. The block
+ // iterator yields blocks in key order, so the first MaxBatchSize
distinct elements
+ // are the ordered top-N. This mirrors the streaming path, which stops
via consumer
+ // backpressure; the sync path has none, so without this it decodes
every matching
+ // block. MaxBatchSize<=0 keeps the full scan.
+ var budgetCounter *distinctDataCounter
+ if req.MaxBatchSize > 0 {
+ budgetCounter = newDistinctDataCounter()
+ }
+ budgetReached := false
consume := func(batch *blockScanResultBatch) error {
defer releaseBlockScanResultBatch(batch)
if batch.err != nil {
@@ -374,16 +389,29 @@ func (s *sidx) processSyncLoop(ctx context.Context, req
QueryRequest, resources
return mergeErr
}
results = append(results, chunks...)
+ if budgetCounter != nil {
+ for _, chunk := range chunks {
+ budgetCounter.add(chunk)
+ }
+ if budgetCounter.count >= req.MaxBatchSize {
+ budgetReached = true
+ return errSyncBudgetReached
+ }
+ }
return nil
}
- if scanErr := resources.scanner.scanSync(ctx, consume); scanErr != nil {
+ if scanErr := resources.scanner.scanSync(ctx, consume); scanErr != nil
&& !errors.Is(scanErr, errSyncBudgetReached) {
return nil, scanErr
}
- chunks, mergeErr := resources.heap.mergeSync(ctx, req.MaxBatchSize,
metrics)
- if mergeErr != nil {
- return nil, mergeErr
+ // When the budget was reached the last consume already drained the
heap; only the
+ // completed-scan path needs a final flush of any cursors left in the
heap.
+ if !budgetReached {
+ chunks, mergeErr := resources.heap.mergeSync(ctx,
req.MaxBatchSize, metrics)
+ if mergeErr != nil {
+ return nil, mergeErr
+ }
+ results = append(results, chunks...)
}
- results = append(results, chunks...)
return results, nil
}
diff --git a/banyand/internal/sidx/query_test.go
b/banyand/internal/sidx/query_test.go
index 59a758f38..6516a0f5b 100644
--- a/banyand/internal/sidx/query_test.go
+++ b/banyand/internal/sidx/query_test.go
@@ -19,6 +19,7 @@ package sidx
import (
"context"
+ "fmt"
"sort"
"strings"
"testing"
@@ -103,6 +104,60 @@ func TestSIDX_QuerySyncMatchesStreamingQuery(t *testing.T)
{
require.Equal(t, flattenQueryRows(streamingRows),
flattenQueryRows(syncRows))
}
+// TestSIDX_QuerySync_EarlyStopAtMaxBatchSize verifies the sync query path
stops
+// scanning once MaxBatchSize distinct results are collected, while still
returning
+// the correct ordered top-N. Blocks are per series, so writing many
single-element
+// series forces the scan across multiple batches (blockScannerBatchSize); the
block
+// iterator yields them in ascending key order, so the bounded result's prefix
must
+// equal the uncapped, fully-ordered scan's prefix.
+func TestSIDX_QuerySync_EarlyStopAtMaxBatchSize(t *testing.T) {
+ s := createTestSIDX(t)
+ defer func() {
+ assert.NoError(t, s.Close())
+ }()
+ ctx := context.Background()
+
+ const seriesCount = 80
+ seriesIDs := make([]common.SeriesID, 0, seriesCount)
+ for i := range seriesCount {
+ key := int64(i + 1)
+ series := common.SeriesID(i + 1)
+ seriesIDs = append(seriesIDs, series)
+ writeTestData(t, s, []WriteRequest{
+ createTestWriteRequest(series, key,
fmt.Sprintf("data%03d", key)),
+ }, 44, uint64(i+1))
+ }
+ waitForIntroducerLoop()
+
+ syncQuerier, ok := s.(interface {
+ QuerySync(context.Context, QueryRequest) ([]*QueryResponse,
error)
+ })
+ require.True(t, ok)
+
+ baseReq := createTestQueryRequest(seriesIDs...)
+
+ // Uncapped scan reads every block and returns all elements in key
order.
+ uncapped, err := syncQuerier.QuerySync(ctx, baseReq)
+ require.NoError(t, err)
+ uncappedRows := flattenQueryRows(uncapped)
+ require.Len(t, uncappedRows, seriesCount)
+
+ const maxBatch = 10
+ cappedReq := baseReq
+ cappedReq.MaxBatchSize = maxBatch
+ capped, err := syncQuerier.QuerySync(ctx, cappedReq)
+ require.NoError(t, err)
+ cappedRows := flattenQueryRows(capped)
+
+ // Bounded query collected at least MaxBatchSize and stopped before
reading every block.
+ require.GreaterOrEqual(t, len(cappedRows), maxBatch, "must collect at
least MaxBatchSize results")
+ require.Less(t, len(cappedRows), seriesCount, "must stop scanning
before reading every block")
+
+ // The bounded result is the correct ordered top-N: its first
MaxBatchSize rows
+ // equal the first MaxBatchSize rows of the uncapped, fully-ordered
scan.
+ require.Equal(t, uncappedRows[:maxBatch], cappedRows[:maxBatch])
+}
+
func TestSIDX_Query_EmptyResult(t *testing.T) {
sidx := createTestSIDX(t)
defer func() {
diff --git a/banyand/internal/sidx/sidx.go b/banyand/internal/sidx/sidx.go
index d902d145a..7f5e118b2 100644
--- a/banyand/internal/sidx/sidx.go
+++ b/banyand/internal/sidx/sidx.go
@@ -930,6 +930,43 @@ func (bch *blockCursorHeap) mergeSync(ctx context.Context,
batchSize int, metric
return results, nil
}
+// distinctDataCounter accumulates distinct data values (= trace IDs) across
sidx
+// merge chunks using hash buckets with collision checks. mergeSync dedups data
+// only within a single batch, so the sync query loop uses this to count
distinct
+// results across batches and bound scanning the same way the streaming
consumer
+// terminates on distinct-trace count.
+type distinctDataCounter struct {
+ seen map[uint64][][]byte
+ count int
+}
+
+func newDistinctDataCounter() *distinctDataCounter {
+ return &distinctDataCounter{seen: make(map[uint64][][]byte)}
+}
+
+// add records every data value in chunk and returns the running distinct
count.
+func (d *distinctDataCounter) add(chunk *QueryResponse) int {
+ if chunk == nil {
+ return d.count
+ }
+ for _, data := range chunk.Data {
+ h := convert.Hash(data)
+ bucket := d.seen[h]
+ duplicate := false
+ for _, existing := range bucket {
+ if bytes.Equal(existing, data) {
+ duplicate = true
+ break
+ }
+ }
+ if !duplicate {
+ d.seen[h] = append(bucket, data)
+ d.count++
+ }
+ }
+ return d.count
+}
+
var blockCursorHeapPool =
pool.Register[*blockCursorHeap]("sidx-blockCursorHeap")
func generateBlockCursorHeap(asc bool) *blockCursorHeap {
diff --git a/banyand/measure/measure_suite_test.go
b/banyand/measure/measure_suite_test.go
index 25565ebfd..7b5b08596 100644
--- a/banyand/measure/measure_suite_test.go
+++ b/banyand/measure/measure_suite_test.go
@@ -82,7 +82,7 @@ func setUp() (*services, func()) {
measureService, err := measure.NewStandalone(metadataService, pipeline,
nil, metricSvc, pm)
gomega.Expect(err).NotTo(gomega.HaveOccurred())
preloadMeasureSvc := &preloadMeasureService{metaSvc: metadataService}
- querySvc, err := query.NewService(context.TODO(), nil, measureService,
nil, metadataService, pipeline, metricSvc)
+ querySvc, err := query.NewService(context.TODO(), nil, measureService,
nil, metadataService, pipeline, metricSvc, false)
gomega.Expect(err).NotTo(gomega.HaveOccurred())
var flags []string
diff --git a/banyand/measure/migration_e2e_test.go
b/banyand/measure/migration_e2e_test.go
index 5d3b3946b..8e5962f15 100644
--- a/banyand/measure/migration_e2e_test.go
+++ b/banyand/measure/migration_e2e_test.go
@@ -580,7 +580,7 @@ func setUpMigrationTarget(measureRootPath string)
(*services, func()) {
measureService, err := measure.NewStandalone(metadataService, pipeline,
nil, metricSvc, pm)
Expect(err).NotTo(HaveOccurred())
preloadSvc := &preloadMeasureService{metaSvc: metadataService}
- querySvc, err := query.NewService(context.TODO(), nil, measureService,
nil, metadataService, pipeline, metricSvc)
+ querySvc, err := query.NewService(context.TODO(), nil, measureService,
nil, metadataService, pipeline, metricSvc, false)
Expect(err).NotTo(HaveOccurred())
metaPath, metaDeferFunc, err := test.NewSpace()
diff --git a/banyand/query/processor.go b/banyand/query/processor.go
index 57ffa1cf3..0c0cc01fa 100644
--- a/banyand/query/processor.go
+++ b/banyand/query/processor.go
@@ -41,6 +41,7 @@ import (
"github.com/apache/skywalking-banyandb/pkg/bus"
"github.com/apache/skywalking-banyandb/pkg/iter"
"github.com/apache/skywalking-banyandb/pkg/logger"
+ pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
"github.com/apache/skywalking-banyandb/pkg/query"
"github.com/apache/skywalking-banyandb/pkg/query/executor"
"github.com/apache/skywalking-banyandb/pkg/query/logical"
@@ -661,6 +662,12 @@ type traceQueryProcessor struct {
traceService trace.Service
*queryService
*bus.UnImplementedHealthyListener
+ // distributed is true on a distributed data node, whose query response
is
+ // sent over the wire to a remote liaison that decodes the native
columnar
+ // frame. It is false in standalone, where the local grpc service
consumes the
+ // response directly and expects the proto body — so the frame must not
be
+ // emitted there.
+ distributed bool
}
func (p *traceQueryProcessor) Rev(ctx context.Context, message bus.Message)
(resp bus.Message) {
@@ -900,20 +907,104 @@ func (p *traceQueryProcessor) buildTraceTags(result
*model.TraceResult, queryCri
return traceTags
}
-func (p *traceQueryProcessor) logSlowQuery(queryCriteria
*tracev1.QueryRequest, traces []*tracev1.InternalTrace, startTime time.Time) {
- if queryCriteria.Trace || p.slowQuery <= 0 {
- return
+// buildFrameTraceResults mirrors processTraceResults+buildTraceTags but
outputs
+// columnar []model.TraceResult (TID/Key/Spans/SpanIDs plus one TagValue column
+// per span) for the native columnar wire frame. It replicates the
projected-tag
+// filter and the traceID/spanID tag augmentation so the liaison reconstructs
the
+// same spans the proto path would have produced.
+func (p *traceQueryProcessor) buildFrameTraceResults(resultIterator
iter.Iterator[model.TraceResult],
+ queryCriteria *tracev1.QueryRequest, execPlan *traceExecutionPlan,
+) ([]model.TraceResult, error) {
+ traceIDInclusionMap := make(map[int]bool)
+ spanIDInclusionMap := make(map[int]bool)
+ for i, tagName := range execPlan.traceIDTagNames {
+ if slices.Contains(queryCriteria.TagProjection, tagName) {
+ traceIDInclusionMap[i] = true
+ }
+ }
+ for i, tagName := range execPlan.spanIDTagNames {
+ if slices.Contains(queryCriteria.TagProjection, tagName) {
+ spanIDInclusionMap[i] = true
+ }
}
- latency := time.Since(startTime)
- if latency <= p.slowQuery {
- return
+ var results []model.TraceResult
+ for {
+ result, hasNext := resultIterator.Next()
+ if !hasNext {
+ break
+ }
+ if result.Error != nil {
+ return nil, result.Error
+ }
+ if result.TID == "" {
+ continue
+ }
+
+ spanCount := len(result.Spans)
+ frameResult := model.TraceResult{
+ TID: result.TID,
+ Key: result.Key,
+ Spans: result.Spans,
+ SpanIDs: result.SpanIDs,
+ }
+ if result.Tags != nil && len(queryCriteria.TagProjection) > 0 {
+ for _, tag := range result.Tags {
+ if
!slices.Contains(queryCriteria.TagProjection, tag.Name) {
+ continue
+ }
+ frameResult.Tags = append(frameResult.Tags, tag)
+ }
+ }
+ if traceIDInclusionMap[result.GroupIndex] {
+ values := make([]*modelv1.TagValue, spanCount)
+ tidValue := &modelv1.TagValue{Value:
&modelv1.TagValue_Str{Str: &modelv1.Str{Value: result.TID}}}
+ for spanIdx := range values {
+ values[spanIdx] = tidValue
+ }
+ frameResult.Tags = append(frameResult.Tags,
model.Tag{Name: execPlan.traceIDTagNames[result.GroupIndex], Values: values})
+ }
+ if spanIDInclusionMap[result.GroupIndex] {
+ values := make([]*modelv1.TagValue, spanCount)
+ for spanIdx := 0; spanIdx < spanCount; spanIdx++ {
+ if spanIdx < len(result.SpanIDs) {
+ values[spanIdx] =
&modelv1.TagValue{Value: &modelv1.TagValue_Str{Str: &modelv1.Str{Value:
result.SpanIDs[spanIdx]}}}
+ } else {
+ values[spanIdx] = pbv1.NullTagValue
+ }
+ }
+ frameResult.Tags = append(frameResult.Tags,
model.Tag{Name: execPlan.spanIDTagNames[result.GroupIndex], Values: values})
+ }
+ results = append(results, frameResult)
}
+ return results, nil
+}
+
+func (p *traceQueryProcessor) logSlowQuery(queryCriteria
*tracev1.QueryRequest, traces []*tracev1.InternalTrace, startTime time.Time) {
spanCount := 0
for _, trace := range traces {
spanCount += len(trace.Spans)
}
+ p.logSlowTraceQuery(queryCriteria, spanCount, startTime)
+}
+
+func (p *traceQueryProcessor) logSlowFrameQuery(queryCriteria
*tracev1.QueryRequest, results []model.TraceResult, startTime time.Time) {
+ spanCount := 0
+ for resultIdx := range results {
+ spanCount += len(results[resultIdx].Spans)
+ }
+ p.logSlowTraceQuery(queryCriteria, spanCount, startTime)
+}
+
+func (p *traceQueryProcessor) logSlowTraceQuery(queryCriteria
*tracev1.QueryRequest, spanCount int, startTime time.Time) {
+ if queryCriteria.Trace || p.slowQuery <= 0 {
+ return
+ }
+ latency := time.Since(startTime)
+ if latency <= p.slowQuery {
+ return
+ }
p.log.Warn().Dur("latency", latency).RawJSON("req",
logger.Proto(queryCriteria)).Int("resp_count", spanCount).Msg("trace slow
query")
}
@@ -961,6 +1052,22 @@ func (p *traceQueryProcessor) executeQuery(ctx
context.Context, queryCriteria *t
return
}
+ // Native wire mode (flag-on, no tracing): emit a columnar frame body
so the
+ // send path passes it through as opaque bytes and the liaison decodes
it
+ // without the protobuf message-slice/oneof machinery. The tracing path
keeps
+ // the proto body (the traceMonitor defer needs *InternalQueryResponse).
+ if p.distributed && data.TraceWireModeRaw() && traceMonitor == nil {
+ results, buildErr := p.buildFrameTraceResults(resultIterator,
queryCriteria, execPlan)
+ if buildErr != nil {
+ p.log.Error().Err(buildErr).RawJSON("req",
logger.Proto(queryCriteria)).Msg("fail to process trace results")
+ resp = bus.NewMessage(bus.MessageID(now),
common.NewError("process trace results for trace %s: %v",
queryCriteria.GetName(), buildErr))
+ return
+ }
+ resp = bus.NewMessage(bus.MessageID(now),
logical_trace.EncodeTraceResultFrame(results))
+ p.logSlowFrameQuery(queryCriteria, results, n)
+ return
+ }
+
traces, err := p.processTraceResults(resultIterator, queryCriteria,
execPlan)
if err != nil {
p.log.Error().Err(err).RawJSON("req",
logger.Proto(queryCriteria)).Msg("fail to process trace results")
diff --git a/banyand/query/query.go b/banyand/query/query.go
index 8d5a25ae9..acd855ae0 100644
--- a/banyand/query/query.go
+++ b/banyand/query/query.go
@@ -52,7 +52,7 @@ type queryService struct {
// NewService return a new query service.
func NewService(_ context.Context, streamService stream.Service,
measureService measure.Service, traceService trace.Service,
- metaService metadata.Repo, pipeline queue.Server, metricSvc
observability.MetricsRegistry,
+ metaService metadata.Repo, pipeline queue.Server, metricSvc
observability.MetricsRegistry, distributed bool,
) (run.Unit, error) {
svc := &queryService{
metaService: metaService,
@@ -83,6 +83,7 @@ func NewService(_ context.Context, streamService
stream.Service, measureService
svc.tqp = &traceQueryProcessor{
traceService: traceService,
queryService: svc,
+ distributed: distributed,
}
return svc, nil
}
diff --git a/banyand/queue/sub/sub.go b/banyand/queue/sub/sub.go
index 4115d62cb..6dd17bb4e 100644
--- a/banyand/queue/sub/sub.go
+++ b/banyand/queue/sub/sub.go
@@ -314,7 +314,9 @@ func (s *server) marshalResponse(
}
return body, nil
case []byte:
- if topic != data.TopicInternalMeasureQuery ||
!data.MeasureWireModeRaw() {
+ rawAllowed := (topic == data.TopicInternalMeasureQuery &&
data.MeasureWireModeRaw()) ||
+ (topic == data.TopicTraceQuery &&
data.TraceWireModeRaw())
+ if !rawAllowed {
s.replyWithErrType(stream, writeEntity, nil,
fmt.Sprintf("invalid response: unexpected raw body on topic %s", topic),
identity, "marshal_error")
return nil, fmt.Errorf("invalid raw body")
}
diff --git a/banyand/stream/stream_suite_test.go
b/banyand/stream/stream_suite_test.go
index 1c83e0f5b..5c02d073c 100644
--- a/banyand/stream/stream_suite_test.go
+++ b/banyand/stream/stream_suite_test.go
@@ -83,7 +83,7 @@ func setUp() (*services, func()) {
streamService, err := stream.NewService(metadataService, pipeline,
metricSvc, pm, nil)
gomega.Expect(err).NotTo(gomega.HaveOccurred())
preloadStreamSvc := &preloadStreamService{metaSvc: metadataService}
- querySvc, err := query.NewService(context.TODO(), streamService, nil,
nil, metadataService, pipeline, metricSvc)
+ querySvc, err := query.NewService(context.TODO(), streamService, nil,
nil, metadataService, pipeline, metricSvc, false)
gomega.Expect(err).NotTo(gomega.HaveOccurred())
var flags []string
metaPath, metaDeferFunc, err := test.NewSpace()
diff --git a/banyand/trace/query_vectorized.go
b/banyand/trace/query_vectorized.go
index 3e08b2ed6..3dcc873a6 100644
--- a/banyand/trace/query_vectorized.go
+++ b/banyand/trace/query_vectorized.go
@@ -380,10 +380,12 @@ func (t *trace) buildVectorizedPhase1TraceBatch(
if iterErr != nil {
return traceBatch{}, iterErr
}
- // Ordered SIDX path: MaxTraceSize controls SIDX batch size
(MaxBatchSize in the
- // request), not the total result cap. The push path emits all
matching traces in
- // batches; vectorized must do the same. Pass 0 to disable the
Phase-1 trace cap.
- plan, buildErr := vtrace.BuildMergePhase1(iters,
sidxRequestDesc(req), 0, batchSize)
+ // Ordered SIDX path: cap Phase-1 at maxRows (= offset+limit)
so only the top-N
+ // distinct trace IDs in sorted-merge order are carried forward
and materialized.
+ // The merge is globally sorted, so the first maxRows traces
are exactly the window
+ // the outer traceLimit keeps; carrying the full match set
(maxRows=0) wastes span
+ // decode on traces the limit discards. maxRows=0 (no limit
set) stays uncapped.
+ plan, buildErr := vtrace.BuildMergePhase1(iters,
sidxRequestDesc(req), maxRows, batchSize)
if buildErr != nil {
return traceBatch{}, fmt.Errorf("build ordered
vectorized trace phase1: %w", buildErr)
}
diff --git a/banyand/trace/svc_liaison.go b/banyand/trace/svc_liaison.go
index 320919ad1..92a2418f1 100644
--- a/banyand/trace/svc_liaison.go
+++ b/banyand/trace/svc_liaison.go
@@ -176,6 +176,8 @@ func (l *liaison) Role() databasev1.Role {
func (l *liaison) PreRun(ctx context.Context) error {
l.l = logger.GetLogger(l.Name())
+ data.SetTraceWireModeRaw(l.option.vectorized.Enabled)
+ l.l.Info().Bool("trace_wire_mode_raw",
l.option.vectorized.Enabled).Msg("trace wire mode published (liaison)")
l.l.Info().Msg("memory protector is initialized in PreRun")
l.lfs = fs.NewLocalFileSystemWithLoggerAndLimit(l.l, l.pm.GetLimit())
var err error
diff --git a/banyand/trace/svc_standalone.go b/banyand/trace/svc_standalone.go
index f667c5207..4f0f657a6 100644
--- a/banyand/trace/svc_standalone.go
+++ b/banyand/trace/svc_standalone.go
@@ -136,6 +136,10 @@ func (s *standalone) Role() databasev1.Role {
func (s *standalone) PreRun(ctx context.Context) error {
s.l = logger.GetLogger(s.Name())
+ // Native columnar wire frame for the data↔liaison query hop follows the
+ // vectorized flag (mirrors measure's wire-mode wiring).
+ data.SetTraceWireModeRaw(s.option.vectorized.Enabled)
+ s.l.Info().Bool("trace_wire_mode_raw",
s.option.vectorized.Enabled).Msg("trace wire mode published (standalone)")
s.l.Info().Msg("memory protector is initialized in PreRun")
s.lfs = fs.NewLocalFileSystemWithLoggerAndLimit(s.l, s.pm.GetLimit())
var err error
diff --git a/banyand/trace/trace_suite_test.go
b/banyand/trace/trace_suite_test.go
index ee9c48581..1c9b5089d 100644
--- a/banyand/trace/trace_suite_test.go
+++ b/banyand/trace/trace_suite_test.go
@@ -83,7 +83,7 @@ func setUp() (*services, func()) {
gomega.Expect(err).NotTo(gomega.HaveOccurred())
preloadTraceSvc := &preloadTraceService{metaSvc: metadataService}
// Init Query Service for trace queries
- querySvc, err := query.NewService(context.TODO(), nil, nil,
traceService, metadataService, pipeline, metricSvc)
+ querySvc, err := query.NewService(context.TODO(), nil, nil,
traceService, metadataService, pipeline, metricSvc, false)
gomega.Expect(err).NotTo(gomega.HaveOccurred())
var flags []string
metaPath, metaDeferFunc, err := test.NewSpace()
diff --git a/banyand/trace/vectorized_parity_test.go
b/banyand/trace/vectorized_parity_test.go
index 00d08c9f5..6432c4043 100644
--- a/banyand/trace/vectorized_parity_test.go
+++ b/banyand/trace/vectorized_parity_test.go
@@ -239,12 +239,13 @@ func TestVectorizedParityOrderMode(t *testing.T) {
}
tests := []struct {
- name string
- keys []int64
- traceIDs []string
- maxTraceSize int
- wantCount int
- sortDir modelv1.Sort
+ name string
+ keys []int64
+ traceIDs []string
+ maxTraceSize int
+ wantCount int
+ wantPullCount int
+ sortDir modelv1.Sort
}{
{
name: "ASC order all traces",
@@ -261,22 +262,25 @@ func TestVectorizedParityOrderMode(t *testing.T) {
wantCount: 3,
},
{
- // MaxTraceSize is a batch-size hint for ordered SIDX
queries, not a total cap.
- // The push path emits all 3 traces in batches of 2;
vectorized must match.
- name: "ASC order MaxTraceSize=2 with 3 traces
returns all 3",
- sortDir: modelv1.Sort_SORT_ASC,
- keys: []int64{1, 2, 3},
- traceIDs: []string{"trace1", "trace2", "trace3"},
- maxTraceSize: 2,
- wantCount: 3,
+ // MaxTraceSize (= offset+limit) caps the pull path's
materialized trace set to
+ // the sorted top-N. The push path streams lazily and
emits all matches, relying
+ // on the outer traceLimit to cap; so pull is the
MaxTraceSize-prefix of push.
+ name: "ASC order MaxTraceSize=2 caps pull to
top 2 (push streams all 3)",
+ sortDir: modelv1.Sort_SORT_ASC,
+ keys: []int64{1, 2, 3},
+ traceIDs: []string{"trace1", "trace2", "trace3"},
+ maxTraceSize: 2,
+ wantCount: 3,
+ wantPullCount: 2,
},
{
- name: "DESC order MaxTraceSize=2 with 3 traces
returns all 3",
- sortDir: modelv1.Sort_SORT_DESC,
- keys: []int64{3, 2, 1},
- traceIDs: []string{"trace3", "trace2", "trace1"},
- maxTraceSize: 2,
- wantCount: 3,
+ name: "DESC order MaxTraceSize=2 caps pull to
top 2 (push streams all 3)",
+ sortDir: modelv1.Sort_SORT_DESC,
+ keys: []int64{3, 2, 1},
+ traceIDs: []string{"trace3", "trace2", "trace1"},
+ maxTraceSize: 2,
+ wantCount: 3,
+ wantPullCount: 2,
},
}
@@ -315,11 +319,18 @@ func TestVectorizedParityOrderMode(t *testing.T) {
defer pullRes.Release()
pullGot := collectResults(t, pullRes)
+ // The pull path caps the materialized set at
MaxTraceSize; the push path streams
+ // all matches (the outer traceLimit caps it in
production). With no cap the two
+ // are identical; with a cap the pull result is the
sorted MaxTraceSize-prefix.
+ wantPull := tt.wantCount
+ if tt.wantPullCount > 0 {
+ wantPull = tt.wantPullCount
+ }
require.Len(t, pushGot, tt.wantCount, "push path result
count")
- require.Len(t, pullGot, tt.wantCount, "pull path result
count")
+ require.Len(t, pullGot, wantPull, "pull path result
count")
- if diff := cmp.Diff(pushGot, pullGot,
protocmp.Transform()); diff != "" {
- t.Errorf("push vs pull order mismatch (-push
+pull):\n%s", diff)
+ if diff := cmp.Diff(pushGot[:wantPull], pullGot,
protocmp.Transform()); diff != "" {
+ t.Errorf("pull path must equal the sorted
MaxTraceSize-prefix of push (-push +pull):\n%s", diff)
}
})
}
diff --git a/pkg/cmdsetup/data.go b/pkg/cmdsetup/data.go
index cd6f6ba50..93891ff68 100644
--- a/pkg/cmdsetup/data.go
+++ b/pkg/cmdsetup/data.go
@@ -76,7 +76,7 @@ func newDataCmd(runners ...run.Unit) *cobra.Command {
if err != nil {
l.Fatal().Err(err).Msg("failed to initiate trace service")
}
- q, err := query.NewService(ctx, streamSvc, measureSvc, traceSvc,
metaSvc, pipeline, metricSvc)
+ q, err := query.NewService(ctx, streamSvc, measureSvc, traceSvc,
metaSvc, pipeline, metricSvc, true)
if err != nil {
l.Fatal().Err(err).Msg("failed to initiate query processor")
}
diff --git a/pkg/cmdsetup/standalone.go b/pkg/cmdsetup/standalone.go
index b76075607..86d715c91 100644
--- a/pkg/cmdsetup/standalone.go
+++ b/pkg/cmdsetup/standalone.go
@@ -72,7 +72,7 @@ func newStandaloneCmd(runners ...run.Unit) *cobra.Command {
if err != nil {
l.Fatal().Err(err).Msg("failed to initiate measure service")
}
- q, err := query.NewService(ctx, streamSvc, measureSvc, traceSvc,
metaSvc, dataPipeline, metricSvc)
+ q, err := query.NewService(ctx, streamSvc, measureSvc, traceSvc,
metaSvc, dataPipeline, metricSvc, false)
if err != nil {
l.Fatal().Err(err).Msg("failed to initiate query processor")
}
diff --git a/pkg/query/logical/trace/trace_frame.go
b/pkg/query/logical/trace/trace_frame.go
new file mode 100644
index 000000000..42cfe873f
--- /dev/null
+++ b/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)
+ }
+}
+
+// frameCursor is a bounds-checked cursor over a frame body; the first error is
+// sticky so callers can read straight-line and check err once at the end.
+type frameCursor struct {
+ err error
+ b []byte
+ pos int
+}
+
+func (c *frameCursor) uvarint() int {
+ if c.err != nil {
+ return 0
+ }
+ v, n := binary.Uvarint(c.b[c.pos:])
+ if n <= 0 {
+ c.err = fmt.Errorf("trace result frame: bad uvarint at offset
%d", c.pos)
+ return 0
+ }
+ c.pos += n
+ return int(v)
+}
+
+func (c *frameCursor) fixed64() uint64 {
+ if c.err != nil {
+ return 0
+ }
+ if c.pos+8 > len(c.b) {
+ c.err = fmt.Errorf("trace result frame: truncated fixed64 at
offset %d", c.pos)
+ return 0
+ }
+ v := binary.LittleEndian.Uint64(c.b[c.pos:])
+ c.pos += 8
+ return v
+}
+
+func (c *frameCursor) str() string {
+ n := c.uvarint()
+ if c.err != nil {
+ return ""
+ }
+ if n < 0 || c.pos+n > len(c.b) {
+ c.err = fmt.Errorf("trace result frame: truncated string len %d
at offset %d", n, c.pos)
+ return ""
+ }
+ s := string(c.b[c.pos : c.pos+n])
+ c.pos += n
+ return s
+}
+
+func (c *frameCursor) bytesCopy() []byte {
+ n := c.uvarint()
+ if c.err != nil {
+ return nil
+ }
+ if n < 0 || c.pos+n > len(c.b) {
+ c.err = fmt.Errorf("trace result frame: truncated bytes len %d
at offset %d", n, c.pos)
+ return nil
+ }
+ out := make([]byte, n)
+ copy(out, c.b[c.pos:c.pos+n])
+ c.pos += n
+ return out
+}
+
+func (c *frameCursor) tagValue() *modelv1.TagValue {
+ if c.err != nil {
+ return nil
+ }
+ if c.pos >= len(c.b) {
+ c.err = fmt.Errorf("trace result frame: truncated tag value at
offset %d", c.pos)
+ return nil
+ }
+ typ := c.b[c.pos]
+ c.pos++
+ switch typ {
+ case traceTagNull:
+ return &modelv1.TagValue{Value: &modelv1.TagValue_Null{}}
+ case traceTagStr:
+ return &modelv1.TagValue{Value: &modelv1.TagValue_Str{Str:
&modelv1.Str{Value: c.str()}}}
+ case traceTagInt:
+ return &modelv1.TagValue{Value: &modelv1.TagValue_Int{Int:
&modelv1.Int{Value: int64(c.fixed64())}}}
+ case traceTagBinary:
+ return &modelv1.TagValue{Value:
&modelv1.TagValue_BinaryData{BinaryData: c.bytesCopy()}}
+ case traceTagProto:
+ raw := c.bytesCopy()
+ if c.err != nil {
+ return nil
+ }
+ tv := &modelv1.TagValue{}
+ if err := proto.Unmarshal(raw, tv); err != nil {
+ c.err = fmt.Errorf("trace result frame: tag value proto
fallback: %w", err)
+ return nil
+ }
+ return tv
+ default:
+ c.err = fmt.Errorf("trace result frame: unknown tag value type
%d", typ)
+ return nil
+ }
+}
diff --git a/pkg/query/logical/trace/trace_frame_test.go
b/pkg/query/logical/trace/trace_frame_test.go
new file mode 100644
index 000000000..3f6b7e8be
--- /dev/null
+++ b/pkg/query/logical/trace/trace_frame_test.go
@@ -0,0 +1,130 @@
+// 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 (
+ "testing"
+
+ "github.com/stretchr/testify/require"
+ "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"
+)
+
+func strTV(s string) *modelv1.TagValue {
+ return &modelv1.TagValue{Value: &modelv1.TagValue_Str{Str:
&modelv1.Str{Value: s}}}
+}
+
+func intTV(i int64) *modelv1.TagValue {
+ return &modelv1.TagValue{Value: &modelv1.TagValue_Int{Int:
&modelv1.Int{Value: i}}}
+}
+
+func nullTV() *modelv1.TagValue {
+ return &modelv1.TagValue{Value: &modelv1.TagValue_Null{}}
+}
+
+func binTV(b []byte) *modelv1.TagValue {
+ return &modelv1.TagValue{Value:
&modelv1.TagValue_BinaryData{BinaryData: b}}
+}
+
+func intArrTV(vs ...int64) *modelv1.TagValue {
+ return &modelv1.TagValue{Value: &modelv1.TagValue_IntArray{IntArray:
&modelv1.IntArray{Value: vs}}}
+}
+
+func sampleTraceResults() []model.TraceResult {
+ return []model.TraceResult{
+ {
+ TID: "trace-1",
+ Key: -42,
+ Spans: [][]byte{[]byte("payload-1"),
[]byte("payload-2")},
+ SpanIDs: []string{"span-1", "span-2"},
+ Tags: []model.Tag{
+ {Name: "service_id", Values:
[]*modelv1.TagValue{strTV("svc-0"), strTV("svc-1")}},
+ {Name: "duration", Values:
[]*modelv1.TagValue{intTV(123), intTV(456)}},
+ {Name: "missing", Values:
[]*modelv1.TagValue{nullTV(), nullTV()}},
+ {Name: "binary", Values:
[]*modelv1.TagValue{binTV([]byte{0x1, 0x2, 0x3}), nullTV()}},
+ {Name: "ints", Values:
[]*modelv1.TagValue{intArrTV(7, 8, 9), intArrTV(10)}},
+ },
+ },
+ {
+ TID: "trace-2",
+ Key: 100,
+ Spans: [][]byte{[]byte("p")},
+ SpanIDs: []string{"s"},
+ Tags: nil,
+ },
+ }
+}
+
+func requireTraceResultsEqual(t *testing.T, want, got []model.TraceResult) {
+ t.Helper()
+ require.Len(t, got, len(want))
+ for i := range want {
+ require.Equal(t, want[i].TID, got[i].TID, "trace %d TID", i)
+ require.Equal(t, want[i].Key, got[i].Key, "trace %d Key", i)
+ require.Equal(t, want[i].Spans, got[i].Spans, "trace %d Spans",
i)
+ require.Equal(t, want[i].SpanIDs, got[i].SpanIDs, "trace %d
SpanIDs", i)
+ require.Len(t, got[i].Tags, len(want[i].Tags), "trace %d tag
count", i)
+ for j := range want[i].Tags {
+ require.Equal(t, want[i].Tags[j].Name,
got[i].Tags[j].Name, "trace %d tag %d name", i, j)
+ require.Len(t, got[i].Tags[j].Values,
len(want[i].Tags[j].Values), "trace %d tag %d value count", i, j)
+ for k := range want[i].Tags[j].Values {
+ require.True(t,
proto.Equal(want[i].Tags[j].Values[k], got[i].Tags[j].Values[k]),
+ "trace %d tag %d value %d: want %v got
%v", i, j, k, want[i].Tags[j].Values[k], got[i].Tags[j].Values[k])
+ }
+ }
+ }
+}
+
+func TestTraceResultFrame_RoundTrip(t *testing.T) {
+ orig := sampleTraceResults()
+ body := EncodeTraceResultFrame(orig)
+ require.Greater(t, len(body), 0)
+ require.Equal(t, data.RawFrameMagicLeadingByte, body[0], "frame must
carry the raw-frame magic")
+ require.Equal(t, traceFrameVersion, body[1], "frame must carry the
version byte")
+
+ got, err := DecodeTraceResultFrame(body)
+ require.NoError(t, err)
+ requireTraceResultsEqual(t, orig, got)
+}
+
+// TestTraceResultFrame_ExactPreSize locks the invariant that traceFrameSize
+// equals the encoded length, so the encode buffer is pre-sized exactly (no
+// realloc, no over-allocation).
+func TestTraceResultFrame_ExactPreSize(t *testing.T) {
+ orig := sampleTraceResults()
+ body := EncodeTraceResultFrame(orig)
+ require.Equal(t, traceFrameSize(orig), len(body), "traceFrameSize must
equal the encoded length")
+}
+
+func TestTraceResultFrame_Empty(t *testing.T) {
+ body := EncodeTraceResultFrame(nil)
+ require.Equal(t, data.RawFrameMagicLeadingByte, body[0])
+ got, err := DecodeTraceResultFrame(body)
+ require.NoError(t, err)
+ require.Empty(t, got)
+}
+
+func TestTraceResultFrame_BadMagic(t *testing.T) {
+ _, err := DecodeTraceResultFrame([]byte{0xFF, traceFrameVersion})
+ require.Error(t, err)
+ _, err = DecodeTraceResultFrame([]byte{data.RawFrameMagicLeadingByte,
0x09})
+ require.Error(t, err)
+}
diff --git a/pkg/query/logical/trace/trace_plan_distributed.go
b/pkg/query/logical/trace/trace_plan_distributed.go
index 073310e65..902fb6d31 100644
--- a/pkg/query/logical/trace/trace_plan_distributed.go
+++ b/pkg/query/logical/trace/trace_plan_distributed.go
@@ -157,48 +157,38 @@ func (p *distributedPlan) Execute(ctx context.Context)
(iter.Iterator[model.Trac
}
var allErr error
var responseCount int
- var st []sort.Iterator[*comparableTrace]
+ var st []sort.Iterator[*comparableTraceResult]
for _, f := range ff {
- if m, getErr := f.Get(); getErr != nil {
+ m, getErr := f.Get()
+ if getErr != nil {
allErr = multierr.Append(allErr, getErr)
- } else {
- d := m.Data()
- if d == nil {
- continue
- }
- resp := d.(*tracev1.InternalQueryResponse)
- responseCount++
- if span != nil {
- span.AddSubTrace(resp.TraceQueryResult)
- }
- st = append(st,
- newSortableTraces(resp.InternalTraces,
p.sortByTraceID))
+ continue
+ }
+ d := m.Data()
+ if d == nil {
+ continue
}
+ nodeResults, decodeErr := p.decodeNodeResults(d, span)
+ if decodeErr != nil {
+ allErr = multierr.Append(allErr, decodeErr)
+ continue
+ }
+ responseCount++
+ st = append(st,
newSortableTraceResults(iter.FromSlice(nodeResults), p.sortByTraceID))
}
if span != nil {
span.Tagf("response_count", "%d", responseCount)
}
sortIter := sort.NewItemIter(st, p.desc)
- var result []*tracev1.InternalTrace
- seen := make(map[string]*tracev1.InternalTrace)
+ var result []model.TraceResult
+ seen := make(map[string]int)
for sortIter.Next() {
- trace := sortIter.Val().InternalTrace
- if _, ok := seen[trace.TraceId]; !ok {
- seen[trace.TraceId] = trace
- result = append(result, trace)
+ tr := sortIter.Val().result
+ if idx, ok := seen[tr.TID]; !ok {
+ seen[tr.TID] = len(result)
+ result = append(result, tr)
} else {
- for _, span := range trace.Spans {
- spanExists := false
- for _, existingSpan := range
seen[trace.TraceId].Spans {
- if existingSpan.SpanId == span.SpanId {
- spanExists = true
- break
- }
- }
- if !spanExists {
- seen[trace.TraceId].Spans =
append(seen[trace.TraceId].Spans, span)
- }
- }
+ mergeTraceResultSpans(&result[idx], &tr)
}
}
if span != nil {
@@ -212,6 +202,33 @@ func (p *distributedPlan) Execute(ctx context.Context)
(iter.Iterator[model.Trac
}, nil
}
+// decodeNodeResults converts a single node response into columnar
+// []model.TraceResult, accepting either the proto
*tracev1.InternalQueryResponse
+// (flag-off / tracing path) or the native columnar frame ([]byte). The proto
+// branch is converted via internalTraceToResult so both paths feed a unified
+// model.TraceResult merge.
+func (p *distributedPlan) decodeNodeResults(d any, span *query.Span)
([]model.TraceResult, error) {
+ switch payload := d.(type) {
+ case *tracev1.InternalQueryResponse:
+ if span != nil {
+ span.AddSubTrace(payload.TraceQueryResult)
+ }
+ results := make([]model.TraceResult, 0,
len(payload.InternalTraces))
+ for _, internalTrace := range payload.InternalTraces {
+ results = append(results,
internalTraceToResult(internalTrace))
+ }
+ return results, nil
+ case []byte:
+ results, decodeErr := DecodeTraceResultFrame(payload)
+ if decodeErr != nil {
+ return nil, fmt.Errorf("decode trace result frame: %w",
decodeErr)
+ }
+ return results, nil
+ default:
+ return nil, fmt.Errorf("unexpected trace query response type
%T", d)
+ }
+}
+
func (p *distributedPlan) String() string {
return fmt.Sprintf("distributed:%s", p.queryTemplate.String())
}
@@ -228,72 +245,92 @@ 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. Tag columns
+// present only in src are first unioned into dst (NullTagValue-backfilled for
the
+// spans dst already holds) so no src-only tag column is dropped during the
merge.
+func mergeTraceResultSpans(dst, src *model.TraceResult) {
+ existing := make(map[string]struct{}, len(dst.SpanIDs))
+ for _, spanID := range dst.SpanIDs {
+ existing[spanID] = struct{}{}
+ }
+ for _, srcTag := range src.Tags {
+ if _, ok := findTag(dst.Tags, srcTag.Name); ok {
+ continue
+ }
+ backfill := make([]*modelv1.TagValue, len(dst.SpanIDs))
+ for fillIdx := range backfill {
+ backfill[fillIdx] = pbv1.NullTagValue
+ }
+ dst.Tags = append(dst.Tags, model.Tag{Name: srcTag.Name,
Values: backfill})
+ }
+ 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)
+ }
+ }
}
-func (t *sortableTraces) iter(fn func(*tracev1.InternalTrace)
(*comparableTrace, error)) bool {
- if t.index >= len(t.traces) {
- return false
- }
- cur, err := fn(t.traces[t.index])
- t.index++
- if err != nil {
- return t.iter(fn)
+func findTag(tags []model.Tag, name string) (model.Tag, bool) {
+ for _, tag := range tags {
+ if tag.Name == name {
+ return tag, true
+ }
}
- t.cur = cur
- return true
+ return model.Tag{}, false
}
var _ executor.TraceExecutable = (*distributedTraceLimit)(nil)
@@ -357,7 +394,7 @@ func newDistributedTraceLimit(input logical.UnresolvedPlan,
offset, limit uint32
type distributedTraceResultIterator struct {
err error
- traces []*tracev1.InternalTrace
+ traces []model.TraceResult
index int
}
@@ -369,47 +406,7 @@ func (t *distributedTraceResultIterator) Next()
(model.TraceResult, bool) {
return model.TraceResult{}, false
}
- trace := t.traces[t.index]
+ result := t.traces[t.index]
t.index++
-
- result := model.TraceResult{
- TID: trace.TraceId,
- Spans: make([][]byte, 0, len(trace.Spans)),
- SpanIDs: make([]string, 0, len(trace.Spans)),
- Tags: make([]model.Tag, 0, len(trace.Spans)),
- }
- tagValuesByName := make(map[string][]*modelv1.TagValue)
- var tagOrder []string
-
- for spanIdx, span := range trace.Spans {
- // Add span data
- result.Spans = append(result.Spans, span.Span)
- // Extract tags from this span and aggregate by name
- 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)
- }
- result.SpanIDs = append(result.SpanIDs, span.SpanId)
- }
- for _, tagName := range tagOrder {
- result.Tags = append(result.Tags, model.Tag{
- Name: tagName,
- Values: tagValuesByName[tagName],
- })
- }
return result, true
}
diff --git a/pkg/query/logical/trace/trace_plan_distributed_test.go
b/pkg/query/logical/trace/trace_plan_distributed_test.go
index 9464627d2..18561df35 100644
--- a/pkg/query/logical/trace/trace_plan_distributed_test.go
+++ b/pkg/query/logical/trace/trace_plan_distributed_test.go
@@ -25,41 +25,35 @@ import (
modelv1
"github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
tracev1
"github.com/apache/skywalking-banyandb/api/proto/banyandb/trace/v1"
pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
+ "github.com/apache/skywalking-banyandb/pkg/query/model"
)
-func TestDistributedTraceResultIteratorPreservesTagOrderAndSpanAlignment(t
*testing.T) {
- iterator := &distributedTraceResultIterator{
- traces: []*tracev1.InternalTrace{
+func TestInternalTraceToResultPreservesTagOrderAndSpanAlignment(t *testing.T) {
+ got := internalTraceToResult(&tracev1.InternalTrace{
+ TraceId: "trace-1",
+ Spans: []*tracev1.Span{
{
- TraceId: "trace-1",
- Spans: []*tracev1.Span{
- {
- Span: []byte("span-1"),
- SpanId: "span-1",
- Tags: []*modelv1.Tag{
- {Key: "service_id",
Value: strTagValueForDistributedTest("svc")},
- {Key: "trace_id",
Value: pbv1.NullTagValue},
- {Key: "trace_id",
Value: strTagValueForDistributedTest("trace-1")},
- {Key: "span_id", Value:
strTagValueForDistributedTest("span-1")},
- },
- },
- {
- Span: []byte("span-2"),
- SpanId: "span-2",
- Tags: []*modelv1.Tag{
- {Key: "service_id",
Value: strTagValueForDistributedTest("svc")},
- {Key: "trace_id",
Value: strTagValueForDistributedTest("trace-1")},
- {Key: "span_id", Value:
strTagValueForDistributedTest("span-2")},
- },
- },
+ Span: []byte("span-1"),
+ SpanId: "span-1",
+ Tags: []*modelv1.Tag{
+ {Key: "service_id", Value:
strTagValueForDistributedTest("svc")},
+ {Key: "trace_id", Value:
pbv1.NullTagValue},
+ {Key: "trace_id", Value:
strTagValueForDistributedTest("trace-1")},
+ {Key: "span_id", Value:
strTagValueForDistributedTest("span-1")},
+ },
+ },
+ {
+ Span: []byte("span-2"),
+ SpanId: "span-2",
+ Tags: []*modelv1.Tag{
+ {Key: "service_id", Value:
strTagValueForDistributedTest("svc")},
+ {Key: "trace_id", Value:
strTagValueForDistributedTest("trace-1")},
+ {Key: "span_id", Value:
strTagValueForDistributedTest("span-2")},
},
},
},
- }
+ })
- got, ok := iterator.Next()
-
- require.True(t, ok)
require.Len(t, got.Tags, 3)
require.Equal(t, "service_id", got.Tags[0].Name)
require.Equal(t, "trace_id", got.Tags[1].Name)
@@ -70,36 +64,29 @@ func
TestDistributedTraceResultIteratorPreservesTagOrderAndSpanAlignment(t *test
require.Equal(t, "span-2", got.Tags[2].Values[1].GetStr().GetValue())
}
-func TestDistributedTraceResultIteratorNullFillsMissingSpanTags(t *testing.T) {
- iterator := &distributedTraceResultIterator{
- traces: []*tracev1.InternalTrace{
+func TestInternalTraceToResultNullFillsMissingSpanTags(t *testing.T) {
+ got := internalTraceToResult(&tracev1.InternalTrace{
+ TraceId: "trace-1",
+ Spans: []*tracev1.Span{
{
- TraceId: "trace-1",
- Spans: []*tracev1.Span{
- {
- Span: []byte("span-1"),
- SpanId: "span-1",
- Tags: []*modelv1.Tag{
- {Key: "service_id",
Value: strTagValueForDistributedTest("svc")},
- {Key: "only_on_first",
Value: strTagValueForDistributedTest("first")},
- },
- },
- {
- Span: []byte("span-2"),
- SpanId: "span-2",
- Tags: []*modelv1.Tag{
- {Key: "service_id",
Value: strTagValueForDistributedTest("svc")},
- {Key: "only_on_second",
Value: strTagValueForDistributedTest("second")},
- },
- },
+ Span: []byte("span-1"),
+ SpanId: "span-1",
+ Tags: []*modelv1.Tag{
+ {Key: "service_id", Value:
strTagValueForDistributedTest("svc")},
+ {Key: "only_on_first", Value:
strTagValueForDistributedTest("first")},
+ },
+ },
+ {
+ Span: []byte("span-2"),
+ SpanId: "span-2",
+ Tags: []*modelv1.Tag{
+ {Key: "service_id", Value:
strTagValueForDistributedTest("svc")},
+ {Key: "only_on_second", Value:
strTagValueForDistributedTest("second")},
},
},
},
- }
-
- got, ok := iterator.Next()
+ })
- require.True(t, ok)
spanCount := len(got.Spans)
require.Equal(t, 2, spanCount)
@@ -121,6 +108,53 @@ func
TestDistributedTraceResultIteratorNullFillsMissingSpanTags(t *testing.T) {
require.Equal(t, "second",
values["only_on_second"][1].GetStr().GetValue())
}
+func TestMergeTraceResultSpansUnionsSrcOnlyTagColumns(t *testing.T) {
+ // dst (from node A) holds span s1 with only service_id.
+ dst := model.TraceResult{
+ TID: "t1",
+ Spans: [][]byte{[]byte("s1")},
+ SpanIDs: []string{"s1"},
+ Tags: []model.Tag{
+ {Name: "service_id", Values:
[]*modelv1.TagValue{strTagValueForDistributedTest("svc")}},
+ },
+ }
+ // src (from node B) re-reports s1 (duplicate) and adds s2, plus a tag
column
+ // http_method that dst never saw.
+ src := model.TraceResult{
+ TID: "t1",
+ Spans: [][]byte{[]byte("s1"), []byte("s2")},
+ SpanIDs: []string{"s1", "s2"},
+ Tags: []model.Tag{
+ {Name: "service_id", Values:
[]*modelv1.TagValue{strTagValueForDistributedTest("svc"),
strTagValueForDistributedTest("svc")}},
+ {Name: "http_method", Values:
[]*modelv1.TagValue{strTagValueForDistributedTest("GET"),
strTagValueForDistributedTest("POST")}},
+ },
+ }
+
+ mergeTraceResultSpans(&dst, &src)
+
+ // Duplicate span s1 is skipped; only s2 is appended.
+ require.Equal(t, []string{"s1", "s2"}, dst.SpanIDs)
+ require.Len(t, dst.Spans, 2)
+
+ // The src-only http_method column is unioned in (not dropped), and
every
+ // column stays aligned with the span count.
+ values := make(map[string][]*modelv1.TagValue, len(dst.Tags))
+ for _, tag := range dst.Tags {
+ require.Lenf(t, tag.Values, len(dst.SpanIDs), "tag %q must have
one value per span", tag.Name)
+ values[tag.Name] = tag.Values
+ }
+ require.Contains(t, values, "service_id")
+ require.Contains(t, values, "http_method")
+
+ require.Equal(t, "svc", values["service_id"][0].GetStr().GetValue())
+ require.Equal(t, "svc", values["service_id"][1].GetStr().GetValue())
+
+ // s1 predates the src-only column, so it is NULL-backfilled; the newly
+ // appended s2 carries its real value.
+ require.Same(t, pbv1.NullTagValue, values["http_method"][0])
+ require.Equal(t, "POST", values["http_method"][1].GetStr().GetValue())
+}
+
func strTagValueForDistributedTest(value string) *modelv1.TagValue {
return &modelv1.TagValue{
Value: &modelv1.TagValue_Str{
