You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by wu...@apache.org on 2025/10/30 03:33:39 UTC
(skywalking-banyandb) branch main updated: Fix Several Flaws of Trace Query (#830)
This is an automated email from the ASF dual-hosted git repository.
wusheng 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 41eef28c Fix Several Flaws of Trace Query (#830)
41eef28c is described below
commit 41eef28c7bb178c47f65ea99c4c9487fc20b5396
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Thu Oct 30 11:33:33 2025 +0800
Fix Several Flaws of Trace Query (#830)
---
banyand/dquery/measure.go | 3 +-
banyand/dquery/stream.go | 3 +-
banyand/dquery/trace.go | 3 +-
banyand/internal/sidx/interfaces.go | 8 +
banyand/internal/sidx/part_iter.go | 4 +-
banyand/internal/sidx/part_iter_test.go | 156 +++++
banyand/internal/sidx/query.go | 159 ++---
banyand/internal/sidx/query_result.go | 41 +-
banyand/internal/sidx/query_test.go | 108 +---
banyand/internal/sidx/sidx.go | 213 ++++++-
banyand/internal/sidx/sidx_test.go | 776 ++++++++++++++++++++++-
banyand/liaison/grpc/measure.go | 3 +-
banyand/liaison/grpc/stream.go | 3 +-
banyand/liaison/grpc/trace.go | 3 +-
banyand/query/processor.go | 9 +-
banyand/trace/block_writer.go | 101 +--
banyand/trace/block_writer_test.go | 206 ++++++
banyand/trace/flusher.go | 3 +
banyand/trace/merger.go | 4 +-
banyand/trace/merger_bench_test.go | 16 +-
banyand/trace/merger_test.go | 70 +-
banyand/trace/part.go | 12 +-
banyand/trace/part_test.go | 57 +-
banyand/trace/query.go | 99 ++-
banyand/trace/query_test.go | 45 +-
banyand/trace/snapshot_test.go | 67 +-
banyand/trace/streaming_pipeline.go | 171 ++++-
banyand/trace/streaming_pipeline_test.go | 117 ++--
banyand/trace/tracing.go | 108 ++++
banyand/trace/tstable.go | 4 +-
banyand/trace/tstable_test.go | 43 +-
pkg/flow/streaming/sliding_window.go | 3 +-
pkg/query/logical/tag_filter_adapter.go | 56 ++
pkg/query/logical/trace/trace_plan_local.go | 2 +
pkg/query/logical/trace/trace_plan_tag_filter.go | 43 +-
pkg/query/model/model.go | 3 +
pkg/query/model/tag_filter_matcher.go | 41 ++
37 files changed, 2255 insertions(+), 508 deletions(-)
diff --git a/banyand/dquery/measure.go b/banyand/dquery/measure.go
index 77a66584..cbe8fe3e 100644
--- a/banyand/dquery/measure.go
+++ b/banyand/dquery/measure.go
@@ -115,14 +115,15 @@ func (p *measureQueryProcessor) Rev(ctx context.Context, message bus.Message) (r
data := resp.Data()
switch d := data.(type) {
case *measurev1.QueryResponse:
+ span.Stop()
d.Trace = tracer.ToProto()
case *common.Error:
span.Error(errors.New(d.Error()))
+ span.Stop()
resp = bus.NewMessage(bus.MessageID(now), &measurev1.QueryResponse{Trace: tracer.ToProto()})
default:
panic("unexpected data type")
}
- span.Stop()
}()
}
diff --git a/banyand/dquery/stream.go b/banyand/dquery/stream.go
index 646415a9..5c957db3 100644
--- a/banyand/dquery/stream.go
+++ b/banyand/dquery/stream.go
@@ -113,14 +113,15 @@ func (p *streamQueryProcessor) Rev(ctx context.Context, message bus.Message) (re
data := resp.Data()
switch d := data.(type) {
case *streamv1.QueryResponse:
+ span.Stop()
d.Trace = tracer.ToProto()
case *common.Error:
span.Error(errors.New(d.Error()))
+ span.Stop()
resp = bus.NewMessage(bus.MessageID(now), &streamv1.QueryResponse{Trace: tracer.ToProto()})
default:
panic("unexpected data type")
}
- span.Stop()
}()
}
se := plan.(executor.StreamExecutable)
diff --git a/banyand/dquery/trace.go b/banyand/dquery/trace.go
index a4a4dd60..d2349c60 100644
--- a/banyand/dquery/trace.go
+++ b/banyand/dquery/trace.go
@@ -117,14 +117,15 @@ func (p *traceQueryProcessor) Rev(ctx context.Context, message bus.Message) (res
data := resp.Data()
switch d := data.(type) {
case *tracev1.InternalQueryResponse:
+ span.Stop()
d.TraceQueryResult = tracer.ToProto()
case *common.Error:
span.Error(errors.New(d.Error()))
+ span.Stop()
resp = bus.NewMessage(bus.MessageID(now), &tracev1.QueryResponse{TraceQueryResult: tracer.ToProto()})
default:
panic("unexpected data type")
}
- span.Stop()
}()
}
te := plan.(executor.TraceExecutable)
diff --git a/banyand/internal/sidx/interfaces.go b/banyand/internal/sidx/interfaces.go
index bee5c743..5d80d6e2 100644
--- a/banyand/internal/sidx/interfaces.go
+++ b/banyand/internal/sidx/interfaces.go
@@ -74,6 +74,7 @@ type WriteRequest struct {
// QueryRequest specifies parameters for a query operation, following StreamQueryOptions pattern.
type QueryRequest struct {
Filter index.Filter
+ TagFilter model.TagFilterMatcher
Order *index.OrderBy
MinKey *int64
MaxKey *int64
@@ -91,6 +92,7 @@ type QueryResponse struct {
Data [][]byte
Tags [][]Tag
SIDs []common.SeriesID
+ PartIDs []uint64
Metadata ResponseMetadata
}
@@ -106,6 +108,7 @@ func (qr *QueryResponse) Reset() {
qr.Data = qr.Data[:0]
qr.Tags = qr.Tags[:0]
qr.SIDs = qr.SIDs[:0]
+ qr.PartIDs = qr.PartIDs[:0]
qr.Metadata = ResponseMetadata{}
}
@@ -114,6 +117,7 @@ func (qr *QueryResponse) Validate() error {
keysLen := len(qr.Keys)
dataLen := len(qr.Data)
sidsLen := len(qr.SIDs)
+ partIDsLen := len(qr.PartIDs)
if keysLen != dataLen {
return fmt.Errorf("inconsistent array lengths: keys=%d, data=%d", keysLen, dataLen)
@@ -121,6 +125,9 @@ func (qr *QueryResponse) Validate() error {
if keysLen != sidsLen {
return fmt.Errorf("inconsistent array lengths: keys=%d, sids=%d", keysLen, sidsLen)
}
+ if keysLen != partIDsLen {
+ return fmt.Errorf("inconsistent array lengths: keys=%d, partIDs=%d", keysLen, partIDsLen)
+ }
// Validate Tags structure if present
if len(qr.Tags) > 0 {
@@ -146,6 +153,7 @@ func (qr *QueryResponse) CopyFrom(other *QueryResponse) {
// Copy parallel arrays
qr.Keys = append(qr.Keys[:0], other.Keys...)
qr.SIDs = append(qr.SIDs[:0], other.SIDs...)
+ qr.PartIDs = append(qr.PartIDs[:0], other.PartIDs...)
// Deep copy data
if cap(qr.Data) < len(other.Data) {
diff --git a/banyand/internal/sidx/part_iter.go b/banyand/internal/sidx/part_iter.go
index d49e8611..39f3f47c 100644
--- a/banyand/internal/sidx/part_iter.go
+++ b/banyand/internal/sidx/part_iter.go
@@ -245,9 +245,7 @@ func (pi *partIter) findBlock() bool {
return false
}
if shouldSkip {
- if !pi.nextSeriesID() {
- return false
- }
+ bhs = bhs[1:]
continue
}
}
diff --git a/banyand/internal/sidx/part_iter_test.go b/banyand/internal/sidx/part_iter_test.go
index a64fdde5..4b1e2258 100644
--- a/banyand/internal/sidx/part_iter_test.go
+++ b/banyand/internal/sidx/part_iter_test.go
@@ -681,3 +681,159 @@ func (mbf *mockBlockFilter) Execute(_ index.GetSearcher, _ common.SeriesID, _ *i
// Not used in our tests, return empty implementation
return nil, nil, nil
}
+
+func TestPartIterShouldSkip(t *testing.T) {
+ tempDir := t.TempDir()
+ testFS := fs.NewLocalFileSystem()
+
+ // Strategy: Create many elements to exceed maxElementsPerBlock (8192) and force multiple blocks.
+ // We'll create elements in two batches with different tag values but same seriesID.
+
+ const elementsPerBatch = 8200 // Exceed maxElementsPerBlock to force multiple blocks
+ var allElements []testElement
+
+ // First batch: seriesID=1, status="pending", keys 0-8199
+ for i := 0; i < elementsPerBatch; i++ {
+ allElements = append(allElements, testElement{
+ seriesID: 1,
+ userKey: int64(i),
+ data: []byte(fmt.Sprintf("pending_data_%d", i)),
+ tags: []tag{
+ {
+ name: "status",
+ value: []byte("pending"),
+ valueType: pbv1.ValueTypeStr,
+ },
+ },
+ })
+ }
+
+ // Second batch: seriesID=1, status="success", keys 20000-28199
+ // Use a large gap in keys to ensure they're in different blocks
+ for i := 0; i < elementsPerBatch; i++ {
+ allElements = append(allElements, testElement{
+ seriesID: 1,
+ userKey: int64(20000 + i),
+ data: []byte(fmt.Sprintf("success_data_%d", i)),
+ tags: []tag{
+ {
+ name: "status",
+ value: []byte("success"),
+ valueType: pbv1.ValueTypeStr,
+ },
+ },
+ })
+ }
+
+ // Add a third series to verify we don't incorrectly skip to this
+ allElements = append(allElements, testElement{
+ seriesID: 2,
+ userKey: 5000,
+ data: []byte("series2_data"),
+ tags: []tag{
+ {
+ name: "status",
+ value: []byte("other"),
+ valueType: pbv1.ValueTypeStr,
+ },
+ },
+ })
+
+ elements := createTestElements(allElements)
+ defer releaseElements(elements)
+
+ // Create memPart and flush it
+ mp := GenerateMemPart()
+ defer ReleaseMemPart(mp)
+ mp.mustInitFromElements(elements)
+
+ partDir := filepath.Join(tempDir, "test_part")
+ mp.mustFlush(testFS, partDir)
+
+ part := mustOpenPart(1, partDir, testFS)
+ defer part.close()
+
+ // Log the blocks in the part to understand the structure
+ t.Logf("Part has %d primary block metadata entries", len(part.primaryBlockMetadata))
+ for i, pbm := range part.primaryBlockMetadata {
+ t.Logf(" Primary block %d: seriesID=%d, keys [%d-%d]", i, pbm.seriesID, pbm.minKey, pbm.maxKey)
+ }
+
+ // Create a selective mock filter that skips blocks with status="pending"
+ // but allows blocks with status="success"
+ selectiveFilter := &selectiveMockBlockFilter{
+ tagName: "status",
+ skipValue: "pending",
+ skipCallCount: 0,
+ }
+
+ // Test with the selective filter, querying only seriesID=1
+ bma := &blockMetadataArray{}
+ defer bma.reset()
+ pi := &partIter{}
+
+ bma.reset()
+ pi.init(bma, part, []common.SeriesID{1, 2}, 0, 30000, selectiveFilter)
+
+ // Iterate through blocks and collect results
+ var foundBlocks []struct {
+ seriesID common.SeriesID
+ minKey int64
+ maxKey int64
+ }
+
+ for pi.nextBlock() {
+ foundBlocks = append(foundBlocks, struct {
+ seriesID common.SeriesID
+ minKey int64
+ maxKey int64
+ }{
+ seriesID: pi.curBlock.seriesID,
+ minKey: pi.curBlock.minKey,
+ maxKey: pi.curBlock.maxKey,
+ })
+ t.Logf("Found block: seriesID=%d, minKey=%d, maxKey=%d",
+ pi.curBlock.seriesID, pi.curBlock.minKey, pi.curBlock.maxKey)
+ }
+
+ require.NoError(t, pi.error())
+
+ // Verify the filter was called at least once
+ assert.Greater(t, selectiveFilter.skipCallCount, 0, "filter should have been called")
+
+ foundSeries1Success := false
+ for _, block := range foundBlocks {
+ if block.seriesID == 1 && block.minKey >= 20000 {
+ foundSeries1Success = true
+ t.Logf("✓ Found the expected seriesID=1 success block: keys [%d-%d]", block.minKey, block.maxKey)
+ }
+ }
+
+ // This assertion will FAIL with the bug, demonstrating the issue
+ assert.True(t, foundSeries1Success,
+ "BUG: Should find seriesID=1 block with status='success' (minKey >= 20000), "+
+ "but the iterator incorrectly skipped to next series after first block with status='pending' failed the filter. "+
+ "Found %d total blocks", len(foundBlocks))
+}
+
+// selectiveMockBlockFilter is a mock filter that selectively skips blocks based on tag values.
+type selectiveMockBlockFilter struct {
+ tagName string
+ skipValue string
+ skipCallCount int
+}
+
+func (smf *selectiveMockBlockFilter) ShouldSkip(filterOp index.FilterOp) (bool, error) {
+ smf.skipCallCount++
+ // Check if the block contains the skip value for the specified tag
+ hasSkipValue := filterOp.Eq(smf.tagName, smf.skipValue)
+ return hasSkipValue, nil
+}
+
+func (smf *selectiveMockBlockFilter) String() string {
+ return fmt.Sprintf("selectiveMockBlockFilter(tag=%s, skipValue=%s)", smf.tagName, smf.skipValue)
+}
+
+func (smf *selectiveMockBlockFilter) Execute(_ index.GetSearcher, _ common.SeriesID, _ *index.RangeOpts) (posting.List, posting.List, error) {
+ return nil, nil, nil
+}
diff --git a/banyand/internal/sidx/query.go b/banyand/internal/sidx/query.go
index a6cea976..c9618f58 100644
--- a/banyand/internal/sidx/query.go
+++ b/banyand/internal/sidx/query.go
@@ -19,9 +19,11 @@ package sidx
import (
"context"
+ "errors"
"math"
"sort"
"sync"
+ "sync/atomic"
"github.com/apache/skywalking-banyandb/api/common"
modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
@@ -31,12 +33,7 @@ import (
// StreamingQuery implements the streaming query API defined on SIDX.
func (s *sidx) StreamingQuery(ctx context.Context, req QueryRequest) (<-chan *QueryResponse, <-chan error) {
- chanSize := req.MaxBatchSize
- if chanSize < 0 {
- chanSize = 0
- }
-
- resultsCh := make(chan *QueryResponse, chanSize)
+ resultsCh := make(chan *QueryResponse)
errCh := make(chan error, 1)
go func() {
@@ -57,38 +54,34 @@ func (s *sidx) StreamingQuery(ctx context.Context, req QueryRequest) (<-chan *Qu
return resultsCh, errCh
}
-type streamingStats struct {
- batches int
- elements int
+type batchMetrics struct {
+ totalBlocksScanned int
+ totalCursorsCreated int
+ cumulativeHeapSize int
+ inputBatchesProcessed int
+ blockElementsLoaded atomic.Int64
+ outputElementsEmitted atomic.Int64
+ elementsDeduplicated atomic.Int64
+ blocksSkipped atomic.Int64
+ outputBatchCount int
+ outputElementCount int
}
-func (s *streamingStats) record(chunk *QueryResponse) {
+func (s *batchMetrics) record(chunk *QueryResponse) {
if s == nil || chunk == nil {
return
}
- s.batches++
- s.elements += chunk.Len()
-}
-
-type batchMetrics struct {
- totalBatchBlockCount int
- totalCursorsBuilt int
- totalHeapSizeAfterPush int
- batchesProcessed int
+ s.outputBatchCount++
+ s.outputElementCount += chunk.Len()
}
// runStreamingQuery executes the streaming query workflow and pushes batches to resultsCh.
func (s *sidx) runStreamingQuery(ctx context.Context, req QueryRequest, resultsCh chan<- *QueryResponse) (err error) {
s.totalQueries.Add(1)
- var (
- stats streamingStats
- heapInitialized bool
- )
-
ctx, span := startStreamingSpan(ctx, req)
defer func() {
- finalizeStreamingSpan(span, stats, heapInitialized, &err)
+ finalizeStreamingSpan(span, &err)
}()
snap := s.currentSnapshot()
@@ -106,13 +99,7 @@ func (s *sidx) runStreamingQuery(ctx context.Context, req QueryRequest, resultsC
}
defer resources.cleanup()
- if loopErr := s.processStreamingLoop(ctx, req, resources, resultsCh, &stats); loopErr != nil {
- heapInitialized = resources.heap.initialized
- return loopErr
- }
- heapInitialized = resources.heap.initialized
-
- return nil
+ return s.processStreamingLoop(ctx, req, resources, resultsCh)
}
type streamingQueryResources struct {
@@ -149,13 +136,10 @@ func startStreamingSpan(ctx context.Context, req QueryRequest) (context.Context,
return spanCtx, span
}
-func finalizeStreamingSpan(span *query.Span, stats streamingStats, heapInitialized bool, errPtr *error) {
+func finalizeStreamingSpan(span *query.Span, errPtr *error) {
if span == nil {
return
}
- span.Tagf("heap_initialized", "%t", heapInitialized)
- span.Tagf("responses_emitted", "%d", stats.batches)
- span.Tagf("elements_emitted", "%d", stats.elements)
if errPtr != nil && *errPtr != nil {
span.Error(*errPtr)
}
@@ -189,6 +173,14 @@ func (s *sidx) prepareStreamingResources(
prepareSpan.Tagf("max_key", "%d", maxKey)
prepareSpan.Tagf("ascending", "%t", asc)
prepareSpan.Tagf("part_count", "%d", len(parts))
+ // Sample the first few part paths
+ sampleSize := 5
+ if len(parts) < sampleSize {
+ sampleSize = len(parts)
+ }
+ for i := 0; i < sampleSize; i++ {
+ prepareSpan.Tag("part_path_"+string(rune('0'+i)), parts[i].path)
+ }
}
if len(parts) == 0 {
return nil, false
@@ -252,7 +244,6 @@ func (s *sidx) processStreamingLoop(
req QueryRequest,
resources *streamingQueryResources,
resultsCh chan<- *QueryResponse,
- stats *streamingStats,
) error {
var loopSpan *query.Span
var metrics *batchMetrics
@@ -261,36 +252,40 @@ func (s *sidx) processStreamingLoop(
loopSpan, ctx = tracer.StartSpan(ctx, "sidx.process-streaming-loop")
defer func() {
if loopSpan != nil {
- loopSpan.Tagf("total_batches_processed", "%d", stats.batches)
s.addBatchMetricsToSpan(loopSpan, metrics)
loopSpan.Stop()
}
}()
}
- batchCount := 0
+ scannerBatchCount := 0
for {
select {
case <-ctx.Done():
if loopSpan != nil {
loopSpan.Tag("termination_reason", "context_canceled")
- loopSpan.Tagf("batches_before_cancel", "%d", batchCount)
+ loopSpan.Tagf("scanner_batches_before_cancel", "%d", scannerBatchCount)
}
return ctx.Err()
case batch, ok := <-resources.blockCh:
if !ok {
if loopSpan != nil {
loopSpan.Tag("termination_reason", "channel_closed")
- loopSpan.Tagf("total_batches", "%d", batchCount)
+ loopSpan.Tagf("total_scanner_batches", "%d", scannerBatchCount)
}
- return resources.heap.merge(ctx, req.MaxBatchSize, resultsCh, stats)
+ return resources.heap.merge(ctx, req.MaxBatchSize, resultsCh, metrics)
}
- batchCount++
- if err := s.handleStreamingBatch(ctx, batch, resources, req, resultsCh, stats, metrics); err != nil {
+ scannerBatchCount++
+ if err := s.handleStreamingBatch(ctx, batch, resources, req, resultsCh, metrics); err != nil {
if loopSpan != nil {
- loopSpan.Tag("termination_reason", "batch_error")
- loopSpan.Tagf("batches_before_error", "%d", batchCount)
- loopSpan.Error(err)
+ if errors.Is(err, context.Canceled) {
+ loopSpan.Tag("termination_reason", "context_canceled")
+ loopSpan.Tagf("scanner_batches_before_cancel", "%d", scannerBatchCount)
+ } else {
+ loopSpan.Tag("termination_reason", "batch_error")
+ loopSpan.Tagf("scanner_batches_before_error", "%d", scannerBatchCount)
+ loopSpan.Error(err)
+ }
}
return err
}
@@ -304,7 +299,6 @@ func (s *sidx) handleStreamingBatch(
resources *streamingQueryResources,
req QueryRequest,
resultsCh chan<- *QueryResponse,
- stats *streamingStats,
metrics *batchMetrics,
) error {
if batch.err != nil {
@@ -316,10 +310,10 @@ func (s *sidx) handleStreamingBatch(
// Collect batch block count metric
if metrics != nil {
batchBlockCount := len(batch.bss)
- metrics.totalBatchBlockCount += batchBlockCount
+ metrics.totalBlocksScanned += batchBlockCount
}
- cursors, cursorsErr := s.buildCursorsForBatch(ctx, batch, resources.tagsToLoad, req, resources.asc)
+ cursors, cursorsErr := s.buildCursorsForBatch(ctx, batch, resources.tagsToLoad, req, resources.asc, metrics)
releaseBlockScanResultBatch(batch)
if cursorsErr != nil {
return cursorsErr
@@ -328,7 +322,7 @@ func (s *sidx) handleStreamingBatch(
// Collect cursors built metric
if metrics != nil {
cursorsBuilt := len(cursors)
- metrics.totalCursorsBuilt += cursorsBuilt
+ metrics.totalCursorsCreated += cursorsBuilt
}
resources.heap.pushCursors(cursors)
@@ -336,26 +330,32 @@ func (s *sidx) handleStreamingBatch(
// Collect heap size metric
if metrics != nil {
heapSize := resources.heap.Len()
- metrics.totalHeapSizeAfterPush += heapSize
- metrics.batchesProcessed++
+ metrics.cumulativeHeapSize += heapSize
+ metrics.inputBatchesProcessed++
}
- return resources.heap.merge(ctx, req.MaxBatchSize, resultsCh, stats)
+ return resources.heap.merge(ctx, req.MaxBatchSize, resultsCh, metrics)
}
func (s *sidx) addBatchMetricsToSpan(span *query.Span, metrics *batchMetrics) {
if span == nil || metrics == nil {
return
}
- span.Tagf("batches_processed", "%d", metrics.batchesProcessed)
- span.Tagf("total_batch_block_count", "%d", metrics.totalBatchBlockCount)
- span.Tagf("total_cursors_built", "%d", metrics.totalCursorsBuilt)
- span.Tagf("total_heap_size_after_push", "%d", metrics.totalHeapSizeAfterPush)
- if metrics.batchesProcessed > 0 {
- span.Tagf("avg_batch_block_count", "%.2f", float64(metrics.totalBatchBlockCount)/float64(metrics.batchesProcessed))
- span.Tagf("avg_cursors_built", "%.2f", float64(metrics.totalCursorsBuilt)/float64(metrics.batchesProcessed))
- span.Tagf("avg_heap_size_after_push", "%.2f", float64(metrics.totalHeapSizeAfterPush)/float64(metrics.batchesProcessed))
- }
+ span.Tagf("input_batches_processed", "%d", metrics.inputBatchesProcessed)
+ span.Tagf("total_blocks_scanned", "%d", metrics.totalBlocksScanned)
+ span.Tagf("total_cursors_created", "%d", metrics.totalCursorsCreated)
+ span.Tagf("cumulative_heap_size", "%d", metrics.cumulativeHeapSize)
+ if metrics.inputBatchesProcessed > 0 {
+ span.Tagf("avg_blocks_scanned_per_batch", "%.2f", float64(metrics.totalBlocksScanned)/float64(metrics.inputBatchesProcessed))
+ span.Tagf("avg_cursors_created_per_batch", "%.2f", float64(metrics.totalCursorsCreated)/float64(metrics.inputBatchesProcessed))
+ span.Tagf("avg_heap_size", "%.2f", float64(metrics.cumulativeHeapSize)/float64(metrics.inputBatchesProcessed))
+ }
+ span.Tagf("block_elements_loaded", "%d", metrics.blockElementsLoaded.Load())
+ span.Tagf("output_elements_emitted", "%d", metrics.outputElementsEmitted.Load())
+ span.Tagf("elements_deduplicated", "%d", metrics.elementsDeduplicated.Load())
+ span.Tagf("blocks_skipped", "%d", metrics.blocksSkipped.Load())
+ span.Tagf("output_element_count", "%d", metrics.outputElementCount)
+ span.Tagf("output_batch_count", "%d", metrics.outputBatchCount)
}
func determineTagsToLoad(req QueryRequest) map[string]struct{} {
@@ -379,21 +379,9 @@ func (s *sidx) buildCursorsForBatch(
tagsToLoad map[string]struct{},
req QueryRequest,
asc bool,
+ metrics *batchMetrics,
) ([]*blockCursor, error) {
- var buildSpan *query.Span
- if tracer := query.GetTracer(ctx); tracer != nil {
- buildSpan, ctx = tracer.StartSpan(ctx, "sidx.build-cursors-for-batch")
- defer func() {
- if buildSpan != nil {
- buildSpan.Stop()
- }
- }()
- }
-
if len(batch.bss) == 0 {
- if buildSpan != nil {
- buildSpan.Tag("empty_batch", "true")
- }
return nil, nil
}
@@ -405,12 +393,6 @@ func (s *sidx) buildCursorsForBatch(
workerCount = len(batch.bss)
}
- if buildSpan != nil {
- buildSpan.Tagf("block_scan_results", "%d", len(batch.bss))
- buildSpan.Tagf("worker_count", "%d", workerCount)
- buildSpan.Tagf("tags_to_load", "%d", len(tagsToLoad))
- }
-
jobCh := make(chan blockScanResult, workerCount)
resultCh := make(chan *blockCursor, len(batch.bss))
@@ -436,7 +418,7 @@ func (s *sidx) buildCursorsForBatch(
bc := generateBlockCursor()
bc.init(bsResult.p, &bsResult.bm, req)
- if s.loadBlockCursor(bc, tmpBlock, bsResult, tagsToLoad, req, s.pm) {
+ if s.loadBlockCursor(bc, tmpBlock, bsResult, tagsToLoad, req, s.pm, metrics) {
if asc {
bc.idx = 0
} else {
@@ -466,9 +448,6 @@ func (s *sidx) buildCursorsForBatch(
for bc := range resultCh {
releaseBlockCursor(bc)
}
- if buildSpan != nil {
- buildSpan.Tag("canceled", "job_distribution")
- }
return nil, ctx.Err()
case jobCh <- batch.bss[i]:
}
@@ -483,18 +462,10 @@ func (s *sidx) buildCursorsForBatch(
cursors = append(cursors, bc)
}
- if buildSpan != nil {
- buildSpan.Tagf("cursors_created", "%d", len(cursors))
- }
-
if err := ctx.Err(); err != nil {
for i := range cursors {
releaseBlockCursor(cursors[i])
}
- if buildSpan != nil {
- buildSpan.Tag("canceled", "after_workers")
- buildSpan.Error(err)
- }
return nil, err
}
diff --git a/banyand/internal/sidx/query_result.go b/banyand/internal/sidx/query_result.go
index 0024bca8..ab45ee88 100644
--- a/banyand/internal/sidx/query_result.go
+++ b/banyand/internal/sidx/query_result.go
@@ -207,7 +207,7 @@ func (qr *queryResult) loadTagData(tmpBlock *block, p *part, tagName string, tag
}
// convertBlockToResponse converts SIDX block data to QueryResponse format.
-func (qr *queryResult) convertBlockToResponse(block *block, seriesID common.SeriesID, result *QueryResponse) {
+func (qr *queryResult) convertBlockToResponse(block *block, seriesID common.SeriesID, partID uint64, result *QueryResponse) {
elemCount := len(block.userKeys)
for i := 0; i < elemCount; i++ {
@@ -229,6 +229,7 @@ func (qr *queryResult) convertBlockToResponse(block *block, seriesID common.Seri
result.Keys = append(result.Keys, key)
result.Data = append(result.Data, block.data[i])
result.SIDs = append(result.SIDs, seriesID)
+ result.PartIDs = append(result.PartIDs, partID)
// Convert tag map to tag slice for this element
elementTags := qr.extractElementTags(block, i)
@@ -288,10 +289,11 @@ func mergeQueryResponseShards(shards []*QueryResponse, maxElements int) *QueryRe
if len(qrh.cursors) == 0 {
return &QueryResponse{
- Keys: make([]int64, 0),
- Data: make([][]byte, 0),
- Tags: make([][]Tag, 0),
- SIDs: make([]common.SeriesID, 0),
+ Keys: make([]int64, 0),
+ Data: make([][]byte, 0),
+ Tags: make([][]Tag, 0),
+ SIDs: make([]common.SeriesID, 0),
+ PartIDs: make([]uint64, 0),
}
}
@@ -324,10 +326,11 @@ func mergeQueryResponseShardsDesc(shards []*QueryResponse, maxElements int) *Que
if len(qrh.cursors) == 0 {
return &QueryResponse{
- Keys: make([]int64, 0),
- Data: make([][]byte, 0),
- Tags: make([][]Tag, 0),
- SIDs: make([]common.SeriesID, 0),
+ Keys: make([]int64, 0),
+ Data: make([][]byte, 0),
+ Tags: make([][]Tag, 0),
+ SIDs: make([]common.SeriesID, 0),
+ PartIDs: make([]uint64, 0),
}
}
@@ -393,10 +396,11 @@ func (qrh *QueryResponseHeap) reset() {
// mergeWithHeap performs heap-based merge of QueryResponse shards.
func (qrh *QueryResponseHeap) mergeWithHeap(limit int) *QueryResponse {
result := &QueryResponse{
- Keys: make([]int64, 0, limit),
- Data: make([][]byte, 0, limit),
- Tags: make([][]Tag, 0, limit),
- SIDs: make([]common.SeriesID, 0, limit),
+ Keys: make([]int64, 0, limit),
+ Data: make([][]byte, 0, limit),
+ Tags: make([][]Tag, 0, limit),
+ SIDs: make([]common.SeriesID, 0, limit),
+ PartIDs: make([]uint64, 0, limit),
}
step := -1
@@ -419,9 +423,10 @@ func (qrh *QueryResponseHeap) mergeWithHeap(limit int) *QueryResponse {
}
var (
- data []byte
- tags []Tag
- sid common.SeriesID
+ data []byte
+ tags []Tag
+ sid common.SeriesID
+ partID uint64
)
if idx < len(resp.Data) {
data = resp.Data[idx]
@@ -432,12 +437,16 @@ func (qrh *QueryResponseHeap) mergeWithHeap(limit int) *QueryResponse {
if idx < len(resp.SIDs) {
sid = resp.SIDs[idx]
}
+ if idx < len(resp.PartIDs) {
+ partID = resp.PartIDs[idx]
+ }
// Copy element from top cursor
result.Keys = append(result.Keys, resp.Keys[idx])
result.Data = append(result.Data, data)
result.Tags = append(result.Tags, tags)
result.SIDs = append(result.SIDs, sid)
+ result.PartIDs = append(result.PartIDs, partID)
// Advance cursor
topCursor.idx += step
diff --git a/banyand/internal/sidx/query_test.go b/banyand/internal/sidx/query_test.go
index e390cd26..2c841250 100644
--- a/banyand/internal/sidx/query_test.go
+++ b/banyand/internal/sidx/query_test.go
@@ -20,7 +20,6 @@ package sidx
import (
"context"
"sort"
- "strconv"
"strings"
"testing"
"time"
@@ -29,12 +28,10 @@ import (
"github.com/stretchr/testify/require"
"github.com/apache/skywalking-banyandb/api/common"
- 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/banyand/internal/test"
"github.com/apache/skywalking-banyandb/pkg/index"
pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
- "github.com/apache/skywalking-banyandb/pkg/query"
"github.com/apache/skywalking-banyandb/pkg/query/model"
)
@@ -352,8 +349,9 @@ func TestSIDX_Query_Validation(t *testing.T) {
tt := tt
t.Run(tt.name, func(t *testing.T) {
resultsCh, errCh := sidx.StreamingQuery(ctx, tt.req)
- // revive:disable-next-line:empty-block
+ // Drain resultsCh to wait for query to close
for range resultsCh {
+ _ = 0
}
err, ok := <-errCh
if tt.expectErr {
@@ -610,70 +608,6 @@ func TestSIDX_StreamingQuery_ChannelLifecycle(t *testing.T) {
}
}
-func TestSIDX_StreamingQuery_Tracing(t *testing.T) {
- sidx := createTestSIDX(t)
- defer func() {
- assert.NoError(t, sidx.Close())
- }()
-
- reqs := []WriteRequest{
- createTestWriteRequest(1, 100, "trace-100"),
- createTestWriteRequest(1, 200, "trace-200"),
- createTestWriteRequest(1, 300, "trace-300"),
- }
- writeTestData(t, sidx, reqs, 10, 10)
- waitForIntroducerLoop()
-
- tracer, ctx := query.NewTracer(context.Background(), "sidx-streaming-query")
-
- minKey := int64(100)
- maxKey := int64(300)
- queryReq := QueryRequest{
- SeriesIDs: []common.SeriesID{1},
- Order: &index.OrderBy{Sort: modelv1.Sort_SORT_ASC},
- MaxBatchSize: 1,
- MinKey: &minKey,
- MaxKey: &maxKey,
- }
-
- resultsCh, errCh := sidx.StreamingQuery(ctx, queryReq)
-
- var (
- responseCount int
- elementCount int
- )
- for res := range resultsCh {
- require.NoError(t, res.Error)
- responseCount++
- elementCount += res.Len()
- }
-
- for err := range errCh {
- require.NoError(t, err)
- }
-
- traceProto := tracer.ToProto()
- require.NotNil(t, traceProto)
-
- span := findSpanByMessage(traceProto.GetSpans(), "sidx.run-streaming-query")
- require.NotNil(t, span)
- require.False(t, span.GetError())
-
- tags := spanTagsToMap(span)
-
- require.Equal(t, "false", requireTag(t, tags, "filter_present"))
- require.Equal(t, queryReq.Order.Sort.String(), requireTag(t, tags, "order_sort"))
- require.Equal(t, "0", requireTag(t, tags, "order_type"))
- require.Equal(t, strconv.Itoa(len(queryReq.SeriesIDs)), requireTag(t, tags, "series_id_count"))
- require.Equal(t, strconv.Itoa(0), requireTag(t, tags, "projected_tags"))
- require.Equal(t, strconv.FormatInt(minKey, 10), requireTag(t, tags, "min_key"))
- require.Equal(t, strconv.FormatInt(maxKey, 10), requireTag(t, tags, "max_key"))
- require.Equal(t, strconv.Itoa(queryReq.MaxBatchSize), requireTag(t, tags, "max_batch_size"))
- require.Equal(t, strconv.Itoa(responseCount), requireTag(t, tags, "responses_emitted"))
- require.Equal(t, strconv.Itoa(elementCount), requireTag(t, tags, "elements_emitted"))
- require.Equal(t, "true", requireTag(t, tags, "heap_initialized"))
-}
-
func TestSIDX_StreamingQuery_ErrorPropagation(t *testing.T) {
t.Run("validation_error", func(t *testing.T) {
sidx := createTestSIDX(t)
@@ -692,6 +626,10 @@ func TestSIDX_StreamingQuery_ErrorPropagation(t *testing.T) {
t.Fatal("expected validation error but errCh did not emit")
}
+ // Drain resultsCh to wait for query to close
+ for range resultsCh {
+ _ = 0
+ }
_, ok := <-resultsCh
require.False(t, ok)
})
@@ -719,6 +657,7 @@ func TestSIDX_StreamingQuery_ErrorPropagation(t *testing.T) {
MaxBatchSize: 0,
})
+ // Drain resultsCh to wait for query to close
for range resultsCh {
t.Fatal("expected no results for quota exceeded")
}
@@ -754,6 +693,7 @@ func TestSIDX_StreamingQuery_EdgeCases(t *testing.T) {
MaxBatchSize: 2,
})
+ // Drain resultsCh to wait for query to close
for range resultsCh {
t.Fatal("expected no results for missing series")
}
@@ -940,8 +880,9 @@ func TestSIDX_StreamingQuery_ContextCancellation(t *testing.T) {
done := make(chan struct{})
go func() {
- //revive:disable-next-line:empty-block
- for range resultsCh { // Drain the channel
+ // Drain resultsCh to wait for query to close
+ for range resultsCh {
+ _ = 0
}
close(done)
}()
@@ -961,30 +902,3 @@ func TestSIDX_StreamingQuery_ContextCancellation(t *testing.T) {
require.ErrorIs(t, err, context.Canceled)
}
}
-
-func findSpanByMessage(spans []*commonv1.Span, message string) *commonv1.Span {
- for _, span := range spans {
- if span.GetMessage() == message {
- return span
- }
- if child := findSpanByMessage(span.GetChildren(), message); child != nil {
- return child
- }
- }
- return nil
-}
-
-func spanTagsToMap(span *commonv1.Span) map[string]string {
- tags := make(map[string]string, len(span.GetTags()))
- for _, tag := range span.GetTags() {
- tags[tag.GetKey()] = tag.GetValue()
- }
- return tags
-}
-
-func requireTag(t *testing.T, tags map[string]string, key string) string {
- t.Helper()
- val, ok := tags[key]
- require.True(t, ok, "expected tag %q to be present, tags=%v", key, tags)
- return val
-}
diff --git a/banyand/internal/sidx/sidx.go b/banyand/internal/sidx/sidx.go
index f73e43f3..e48a7f32 100644
--- a/banyand/internal/sidx/sidx.go
+++ b/banyand/internal/sidx/sidx.go
@@ -18,6 +18,7 @@
package sidx
import (
+ "bytes"
"container/heap"
"context"
"os"
@@ -26,7 +27,9 @@ import (
"sync/atomic"
"github.com/apache/skywalking-banyandb/api/common"
+ modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
"github.com/apache/skywalking-banyandb/banyand/protector"
+ "github.com/apache/skywalking-banyandb/pkg/convert"
"github.com/apache/skywalking-banyandb/pkg/fs"
"github.com/apache/skywalking-banyandb/pkg/logger"
"github.com/apache/skywalking-banyandb/pkg/pool"
@@ -207,7 +210,9 @@ func (bc *blockCursor) init(p *part, bm *blockMetadata, req QueryRequest) {
}
// loadBlockCursor loads block data into the cursor, similar to loadBlockCursor in query_by_ts.go.
-func (s *sidx) loadBlockCursor(bc *blockCursor, tmpBlock *block, bs blockScanResult, tagsToLoad map[string]struct{}, req QueryRequest, pm protector.Memory) bool {
+func (s *sidx) loadBlockCursor(bc *blockCursor, tmpBlock *block, bs blockScanResult,
+ tagsToLoad map[string]struct{}, req QueryRequest, pm protector.Memory, metrics *batchMetrics,
+) bool {
tmpBlock.reset()
// Create a temporary queryResult to reuse existing logic
@@ -223,28 +228,145 @@ func (s *sidx) loadBlockCursor(bc *blockCursor, tmpBlock *block, bs blockScanRes
return false
}
- // Copy data to block cursor
- bc.userKeys = make([]int64, len(tmpBlock.userKeys))
- copy(bc.userKeys, tmpBlock.userKeys)
+ if metrics != nil {
+ metrics.blockElementsLoaded.Add(int64(len(tmpBlock.userKeys)))
+ }
- bc.data = make([][]byte, len(tmpBlock.data))
- for i, data := range tmpBlock.data {
- bc.data[i] = make([]byte, len(data))
- copy(bc.data[i], data)
+ totalElements := len(tmpBlock.userKeys)
+ if totalElements == 0 {
+ return false
}
- // Copy tags
+ // Pre-allocate slices for filtered data (optimize for common case where most elements match)
+ bc.userKeys = make([]int64, 0, totalElements)
+ bc.data = make([][]byte, 0, totalElements)
+
+ // Pre-allocate tag slices
bc.tags = make(map[string][]Tag)
- for tagName, tagData := range tmpBlock.tags {
- tagSlice := make([]Tag, len(tagData.values))
- for i, value := range tagData.values {
- tagSlice[i] = Tag{
- Name: tagName,
- Value: value,
- ValueType: tagData.valueType,
+ for tagName := range tmpBlock.tags {
+ bc.tags[tagName] = make([]Tag, 0, totalElements)
+ }
+
+ // Track seen data for deduplication using hash buckets with collision checks
+ seenData := make(map[uint64][][]byte)
+
+ // Single loop: filter and copy data in one pass
+ if req.TagFilter != nil {
+ tags := make([]*modelv1.Tag, 0, len(tmpBlock.tags))
+ decoder := req.TagFilter.GetDecoder()
+
+ for i := 0; i < totalElements; i++ {
+ // Check for duplicate data before processing via hash + bytes.Equal on collisions
+ dataBytes := tmpBlock.data[i]
+ h := convert.Hash(dataBytes)
+ bucket := seenData[h]
+ duplicate := false
+ for _, b := range bucket {
+ if bytes.Equal(b, dataBytes) {
+ duplicate = true
+ break
+ }
+ }
+ if duplicate {
+ if metrics != nil {
+ metrics.elementsDeduplicated.Add(1)
+ }
+ continue
+ }
+
+ // Build tags slice for this element
+ tags = tags[:0]
+ for tagName, tagData := range tmpBlock.tags {
+ if i < len(tagData.values) && tagData.values[i] != nil {
+ // Decode []byte to *modelv1.TagValue using the provided decoder
+ tagValue := decoder(tagData.valueType, tagData.values[i])
+ if tagValue != nil {
+ tags = append(tags, &modelv1.Tag{
+ Key: tagName,
+ Value: tagValue,
+ })
+ }
+ }
+ }
+
+ // Apply filter
+ matched, err := req.TagFilter.Match(tags)
+ if err != nil {
+ s.l.Error().Err(err).Msg("tag filter match error")
+ return false
+ }
+
+ if matched {
+ // Mark data as seen
+ seenData[h] = append(bucket, dataBytes)
+
+ // Copy userKey
+ bc.userKeys = append(bc.userKeys, tmpBlock.userKeys[i])
+
+ // Copy data
+ dataCopy := make([]byte, len(tmpBlock.data[i]))
+ copy(dataCopy, tmpBlock.data[i])
+ bc.data = append(bc.data, dataCopy)
+
+ // Copy tags
+ for tagName, tagData := range tmpBlock.tags {
+ if i < len(tagData.values) {
+ bc.tags[tagName] = append(bc.tags[tagName], Tag{
+ Name: tagName,
+ Value: tagData.values[i],
+ ValueType: tagData.valueType,
+ })
+ }
+ }
+ }
+ }
+ } else {
+ // No filter - copy all elements but skip duplicates
+ for i := 0; i < totalElements; i++ {
+ // Check for duplicate data via hash + bytes.Equal on collisions
+ dataBytes := tmpBlock.data[i]
+ h := convert.Hash(dataBytes)
+ bucket := seenData[h]
+ duplicate := false
+ for _, b := range bucket {
+ if bytes.Equal(b, dataBytes) {
+ duplicate = true
+ break
+ }
}
+ if duplicate {
+ continue
+ }
+
+ // Mark data as seen
+ seenData[h] = append(bucket, dataBytes)
+
+ // Copy userKey
+ bc.userKeys = append(bc.userKeys, tmpBlock.userKeys[i])
+
+ // Copy data
+ dataCopy := make([]byte, len(tmpBlock.data[i]))
+ copy(dataCopy, tmpBlock.data[i])
+ bc.data = append(bc.data, dataCopy)
+
+ // Copy tags
+ for tagName, tagData := range tmpBlock.tags {
+ if i < len(tagData.values) {
+ bc.tags[tagName] = append(bc.tags[tagName], Tag{
+ Name: tagName,
+ Value: tagData.values[i],
+ ValueType: tagData.valueType,
+ })
+ }
+ }
+ }
+ }
+
+ if metrics != nil {
+ metrics.outputElementsEmitted.Add(int64(len(bc.userKeys)))
+ if len(bc.userKeys) == 0 {
+ metrics.blocksSkipped.Add(1)
}
- bc.tags[tagName] = tagSlice
}
return len(bc.userKeys) > 0
@@ -268,6 +390,7 @@ func (bc *blockCursor) copyTo(result *QueryResponse) bool {
result.Keys = append(result.Keys, key)
result.Data = append(result.Data, bc.data[bc.idx])
result.SIDs = append(result.SIDs, bc.seriesID)
+ result.PartIDs = append(result.PartIDs, bc.p.ID())
// Copy tags for this element
var elementTags []Tag
@@ -352,7 +475,7 @@ func (bch *blockCursorHeap) pushCursors(cursors []*blockCursor) {
// merge performs heap-based merge similar to query_by_ts.go.
// It returns a QueryResponse along with a flag indicating whether the merge
// stopped because the MaxBatchSize limit has been reached.
-func (bch *blockCursorHeap) merge(ctx context.Context, batchSize int, resultsCh chan<- *QueryResponse, stats *streamingStats) error {
+func (bch *blockCursorHeap) merge(ctx context.Context, batchSize int, resultsCh chan<- *QueryResponse, metrics *batchMetrics) error {
if !bch.initialized || bch.Len() == 0 {
return nil
}
@@ -364,12 +487,16 @@ func (bch *blockCursorHeap) merge(ctx context.Context, batchSize int, resultsCh
// Initialize first batch
batch := &QueryResponse{
- Keys: make([]int64, 0),
- Data: make([][]byte, 0),
- Tags: make([][]Tag, 0),
- SIDs: make([]common.SeriesID, 0),
+ Keys: make([]int64, 0),
+ Data: make([][]byte, 0),
+ Tags: make([][]Tag, 0),
+ SIDs: make([]common.SeriesID, 0),
+ PartIDs: make([]uint64, 0),
}
+ // Track seen data for deduplication across all batches using hash buckets with collision checks
+ seenData := make(map[uint64][][]byte)
+
for bch.Len() > 0 {
topBC := bch.bcc[0]
if topBC.idx < 0 || topBC.idx >= len(topBC.userKeys) {
@@ -377,8 +504,26 @@ func (bch *blockCursorHeap) merge(ctx context.Context, batchSize int, resultsCh
continue
}
- // Copy the element (may be filtered out by key range)
- topBC.copyTo(batch)
+ // Check for duplicate data before copying via hash + bytes.Equal on collisions
+ currentData := topBC.data[topBC.idx]
+ h := convert.Hash(currentData)
+ bucket := seenData[h]
+ duplicate := false
+ for _, b := range bucket {
+ if bytes.Equal(b, currentData) {
+ duplicate = true
+ break
+ }
+ }
+
+ // Only copy if this data hasn't been seen across all batches
+ if !duplicate {
+ // Copy the element (may be filtered out by key range)
+ if topBC.copyTo(batch) {
+ // Mark this data as seen
+ seenData[h] = append(bucket, currentData)
+ }
+ }
topBC.idx += step
@@ -398,8 +543,8 @@ func (bch *blockCursorHeap) merge(ctx context.Context, batchSize int, resultsCh
// Send the batch when it reaches batchSize
if batchSize > 0 && batch.Len() >= batchSize {
- if stats != nil {
- stats.record(batch)
+ if metrics != nil {
+ metrics.record(batch)
}
select {
@@ -408,20 +553,22 @@ func (bch *blockCursorHeap) merge(ctx context.Context, batchSize int, resultsCh
case resultsCh <- batch:
}
- // Create a new batch and continue
+ // Create a new batch but keep the deduplication map for global deduplication
batch = &QueryResponse{
- Keys: make([]int64, 0),
- Data: make([][]byte, 0),
- Tags: make([][]Tag, 0),
- SIDs: make([]common.SeriesID, 0),
+ Keys: make([]int64, 0),
+ Data: make([][]byte, 0),
+ Tags: make([][]Tag, 0),
+ SIDs: make([]common.SeriesID, 0),
+ PartIDs: make([]uint64, 0),
}
+ // Do NOT reset seenData here - maintain it across batches for global deduplication
}
}
// Send remaining elements in the last batch
if batch.Len() > 0 {
- if stats != nil {
- stats.record(batch)
+ if metrics != nil {
+ metrics.record(batch)
}
select {
diff --git a/banyand/internal/sidx/sidx_test.go b/banyand/internal/sidx/sidx_test.go
index 22b345fc..aad90147 100644
--- a/banyand/internal/sidx/sidx_test.go
+++ b/banyand/internal/sidx/sidx_test.go
@@ -28,10 +28,16 @@ import (
"github.com/stretchr/testify/require"
"github.com/apache/skywalking-banyandb/api/common"
+ modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
"github.com/apache/skywalking-banyandb/banyand/observability"
"github.com/apache/skywalking-banyandb/banyand/protector"
"github.com/apache/skywalking-banyandb/pkg/fs"
pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
+ "github.com/apache/skywalking-banyandb/pkg/query/model"
+)
+
+const (
+ testTagStatus = "status"
)
// Test helper functions.
@@ -198,7 +204,7 @@ func TestSIDX_Write_WithTags(t *testing.T) {
tags := []Tag{
createTestTag("service", "user-service"),
createTestTag("endpoint", "/api/users"),
- createTestTag("status", "200"),
+ createTestTag(testTagStatus, "200"),
}
reqs := []WriteRequest{
@@ -594,13 +600,14 @@ func TestQueryResult_MaxBatchSize(t *testing.T) {
}
result := &QueryResponse{
- Keys: make([]int64, 0),
- Data: make([][]byte, 0),
- Tags: make([][]Tag, 0),
- SIDs: make([]common.SeriesID, 0),
+ Keys: make([]int64, 0),
+ Data: make([][]byte, 0),
+ Tags: make([][]Tag, 0),
+ SIDs: make([]common.SeriesID, 0),
+ PartIDs: make([]uint64, 0),
}
- qr.convertBlockToResponse(block, 1, result)
+ qr.convertBlockToResponse(block, 1, 1, result)
assert.Equal(t, tt.expectedKeys, result.Keys)
assert.Equal(t, tt.expectedData, result.Data)
@@ -608,6 +615,7 @@ func TestQueryResult_MaxBatchSize(t *testing.T) {
assert.Equal(t, len(result.Keys), len(result.Data), "Keys and Data arrays should have same length")
assert.Equal(t, len(result.Keys), len(result.SIDs), "Keys and SIDs arrays should have same length")
assert.Equal(t, len(result.Keys), len(result.Tags), "Keys and Tags arrays should have same length")
+ assert.Equal(t, len(result.Keys), len(result.PartIDs), "Keys and PartIDs arrays should have same length")
})
}
}
@@ -621,10 +629,11 @@ func TestQueryResult_ConvertBlockToResponse_RespectsLimitAcrossCalls(t *testing.
}
result := &QueryResponse{
- Keys: make([]int64, 0),
- Data: make([][]byte, 0),
- Tags: make([][]Tag, 0),
- SIDs: make([]common.SeriesID, 0),
+ Keys: make([]int64, 0),
+ Data: make([][]byte, 0),
+ Tags: make([][]Tag, 0),
+ SIDs: make([]common.SeriesID, 0),
+ PartIDs: make([]uint64, 0),
}
// First call: add trace1 data
@@ -633,7 +642,7 @@ func TestQueryResult_ConvertBlockToResponse_RespectsLimitAcrossCalls(t *testing.
data: [][]byte{[]byte("trace1"), []byte("trace1")},
tags: make(map[string]*tagData),
}
- qr.convertBlockToResponse(block1, 1, result)
+ qr.convertBlockToResponse(block1, 1, 1, result)
// Verify first call results
assert.Equal(t, 2, result.Len(), "First call should add 2 elements")
@@ -644,7 +653,7 @@ func TestQueryResult_ConvertBlockToResponse_RespectsLimitAcrossCalls(t *testing.
data: [][]byte{[]byte("trace2"), []byte("trace2")},
tags: make(map[string]*tagData),
}
- qr.convertBlockToResponse(block2, 2, result)
+ qr.convertBlockToResponse(block2, 2, 2, result)
assert.Equal(t, 2, result.Len(), "Second call should not add elements beyond limit")
@@ -654,7 +663,7 @@ func TestQueryResult_ConvertBlockToResponse_RespectsLimitAcrossCalls(t *testing.
data: [][]byte{[]byte("trace3"), []byte("trace3")},
tags: make(map[string]*tagData),
}
- qr.convertBlockToResponse(block3, 3, result)
+ qr.convertBlockToResponse(block3, 3, 3, result)
assert.Equal(t, 2, result.Len(), "Result length should remain capped at the limit")
assert.Equal(t, [][]byte{[]byte("trace1"), []byte("trace1")}, result.Data)
@@ -781,3 +790,744 @@ func TestQueryResponse_Reset(t *testing.T) {
assert.Equal(t, 0, len(result.SIDs))
assert.Equal(t, ResponseMetadata{}, result.Metadata)
}
+
+type mockTagFilterMatcher struct {
+ matchFunc func(tags []*modelv1.Tag) (bool, error)
+ decoder model.TagValueDecoder
+}
+
+func (m *mockTagFilterMatcher) Match(tags []*modelv1.Tag) (bool, error) {
+ if m.matchFunc == nil {
+ return true, nil
+ }
+ return m.matchFunc(tags)
+}
+
+func (m *mockTagFilterMatcher) GetDecoder() model.TagValueDecoder {
+ return m.decoder
+}
+
+func testTagValueDecoder(valueType pbv1.ValueType, value []byte) *modelv1.TagValue {
+ if value == nil {
+ return pbv1.NullTagValue
+ }
+ switch valueType {
+ case pbv1.ValueTypeStr:
+ return &modelv1.TagValue{
+ Value: &modelv1.TagValue_Str{
+ Str: &modelv1.Str{Value: string(value)},
+ },
+ }
+ case pbv1.ValueTypeInt64:
+ // For testing, we'll parse the string as int
+ var intVal int64
+ fmt.Sscanf(string(value), "%d", &intVal)
+ return &modelv1.TagValue{
+ Value: &modelv1.TagValue_Int{
+ Int: &modelv1.Int{Value: intVal},
+ },
+ }
+ default:
+ return pbv1.NullTagValue
+ }
+}
+
+func TestSIDX_TagFilter(t *testing.T) {
+ sidx := createTestSIDX(t)
+ defer func() {
+ assert.NoError(t, sidx.Close())
+ }()
+
+ ctx := context.Background()
+
+ reqs := []WriteRequest{
+ createTestWriteRequest(1, 100, "trace1-span1",
+ Tag{Name: testTagStatus, Value: []byte("200"), ValueType: pbv1.ValueTypeStr},
+ Tag{Name: "service", Value: []byte("user-service"), ValueType: pbv1.ValueTypeStr},
+ ),
+ createTestWriteRequest(1, 101, "trace1-span2",
+ Tag{Name: testTagStatus, Value: []byte("404"), ValueType: pbv1.ValueTypeStr},
+ Tag{Name: "service", Value: []byte("user-service"), ValueType: pbv1.ValueTypeStr},
+ ),
+ createTestWriteRequest(1, 102, "trace1-span3",
+ Tag{Name: testTagStatus, Value: []byte("200"), ValueType: pbv1.ValueTypeStr},
+ Tag{Name: "service", Value: []byte("order-service"), ValueType: pbv1.ValueTypeStr},
+ ),
+ createTestWriteRequest(1, 103, "trace1-span4",
+ Tag{Name: testTagStatus, Value: []byte("500"), ValueType: pbv1.ValueTypeStr},
+ Tag{Name: "service", Value: []byte("payment-service"), ValueType: pbv1.ValueTypeStr},
+ ),
+ }
+
+ writeTestData(t, sidx, reqs, 1, 1)
+
+ tests := []struct {
+ filterFunc func(tags []*modelv1.Tag) (bool, error)
+ name string
+ description string
+ expectedCount int
+ }{
+ {
+ name: "filter_by_status_200",
+ filterFunc: func(tags []*modelv1.Tag) (bool, error) {
+ for _, tag := range tags {
+ if tag.Key == testTagStatus && tag.Value.GetStr() != nil && tag.Value.GetStr().Value == "200" {
+ return true, nil
+ }
+ }
+ return false, nil
+ },
+ expectedCount: 2, // span1 and span3 have status=200
+ description: "Should return only spans with status=200",
+ },
+ {
+ name: "filter_by_service_user_service",
+ filterFunc: func(tags []*modelv1.Tag) (bool, error) {
+ for _, tag := range tags {
+ if tag.Key == "service" && tag.Value.GetStr() != nil && tag.Value.GetStr().Value == "user-service" {
+ return true, nil
+ }
+ }
+ return false, nil
+ },
+ expectedCount: 2, // span1 and span2 have service=user-service
+ description: "Should return only spans with service=user-service",
+ },
+ {
+ name: "filter_by_status_500",
+ filterFunc: func(tags []*modelv1.Tag) (bool, error) {
+ for _, tag := range tags {
+ if tag.Key == testTagStatus && tag.Value.GetStr() != nil && tag.Value.GetStr().Value == "500" {
+ return true, nil
+ }
+ }
+ return false, nil
+ },
+ expectedCount: 1, // only span4 has status=500
+ description: "Should return only spans with status=500",
+ },
+ {
+ name: "filter_excludes_all",
+ filterFunc: func(tags []*modelv1.Tag) (bool, error) {
+ for _, tag := range tags {
+ if tag.Key == testTagStatus && tag.Value.GetStr() != nil && tag.Value.GetStr().Value == "999" {
+ return true, nil
+ }
+ }
+ return false, nil
+ },
+ expectedCount: 0, // no spans match
+ description: "Should return no spans when filter matches nothing",
+ },
+ {
+ name: "no_filter",
+ filterFunc: func(_ []*modelv1.Tag) (bool, error) {
+ return true, nil
+ },
+ expectedCount: 4, // all spans match
+ description: "Should return all spans when filter always returns true",
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ // Create tag filter matcher
+ tagFilter := &mockTagFilterMatcher{
+ matchFunc: tt.filterFunc,
+ decoder: testTagValueDecoder,
+ }
+
+ // Create query request with tag filter
+ queryReq := QueryRequest{
+ SeriesIDs: []common.SeriesID{1},
+ TagFilter: tagFilter,
+ }
+
+ // Execute query
+ resultsCh, errCh := sidx.StreamingQuery(ctx, queryReq)
+
+ var totalElements int
+ var allKeys []int64
+ for res := range resultsCh {
+ require.NoError(t, res.Error)
+ totalElements += res.Len()
+ allKeys = append(allKeys, res.Keys...)
+ }
+ if err, ok := <-errCh; ok {
+ require.NoError(t, err)
+ }
+
+ // Verify the filtered results
+ assert.Equal(t, tt.expectedCount, totalElements, tt.description)
+
+ // Verify keys are unique
+ keySet := make(map[int64]bool)
+ for _, key := range allKeys {
+ assert.False(t, keySet[key], "Duplicate key found: %d", key)
+ keySet[key] = true
+ }
+ })
+ }
+}
+
+func TestSIDX_TagFilterWithMultipleTags(t *testing.T) {
+ sidx := createTestSIDX(t)
+ defer func() {
+ assert.NoError(t, sidx.Close())
+ }()
+
+ ctx := context.Background()
+
+ reqs := []WriteRequest{
+ createTestWriteRequest(1, 200, "trace2-span1",
+ Tag{Name: testTagStatus, Value: []byte("200"), ValueType: pbv1.ValueTypeStr},
+ Tag{Name: "method", Value: []byte("GET"), ValueType: pbv1.ValueTypeStr},
+ Tag{Name: "endpoint", Value: []byte("/api/users"), ValueType: pbv1.ValueTypeStr},
+ ),
+ createTestWriteRequest(1, 201, "trace2-span2",
+ Tag{Name: testTagStatus, Value: []byte("200"), ValueType: pbv1.ValueTypeStr},
+ Tag{Name: "method", Value: []byte("POST"), ValueType: pbv1.ValueTypeStr},
+ Tag{Name: "endpoint", Value: []byte("/api/users"), ValueType: pbv1.ValueTypeStr},
+ ),
+ createTestWriteRequest(1, 202, "trace2-span3",
+ Tag{Name: testTagStatus, Value: []byte("404"), ValueType: pbv1.ValueTypeStr},
+ Tag{Name: "method", Value: []byte("GET"), ValueType: pbv1.ValueTypeStr},
+ Tag{Name: "endpoint", Value: []byte("/api/orders"), ValueType: pbv1.ValueTypeStr},
+ ),
+ }
+
+ writeTestData(t, sidx, reqs, 2, 2)
+
+ // Test: Filter by status=200 AND method=GET
+ tagFilter := &mockTagFilterMatcher{
+ matchFunc: func(tags []*modelv1.Tag) (bool, error) {
+ hasStatus200 := false
+ hasMethodGET := false
+ for _, tag := range tags {
+ if tag.Key == testTagStatus && tag.Value.GetStr() != nil && tag.Value.GetStr().Value == "200" {
+ hasStatus200 = true
+ }
+ if tag.Key == "method" && tag.Value.GetStr() != nil && tag.Value.GetStr().Value == "GET" {
+ hasMethodGET = true
+ }
+ }
+ return hasStatus200 && hasMethodGET, nil
+ },
+ decoder: testTagValueDecoder,
+ }
+
+ queryReq := QueryRequest{
+ SeriesIDs: []common.SeriesID{1},
+ TagFilter: tagFilter,
+ }
+
+ resultsCh, errCh := sidx.StreamingQuery(ctx, queryReq)
+
+ var totalElements int
+ var keys []int64
+ for res := range resultsCh {
+ require.NoError(t, res.Error)
+ totalElements += res.Len()
+ keys = append(keys, res.Keys...)
+ }
+ if err, ok := <-errCh; ok {
+ require.NoError(t, err)
+ }
+
+ // Only span1 should match (status=200 AND method=GET)
+ assert.Equal(t, 1, totalElements, "Should return only one span matching both conditions")
+ assert.Equal(t, []int64{200}, keys, "Should return key 200")
+}
+
+func TestSIDX_TagFilterWithEmptyTags(t *testing.T) {
+ sidx := createTestSIDX(t)
+ defer func() {
+ assert.NoError(t, sidx.Close())
+ }()
+
+ ctx := context.Background()
+
+ reqs := []WriteRequest{
+ createTestWriteRequest(1, 300, "trace3-span1",
+ Tag{Name: testTagStatus, Value: []byte("200"), ValueType: pbv1.ValueTypeStr},
+ ),
+ createTestWriteRequest(1, 301, "trace3-span2"), // No tags
+ createTestWriteRequest(1, 302, "trace3-span3",
+ Tag{Name: testTagStatus, Value: []byte("404"), ValueType: pbv1.ValueTypeStr},
+ ),
+ }
+
+ writeTestData(t, sidx, reqs, 3, 3)
+
+ // Test: Filter by status=200
+ tagFilter := &mockTagFilterMatcher{
+ matchFunc: func(tags []*modelv1.Tag) (bool, error) {
+ for _, tag := range tags {
+ if tag.Key == testTagStatus && tag.Value.GetStr() != nil && tag.Value.GetStr().Value == "200" {
+ return true, nil
+ }
+ }
+ return false, nil
+ },
+ decoder: testTagValueDecoder,
+ }
+
+ queryReq := QueryRequest{
+ SeriesIDs: []common.SeriesID{1},
+ TagFilter: tagFilter,
+ }
+
+ resultsCh, errCh := sidx.StreamingQuery(ctx, queryReq)
+
+ var totalElements int
+ for res := range resultsCh {
+ require.NoError(t, res.Error)
+ totalElements += res.Len()
+ }
+ if err, ok := <-errCh; ok {
+ require.NoError(t, err)
+ }
+
+ // Only span1 should match (span2 has no tags, span3 has status=404)
+ assert.Equal(t, 1, totalElements, "Should filter out spans with no matching tags")
+}
+
+func TestBlockCursorHeap_Deduplication(t *testing.T) {
+ // Helper to create a test part
+ createTestPart := func(id uint64) *part {
+ return &part{
+ partMetadata: &partMetadata{
+ ID: id,
+ },
+ }
+ }
+
+ tests := []struct {
+ name string
+ cursors []*blockCursor
+ expectedData []string
+ expectedKeys []int64
+ batchSize int
+ expectedCount int
+ asc bool
+ }{
+ {
+ name: "duplicate data with different keys - ascending",
+ cursors: []*blockCursor{
+ {
+ userKeys: []int64{100, 101, 102},
+ data: [][]byte{[]byte("data1"), []byte("data2"), []byte("data1")},
+ seriesID: 1,
+ idx: 0,
+ p: createTestPart(1),
+ request: QueryRequest{},
+ },
+ {
+ userKeys: []int64{103, 104},
+ data: [][]byte{[]byte("data2"), []byte("data3")},
+ seriesID: 1,
+ idx: 0,
+ p: createTestPart(1),
+ request: QueryRequest{},
+ },
+ },
+ batchSize: 10,
+ asc: true,
+ expectedData: []string{"data1", "data2", "data3"},
+ expectedKeys: []int64{100, 101, 104},
+ expectedCount: 3,
+ },
+ {
+ name: "duplicate data with different keys - descending",
+ cursors: []*blockCursor{
+ {
+ userKeys: []int64{100, 101, 102},
+ data: [][]byte{[]byte("data1"), []byte("data2"), []byte("data1")},
+ seriesID: 1,
+ idx: 2, // Start from end for descending
+ p: createTestPart(1),
+ request: QueryRequest{},
+ },
+ {
+ userKeys: []int64{103, 104},
+ data: [][]byte{[]byte("data2"), []byte("data3")},
+ seriesID: 1,
+ idx: 1, // Start from end for descending
+ p: createTestPart(1),
+ request: QueryRequest{},
+ },
+ },
+ batchSize: 10,
+ asc: false,
+ expectedData: []string{"data3", "data2", "data1"},
+ expectedKeys: []int64{104, 103, 102},
+ expectedCount: 3,
+ },
+ {
+ name: "all duplicate data",
+ cursors: []*blockCursor{
+ {
+ userKeys: []int64{100, 101, 102},
+ data: [][]byte{[]byte("data1"), []byte("data1"), []byte("data1")},
+ seriesID: 1,
+ idx: 0,
+ p: createTestPart(1),
+ request: QueryRequest{},
+ },
+ },
+ batchSize: 10,
+ asc: true,
+ expectedData: []string{"data1"},
+ expectedKeys: []int64{100},
+ expectedCount: 1,
+ },
+ {
+ name: "no duplicates",
+ cursors: []*blockCursor{
+ {
+ userKeys: []int64{100, 101, 102},
+ data: [][]byte{[]byte("data1"), []byte("data2"), []byte("data3")},
+ seriesID: 1,
+ idx: 0,
+ p: createTestPart(1),
+ request: QueryRequest{},
+ },
+ },
+ batchSize: 10,
+ asc: true,
+ expectedData: []string{"data1", "data2", "data3"},
+ expectedKeys: []int64{100, 101, 102},
+ expectedCount: 3,
+ },
+ {
+ name: "duplicate data across batch boundaries",
+ cursors: []*blockCursor{
+ {
+ userKeys: []int64{100, 101, 102, 103},
+ data: [][]byte{[]byte("data1"), []byte("data2"), []byte("data1"), []byte("data3")},
+ seriesID: 1,
+ idx: 0,
+ p: createTestPart(1),
+ request: QueryRequest{},
+ },
+ },
+ batchSize: 2, // Small batch to test cross-batch behavior
+ asc: true,
+ expectedData: []string{"data1", "data2", "data3"},
+ expectedKeys: []int64{100, 101, 103},
+ expectedCount: 3,
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ ctx := context.Background()
+ resultsCh := make(chan *QueryResponse, 10)
+
+ // Create heap
+ bch := generateBlockCursorHeap(tt.asc)
+ defer releaseBlockCursorHeap(bch)
+
+ // Push cursors
+ bch.pushCursors(tt.cursors)
+
+ // Run merge in a goroutine
+ go func() {
+ err := bch.merge(ctx, tt.batchSize, resultsCh, nil)
+ require.NoError(t, err)
+ close(resultsCh)
+ }()
+
+ // Collect results
+ var allData []string
+ var allKeys []int64
+ for batch := range resultsCh {
+ for i, data := range batch.Data {
+ allData = append(allData, string(data))
+ if i < len(batch.Keys) {
+ allKeys = append(allKeys, batch.Keys[i])
+ }
+ }
+ }
+
+ // Verify results
+ assert.Equal(t, tt.expectedCount, len(allData), "Expected count mismatch")
+ assert.Equal(t, tt.expectedData, allData, "Expected data mismatch")
+ assert.Equal(t, tt.expectedKeys, allKeys, "Expected keys mismatch")
+ })
+ }
+}
+
+func TestBlockCursorHeap_DeduplicationIntegration(t *testing.T) {
+ sidx := createTestSIDX(t)
+ defer func() {
+ assert.NoError(t, sidx.Close())
+ }()
+
+ ctx := context.Background()
+
+ // Write test data with duplicates (same data, different keys)
+ reqs := []WriteRequest{
+ createTestWriteRequest(1, 100, "duplicate_data"),
+ createTestWriteRequest(1, 101, "unique_data_1"),
+ createTestWriteRequest(1, 102, "duplicate_data"), // Same data as key 100
+ createTestWriteRequest(1, 103, "unique_data_2"),
+ createTestWriteRequest(1, 104, "duplicate_data"), // Same data as key 100 and 102
+ }
+ writeTestData(t, sidx, reqs, 1, 1)
+
+ waitForIntroducerLoop()
+
+ queryReq := createTestQueryRequest(1)
+ resultsCh, errCh := sidx.StreamingQuery(ctx, queryReq)
+
+ // Collect all data
+ dataSet := make(map[string]int)
+ for res := range resultsCh {
+ require.NoError(t, res.Error)
+ for _, data := range res.Data {
+ dataSet[string(data)]++
+ }
+ }
+ if err, ok := <-errCh; ok {
+ require.NoError(t, err)
+ }
+
+ // Verify deduplication: each unique data should appear only once
+ assert.Equal(t, 1, dataSet["duplicate_data"], "duplicate_data should appear only once")
+ assert.Equal(t, 1, dataSet["unique_data_1"], "unique_data_1 should appear only once")
+ assert.Equal(t, 1, dataSet["unique_data_2"], "unique_data_2 should appear only once")
+ assert.Equal(t, 3, len(dataSet), "Should have exactly 3 unique data entries")
+
+ // Verify that we have 3 unique results (not 5)
+ totalCount := 0
+ for _, count := range dataSet {
+ totalCount += count
+ }
+ assert.Equal(t, 3, totalCount, "Should have 3 total results after deduplication")
+}
+
+func TestLoadBlockCursor_DeduplicationWithoutFilter(t *testing.T) {
+ sidx := createTestSIDX(t)
+ defer func() {
+ assert.NoError(t, sidx.Close())
+ }()
+
+ ctx := context.Background()
+
+ // Write test data with duplicates
+ reqs := []WriteRequest{
+ createTestWriteRequest(1, 100, "data1",
+ Tag{Name: "tag1", Value: []byte("value1"), ValueType: pbv1.ValueTypeStr},
+ ),
+ createTestWriteRequest(1, 101, "data2",
+ Tag{Name: "tag1", Value: []byte("value2"), ValueType: pbv1.ValueTypeStr},
+ ),
+ createTestWriteRequest(1, 102, "data1", // Duplicate data
+ Tag{Name: "tag1", Value: []byte("value3"), ValueType: pbv1.ValueTypeStr},
+ ),
+ createTestWriteRequest(1, 103, "data3",
+ Tag{Name: "tag1", Value: []byte("value4"), ValueType: pbv1.ValueTypeStr},
+ ),
+ createTestWriteRequest(1, 104, "data2", // Duplicate data
+ Tag{Name: "tag1", Value: []byte("value5"), ValueType: pbv1.ValueTypeStr},
+ ),
+ }
+
+ writeTestData(t, sidx, reqs, 5, 5)
+
+ // Query without filter
+ queryReq := createTestQueryRequest(1)
+ resultsCh, errCh := sidx.StreamingQuery(ctx, queryReq)
+
+ // Collect all data
+ dataSet := make(map[string]int)
+ for res := range resultsCh {
+ require.NoError(t, res.Error)
+ for _, data := range res.Data {
+ dataSet[string(data)]++
+ }
+ }
+ if err, ok := <-errCh; ok {
+ require.NoError(t, err)
+ }
+
+ // Verify deduplication: each unique data should appear only once
+ assert.Equal(t, 1, dataSet["data1"], "data1 should appear only once despite being written twice")
+ assert.Equal(t, 1, dataSet["data2"], "data2 should appear only once despite being written twice")
+ assert.Equal(t, 1, dataSet["data3"], "data3 should appear only once")
+ assert.Equal(t, 3, len(dataSet), "Should have exactly 3 unique data entries")
+}
+
+func TestLoadBlockCursor_DeduplicationWithFilter(t *testing.T) {
+ sidx := createTestSIDX(t)
+ defer func() {
+ assert.NoError(t, sidx.Close())
+ }()
+
+ ctx := context.Background()
+
+ // Write test data with duplicates and different tag values
+ reqs := []WriteRequest{
+ createTestWriteRequest(1, 100, "data1",
+ Tag{Name: testTagStatus, Value: []byte("200"), ValueType: pbv1.ValueTypeStr},
+ ),
+ createTestWriteRequest(1, 101, "data2",
+ Tag{Name: testTagStatus, Value: []byte("404"), ValueType: pbv1.ValueTypeStr},
+ ),
+ createTestWriteRequest(1, 102, "data1", // Duplicate data with status=200
+ Tag{Name: testTagStatus, Value: []byte("200"), ValueType: pbv1.ValueTypeStr},
+ ),
+ createTestWriteRequest(1, 103, "data3",
+ Tag{Name: testTagStatus, Value: []byte("200"), ValueType: pbv1.ValueTypeStr},
+ ),
+ createTestWriteRequest(1, 104, "data1", // Duplicate data with status=404
+ Tag{Name: testTagStatus, Value: []byte("404"), ValueType: pbv1.ValueTypeStr},
+ ),
+ createTestWriteRequest(1, 105, "data3", // Duplicate data3 with status=200
+ Tag{Name: testTagStatus, Value: []byte("200"), ValueType: pbv1.ValueTypeStr},
+ ),
+ }
+
+ writeTestData(t, sidx, reqs, 6, 6)
+
+ // Query with filter for status=200
+ tagFilter := &mockTagFilterMatcher{
+ matchFunc: func(tags []*modelv1.Tag) (bool, error) {
+ for _, tag := range tags {
+ if tag.Key == testTagStatus && tag.Value.GetStr() != nil && tag.Value.GetStr().Value == "200" {
+ return true, nil
+ }
+ }
+ return false, nil
+ },
+ decoder: testTagValueDecoder,
+ }
+
+ queryReq := QueryRequest{
+ SeriesIDs: []common.SeriesID{1},
+ TagFilter: tagFilter,
+ }
+
+ resultsCh, errCh := sidx.StreamingQuery(ctx, queryReq)
+
+ // Collect all data
+ dataSet := make(map[string]int)
+ for res := range resultsCh {
+ require.NoError(t, res.Error)
+ for _, data := range res.Data {
+ dataSet[string(data)]++
+ }
+ }
+ if err, ok := <-errCh; ok {
+ require.NoError(t, err)
+ }
+
+ // Verify deduplication with filter:
+ // - data1 (keys 100, 102) with status=200 → only first occurrence should be kept
+ // - data3 (keys 103, 105) with status=200 → only first occurrence should be kept
+ // - data2 (key 101) with status=404 → filtered out
+ // - data1 (key 104) with status=404 → filtered out
+ assert.Equal(t, 1, dataSet["data1"], "data1 should appear only once even with multiple status=200 entries")
+ assert.Equal(t, 1, dataSet["data3"], "data3 should appear only once even with multiple status=200 entries")
+ assert.Equal(t, 0, dataSet["data2"], "data2 should not appear (filtered out by status)")
+ assert.Equal(t, 2, len(dataSet), "Should have exactly 2 unique data entries matching the filter")
+}
+
+func TestLoadBlockCursor_DeduplicationEdgeCases(t *testing.T) {
+ tests := []struct {
+ expectedData map[string]int
+ name string
+ description string
+ requests []WriteRequest
+ expectedCount int
+ }{
+ {
+ name: "all_duplicates",
+ requests: []WriteRequest{
+ createTestWriteRequest(1, 100, "same_data"),
+ createTestWriteRequest(1, 101, "same_data"),
+ createTestWriteRequest(1, 102, "same_data"),
+ createTestWriteRequest(1, 103, "same_data"),
+ },
+ expectedData: map[string]int{
+ "same_data": 1,
+ },
+ expectedCount: 1,
+ description: "All duplicate data should result in single entry",
+ },
+ {
+ name: "no_duplicates",
+ requests: []WriteRequest{
+ createTestWriteRequest(1, 100, "data1"),
+ createTestWriteRequest(1, 101, "data2"),
+ createTestWriteRequest(1, 102, "data3"),
+ },
+ expectedData: map[string]int{
+ "data1": 1,
+ "data2": 1,
+ "data3": 1,
+ },
+ expectedCount: 3,
+ description: "No duplicates should result in all entries",
+ },
+ {
+ name: "alternating_duplicates",
+ requests: []WriteRequest{
+ createTestWriteRequest(1, 100, "dataA"),
+ createTestWriteRequest(1, 101, "dataB"),
+ createTestWriteRequest(1, 102, "dataA"),
+ createTestWriteRequest(1, 103, "dataB"),
+ createTestWriteRequest(1, 104, "dataA"),
+ },
+ expectedData: map[string]int{
+ "dataA": 1,
+ "dataB": 1,
+ },
+ expectedCount: 2,
+ description: "Alternating duplicates should result in unique entries only",
+ },
+ }
+
+ for i, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ // Create a fresh sidx instance for each subtest
+ sidx := createTestSIDX(t)
+ defer func() {
+ assert.NoError(t, sidx.Close())
+ }()
+
+ ctx := context.Background()
+
+ // Use different segment and part IDs for each test
+ segmentID := int64(100 + i)
+ partID := uint64(100 + i)
+
+ writeTestData(t, sidx, tt.requests, segmentID, partID)
+
+ queryReq := createTestQueryRequest(1)
+ resultsCh, errCh := sidx.StreamingQuery(ctx, queryReq)
+
+ dataSet := make(map[string]int)
+ totalCount := 0
+ for res := range resultsCh {
+ require.NoError(t, res.Error)
+ for _, data := range res.Data {
+ dataSet[string(data)]++
+ totalCount++
+ }
+ }
+ if err, ok := <-errCh; ok {
+ require.NoError(t, err)
+ }
+
+ // Verify the expected data set
+ for expectedKey, expectedValue := range tt.expectedData {
+ assert.Equal(t, expectedValue, dataSet[expectedKey],
+ "%s: %s count mismatch", tt.description, expectedKey)
+ }
+ assert.Equal(t, tt.expectedCount, len(dataSet),
+ "%s: unique data count mismatch", tt.description)
+ })
+ }
+}
diff --git a/banyand/liaison/grpc/measure.go b/banyand/liaison/grpc/measure.go
index 2f9eafba..9cb32f9d 100644
--- a/banyand/liaison/grpc/measure.go
+++ b/banyand/liaison/grpc/measure.go
@@ -304,11 +304,12 @@ func (ms *measureService) Query(ctx context.Context, req *measurev1.QueryRequest
defer func() {
if err != nil {
span.Error(err)
+ span.Stop()
} else {
span.AddSubTrace(resp.Trace)
+ span.Stop()
resp.Trace = tracer.ToProto()
}
- span.Stop()
}()
}
feat, err := ms.broadcaster.Publish(ctx, data.TopicMeasureQuery, bus.NewMessage(bus.MessageID(now.UnixNano()), req))
diff --git a/banyand/liaison/grpc/stream.go b/banyand/liaison/grpc/stream.go
index 5cc5a46a..c19b6038 100644
--- a/banyand/liaison/grpc/stream.go
+++ b/banyand/liaison/grpc/stream.go
@@ -286,11 +286,12 @@ func (s *streamService) Query(ctx context.Context, req *streamv1.QueryRequest) (
defer func() {
if err != nil {
span.Error(err)
+ span.Stop()
} else {
span.AddSubTrace(resp.Trace)
+ span.Stop()
resp.Trace = tracer.ToProto()
}
- span.Stop()
}()
}
message := bus.NewMessage(bus.MessageID(now.UnixNano()), req)
diff --git a/banyand/liaison/grpc/trace.go b/banyand/liaison/grpc/trace.go
index 73c5e717..81000319 100644
--- a/banyand/liaison/grpc/trace.go
+++ b/banyand/liaison/grpc/trace.go
@@ -349,11 +349,12 @@ func (s *traceService) Query(ctx context.Context, req *tracev1.QueryRequest) (re
defer func() {
if err != nil {
span.Error(err)
+ span.Stop()
} else if resp != nil && resp != emptyTraceQueryResponse {
span.AddSubTrace(resp.TraceQueryResult)
+ span.Stop()
resp.TraceQueryResult = tracer.ToProto()
}
- span.Stop()
}()
}
message := bus.NewMessage(bus.MessageID(now.UnixNano()), req)
diff --git a/banyand/query/processor.go b/banyand/query/processor.go
index 3d8a9b38..65808e25 100644
--- a/banyand/query/processor.go
+++ b/banyand/query/processor.go
@@ -123,14 +123,15 @@ func (p *streamQueryProcessor) Rev(ctx context.Context, message bus.Message) (re
data := resp.Data()
switch d := data.(type) {
case *streamv1.QueryResponse:
+ span.Stop()
d.Trace = tracer.ToProto()
case *common.Error:
span.Error(errors.New(d.Error()))
+ span.Stop()
resp = bus.NewMessage(bus.MessageID(now), &measurev1.QueryResponse{Trace: tracer.ToProto()})
default:
panic("unexpected data type")
}
- span.Stop()
}()
}
se := plan.(executor.StreamExecutable)
@@ -272,13 +273,14 @@ func (p *measureQueryProcessor) executeQuery(ctx context.Context, queryCriteria
switch d := data.(type) {
case *measurev1.QueryResponse:
d.Trace = tracer.ToProto()
+ span.Stop()
case *common.Error:
span.Error(errors.New(d.Error()))
+ span.Stop()
resp = bus.NewMessage(bus.MessageID(now), &measurev1.QueryResponse{Trace: tracer.ToProto()})
default:
panic("unexpected data type")
}
- span.Stop()
}()
}
@@ -549,14 +551,15 @@ func (tm *traceMonitor) finishTrace(resp *bus.Message, messageID int64) {
data := resp.Data()
switch d := data.(type) {
case *tracev1.InternalQueryResponse:
+ tm.span.Stop()
d.TraceQueryResult = tm.tracer.ToProto()
case *common.Error:
tm.span.Error(errors.New(d.Error()))
+ tm.span.Stop()
*resp = bus.NewMessage(bus.MessageID(messageID), &tracev1.QueryResponse{TraceQueryResult: tm.tracer.ToProto()})
default:
panic("unexpected data type")
}
- tm.span.Stop()
}
func (p *traceQueryProcessor) processTraceResults(resultIterator iter.Iterator[model.TraceResult],
diff --git a/banyand/trace/block_writer.go b/banyand/trace/block_writer.go
index adcb45dd..a3993dde 100644
--- a/banyand/trace/block_writer.go
+++ b/banyand/trace/block_writer.go
@@ -135,25 +135,31 @@ func (sw *writers) getWriters(tagName string) (*writer, *writer) {
}
type blockWriter struct {
- writers writers
- traceIDs []string
+ traceIDFilter *traceIDFilter
tagType tagType
+ writers writers
+ tidLast string
+ tidFirst string
metaData []byte
primaryBlockData []byte
primaryBlockMetadata primaryBlockMetadata
- totalBlocksCount uint64
- maxTimestamp int64
totalUncompressedSpanSizeBytes uint64
totalCount uint64
minTimestamp int64
totalMinTimestamp int64
totalMaxTimestamp int64
minTimestampLast int64
+ maxTimestamp int64
+ totalBlocksCount uint64
+ hasWrittenBlocks bool
}
func (bw *blockWriter) reset() {
bw.writers.reset()
- bw.traceIDs = bw.traceIDs[:0]
+ if bw.traceIDFilter != nil {
+ bw.traceIDFilter.reset()
+ bw.traceIDFilter = nil
+ }
if bw.tagType == nil {
bw.tagType = make(tagType)
} else {
@@ -170,17 +176,27 @@ func (bw *blockWriter) reset() {
bw.primaryBlockData = bw.primaryBlockData[:0]
bw.metaData = bw.metaData[:0]
bw.primaryBlockMetadata.reset()
+ bw.tidFirst = ""
+ bw.tidLast = ""
+ bw.hasWrittenBlocks = false
}
-func (bw *blockWriter) MustInitForMemPart(mp *memPart) {
+func (bw *blockWriter) MustInitForMemPart(mp *memPart, traceSize int) {
bw.reset()
bw.writers.mustCreateTagWriters = mp.mustCreateMemTagWriters
bw.writers.metaWriter.init(&mp.meta)
bw.writers.primaryWriter.init(&mp.primary)
bw.writers.spanWriter.init(&mp.spans)
+ if traceSize > 0 {
+ bw.traceIDFilter = &traceIDFilter{
+ filter: generateTraceIDBloomFilter(),
+ }
+ bw.traceIDFilter.filter.SetN(traceSize)
+ bw.traceIDFilter.filter.ResizeBits((traceSize*filter.B + 63) / 64)
+ }
}
-func (bw *blockWriter) mustInitForFilePart(fileSystem fs.FileSystem, path string, shouldCache bool) {
+func (bw *blockWriter) mustInitForFilePart(fileSystem fs.FileSystem, path string, shouldCache bool, traceSize int) {
bw.reset()
fileSystem.MkdirPanicIfExist(path, storage.DirPerm)
bw.writers.mustCreateTagWriters = func(name string) (fs.Writer, fs.Writer) {
@@ -193,6 +209,13 @@ func (bw *blockWriter) mustInitForFilePart(fileSystem fs.FileSystem, path string
bw.writers.metaWriter.init(fs.MustCreateFile(fileSystem, filepath.Join(path, metaFilename), storage.FilePerm, shouldCache))
bw.writers.primaryWriter.init(fs.MustCreateFile(fileSystem, filepath.Join(path, primaryFilename), storage.FilePerm, shouldCache))
bw.writers.spanWriter.init(fs.MustCreateFile(fileSystem, filepath.Join(path, spansFilename), storage.FilePerm, shouldCache))
+ if traceSize > 0 {
+ bw.traceIDFilter = &traceIDFilter{
+ filter: generateTraceIDBloomFilter(),
+ }
+ bw.traceIDFilter.filter.SetN(traceSize)
+ bw.traceIDFilter.filter.ResizeBits((traceSize*filter.B + 63) / 64)
+ }
}
func (bw *blockWriter) MustWriteTrace(tid string, spans [][]byte, tags [][]*tagValue, timestamps []int64, spanIDs []string) {
@@ -210,21 +233,22 @@ func (bw *blockWriter) mustWriteBlock(tid string, b *block) {
if b.Len() == 0 {
return
}
- hasWrittenBlocks := len(bw.traceIDs) > 0
- var tidLast string
- var isSeenTid bool
-
- if hasWrittenBlocks {
- tidLast = bw.traceIDs[len(bw.traceIDs)-1]
- if tid < tidLast {
- logger.Panicf("the tid=%s cannot be smaller than the previously written tid=%s", tid, tidLast)
- }
- isSeenTid = tid == tidLast
+ if tid < bw.tidLast {
+ logger.Panicf("the tid=%s cannot be smaller than the previously written tid=%s", tid, bw.tidLast)
+ }
+ hasWrittenBlocks := bw.hasWrittenBlocks
+ if !hasWrittenBlocks {
+ bw.tidFirst = tid
+ bw.hasWrittenBlocks = true
}
+ isSeenTid := tid == bw.tidLast
+ bw.tidLast = tid
bm := generateBlockMetadata()
b.mustWriteTo(tid, bm, &bw.writers)
- bw.traceIDs = append(bw.traceIDs, tid)
+ if bw.traceIDFilter != nil && bw.traceIDFilter.filter != nil {
+ bw.traceIDFilter.filter.Add(convert.StringToBytes(tid))
+ }
bw.tagType.copyFrom(bm.tagType)
tm := &bm.timestamps
if bw.totalCount == 0 || tm.min < bw.totalMinTimestamp {
@@ -258,11 +282,13 @@ func (bw *blockWriter) mustWriteBlock(tid string, b *block) {
func (bw *blockWriter) mustFlushPrimaryBlock(data []byte) {
if len(data) > 0 {
- bw.primaryBlockMetadata.mustWriteBlock(data, bw.traceIDs[0], &bw.writers)
+ bw.primaryBlockMetadata.mustWriteBlock(data, bw.tidFirst, &bw.writers)
bw.metaData = bw.primaryBlockMetadata.marshal(bw.metaData)
}
+ bw.hasWrittenBlocks = false
bw.minTimestamp = 0
bw.maxTimestamp = 0
+ bw.tidFirst = ""
}
func (bw *blockWriter) mustWriteRawBlock(r *rawBlock) {
@@ -270,18 +296,19 @@ func (bw *blockWriter) mustWriteRawBlock(r *rawBlock) {
if bm.count == 0 {
return
}
- hasWrittenBlocks := len(bw.traceIDs) > 0
- var tidLast string
- var isSeenTid bool
-
- if hasWrittenBlocks {
- tidLast = bw.traceIDs[len(bw.traceIDs)-1]
- if bm.traceID < tidLast {
- logger.Panicf("the tid=%s cannot be smaller than the previously written tid=%s", bm.traceID, tidLast)
- }
- isSeenTid = bm.traceID == tidLast
+ if bm.traceID < bw.tidLast {
+ logger.Panicf("the tid=%s cannot be smaller than the previously written tid=%s", bm.traceID, bw.tidLast)
+ }
+ hasWrittenBlocks := bw.hasWrittenBlocks
+ if !hasWrittenBlocks {
+ bw.tidFirst = bm.traceID
+ bw.hasWrittenBlocks = true
+ }
+ isSeenTid := bm.traceID == bw.tidLast
+ bw.tidLast = bm.traceID
+ if bw.traceIDFilter != nil && bw.traceIDFilter.filter != nil {
+ bw.traceIDFilter.filter.Add(convert.StringToBytes(bm.traceID))
}
- bw.traceIDs = append(bw.traceIDs, bm.traceID)
bw.tagType.copyFrom(bm.tagType)
tm := &bm.timestamps
@@ -361,15 +388,9 @@ func (bw *blockWriter) Flush(pm *partMetadata, tf *traceIDFilter, tt *tagType) {
pm.CompressedSizeBytes = bw.writers.totalBytesWritten()
- if len(bw.traceIDs) > 0 {
- if tf.filter == nil {
- tf.filter = generateTraceIDBloomFilter()
- }
- tf.filter.SetN(len(bw.traceIDs))
- tf.filter.ResizeBits((len(bw.traceIDs)*filter.B + 63) / 64)
- for _, traceID := range bw.traceIDs {
- tf.filter.Add(convert.StringToBytes(traceID))
- }
+ if bw.traceIDFilter != nil && bw.traceIDFilter.filter != nil {
+ tf.filter = bw.traceIDFilter.filter
+ bw.traceIDFilter.filter = nil
}
tt.copyFrom(bw.tagType)
@@ -380,7 +401,7 @@ func (bw *blockWriter) Flush(pm *partMetadata, tf *traceIDFilter, tt *tagType) {
func generateBlockWriter() *blockWriter {
v := blockWriterPool.Get()
if v == nil {
- return &blockWriter{
+ v = &blockWriter{
writers: writers{
tagMetadataWriters: make(map[string]*writer),
tagWriters: make(map[string]*writer),
diff --git a/banyand/trace/block_writer_test.go b/banyand/trace/block_writer_test.go
new file mode 100644
index 00000000..754b6ec7
--- /dev/null
+++ b/banyand/trace/block_writer_test.go
@@ -0,0 +1,206 @@
+// 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 (
+ "fmt"
+ "sort"
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ "github.com/apache/skywalking-banyandb/pkg/convert"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
+ pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
+ "github.com/apache/skywalking-banyandb/pkg/test"
+)
+
+func TestMultiplePrimaryBlockFlushWithPartIter(t *testing.T) {
+ // Create enough trace data to trigger multiple primary block flushes
+ // maxUncompressedPrimaryBlockSize = 128 * 1024 bytes
+ // We need to create enough blocks to exceed this size multiple times
+
+ traces := generateLargeTraceSet()
+
+ // Create a memPart and write all traces
+ mp := &memPart{}
+ mp.mustInitFromTraces(traces)
+
+ // Flush to disk
+ tmpPath, defFn := test.Space(require.New(t))
+ defer defFn()
+ epoch := uint64(1)
+ path := partPath(tmpPath, epoch)
+ fileSystem := fs.NewLocalFileSystem()
+ mp.mustFlush(fileSystem, path)
+
+ // Open the part
+ part := mustOpenFilePart(epoch, tmpPath, fileSystem)
+ defer part.close()
+
+ // Verify we have multiple primary block metadata entries
+ require.Greater(t, len(part.primaryBlockMetadata), 1,
+ "Test requires multiple primary blocks to verify the bug")
+
+ // Check that primary block metadata has correct trace IDs
+ // BUG: All primary blocks will have the same traceID because bw.traceIDs is not cleared
+ for i, pbm := range part.primaryBlockMetadata {
+ t.Logf("Primary block %d: traceID=%s", i, pbm.traceID)
+ }
+
+ // Check if all primary blocks have the same traceID (this indicates the bug)
+ allSame := true
+ firstTraceIDInPrimary := part.primaryBlockMetadata[0].traceID
+ for i := 1; i < len(part.primaryBlockMetadata); i++ {
+ if part.primaryBlockMetadata[i].traceID != firstTraceIDInPrimary {
+ allSame = false
+ break
+ }
+ }
+ if allSame {
+ t.Logf("BUG DETECTED: All %d primary blocks have the same traceID: %s",
+ len(part.primaryBlockMetadata), firstTraceIDInPrimary)
+ t.Logf("This is because mustFlushPrimaryBlock uses bw.traceIDs[0] which is never reset")
+ t.Logf("FIX: Add tidFirst field (like stream/block_writer.go) and reset it after flush")
+ }
+
+ // Now use partIter to search for traces
+ // We'll search for trace IDs that should be in different primary blocks
+ firstTraceID := traces.traceIDs[0]
+ lastTraceID := traces.traceIDs[len(traces.traceIDs)-1]
+
+ // Get unique trace IDs and sort them
+ uniqueTraceIDs := make(map[string]bool)
+ for _, tid := range traces.traceIDs {
+ uniqueTraceIDs[tid] = true
+ }
+ sortedTraceIDs := make([]string, 0, len(uniqueTraceIDs))
+ for tid := range uniqueTraceIDs {
+ sortedTraceIDs = append(sortedTraceIDs, tid)
+ }
+ sort.Strings(sortedTraceIDs)
+
+ t.Logf("First trace ID: %s", firstTraceID)
+ t.Logf("Last trace ID: %s", lastTraceID)
+ t.Logf("Total unique trace IDs: %d", len(sortedTraceIDs))
+
+ // Test 1: Search for the first trace ID
+ pi := &partIter{}
+ bma := &blockMetadataArray{}
+ pi.init(bma, part, []string{firstTraceID})
+
+ foundFirst := false
+ for pi.nextBlock() {
+ if pi.curBlock.traceID == firstTraceID {
+ foundFirst = true
+ break
+ }
+ }
+ assert.NoError(t, pi.error())
+ assert.True(t, foundFirst, "Should find first trace ID: %s", firstTraceID)
+
+ // Test 2: Search for the last trace ID
+ // This will fail because of the bug: if the last trace ID is in a later primary block,
+ // that primary block metadata will have the wrong traceID (the first one),
+ // so partIter won't be able to locate it
+ pi2 := &partIter{}
+ bma2 := &blockMetadataArray{}
+ pi2.init(bma2, part, []string{lastTraceID})
+
+ foundLast := false
+ for pi2.nextBlock() {
+ if pi2.curBlock.traceID == lastTraceID {
+ foundLast = true
+ break
+ }
+ }
+ assert.NoError(t, pi2.error())
+ assert.True(t, foundLast, "Should find last trace ID: %s", lastTraceID)
+
+ // Test 3: Search for a middle trace ID (if we have enough)
+ if len(sortedTraceIDs) > 10 {
+ middleTraceID := sortedTraceIDs[len(sortedTraceIDs)/2]
+ t.Logf("Middle trace ID: %s", middleTraceID)
+
+ pi3 := &partIter{}
+ bma3 := &blockMetadataArray{}
+ pi3.init(bma3, part, []string{middleTraceID})
+
+ foundMiddle := false
+ for pi3.nextBlock() {
+ if pi3.curBlock.traceID == middleTraceID {
+ foundMiddle = true
+ break
+ }
+ }
+ assert.NoError(t, pi3.error())
+ assert.True(t, foundMiddle, "Should find middle trace ID: %s", middleTraceID)
+ }
+
+ // Test 4: Verify primary block metadata correctness
+ // Each primary block metadata should contain the minimum traceID of the blocks it contains
+ // With the bug, all primary blocks will have the same traceID (the first one)
+ for i := 0; i < len(part.primaryBlockMetadata)-1; i++ {
+ curr := part.primaryBlockMetadata[i]
+ next := part.primaryBlockMetadata[i+1]
+ assert.LessOrEqual(t, curr.traceID, next.traceID,
+ "Primary block metadata should be in ascending order: block[%d].traceID=%s should be <= block[%d].traceID=%s",
+ i, curr.traceID, i+1, next.traceID)
+ }
+}
+
+func generateLargeTraceSet() *traces {
+ // maxUncompressedPrimaryBlockSize = 128KB
+ // Block metadata contains: traceID, timestamps, count, uncompressed size, and tag metadata
+ // Each block metadata is roughly 200-500 bytes when marshaled (depending on tags)
+ // To exceed 128KB multiple times, we need more blocks
+ // Let's create 5000 blocks with more tags to ensure larger metadata
+
+ numTraces := 5000
+ traces := &traces{
+ traceIDs: make([]string, 0, numTraces),
+ timestamps: make([]int64, 0, numTraces),
+ tags: make([][]*tagValue, 0, numTraces),
+ spans: make([][]byte, 0, numTraces),
+ spanIDs: make([]string, 0, numTraces),
+ }
+
+ for i := 0; i < numTraces; i++ {
+ // Create trace IDs that are well distributed
+ // Using a format that ensures good lexicographic distribution
+ traceID := fmt.Sprintf("trace-%08d", i)
+ spanData := []byte(fmt.Sprintf("span-data-%d-with-some-more-content-to-make-it-larger", i))
+ spanID := fmt.Sprintf("span-%d", i)
+
+ traces.traceIDs = append(traces.traceIDs, traceID)
+ traces.timestamps = append(traces.timestamps, int64(i))
+ traces.tags = append(traces.tags, []*tagValue{
+ {tag: "strTag1", valueType: pbv1.ValueTypeStr, value: []byte(fmt.Sprintf("value-%d", i))},
+ {tag: "strTag2", valueType: pbv1.ValueTypeStr, value: []byte(fmt.Sprintf("another-value-%d", i))},
+ {tag: "strTag3", valueType: pbv1.ValueTypeStr, value: []byte(fmt.Sprintf("yet-another-value-%d", i))},
+ {tag: "intTag1", valueType: pbv1.ValueTypeInt64, value: convert.Int64ToBytes(int64(i))},
+ {tag: "intTag2", valueType: pbv1.ValueTypeInt64, value: convert.Int64ToBytes(int64(i * 2))},
+ {tag: "intTag3", valueType: pbv1.ValueTypeInt64, value: convert.Int64ToBytes(int64(i * 3))},
+ })
+ traces.spans = append(traces.spans, spanData)
+ traces.spanIDs = append(traces.spanIDs, spanID)
+ }
+
+ return traces
+}
diff --git a/banyand/trace/flusher.go b/banyand/trace/flusher.go
index 194f8a0d..51d82dc5 100644
--- a/banyand/trace/flusher.go
+++ b/banyand/trace/flusher.go
@@ -255,6 +255,9 @@ func (tst *tsTable) flush(snapshot *snapshot, flushCh chan *flusherIntroduction)
select {
case flushCh <- ind:
case <-tst.loopCloser.CloseNotify():
+ for _, pw := range ind.flushed {
+ pw.decRef()
+ }
return
}
select {
diff --git a/banyand/trace/merger.go b/banyand/trace/merger.go
index 2e496876..6ddc1182 100644
--- a/banyand/trace/merger.go
+++ b/banyand/trace/merger.go
@@ -276,18 +276,20 @@ func (tst *tsTable) mergeParts(fileSystem fs.FileSystem, closeCh <-chan struct{}
}
dstPath := partPath(root, partID)
var totalSize int64
+ var traceSize uint64
pii := make([]*partMergeIter, 0, len(parts))
for i := range parts {
pmi := generatePartMergeIter()
pmi.mustInitFromPart(parts[i].p)
pii = append(pii, pmi)
totalSize += int64(parts[i].p.partMetadata.CompressedSizeBytes)
+ traceSize += parts[i].p.partMetadata.BlocksCount
}
shouldCache := tst.pm.ShouldCache(totalSize)
br := generateBlockReader()
br.init(pii)
bw := generateBlockWriter()
- bw.mustInitForFilePart(fileSystem, dstPath, shouldCache)
+ bw.mustInitForFilePart(fileSystem, dstPath, shouldCache, int(traceSize))
var minTimestamp, maxTimestamp int64
for i, pw := range parts {
diff --git a/banyand/trace/merger_bench_test.go b/banyand/trace/merger_bench_test.go
index 607c5d94..fb2aae7f 100644
--- a/banyand/trace/merger_bench_test.go
+++ b/banyand/trace/merger_bench_test.go
@@ -103,17 +103,19 @@ func Benchmark_mergeBlocks_FastRawMerge(b *testing.B) {
for i := 0; i < b.N; i++ {
// Create new iterators for each benchmark iteration
var pmi []*partMergeIter
+ var traceSize uint64
for _, p := range parts {
iter := generatePartMergeIter()
iter.mustInitFromPart(p)
pmi = append(pmi, iter)
+ traceSize += p.partMetadata.TotalCount
}
br := generateBlockReader()
br.init(pmi)
bw := generateBlockWriter()
dstPath := partPath(tmpPath, uint64(10000+i))
- bw.mustInitForFilePart(fileSystem, dstPath, false)
+ bw.mustInitForFilePart(fileSystem, dstPath, false, int(traceSize))
closeCh := make(chan struct{})
@@ -208,17 +210,19 @@ func Benchmark_mergeBlocks_OriginalMerge(b *testing.B) {
for i := 0; i < b.N; i++ {
// Create new iterators for each benchmark iteration
var pmi []*partMergeIter
+ var traceSize uint64
for _, p := range parts {
iter := generatePartMergeIter()
iter.mustInitFromPart(p)
pmi = append(pmi, iter)
+ traceSize += p.partMetadata.TotalCount
}
br := generateBlockReader()
br.init(pmi)
bw := generateBlockWriter()
dstPath := partPath(tmpPath, uint64(20000+i))
- bw.mustInitForFilePart(fileSystem, dstPath, false)
+ bw.mustInitForFilePart(fileSystem, dstPath, false, int(traceSize))
closeCh := make(chan struct{})
@@ -311,17 +315,19 @@ func Benchmark_mergeBlocks_Comparison(b *testing.B) {
for i := 0; i < b.N; i++ {
// Create new iterators for each benchmark iteration
var pmi []*partMergeIter
+ var traceSize uint64
for _, p := range parts {
iter := generatePartMergeIter()
iter.mustInitFromPart(p)
pmi = append(pmi, iter)
+ traceSize += p.partMetadata.TotalCount
}
br := generateBlockReader()
br.init(pmi)
bw := generateBlockWriter()
dstPath := partPath(tmpPathFast, uint64(30000+i))
- bw.mustInitForFilePart(fileSystem, dstPath, false)
+ bw.mustInitForFilePart(fileSystem, dstPath, false, int(traceSize))
closeCh := make(chan struct{})
@@ -355,17 +361,19 @@ func Benchmark_mergeBlocks_Comparison(b *testing.B) {
for i := 0; i < b.N; i++ {
// Create new iterators for each benchmark iteration
var pmi []*partMergeIter
+ var traceSize uint64
for _, p := range parts {
iter := generatePartMergeIter()
iter.mustInitFromPart(p)
pmi = append(pmi, iter)
+ traceSize += p.partMetadata.TotalCount
}
br := generateBlockReader()
br.init(pmi)
bw := generateBlockWriter()
dstPath := partPath(tmpPathSlow, uint64(40000+i))
- bw.mustInitForFilePart(fileSystem, dstPath, false)
+ bw.mustInitForFilePart(fileSystem, dstPath, false, int(traceSize))
closeCh := make(chan struct{})
diff --git a/banyand/trace/merger_test.go b/banyand/trace/merger_test.go
index 7196f85a..a01f2c47 100644
--- a/banyand/trace/merger_test.go
+++ b/banyand/trace/merger_test.go
@@ -27,6 +27,7 @@ import (
"github.com/stretchr/testify/require"
"github.com/apache/skywalking-banyandb/banyand/protector"
+ "github.com/apache/skywalking-banyandb/pkg/convert"
"github.com/apache/skywalking-banyandb/pkg/encoding"
"github.com/apache/skywalking-banyandb/pkg/fs"
pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
@@ -240,6 +241,7 @@ func Test_mergeBlocks_fastPath(t *testing.T) {
// Create parts
var pmi []*partMergeIter
+ var traceSize uint64
for i, traces := range tt.parts {
mp := generateMemPart()
mp.mustInitFromTraces(traces)
@@ -248,6 +250,7 @@ func Test_mergeBlocks_fastPath(t *testing.T) {
iter := generatePartMergeIter()
iter.mustInitFromPart(p)
pmi = append(pmi, iter)
+ traceSize += p.partMetadata.TotalCount
releaseMemPart(mp)
}
@@ -256,7 +259,7 @@ func Test_mergeBlocks_fastPath(t *testing.T) {
br.init(pmi)
bw := generateBlockWriter()
dstPath := partPath(tmpPath, 9999)
- bw.mustInitForFilePart(fileSystem, dstPath, false)
+ bw.mustInitForFilePart(fileSystem, dstPath, false, int(traceSize))
closeCh := make(chan struct{})
defer close(closeCh)
@@ -267,6 +270,27 @@ func Test_mergeBlocks_fastPath(t *testing.T) {
require.NotNil(t, tf)
require.NotNil(t, tagTypes)
+ // Verify bloom filter before flushing
+ require.NotNil(t, tf.filter, "Trace ID bloom filter should not be nil")
+
+ // Collect all unique trace IDs from the test data
+ uniqueTraceIDs := make(map[string]bool)
+ for _, traces := range tt.parts {
+ for _, tid := range traces.traceIDs {
+ uniqueTraceIDs[tid] = true
+ }
+ }
+
+ // Verify expected trace IDs are in the bloom filter
+ for traceID := range uniqueTraceIDs {
+ require.True(t, tf.filter.MightContain(convert.StringToBytes(traceID)),
+ "Expected trace ID %s to be in bloom filter", traceID)
+ }
+
+ // Verify bloom filter N is at least the unique trace count (it may be higher due to over-estimation)
+ require.GreaterOrEqual(t, tf.filter.N(), len(uniqueTraceIDs),
+ "Bloom filter N should be at least the number of unique trace IDs")
+
releaseBlockWriter(bw)
releaseBlockReader(br)
for _, iter := range pmi {
@@ -281,6 +305,16 @@ func Test_mergeBlocks_fastPath(t *testing.T) {
// Verify merged results by reading back
mergedPart := mustOpenFilePart(9999, tmpPath, fileSystem)
+
+ // Verify bloom filter persists after reload
+ require.NotNil(t, mergedPart.traceIDFilter.filter, "Trace ID bloom filter should persist after reload")
+ for traceID := range uniqueTraceIDs {
+ require.True(t, mergedPart.traceIDFilter.filter.MightContain(convert.StringToBytes(traceID)),
+ "Expected trace ID %s to be in bloom filter after reload", traceID)
+ }
+ require.GreaterOrEqual(t, mergedPart.traceIDFilter.filter.N(), len(uniqueTraceIDs),
+ "Bloom filter N should be at least the unique trace count after reload")
+
mergedIter := generatePartMergeIter()
mergedIter.mustInitFromPart(mergedPart)
@@ -337,6 +371,16 @@ func verifyPartContainsTraces(t *testing.T, partID uint64, tmpPath string, fileS
fs.MustClose(p.spans)
}()
+ // Verify bloom filter contains all expected trace IDs
+ require.NotNil(t, p.traceIDFilter.filter, "Part %d should have a bloom filter", partID)
+ for traceID := range expectedTraces {
+ require.True(t, p.traceIDFilter.filter.MightContain(convert.StringToBytes(traceID)),
+ "Part %d: Expected trace ID %s to be in bloom filter", partID, traceID)
+ }
+ // Verify bloom filter N is at least the unique trace count (may be higher due to over-estimation)
+ require.GreaterOrEqual(t, p.traceIDFilter.filter.N(), len(expectedTraces),
+ "Part %d: Bloom filter N should be at least the number of unique trace IDs", partID)
+
// Create iterator to read blocks
pmi := generatePartMergeIter()
pmi.mustInitFromPart(p)
@@ -1197,6 +1241,30 @@ func Test_mergeParts(t *testing.T) {
return
}
defer p.decRef()
+
+ // Verify bloom filter
+ if len(tt.tsList) > 0 {
+ require.NotNil(t, p.p.traceIDFilter.filter, "Merged part should have a bloom filter")
+
+ // Collect all unique trace IDs from test data
+ uniqueTraceIDs := make(map[string]bool)
+ for _, traces := range tt.tsList {
+ for _, tid := range traces.traceIDs {
+ uniqueTraceIDs[tid] = true
+ }
+ }
+
+ // Verify all expected trace IDs are in the bloom filter
+ for traceID := range uniqueTraceIDs {
+ require.True(t, p.p.traceIDFilter.filter.MightContain(convert.StringToBytes(traceID)),
+ "Expected trace ID %s to be in bloom filter", traceID)
+ }
+
+ // Verify bloom filter N is at least the unique trace count (may be higher due to over-estimation)
+ require.GreaterOrEqual(t, p.p.traceIDFilter.filter.N(), len(uniqueTraceIDs),
+ "Bloom filter N should be at least the number of unique trace IDs")
+ }
+
pmi := &partMergeIter{}
pmi.mustInitFromPart(p.p)
reader := &blockReader{}
diff --git a/banyand/trace/part.go b/banyand/trace/part.go
index 1dbda89c..2f744b61 100644
--- a/banyand/trace/part.go
+++ b/banyand/trace/part.go
@@ -235,8 +235,18 @@ func (mp *memPart) mustInitFromTraces(ts *traces) {
sort.Sort(ts)
+ // Count unique trace IDs
+ traceSize := 0
+ var tidPrevCount string
+ for _, tid := range ts.traceIDs {
+ if tid != tidPrevCount {
+ traceSize++
+ tidPrevCount = tid
+ }
+ }
+
bsw := generateBlockWriter()
- bsw.MustInitForMemPart(mp)
+ bsw.MustInitForMemPart(mp, traceSize)
var tidPrev string
uncompressedSpansSizeBytes := uint64(0)
diff --git a/banyand/trace/part_test.go b/banyand/trace/part_test.go
index 57cf63f4..eb2fa840 100644
--- a/banyand/trace/part_test.go
+++ b/banyand/trace/part_test.go
@@ -31,9 +31,12 @@ import (
func TestMustInitFromTraces(t *testing.T) {
tests := []struct {
- ts *traces
- name string
- want partMetadata
+ ts *traces
+ name string
+ expectedTraceIDs []string
+ notExpectedTraceIDs []string
+ want partMetadata
+ expectEmptyBloomFilter bool
}{
{
name: "Test with empty traces",
@@ -44,7 +47,10 @@ func TestMustInitFromTraces(t *testing.T) {
spans: [][]byte{},
spanIDs: []string{},
},
- want: partMetadata{},
+ want: partMetadata{},
+ expectedTraceIDs: []string{},
+ notExpectedTraceIDs: []string{"trace1", "trace2"},
+ expectEmptyBloomFilter: true,
},
{
name: "Test with one item in traces",
@@ -66,6 +72,9 @@ func TestMustInitFromTraces(t *testing.T) {
MaxTimestamp: 1,
TotalCount: 1,
},
+ expectedTraceIDs: []string{"trace1"},
+ notExpectedTraceIDs: []string{"trace2", "trace3", "trace0"},
+ expectEmptyBloomFilter: false,
},
{
name: "Test with multiple items in traces",
@@ -76,6 +85,9 @@ func TestMustInitFromTraces(t *testing.T) {
MaxTimestamp: 220,
TotalCount: 6,
},
+ expectedTraceIDs: []string{"trace1", "trace2", "trace3"},
+ notExpectedTraceIDs: []string{"trace0", "trace4", "trace5", "nonexistent"},
+ expectEmptyBloomFilter: false,
},
}
for _, tt := range tests {
@@ -87,6 +99,27 @@ func TestMustInitFromTraces(t *testing.T) {
assert.Equal(t, tt.want.MaxTimestamp, mp.partMetadata.MaxTimestamp)
assert.Equal(t, tt.want.TotalCount, mp.partMetadata.TotalCount)
assert.Equal(t, len(mp.tags), len(mp.tagMetadata))
+
+ // Verify bloom filter in memPart
+ if tt.expectEmptyBloomFilter {
+ assert.Nil(t, mp.traceIDFilter.filter, "Expected nil bloom filter for empty traces")
+ } else {
+ require.NotNil(t, mp.traceIDFilter.filter, "Expected non-nil bloom filter")
+ // Verify expected trace IDs are in the filter
+ for _, traceID := range tt.expectedTraceIDs {
+ assert.True(t, mp.traceIDFilter.filter.MightContain(convert.StringToBytes(traceID)),
+ "Expected trace ID %s to be in bloom filter", traceID)
+ }
+ // Verify not expected trace IDs are not in the filter (or might be due to false positives)
+ for _, traceID := range tt.notExpectedTraceIDs {
+ // Note: We can't assert false here due to potential false positives in bloom filters
+ // But we log it for debugging purposes
+ if mp.traceIDFilter.filter.MightContain(convert.StringToBytes(traceID)) {
+ t.Logf("Bloom filter returned potential false positive for trace ID: %s", traceID)
+ }
+ }
+ }
+
tmpPath, defFn := test.Space(require.New(t))
defer defFn()
epoch := uint64(1)
@@ -100,6 +133,22 @@ func TestMustInitFromTraces(t *testing.T) {
assert.Equal(t, tt.want.MinTimestamp, p.partMetadata.MinTimestamp)
assert.Equal(t, tt.want.MaxTimestamp, p.partMetadata.MaxTimestamp)
assert.Equal(t, tt.want.TotalCount, p.partMetadata.TotalCount)
+
+ // Verify bloom filter in filePart after flush/load
+ if tt.expectEmptyBloomFilter {
+ assert.Nil(t, p.traceIDFilter.filter, "Expected nil bloom filter for empty traces after flush")
+ } else {
+ require.NotNil(t, p.traceIDFilter.filter, "Expected non-nil bloom filter after flush")
+ // Verify expected trace IDs are in the filter
+ for _, traceID := range tt.expectedTraceIDs {
+ assert.True(t, p.traceIDFilter.filter.MightContain(convert.StringToBytes(traceID)),
+ "Expected trace ID %s to be in bloom filter after flush", traceID)
+ }
+ // Verify bloom filter properties
+ assert.Equal(t, len(tt.expectedTraceIDs), p.traceIDFilter.filter.N(),
+ "Bloom filter N should match number of unique trace IDs")
+ }
+
if len(mp.tags) > 0 {
for k := range mp.tags {
_, ok := mp.tagMetadata[k]
diff --git a/banyand/trace/query.go b/banyand/trace/query.go
index cce521b5..2b0d9ded 100644
--- a/banyand/trace/query.go
+++ b/banyand/trace/query.go
@@ -47,8 +47,6 @@ type queryOptions struct {
seriesToEntity map[common.SeriesID][]*modelv1.TagValue
traceIDs []string
model.TraceQueryOptions
- minTimestamp int64
- maxTimestamp int64
}
func (t *trace) Query(ctx context.Context, tqo model.TraceQueryOptions) (model.TraceQueryResult, error) {
@@ -86,8 +84,6 @@ func (t *trace) Query(ctx context.Context, tqo model.TraceQueryOptions) (model.T
qo := queryOptions{
TraceQueryOptions: tqo,
traceIDs: tqo.TraceIDs,
- minTimestamp: tqo.TimeRange.Start.UnixNano(),
- maxTimestamp: tqo.TimeRange.End.UnixNano(),
}
if err = t.resolveSeriesEntities(ctx, segments, &qo, tqo.Name, tqo.Entities); err != nil {
@@ -101,8 +97,6 @@ func (t *trace) Query(ctx context.Context, tqo model.TraceQueryOptions) (model.T
return nilResult, nil
}
- parts := t.attachSnapshots(&result, tables, qo.minTimestamp, qo.maxTimestamp, qo.traceIDs)
-
pipelineCtx, cancel := context.WithTimeout(ctx, queryTimeout)
result.ctx = pipelineCtx
result.cancel = cancel
@@ -116,12 +110,14 @@ func (t *trace) Query(ctx context.Context, tqo model.TraceQueryOptions) (model.T
case len(qo.traceIDs) > 0:
traceBatchCh = staticTraceBatchSource(pipelineCtx, qo.traceIDs, tqo.MaxTraceSize, result.keys)
case useSIDXStreaming:
- traceBatchCh = t.streamSIDXTraceBatches(pipelineCtx, sidxInstances, sidxQueryRequest, tqo.MaxTraceSize)
+ var streamDone <-chan struct{}
+ traceBatchCh, streamDone = t.streamSIDXTraceBatches(pipelineCtx, sidxInstances, sidxQueryRequest, tqo.MaxTraceSize)
+ result.streamDone = streamDone
default:
return nilResult, errors.New("invalid query options: either traceIDs or order must be specified")
}
- result.cursorBatchCh = t.startBlockScanStage(pipelineCtx, parts, qo, traceBatchCh)
+ result.cursorBatchCh = t.startBlockScanStage(pipelineCtx, tables, qo, traceBatchCh)
return &result, nil
}
@@ -136,6 +132,10 @@ func validateTraceQueryOptions(tqo model.TraceQueryOptions) error {
return nil
}
+func (t *trace) GetTagValueDecoder() model.TagValueDecoder {
+ return mustDecodeTagValue
+}
+
func (t *trace) ensureTSDB() (storage.TSDB[*tsTable, option], error) {
if db := t.tsdb.Load(); db != nil {
return db.(storage.TSDB[*tsTable, option]), nil
@@ -226,6 +226,7 @@ func (t *trace) prepareSIDXStreaming(
req := sidx.QueryRequest{
Filter: tqo.SkippingFilter,
+ TagFilter: tqo.TagFilter,
Order: tqo.Order,
MaxBatchSize: tqo.MaxTraceSize,
MinKey: &tqo.MinVal,
@@ -239,42 +240,17 @@ func (t *trace) prepareSIDXStreaming(
return sidxInstances, req, true
}
-func (t *trace) attachSnapshots(
- qr *queryResult,
- tables []*tsTable,
- minTimestamp int64,
- maxTimestamp int64,
- traceIDs []string,
-) []*part {
- parts := make([]*part, 0)
- for i := range tables {
- s := tables[i].currentSnapshot()
- if s == nil {
- continue
- }
-
- var count int
- parts, count = s.getParts(parts, minTimestamp, maxTimestamp, traceIDs)
- if count < 1 {
- s.decRef()
- continue
- }
-
- qr.snapshots = append(qr.snapshots, s)
- }
- return parts
-}
-
type queryResult struct {
ctx context.Context
err error
- cancel context.CancelFunc
+ currentBatch *scanBatch
tagProjection *model.TagProjection
keys map[string]int64
cursorBatchCh <-chan *scanBatch
- currentBatch *scanBatch
+ cancel context.CancelFunc
currentCursorGroups map[string][]*blockCursor
- snapshots []*snapshot
+ streamDone <-chan struct{}
+ currentTraceIDs []string
segments []storage.Segment[*tsTable, option]
currentIndex int
hit int
@@ -299,12 +275,12 @@ func (qr *queryResult) Pull() *model.TraceResult {
return nil
}
- if qr.currentBatch == nil || qr.currentIndex >= len(qr.currentBatch.traceIDs) {
+ if qr.currentBatch == nil || qr.currentIndex >= len(qr.currentTraceIDs) {
qr.releaseCurrentBatch()
continue
}
- traceID := qr.currentBatch.traceIDs[qr.currentIndex]
+ traceID := qr.currentTraceIDs[qr.currentIndex]
cursors := qr.currentCursorGroups[traceID]
if len(cursors) == 0 {
@@ -341,7 +317,7 @@ func (qr *queryResult) Pull() *model.TraceResult {
}
func (qr *queryResult) ensureCurrentBatch() bool {
- if qr.currentBatch != nil && qr.currentIndex < len(qr.currentBatch.traceIDs) {
+ if qr.currentBatch != nil && qr.currentIndex < len(qr.currentTraceIDs) {
return true
}
@@ -364,7 +340,11 @@ func (qr *queryResult) ensureCurrentBatch() bool {
qr.currentBatch = batch
qr.currentIndex = 0
- qr.currentCursorGroups = make(map[string][]*blockCursor, len(batch.traceIDs))
+
+ // Use the ordered list of trace IDs to maintain the sorted order from SIDX stream
+ qr.currentTraceIDs = batch.traceIDsOrder
+
+ qr.currentCursorGroups = make(map[string][]*blockCursor, len(qr.currentTraceIDs))
// Stream cursors from channel and group by traceID
if batch.cursorCh != nil {
@@ -378,6 +358,11 @@ func (qr *queryResult) ensureCurrentBatch() bool {
}
}
qr.currentCursorGroups = nil
+ // Release snapshots from this batch on error
+ for _, s := range batch.snapshots {
+ s.decRef()
+ }
+ qr.currentBatch = nil
return true
}
if result.cursor != nil {
@@ -464,10 +449,15 @@ func (qr *queryResult) releaseCurrentBatch() {
qr.currentCursorGroups = nil
}
if qr.currentBatch != nil {
+ // Release snapshots from this batch
+ for _, s := range qr.currentBatch.snapshots {
+ s.decRef()
+ }
// Note: cursorCh should be fully consumed by ensureCurrentBatch
// so no need to drain it here
qr.currentBatch = nil
}
+ qr.currentTraceIDs = nil
qr.currentIndex = 0
}
@@ -476,26 +466,33 @@ func (qr *queryResult) Release() {
qr.cancel()
}
+ // Wait for streamSIDXTraceBatches goroutine to exit if it was used
+ if qr.streamDone != nil {
+ <-qr.streamDone
+ }
+
// Drain all batches and their cursor channels to ensure scanTraceIDsInline completes
for batch := range qr.cursorBatchCh {
- if batch != nil && batch.cursorCh != nil {
- // Drain the cursor channel to ensure scanTraceIDsInline goroutine finishes
- for result := range batch.cursorCh {
- if result.cursor != nil {
- releaseBlockCursor(result.cursor)
+ if batch != nil {
+ if batch.cursorCh != nil {
+ // Drain the cursor channel to ensure scanTraceIDsInline goroutine finishes
+ for result := range batch.cursorCh {
+ if result.cursor != nil {
+ releaseBlockCursor(result.cursor)
+ }
}
}
+ // Release snapshots from drained batches
+ for _, s := range batch.snapshots {
+ s.decRef()
+ }
}
}
qr.cursorBatchCh = nil
qr.releaseCurrentBatch()
- // Now safe to release snapshots - all workers have finished
- for i := range qr.snapshots {
- qr.snapshots[i].decRef()
- }
- qr.snapshots = qr.snapshots[:0]
+ // Release segments
for i := range qr.segments {
qr.segments[i].DecRef()
}
diff --git a/banyand/trace/query_test.go b/banyand/trace/query_test.go
index c1cb8b56..bff0798d 100644
--- a/banyand/trace/query_test.go
+++ b/banyand/trace/query_test.go
@@ -91,18 +91,19 @@ func TestQueryResult(t *testing.T) {
t.Run(tt.name, func(t *testing.T) {
verify := func(t *testing.T, tst *tsTable) {
defer tst.Close()
- queryOpts := queryOptions{
- minTimestamp: tt.minTimestamp,
- maxTimestamp: tt.maxTimestamp,
- }
+ queryOpts := queryOptions{}
s := tst.currentSnapshot()
require.NotNil(t, s)
defer s.decRef()
- pp, _ := s.getParts(nil, queryOpts.minTimestamp, queryOpts.maxTimestamp, []string{tt.traceID})
+ pp, _ := s.getParts(nil, tt.minTimestamp, tt.maxTimestamp, []string{tt.traceID})
bma := generateBlockMetadataArray()
defer releaseBlockMetadataArray(bma)
ti := &tstIter{}
- ti.init(bma, pp, []string{tt.traceID})
+ groupedTids := make([][]string, len(pp))
+ for i := range groupedTids {
+ groupedTids[i] = []string{tt.traceID}
+ }
+ ti.init(bma, pp, groupedTids)
var (
result queryResult
@@ -129,10 +130,14 @@ func TestQueryResult(t *testing.T) {
close(cursorCh)
cursorBatch := make(chan *scanBatch, 1)
+ // Create trace batch with map structure
+ traceIDMap := make(map[uint64][]string)
+ traceIDMap[0] = []string{tt.traceID}
cursorBatch <- &scanBatch{
traceBatch: traceBatch{
- traceIDs: []string{tt.traceID},
- keys: map[string]int64{tt.traceID: 0},
+ traceIDs: traceIDMap,
+ traceIDsOrder: []string{tt.traceID},
+ keys: map[string]int64{tt.traceID: 0},
},
cursorCh: cursorCh,
}
@@ -333,10 +338,7 @@ func TestQueryResultMultipleBatches(t *testing.T) {
t.Run(tt.name, func(t *testing.T) {
verify := func(t *testing.T, tst *tsTable) {
defer tst.Close()
- queryOpts := queryOptions{
- minTimestamp: tt.minTimestamp,
- maxTimestamp: tt.maxTimestamp,
- }
+ queryOpts := queryOptions{}
s := tst.currentSnapshot()
require.NotNil(t, s)
defer s.decRef()
@@ -347,7 +349,7 @@ func TestQueryResultMultipleBatches(t *testing.T) {
allTraceIDs = append(allTraceIDs, batchTraceIDs...)
}
- pp, _ := s.getParts(nil, queryOpts.minTimestamp, queryOpts.maxTimestamp, allTraceIDs)
+ pp, _ := s.getParts(nil, tt.minTimestamp, tt.maxTimestamp, allTraceIDs)
// Create multiple batches, each simulating a batch from sidx
cursorBatch := make(chan *scanBatch, len(tt.traceIDs))
@@ -355,7 +357,11 @@ func TestQueryResultMultipleBatches(t *testing.T) {
for batchIdx, batchTraceIDs := range tt.traceIDs {
bma := generateBlockMetadataArray()
ti := &tstIter{}
- ti.init(bma, pp, batchTraceIDs)
+ groupedTids := make([][]string, len(pp))
+ for i := range groupedTids {
+ groupedTids[i] = batchTraceIDs
+ }
+ ti.init(bma, pp, groupedTids)
var cursors []*blockCursor
for ti.nextBlock() {
@@ -381,12 +387,17 @@ func TestQueryResultMultipleBatches(t *testing.T) {
keys[traceID] = int64(batchIdx)
}
+ // Create trace ID map for this batch
+ traceIDMap := make(map[uint64][]string)
+ traceIDMap[uint64(batchIdx)] = batchTraceIDs
+
// Add batch to channel
cursorBatch <- &scanBatch{
traceBatch: traceBatch{
- traceIDs: batchTraceIDs,
- keys: keys,
- seq: batchIdx,
+ traceIDs: traceIDMap,
+ traceIDsOrder: batchTraceIDs,
+ keys: keys,
+ seq: batchIdx,
},
cursorCh: cursorCh,
}
diff --git a/banyand/trace/snapshot_test.go b/banyand/trace/snapshot_test.go
index 65082ac2..f028f66e 100644
--- a/banyand/trace/snapshot_test.go
+++ b/banyand/trace/snapshot_test.go
@@ -40,25 +40,26 @@ import (
func TestSnapshotGetParts(t *testing.T) {
tests := []struct {
- snapshot *snapshot
- name string
- dst []*part
- expected []*part
- opts queryOptions
- count int
+ snapshot *snapshot
+ name string
+ dst []*part
+ expected []*part
+ opts queryOptions
+ minTimestamp int64
+ maxTimestamp int64
+ count int
}{
{
name: "Test with empty snapshot",
snapshot: &snapshot{
parts: []*partWrapper{},
},
- dst: []*part{},
- opts: queryOptions{
- minTimestamp: 0,
- maxTimestamp: 10,
- },
- expected: []*part{},
- count: 0,
+ dst: []*part{},
+ opts: queryOptions{},
+ minTimestamp: 0,
+ maxTimestamp: 10,
+ expected: []*part{},
+ count: 0,
},
{
name: "Test with non-empty snapshot and no matching parts",
@@ -78,13 +79,12 @@ func TestSnapshotGetParts(t *testing.T) {
},
},
},
- dst: []*part{},
- opts: queryOptions{
- minTimestamp: 11,
- maxTimestamp: 15,
- },
- expected: []*part{},
- count: 0,
+ dst: []*part{},
+ opts: queryOptions{},
+ minTimestamp: 11,
+ maxTimestamp: 15,
+ expected: []*part{},
+ count: 0,
},
{
name: "Test with non-empty snapshot and some matching parts",
@@ -104,11 +104,10 @@ func TestSnapshotGetParts(t *testing.T) {
},
},
},
- dst: []*part{},
- opts: queryOptions{
- minTimestamp: 5,
- maxTimestamp: 10,
- },
+ dst: []*part{},
+ opts: queryOptions{},
+ minTimestamp: 5,
+ maxTimestamp: 10,
expected: []*part{
{partMetadata: partMetadata{
MinTimestamp: 0,
@@ -164,10 +163,10 @@ func TestSnapshotGetParts(t *testing.T) {
}(),
dst: []*part{},
opts: queryOptions{
- minTimestamp: 0,
- maxTimestamp: 10,
- traceIDs: []string{"trace1"},
+ traceIDs: []string{"trace1"},
},
+ minTimestamp: 0,
+ maxTimestamp: 10,
expected: []*part{
{
partMetadata: partMetadata{
@@ -205,18 +204,18 @@ func TestSnapshotGetParts(t *testing.T) {
}(),
dst: []*part{},
opts: queryOptions{
- minTimestamp: 0,
- maxTimestamp: 10,
- traceIDs: []string{"trace0"},
+ traceIDs: []string{"trace0"},
},
- expected: []*part{},
- count: 0,
+ minTimestamp: 0,
+ maxTimestamp: 10,
+ expected: []*part{},
+ count: 0,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
- result, count := tt.snapshot.getParts(tt.dst, tt.opts.minTimestamp, tt.opts.maxTimestamp, tt.opts.traceIDs)
+ result, count := tt.snapshot.getParts(tt.dst, tt.minTimestamp, tt.maxTimestamp, tt.opts.traceIDs)
assert.Equal(t, tt.count, count)
require.Equal(t, len(tt.expected), len(result))
for i := range tt.expected {
diff --git a/banyand/trace/streaming_pipeline.go b/banyand/trace/streaming_pipeline.go
index 0fabfa6d..eb5d7973 100644
--- a/banyand/trace/streaming_pipeline.go
+++ b/banyand/trace/streaming_pipeline.go
@@ -22,6 +22,7 @@ import (
"context"
stdErrors "errors"
"fmt"
+ "slices"
"sort"
"sync"
"sync/atomic"
@@ -30,26 +31,29 @@ import (
modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
"github.com/apache/skywalking-banyandb/banyand/internal/sidx"
+ "github.com/apache/skywalking-banyandb/pkg/convert"
"github.com/apache/skywalking-banyandb/pkg/query"
)
type traceBatch struct {
- err error
- keys map[string]int64
- traceIDs []string
- seq int
+ err error
+ keys map[string]int64
+ traceIDs map[uint64][]string
+ traceIDsOrder []string // ordered list of trace IDs as they were added
+ seq int
}
type scanBatch struct {
- err error
- cursorCh <-chan scanCursorResult
+ err error
+ cursorCh <-chan scanCursorResult
+ snapshots []*snapshot
traceBatch
}
func newTraceBatch(seq int, capacity int) traceBatch {
tb := traceBatch{
seq: seq,
- traceIDs: make([]string, 0, capacity),
+ traceIDs: make(map[uint64][]string),
}
if capacity > 0 {
tb.keys = make(map[string]int64, capacity)
@@ -87,13 +91,19 @@ func staticTraceBatchSource(ctx context.Context, traceIDs []string, maxTraceSize
end = len(orderedIDs)
}
+ // Static sources don't have partID information, use sentinel value 0
+ batchIDs := append([]string(nil), orderedIDs[start:end]...)
+ traceIDMap := make(map[uint64][]string)
+ traceIDMap[0] = batchIDs
+
select {
case <-ctx.Done():
return
case out <- traceBatch{
- seq: seq,
- traceIDs: append([]string(nil), orderedIDs[start:end]...),
- keys: keys,
+ seq: seq,
+ traceIDs: traceIDMap,
+ traceIDsOrder: batchIDs,
+ keys: keys,
}:
seq++
}
@@ -235,12 +245,14 @@ func (t *trace) streamSIDXTraceBatches(
sidxInstances []sidx.SIDX,
req sidx.QueryRequest,
maxTraceSize int,
-) <-chan traceBatch {
+) (<-chan traceBatch, <-chan struct{}) {
out := make(chan traceBatch)
+ done := make(chan struct{})
if len(sidxInstances) == 0 {
close(out)
- return out
+ close(done)
+ return out, done
}
tracer := query.GetTracer(ctx)
@@ -267,6 +279,7 @@ func (t *trace) streamSIDXTraceBatches(
}
cancel()
close(out)
+ close(done)
}()
if err := runner.prepare(sidxInstances); err != nil {
@@ -277,7 +290,7 @@ func (t *trace) streamSIDXTraceBatches(
runner.run(out)
}()
- return out
+ return out, done
}
type sidxStreamRunner struct {
@@ -463,9 +476,16 @@ func (r *sidxStreamRunner) run(out chan<- traceBatch) {
return
}
- if added && len(r.batch.traceIDs) >= r.batchSize {
- if !r.emitBatch(out) {
- return
+ if added {
+ // Count total trace IDs across all partIDs
+ totalTraceIDs := 0
+ for _, ids := range r.batch.traceIDs {
+ totalTraceIDs += len(ids)
+ }
+ if totalTraceIDs >= r.batchSize {
+ if !r.emitBatch(out) {
+ return
+ }
}
}
@@ -475,7 +495,15 @@ func (r *sidxStreamRunner) run(out chan<- traceBatch) {
}
}
- if len(r.batch.traceIDs) > 0 {
+ // Emit remaining batch if it has any trace IDs
+ hasTraceIDs := false
+ for _, ids := range r.batch.traceIDs {
+ if len(ids) > 0 {
+ hasTraceIDs = true
+ break
+ }
+ }
+ if hasTraceIDs {
if !r.emitBatch(out) {
return
}
@@ -529,8 +557,12 @@ func (r *sidxStreamRunner) consumeShard(shard *sidxStreamShard) (bool, error) {
return false, nil
}
+ // Extract partID from SIDX response
+ partID := shard.response.PartIDs[shard.idx]
+
r.seenTraceIDs[traceID] = struct{}{}
- r.batch.traceIDs = append(r.batch.traceIDs, traceID)
+ r.batch.traceIDs[partID] = append(r.batch.traceIDs[partID], traceID)
+ r.batch.traceIDsOrder = append(r.batch.traceIDsOrder, traceID)
if r.batch.keys == nil {
r.batch.keys = make(map[string]int64)
}
@@ -652,7 +684,7 @@ func (r *sidxStreamRunner) cancel() {
func (t *trace) startBlockScanStage(
ctx context.Context,
- parts []*part,
+ tables []*tsTable,
qo queryOptions,
batches <-chan traceBatch,
) <-chan *scanBatch {
@@ -671,23 +703,96 @@ func (t *trace) startBlockScanStage(
continue
}
+ // Acquire snapshots from all tables
+ snapshots := make([]*snapshot, 0, len(tables))
+ for _, table := range tables {
+ s := table.currentSnapshot()
+ if s == nil {
+ continue
+ }
+ snapshots = append(snapshots, s)
+ }
+
+ // If no snapshots available, skip this batch
+ if len(snapshots) == 0 {
+ select {
+ case <-ctx.Done():
+ return
+ case out <- &scanBatch{traceBatch: batch}:
+ }
+ continue
+ }
+
+ // Start part selection span
+ partSelectionCtx, finishPartSelection := startPartSelectionSpan(ctx, &batch, snapshots)
+
+ parts := make([]*part, 0)
+ groupedIDs := make([][]string, 0)
+
+ allTraceIDs := make([]string, 0)
+ for _, ids := range batch.traceIDs {
+ allTraceIDs = append(allTraceIDs, ids...)
+ }
+ sort.Strings(allTraceIDs)
+
+ bloomFilteredPartIDs := make([]uint64, 0)
+ totalGroupedIDs := 0
+
+ for _, s := range snapshots {
+ for _, pw := range s.parts {
+ p := pw.p
+ partID := p.partMetadata.ID
+
+ var idsFromSIDX []string
+ if traceIDsFromSIDX, exists := batch.traceIDs[partID]; exists {
+ idsFromSIDX = append([]string(nil), traceIDsFromSIDX...)
+ }
+ var idsForPart []string
+ for _, traceID := range allTraceIDs {
+ if slices.Contains(idsFromSIDX, traceID) || p.traceIDFilter.filter.MightContain(convert.StringToBytes(traceID)) {
+ idsForPart = append(idsForPart, traceID)
+ }
+ }
+
+ if len(idsForPart) > 0 {
+ parts = append(parts, p)
+ groupedIDs = append(groupedIDs, idsForPart)
+ totalGroupedIDs += len(idsForPart)
+ } else {
+ bloomFilteredPartIDs = append(bloomFilteredPartIDs, partID)
+ }
+ }
+ }
+
+ if finishPartSelection != nil {
+ finishPartSelection(&partSelectionMetrics{
+ bloomFilteredPartIDs: bloomFilteredPartIDs,
+ totalGroupedIDs: totalGroupedIDs,
+ }, len(parts))
+ }
+
// Create the cursor channel and scanBatch
cursorCh := make(chan scanCursorResult)
sb := &scanBatch{
traceBatch: batch,
cursorCh: cursorCh,
+ snapshots: snapshots,
}
// Send batch downstream first so consumer can start reading
select {
case <-ctx.Done():
close(cursorCh)
+ // Release snapshots on early exit
+ for _, s := range snapshots {
+ s.decRef()
+ }
return
case out <- sb:
}
// Now scan inline and populate the channel
- t.scanTraceIDsInline(ctx, parts, qo, batch.traceIDs, cursorCh)
+ t.scanPartsInline(partSelectionCtx, parts, groupedIDs, qo, cursorCh)
close(cursorCh)
}
}()
@@ -700,18 +805,12 @@ type scanCursorResult struct {
err error
}
-func (t *trace) scanTraceIDsInline(ctx context.Context, parts []*part, qo queryOptions, traceIDs []string, out chan<- scanCursorResult) {
- if len(parts) == 0 || len(traceIDs) == 0 {
+func (t *trace) scanPartsInline(ctx context.Context, parts []*part, groupedIDs [][]string, qo queryOptions, out chan<- scanCursorResult) {
+ if len(parts) == 0 {
return
}
- sortedIDs := append([]string(nil), traceIDs...)
- sort.Strings(sortedIDs)
-
- recordBlock, finish := startBlockScanSpan(ctx, sortedIDs, parts)
- if finish == nil {
- finish = func(error) {}
- }
+ recordBlock, finishSpan := startAggregatedBlockScanSpan(ctx, groupedIDs, parts)
var (
spanErr error
@@ -719,17 +818,19 @@ func (t *trace) scanTraceIDsInline(ctx context.Context, parts []*part, qo queryO
cursorCount int
)
- defer func() {
- finish(spanErr)
- }()
-
bma := generateBlockMetadataArray()
defer releaseBlockMetadataArray(bma)
tstIter := generateTstIter()
defer releaseTstIter(tstIter)
- tstIter.init(bma, parts, sortedIDs)
+ if finishSpan != nil {
+ defer func() {
+ finishSpan(cursorCount, spanBlockBytes, spanErr)
+ }()
+ }
+
+ tstIter.init(bma, parts, groupedIDs)
if initErr := tstIter.Error(); initErr != nil {
spanErr = fmt.Errorf("cannot init tstIter: %w", initErr)
select {
@@ -778,7 +879,7 @@ func (t *trace) scanTraceIDsInline(ctx context.Context, parts []*part, qo queryO
// Quota OK, send cursor
if recordBlock != nil {
- recordBlock(bc)
+ recordBlock(bc, blockSize)
}
spanBlockBytes += blockSize
cursorCount++
diff --git a/banyand/trace/streaming_pipeline_test.go b/banyand/trace/streaming_pipeline_test.go
index bd1da3ec..f361ded5 100644
--- a/banyand/trace/streaming_pipeline_test.go
+++ b/banyand/trace/streaming_pipeline_test.go
@@ -135,6 +135,7 @@ func TestStreamSIDXTraceBatches_ProducesOrderedBatches(t *testing.T) {
encodeTraceIDForTest("a"),
encodeTraceIDForTest("b"),
},
+ PartIDs: []uint64{100, 101},
},
{
Keys: []int64{2, 3},
@@ -142,6 +143,7 @@ func TestStreamSIDXTraceBatches_ProducesOrderedBatches(t *testing.T) {
encodeTraceIDForTest("b"),
encodeTraceIDForTest("c"),
},
+ PartIDs: []uint64{101, 102},
},
}
@@ -149,7 +151,7 @@ func TestStreamSIDXTraceBatches_ProducesOrderedBatches(t *testing.T) {
defer cancel()
var tr trace
- batchCh := tr.streamSIDXTraceBatches(ctx, []sidx.SIDX{&fakeSIDX{responses: responses}}, req, 3)
+ batchCh, _ := tr.streamSIDXTraceBatches(ctx, []sidx.SIDX{&fakeSIDX{responses: responses}}, req, 3)
var batches []traceBatch
for batch := range batchCh {
@@ -163,18 +165,30 @@ func TestStreamSIDXTraceBatches_ProducesOrderedBatches(t *testing.T) {
t.Fatalf("expected 2 batches, got %d", len(batches))
}
- if diff := cmp.Diff([]string{"a", "b"}, batches[0].traceIDs); diff != "" {
- t.Fatalf("first batch mismatch (-got +want):\n%s", diff)
+ // First batch should have trace IDs grouped by partID
+ wantBatch0 := map[uint64][]string{
+ 100: {"a"},
+ 101: {"b"},
}
- if diff := cmp.Diff([]string{"c"}, batches[1].traceIDs); diff != "" {
- t.Fatalf("second batch mismatch (-got +want):\n%s", diff)
+ if diff := cmp.Diff(wantBatch0, batches[0].traceIDs); diff != "" {
+ t.Fatalf("first batch mismatch (-want +got):\n%s", diff)
+ }
+
+ // Second batch should have trace ID in partID 102
+ wantBatch1 := map[uint64][]string{
+ 102: {"c"},
+ }
+ if diff := cmp.Diff(wantBatch1, batches[1].traceIDs); diff != "" {
+ t.Fatalf("second batch mismatch (-want +got):\n%s", diff)
}
wantKeys := map[string]int64{"a": 1, "b": 2, "c": 3}
for _, batch := range batches {
- for _, tid := range batch.traceIDs {
- if got := batch.keys[tid]; got != wantKeys[tid] {
- t.Fatalf("unexpected key for %s: got %d, want %d", tid, got, wantKeys[tid])
+ for _, ids := range batch.traceIDs {
+ for _, tid := range ids {
+ if got := batch.keys[tid]; got != wantKeys[tid] {
+ t.Fatalf("unexpected key for %s: got %d, want %d", tid, got, wantKeys[tid])
+ }
}
}
}
@@ -193,6 +207,7 @@ func TestStreamSIDXTraceBatches_OrdersDescending(t *testing.T) {
encodeTraceIDForTest("e"),
encodeTraceIDForTest("d"),
},
+ PartIDs: []uint64{200, 201},
},
{
Keys: []int64{4, 3},
@@ -200,6 +215,7 @@ func TestStreamSIDXTraceBatches_OrdersDescending(t *testing.T) {
encodeTraceIDForTest("d"),
encodeTraceIDForTest("c"),
},
+ PartIDs: []uint64{201, 202},
},
}
@@ -207,7 +223,7 @@ func TestStreamSIDXTraceBatches_OrdersDescending(t *testing.T) {
defer cancel()
var tr trace
- batchCh := tr.streamSIDXTraceBatches(ctx, []sidx.SIDX{&fakeSIDX{responses: responses}}, req, 3)
+ batchCh, _ := tr.streamSIDXTraceBatches(ctx, []sidx.SIDX{&fakeSIDX{responses: responses}}, req, 3)
var batches []traceBatch
for batch := range batchCh {
@@ -221,18 +237,30 @@ func TestStreamSIDXTraceBatches_OrdersDescending(t *testing.T) {
t.Fatalf("expected 2 batches, got %d", len(batches))
}
- if diff := cmp.Diff([]string{"e", "d"}, batches[0].traceIDs); diff != "" {
- t.Fatalf("first batch mismatch (-got +want):\n%s", diff)
+ // First batch should have trace IDs grouped by partID
+ wantBatch0 := map[uint64][]string{
+ 200: {"e"},
+ 201: {"d"},
}
- if diff := cmp.Diff([]string{"c"}, batches[1].traceIDs); diff != "" {
- t.Fatalf("second batch mismatch (-got +want):\n%s", diff)
+ if diff := cmp.Diff(wantBatch0, batches[0].traceIDs); diff != "" {
+ t.Fatalf("first batch mismatch (-want +got):\n%s", diff)
+ }
+
+ // Second batch should have trace ID in partID 202
+ wantBatch1 := map[uint64][]string{
+ 202: {"c"},
+ }
+ if diff := cmp.Diff(wantBatch1, batches[1].traceIDs); diff != "" {
+ t.Fatalf("second batch mismatch (-want +got):\n%s", diff)
}
wantKeys := map[string]int64{"e": 5, "d": 4, "c": 3}
for _, batch := range batches {
- for _, tid := range batch.traceIDs {
- if got := batch.keys[tid]; got != wantKeys[tid] {
- t.Fatalf("unexpected key for %s: got %d, want %d", tid, got, wantKeys[tid])
+ for _, ids := range batch.traceIDs {
+ for _, tid := range ids {
+ if got := batch.keys[tid]; got != wantKeys[tid] {
+ t.Fatalf("unexpected key for %s: got %d, want %d", tid, got, wantKeys[tid])
+ }
}
}
}
@@ -254,6 +282,7 @@ func TestStreamSIDXTraceBatches_PropagatesErrorAfterCancellation(t *testing.T) {
Data: [][]byte{
encodeTraceIDForTest("trace-1"),
},
+ PartIDs: []uint64{300},
},
},
},
@@ -264,7 +293,7 @@ func TestStreamSIDXTraceBatches_PropagatesErrorAfterCancellation(t *testing.T) {
defer cancel()
var tr trace
- batchCh := tr.streamSIDXTraceBatches(ctx, []sidx.SIDX{sidxInstance}, req, 0)
+ batchCh, _ := tr.streamSIDXTraceBatches(ctx, []sidx.SIDX{sidxInstance}, req, 0)
var (
dataSeen bool
@@ -278,8 +307,12 @@ func TestStreamSIDXTraceBatches_PropagatesErrorAfterCancellation(t *testing.T) {
errSeen = true
continue
}
- if len(batch.traceIDs) > 0 {
- dataSeen = true
+ // Check if batch has any trace IDs in the map
+ for _, ids := range batch.traceIDs {
+ if len(ids) > 0 {
+ dataSeen = true
+ break
+ }
}
}
@@ -378,7 +411,7 @@ func TestStreamSIDXTraceBatches_PropagatesBlockScannerError(t *testing.T) {
defer cancel()
var tr trace
- batchCh := tr.streamSIDXTraceBatches(ctx, []sidx.SIDX{sidxInstance}, req, 0)
+ batchCh, _ := tr.streamSIDXTraceBatches(ctx, []sidx.SIDX{sidxInstance}, req, 0)
var receivedError error
@@ -425,6 +458,7 @@ func TestStreamSIDXTraceBatches_DrainErrorEventsGuaranteed(t *testing.T) {
Data: [][]byte{
encodeTraceIDForTest("trace-1"),
},
+ PartIDs: []uint64{400},
},
},
},
@@ -436,7 +470,7 @@ func TestStreamSIDXTraceBatches_DrainErrorEventsGuaranteed(t *testing.T) {
cancel()
var tr trace
- batchCh := tr.streamSIDXTraceBatches(ctx, []sidx.SIDX{sidxInstance}, req, 0)
+ batchCh, _ := tr.streamSIDXTraceBatches(ctx, []sidx.SIDX{sidxInstance}, req, 0)
var receivedError error
@@ -473,6 +507,7 @@ func TestStreamSIDXTraceBatches_ErrorEmissionResilience(t *testing.T) {
encodeTraceIDForTest("trace-2"),
encodeTraceIDForTest("trace-3"),
},
+ PartIDs: []uint64{500, 501, 502},
},
},
},
@@ -483,7 +518,7 @@ func TestStreamSIDXTraceBatches_ErrorEmissionResilience(t *testing.T) {
defer cancel()
var tr trace
- batchCh := tr.streamSIDXTraceBatches(ctx, []sidx.SIDX{sidxInstance}, req, 0)
+ batchCh, _ := tr.streamSIDXTraceBatches(ctx, []sidx.SIDX{sidxInstance}, req, 0)
var receivedError error
batchCount := 0
@@ -554,6 +589,7 @@ func (f *fakeSIDXInfinite) StreamingQuery(ctx context.Context, _ sidx.QueryReque
keys := make([]int64, batchSize)
data := make([][]byte, batchSize)
+ partIDs := make([]uint64, batchSize)
for i := 0; i < batchSize; i++ {
keys[i] = key
@@ -563,13 +599,15 @@ func (f *fakeSIDXInfinite) StreamingQuery(ctx context.Context, _ sidx.QueryReque
}
traceID := prefix + "-" + strconv.Itoa(counter)
data[i] = encodeTraceIDForTest(traceID)
+ partIDs[i] = uint64(1000 + counter) // Generate unique partIDs
key++
counter++
}
resp := &sidx.QueryResponse{
- Keys: keys,
- Data: data,
+ Keys: keys,
+ Data: data,
+ PartIDs: partIDs,
}
select {
@@ -629,7 +667,7 @@ func TestStreamSIDXTraceBatches_InfiniteChannelContinuesUntilCanceled(t *testing
defer cancel()
var tr trace
- batchCh := tr.streamSIDXTraceBatches(ctx, []sidx.SIDX{sidxInstance}, req, 0 /* no maxTraceSize limit */)
+ batchCh, _ := tr.streamSIDXTraceBatches(ctx, []sidx.SIDX{sidxInstance}, req, 0 /* no maxTraceSize limit */)
const targetTraceIDs = 50
totalTraceIDs := 0
@@ -644,15 +682,16 @@ func TestStreamSIDXTraceBatches_InfiniteChannelContinuesUntilCanceled(t *testing
t.Fatalf("unexpected error batch: %v", batch.err)
}
- for _, tid := range batch.traceIDs {
- if _, exists := seenIDs[tid]; exists {
- t.Fatalf("duplicate trace ID: %s", tid)
+ for _, ids := range batch.traceIDs {
+ for _, tid := range ids {
+ if _, exists := seenIDs[tid]; exists {
+ t.Fatalf("duplicate trace ID: %s", tid)
+ }
+ seenIDs[tid] = struct{}{}
+ totalTraceIDs++
}
- seenIDs[tid] = struct{}{}
}
- totalTraceIDs += len(batch.traceIDs)
-
// Cancel after we've received enough traces to prove it's streaming
if totalTraceIDs >= targetTraceIDs {
cancel()
@@ -686,7 +725,7 @@ func TestStreamSIDXTraceBatches_InfiniteChannelCancellation(t *testing.T) {
defer cancel()
var tr trace
- batchCh := tr.streamSIDXTraceBatches(ctx, []sidx.SIDX{sidxInstance}, req, 0 /* no maxTraceSize limit */)
+ batchCh, _ := tr.streamSIDXTraceBatches(ctx, []sidx.SIDX{sidxInstance}, req, 0 /* no maxTraceSize limit */)
// Read a few batches, then cancel
batchesRead := 0
@@ -735,7 +774,7 @@ func TestStreamSIDXTraceBatches_InfiniteChannelGoroutineCleanup(t *testing.T) {
defer cancel()
var tr trace
- batchCh := tr.streamSIDXTraceBatches(ctx, []sidx.SIDX{sidxInstance}, req, 0 /* no limit */)
+ batchCh, _ := tr.streamSIDXTraceBatches(ctx, []sidx.SIDX{sidxInstance}, req, 0 /* no limit */)
totalTraceIDs := 0
batchesRead := 0
@@ -750,7 +789,9 @@ func TestStreamSIDXTraceBatches_InfiniteChannelGoroutineCleanup(t *testing.T) {
t.Fatalf("unexpected error: %v", batch.err)
}
- totalTraceIDs += len(batch.traceIDs)
+ for _, ids := range batch.traceIDs {
+ totalTraceIDs += len(ids)
+ }
batchesRead++
// Cancel after reading a few batches
@@ -806,7 +847,7 @@ func TestStreamSIDXTraceBatches_MultipleInfiniteSIDX(t *testing.T) {
defer cancel()
var tr trace
- batchCh := tr.streamSIDXTraceBatches(ctx, sidxInstances, req, 0 /* no limit */)
+ batchCh, _ := tr.streamSIDXTraceBatches(ctx, sidxInstances, req, 0 /* no limit */)
totalTraceIDs := 0
seenIDs := make(map[string]struct{})
@@ -821,16 +862,16 @@ func TestStreamSIDXTraceBatches_MultipleInfiniteSIDX(t *testing.T) {
t.Fatalf("unexpected error batch: %v", batch.err)
}
- for _, tid := range batch.traceIDs {
+ // Use traceIDsOrder to maintain the correct order from SIDX stream
+ for _, tid := range batch.traceIDsOrder {
if _, exists := seenIDs[tid]; exists {
t.Fatalf("duplicate trace ID: %s", tid)
}
seenIDs[tid] = struct{}{}
keys = append(keys, batch.keys[tid])
+ totalTraceIDs++
}
- totalTraceIDs += len(batch.traceIDs)
-
// Cancel after we've received enough to verify the merge is working
if totalTraceIDs >= targetTraceIDs {
cancel()
diff --git a/banyand/trace/tracing.go b/banyand/trace/tracing.go
index 3fdf8fa6..f3ed8ac2 100644
--- a/banyand/trace/tracing.go
+++ b/banyand/trace/tracing.go
@@ -36,6 +36,7 @@ const (
traceIDSampleLimit = 5
blockHeadLimit = 10
blockTailLimit = 10
+ blockSampleLimit = 20
)
type blockInfo struct {
@@ -246,3 +247,110 @@ func startBlockScanSpan(ctx context.Context, traceIDs []string, parts []*part) (
span.Stop()
}
}
+
+// partSelectionMetrics holds the metrics for part selection span.
+type partSelectionMetrics struct {
+ bloomFilteredPartIDs []uint64
+ totalGroupedIDs int
+}
+
+// startPartSelectionSpan creates a span for part selection.
+// It returns the updated context and a finish function to complete the span.
+func startPartSelectionSpan(ctx context.Context, batch *traceBatch, snapshots []*snapshot) (context.Context, func(*partSelectionMetrics, int)) {
+ tracer := query.GetTracer(ctx)
+ if tracer == nil {
+ return ctx, nil
+ }
+
+ // Count total trace IDs
+ totalTraceIDCount := 0
+ for _, ids := range batch.traceIDs {
+ totalTraceIDCount += len(ids)
+ }
+
+ // Count initial parts
+ initialParts := 0
+ for _, s := range snapshots {
+ initialParts += len(s.parts)
+ }
+
+ span, spanCtx := tracer.StartSpan(ctx, "part-selection")
+ span.Tagf("initial_parts", "%d", initialParts)
+ span.Tagf("trace_ids", "%d", totalTraceIDCount)
+ for p := range batch.traceIDs {
+ span.Tagf(fmt.Sprintf("part_%d_trace_ids", p), "%d", len(batch.traceIDs[p]))
+ }
+
+ return spanCtx, func(metrics *partSelectionMetrics, finalParts int) {
+ if metrics != nil {
+ if len(metrics.bloomFilteredPartIDs) > 0 {
+ partIDStrs := make([]string, len(metrics.bloomFilteredPartIDs))
+ for i, id := range metrics.bloomFilteredPartIDs {
+ partIDStrs[i] = strconv.FormatUint(id, 10)
+ }
+ span.Tag("bloom_filtered_part_ids", strings.Join(partIDStrs, ","))
+ }
+ }
+ span.Tagf("final_parts", "%d", finalParts)
+ if finalParts > 0 && metrics != nil {
+ span.Tagf("total_grouped_ids", "%d", metrics.totalGroupedIDs)
+ avgIDsPerPart := float64(metrics.totalGroupedIDs) / float64(finalParts)
+ span.Tagf("avg_ids_per_part", "%.2f", avgIDsPerPart)
+ partsSkipped := initialParts - finalParts
+ if partsSkipped > 0 {
+ span.Tagf("parts_skipped_empty", "%d", partsSkipped)
+ }
+ }
+ span.Stop()
+ }
+}
+
+// startAggregatedBlockScanSpan creates an aggregated span for block scanning with simplified sampling.
+// It returns a callback to record blocks and a finish function to complete the span.
+func startAggregatedBlockScanSpan(ctx context.Context, groupedIDs [][]string, parts []*part) (func(*blockCursor, uint64), func(int, uint64, error)) {
+ tracer := query.GetTracer(ctx)
+ if tracer == nil {
+ return nil, nil
+ }
+
+ var scannedBlocks []blockInfo
+
+ return func(bc *blockCursor, blockSize uint64) {
+ // Sample first N blocks to prevent OOM
+ if len(scannedBlocks) < blockSampleLimit {
+ scannedBlocks = append(scannedBlocks, blockInfo{
+ blockString: bc.String(),
+ size: blockSize,
+ })
+ }
+ }, func(blockCount int, totalBytes uint64, err error) {
+ span, _ := tracer.StartSpan(ctx, "scan-blocks")
+ span.Tag("part_header", partMetadataHeader)
+ span.Tag("part_count", strconv.Itoa(len(parts)))
+ for i := range parts {
+ if parts[i] == nil {
+ continue
+ }
+ span.Tag(fmt.Sprintf("part_%d_%s", parts[i].partMetadata.ID, parts[i].path), parts[i].partMetadata.String())
+ span.Tag(fmt.Sprintf("part_%d_trace_ids", parts[i].partMetadata.ID), strings.Join(groupedIDs[i], ","))
+ }
+
+ // Add aggregated block metrics
+ span.Tag("block_header", blockHeader)
+ span.Tag("block_total_bytes", humanize.Bytes(totalBytes))
+ span.Tag("block_count", strconv.Itoa(blockCount))
+
+ // Add sampled blocks (first N encountered)
+ if len(scannedBlocks) > 0 {
+ span.Tag("block_sample_count", strconv.Itoa(len(scannedBlocks)))
+ for i := range scannedBlocks {
+ span.Tag(fmt.Sprintf("block_%d", i), scannedBlocks[i].blockString)
+ }
+ }
+
+ if err != nil {
+ span.Error(err)
+ }
+ span.Stop()
+ }
+}
diff --git a/banyand/trace/tstable.go b/banyand/trace/tstable.go
index 36849437..4a9645eb 100644
--- a/banyand/trace/tstable.go
+++ b/banyand/trace/tstable.go
@@ -460,7 +460,7 @@ func (ti *tstIter) reset() {
ti.idx = 0
}
-func (ti *tstIter) init(bma *blockMetadataArray, parts []*part, tids []string) {
+func (ti *tstIter) init(bma *blockMetadataArray, parts []*part, groupedTids [][]string) {
ti.reset()
ti.parts = parts
@@ -470,7 +470,7 @@ func (ti *tstIter) init(bma *blockMetadataArray, parts []*part, tids []string) {
ti.piPool = ti.piPool[:len(ti.parts)]
for i, p := range ti.parts {
ti.piPool[i] = &partIter{}
- ti.piPool[i].init(bma, p, tids)
+ ti.piPool[i].init(bma, p, groupedTids[i])
}
if len(ti.piPool) == 0 {
diff --git a/banyand/trace/tstable_test.go b/banyand/trace/tstable_test.go
index 706dc6fd..73d072b3 100644
--- a/banyand/trace/tstable_test.go
+++ b/banyand/trace/tstable_test.go
@@ -140,7 +140,12 @@ func Test_tstIter(t *testing.T) {
pp, n := s.getParts(nil, tt.minTimestamp, tt.maxTimestamp, []string{tt.tid})
require.Equal(t, len(s.parts), n)
ti := &tstIter{}
- ti.init(bma, pp, []string{tt.tid})
+ // Convert trace IDs to grouped format (each part gets all trace IDs for testing)
+ groupedTids := make([][]string, len(pp))
+ for i := range groupedTids {
+ groupedTids[i] = []string{tt.tid}
+ }
+ ti.init(bma, pp, groupedTids)
var got []blockMetadata
for ti.nextBlock() {
if ti.piPool[ti.idx].curBlock.traceID == "" {
@@ -400,7 +405,11 @@ func Test_tstIter_reset_does_not_corrupt_parts(t *testing.T) {
// Initialize and release tstIter
ti := generateTstIter()
- ti.init(bma, parts, []string{"trace1"})
+ groupedTids := make([][]string, len(parts))
+ for i := range groupedTids {
+ groupedTids[i] = []string{"trace1"}
+ }
+ ti.init(bma, parts, groupedTids)
releaseTstIter(ti)
// Verify parts array is intact
@@ -441,7 +450,11 @@ func Test_tstIter_reset_does_not_corrupt_parts(t *testing.T) {
// Simulate scanTraceIDsInline: get tstIter, use it, release it
ti := generateTstIter()
- ti.init(bma, parts, traceIDs)
+ groupedTids := make([][]string, len(parts))
+ for i := range groupedTids {
+ groupedTids[i] = traceIDs
+ }
+ ti.init(bma, parts, groupedTids)
// Verify tstIter was initialized correctly
require.Equal(t, 3, len(ti.parts), "Cycle %d: tstIter should have 3 parts", cycle)
@@ -471,7 +484,11 @@ func Test_tstIter_reset_does_not_corrupt_parts(t *testing.T) {
originalCap := cap(parts)
ti := generateTstIter()
- ti.init(bma, parts, []string{"trace1"})
+ groupedTids := make([][]string, len(parts))
+ for i := range groupedTids {
+ groupedTids[i] = []string{"trace1"}
+ }
+ ti.init(bma, parts, groupedTids)
releaseTstIter(ti)
// Verify length and capacity are preserved
@@ -497,7 +514,11 @@ func Test_tstIter_reset_does_not_corrupt_parts(t *testing.T) {
// First use: with parts1
ti := generateTstIter()
- ti.init(bma, parts1, []string{"a", "b"})
+ groupedTids1 := make([][]string, len(parts1))
+ for i := range groupedTids1 {
+ groupedTids1[i] = []string{"a", "b"}
+ }
+ ti.init(bma, parts1, groupedTids1)
require.Equal(t, 2, len(ti.parts), "First init: tstIter should have 2 parts")
releaseTstIter(ti)
@@ -507,7 +528,11 @@ func Test_tstIter_reset_does_not_corrupt_parts(t *testing.T) {
// Second use: reuse tstIter with parts2 (different size)
ti = generateTstIter()
- ti.init(bma, parts2, []string{"c", "d", "e"})
+ groupedTids2 := make([][]string, len(parts2))
+ for i := range groupedTids2 {
+ groupedTids2[i] = []string{"c", "d", "e"}
+ }
+ ti.init(bma, parts2, groupedTids2)
require.Equal(t, 3, len(ti.parts), "Second init: tstIter should have 3 parts")
releaseTstIter(ti)
@@ -551,7 +576,11 @@ func Test_tstIter_reset_clears_but_preserves_input(t *testing.T) {
// Initialize tstIter
ti := generateTstIter()
- ti.init(bma, parts, []string{"trace1"})
+ groupedTids := make([][]string, len(parts))
+ for i := range groupedTids {
+ groupedTids[i] = []string{"trace1"}
+ }
+ ti.init(bma, parts, groupedTids)
require.Equal(t, 3, len(ti.parts), "After init: tstIter should have 3 parts")
require.NotNil(t, ti.parts[0], "After init: ti.parts[0] should not be nil")
diff --git a/pkg/flow/streaming/sliding_window.go b/pkg/flow/streaming/sliding_window.go
index 9a1e2c62..9cd7bbc5 100644
--- a/pkg/flow/streaming/sliding_window.go
+++ b/pkg/flow/streaming/sliding_window.go
@@ -135,7 +135,8 @@ func (s *tumblingTimeWindows) flushWindow(w timeWindow) {
flushed := s.flushSnapshot(w, snapshot.(flow.AggregationOp))
if s.l != nil {
if e := s.l.Debug(); e.Enabled() {
- e.Stringer("window", w).Bool("flushed", flushed).Msg("flush window")
+ windowsStr := w.String()
+ e.Str("window", windowsStr).Bool("flushed", flushed).Msg("flush window")
}
}
}
diff --git a/pkg/query/logical/tag_filter_adapter.go b/pkg/query/logical/tag_filter_adapter.go
new file mode 100644
index 00000000..f4a52e11
--- /dev/null
+++ b/pkg/query/logical/tag_filter_adapter.go
@@ -0,0 +1,56 @@
+// 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 logical
+
+import (
+ modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
+ "github.com/apache/skywalking-banyandb/pkg/query/model"
+)
+
+type tagFilterAdapter struct {
+ filter TagFilter
+ schema Schema
+ decoder model.TagValueDecoder
+}
+
+// NewTagFilterMatcher creates a TagFilterMatcher from a TagFilter and Schema.
+func NewTagFilterMatcher(filter TagFilter, schema Schema, decoder model.TagValueDecoder) model.TagFilterMatcher {
+ if filter == nil || filter == DummyFilter {
+ return nil
+ }
+ return &tagFilterAdapter{
+ filter: filter,
+ schema: schema,
+ decoder: decoder,
+ }
+}
+
+func (tfa *tagFilterAdapter) Match(tags []*modelv1.Tag) (bool, error) {
+ // Convert tags to TagFamily for logical.TagFilter.Match
+ family := &modelv1.TagFamily{
+ Name: "",
+ Tags: tags,
+ }
+
+ tagFamilies := []*modelv1.TagFamily{family}
+ return tfa.filter.Match(TagFamilies(tagFamilies), tfa.schema)
+}
+
+func (tfa *tagFilterAdapter) GetDecoder() model.TagValueDecoder {
+ return tfa.decoder
+}
diff --git a/pkg/query/logical/trace/trace_plan_local.go b/pkg/query/logical/trace/trace_plan_local.go
index 9dae3174..7f1e950e 100644
--- a/pkg/query/logical/trace/trace_plan_local.go
+++ b/pkg/query/logical/trace/trace_plan_local.go
@@ -42,6 +42,7 @@ var (
type localScan struct {
schema logical.Schema
skippingFilter index.Filter
+ tagFilterMatcher model.TagFilterMatcher
result model.TraceQueryResult
ec executor.TraceExecutionContext
order *logical.OrderBy
@@ -93,6 +94,7 @@ func (i *localScan) Execute(ctx context.Context) (iter.Iterator[model.TraceResul
Name: i.metadata.GetName(),
TimeRange: &i.timeRange,
SkippingFilter: i.skippingFilter,
+ TagFilter: i.tagFilterMatcher,
Order: orderBy,
TagProjection: i.projectionTags,
Entities: i.entities,
diff --git a/pkg/query/logical/trace/trace_plan_tag_filter.go b/pkg/query/logical/trace/trace_plan_tag_filter.go
index 2561d7e4..cf900fdb 100644
--- a/pkg/query/logical/trace/trace_plan_tag_filter.go
+++ b/pkg/query/logical/trace/trace_plan_tag_filter.go
@@ -97,8 +97,19 @@ func (uis *unresolvedTraceTagFilter) Analyze(s logical.Schema) (logical.Plan, er
}
// Add tag names from filter conditions to projection
+ var conditionSchema logical.Schema
if len(conditionTagNames) > 0 {
ctx.projectionTags.Names = append(ctx.projectionTags.Names, conditionTagNames...)
+ tags := make([]*logical.Tag, len(conditionTagNames))
+ for i, tagName := range conditionTagNames {
+ tags[i] = logical.NewTag("", tagName)
+ }
+ var conditionTagRefs [][]*logical.TagRef
+ conditionTagRefs, err = s.CreateTagRef(tags)
+ if err != nil {
+ return nil, err
+ }
+ conditionSchema = s.ProjTags(conditionTagRefs...)
}
// Deduplicate tag names
@@ -116,14 +127,31 @@ func (uis *unresolvedTraceTagFilter) Analyze(s logical.Schema) (logical.Plan, er
return nil, errProject
}
}
- plan := uis.selectTraceScanner(ctx, uis.ec, traceIDs, minVal, maxVal)
+ // Build tag filter and create matcher for SIDX
+ var tagFilterMatcher model.TagFilterMatcher
+ var tagFilter logical.TagFilter
if uis.criteria != nil {
- spanIDFilter := buildSpanIDFilter(uis.criteria, uis.spanIDTagName)
- tagFilter, errFilter := logical.BuildTagFilter(uis.criteria, entityDict, s, s, len(traceIDs) > 0, uis.traceIDTagName, uis.spanIDTagName)
- if errFilter != nil {
- return nil, errFilter
+ var orderByTags []string
+ if ok, indexRule := s.IndexRuleDefined(uis.orderByTag); ok {
+ orderByTags = indexRule.Tags
+ }
+ skippedTagNames := make([]string, 0, len(orderByTags)+2)
+ skippedTagNames = append(skippedTagNames, uis.traceIDTagName, uis.spanIDTagName)
+ skippedTagNames = append(skippedTagNames, orderByTags...)
+ tagFilter, err = logical.BuildTagFilter(uis.criteria, entityDict, s, s, len(traceIDs) > 0, skippedTagNames...)
+ if err != nil {
+ return nil, err
}
- if tagFilter != logical.DummyFilter || spanIDFilter != nil {
+ // Get the decoder from the execution context (trace module)
+ decoder := uis.ec.(model.TagValueDecoderProvider).GetTagValueDecoder()
+ // Create tag filter matcher for SIDX
+ tagFilterMatcher = logical.NewTagFilterMatcher(tagFilter, conditionSchema, decoder)
+ }
+
+ plan := uis.selectTraceScanner(ctx, uis.ec, traceIDs, minVal, maxVal, tagFilterMatcher)
+ if uis.criteria != nil {
+ spanIDFilter := buildSpanIDFilter(uis.criteria, uis.spanIDTagName)
+ if len(traceIDs) > 0 || tagFilter != logical.DummyFilter || spanIDFilter != nil {
// create filter with a projected view
plan = newTraceTagFilter(s.ProjTags(ctx.projTagsRefs...), plan, tagFilter, spanIDFilter)
}
@@ -132,7 +160,7 @@ func (uis *unresolvedTraceTagFilter) Analyze(s logical.Schema) (logical.Plan, er
}
func (uis *unresolvedTraceTagFilter) selectTraceScanner(ctx *traceAnalyzeContext,
- ec executor.TraceExecutionContext, traceIDs []string, minVal, maxVal int64,
+ ec executor.TraceExecutionContext, traceIDs []string, minVal, maxVal int64, tagFilterMatcher model.TagFilterMatcher,
) logical.Plan {
return &localScan{
timeRange: timestamp.NewInclusiveTimeRange(uis.startTime, uis.endTime),
@@ -141,6 +169,7 @@ func (uis *unresolvedTraceTagFilter) selectTraceScanner(ctx *traceAnalyzeContext
projectionTags: ctx.projectionTags,
metadata: uis.metadata,
skippingFilter: ctx.skippingFilter,
+ tagFilterMatcher: tagFilterMatcher,
entities: ctx.entities,
l: logger.GetLogger("query", "trace", "local-scan"),
ec: ec,
diff --git a/pkg/query/model/model.go b/pkg/query/model/model.go
index ae6e165b..baa413d4 100644
--- a/pkg/query/model/model.go
+++ b/pkg/query/model/model.go
@@ -318,6 +318,7 @@ type StreamQueryResult interface {
// TraceQueryOptions is the options of a trace query.
type TraceQueryOptions struct {
SkippingFilter index.Filter
+ TagFilter TagFilterMatcher
TimeRange *timestamp.TimeRange
Order *index.OrderBy
TagProjection *TagProjection
@@ -334,6 +335,7 @@ func (t *TraceQueryOptions) Reset() {
t.Name = ""
t.TimeRange = nil
t.SkippingFilter = nil
+ t.TagFilter = nil
t.Order = nil
t.TagProjection = nil
t.Entities = nil
@@ -346,6 +348,7 @@ func (t *TraceQueryOptions) CopyFrom(other *TraceQueryOptions) {
t.Name = other.Name
t.TimeRange = other.TimeRange
t.SkippingFilter = other.SkippingFilter
+ t.TagFilter = other.TagFilter
t.Order = other.Order
t.TagProjection = other.TagProjection
t.Entities = other.Entities
diff --git a/pkg/query/model/tag_filter_matcher.go b/pkg/query/model/tag_filter_matcher.go
new file mode 100644
index 00000000..993285a0
--- /dev/null
+++ b/pkg/query/model/tag_filter_matcher.go
@@ -0,0 +1,41 @@
+// 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 model
+
+import (
+ modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
+ pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
+)
+
+// TagValueDecoder decodes a byte slice to a TagValue based on the value type.
+type TagValueDecoder func(valueType pbv1.ValueType, value []byte) *modelv1.TagValue
+
+// TagValueDecoderProvider provides a decoder for tag values.
+type TagValueDecoderProvider interface {
+ GetTagValueDecoder() TagValueDecoder
+}
+
+// TagFilterMatcher is a simple interface for matching tags without schema dependency.
+// This avoids cycle imports between logical and sidx packages.
+type TagFilterMatcher interface {
+ // Match returns true if the given tags match the filter criteria.
+ // tags is a slice of tags for a single element/row.
+ Match(tags []*modelv1.Tag) (bool, error)
+ // GetDecoder returns the decoder function for tag values.
+ GetDecoder() TagValueDecoder
+}