You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by ha...@apache.org on 2025/08/29 22:48:32 UTC
(skywalking-banyandb) branch main updated: Integrate sidx writing, flushing and merging into core trace storage (#742)
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
The following commit(s) were added to refs/heads/main by this push:
new d2621864 Integrate sidx writing, flushing and merging into core trace storage (#742)
d2621864 is described below
commit d26218647e3c7bbb9096c1bf3eacfe0327e6ddd8
Author: Huang Youliang <bu...@gmail.com>
AuthorDate: Sat Aug 30 06:48:27 2025 +0800
Integrate sidx writing, flushing and merging into core trace storage (#742)
* Integrate sidx writing, flushing and merging into core trace storage
* Fixup
---
banyand/trace/flusher.go | 4 +++
banyand/trace/merger.go | 4 +++
banyand/trace/part.go | 3 ++
banyand/trace/traces.go | 2 ++
banyand/trace/tstable.go | 17 ++++++++++
banyand/trace/write_liaison.go | 2 +-
banyand/trace/write_standalone.go | 71 +++++++++++++++++++++++++++++++++++----
7 files changed, 95 insertions(+), 8 deletions(-)
diff --git a/banyand/trace/flusher.go b/banyand/trace/flusher.go
index 5e45f448..3798e76c 100644
--- a/banyand/trace/flusher.go
+++ b/banyand/trace/flusher.go
@@ -203,6 +203,10 @@ func (tst *tsTable) flush(snapshot *snapshot, flushCh chan *flusherIntroduction)
if len(ind.flushed) < 1 {
return
}
+ if err := tst.sidx.Flush(); err != nil {
+ tst.l.Warn().Err(err).Msg("sidx flush failed")
+ return
+ }
end := time.Now()
tst.incTotalFlushed(1)
tst.incTotalFlushedMemParts(partsCount)
diff --git a/banyand/trace/merger.go b/banyand/trace/merger.go
index d9dcb3d2..7258f6d2 100644
--- a/banyand/trace/merger.go
+++ b/banyand/trace/merger.go
@@ -111,6 +111,10 @@ func (tst *tsTable) mergePartsThenSendIntroduction(creator snapshotCreator, part
if err != nil {
return nil, err
}
+ if err := tst.sidx.Merge(); err != nil {
+ tst.l.Warn().Err(err).Msg("sidx merge failed")
+ return nil, err
+ }
elapsed := time.Since(start)
tst.incTotalMergeLatency(elapsed.Seconds(), typ)
tst.incTotalMerged(1, typ)
diff --git a/banyand/trace/part.go b/banyand/trace/part.go
index ee61e8c6..ba42c7ac 100644
--- a/banyand/trace/part.go
+++ b/banyand/trace/part.go
@@ -23,6 +23,7 @@ import (
"io"
"path"
"path/filepath"
+ "sort"
"sync/atomic"
"github.com/apache/skywalking-banyandb/banyand/internal/storage"
@@ -438,6 +439,8 @@ func (mp *memPart) mustInitFromTraces(ts *traces) {
return
}
+ sort.Sort(ts)
+
bsw := generateBlockWriter()
bsw.MustInitForMemPart(mp)
for _, tid := range ts.traceIDs {
diff --git a/banyand/trace/traces.go b/banyand/trace/traces.go
index 93361ce2..4f6599f2 100644
--- a/banyand/trace/traces.go
+++ b/banyand/trace/traces.go
@@ -23,6 +23,7 @@ import (
"github.com/pkg/errors"
"github.com/apache/skywalking-banyandb/api/common"
+ "github.com/apache/skywalking-banyandb/banyand/internal/sidx"
"github.com/apache/skywalking-banyandb/banyand/internal/storage"
"github.com/apache/skywalking-banyandb/banyand/internal/wqueue"
pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
@@ -199,6 +200,7 @@ type tracesInTable struct {
tsTable *tsTable
traces *traces
timeRange timestamp.TimeRange
+ sidxReqs []sidx.WriteRequest
shardID common.ShardID
}
diff --git a/banyand/trace/tstable.go b/banyand/trace/tstable.go
index 7443831b..493d29a6 100644
--- a/banyand/trace/tstable.go
+++ b/banyand/trace/tstable.go
@@ -30,6 +30,7 @@ import (
"time"
"github.com/apache/skywalking-banyandb/api/common"
+ "github.com/apache/skywalking-banyandb/banyand/internal/sidx"
"github.com/apache/skywalking-banyandb/banyand/internal/storage"
"github.com/apache/skywalking-banyandb/banyand/protector"
"github.com/apache/skywalking-banyandb/pkg/fs"
@@ -47,6 +48,7 @@ const (
type tsTable struct {
l *logger.Logger
fileSystem fs.FileSystem
+ sidx sidx.SIDX
pm protector.Memory
metrics *metrics
snapshot *snapshot
@@ -199,6 +201,15 @@ func initTSTable(fileSystem fs.FileSystem, rootPath string, p common.Position,
if m != nil {
tst.metrics = m.(*metrics)
}
+ sidxOpts, err := sidx.NewOptions(filepath.Join(rootPath, "sidx"), option.protector)
+ if err != nil {
+ logger.Panicf("cannot create sidx options: %s", err)
+ }
+ sidx, err := sidx.NewSIDX(sidxOpts)
+ if err != nil {
+ logger.Panicf("cannot create sidx: %s", err)
+ }
+ tst.sidx = sidx
tst.gc.init(&tst)
ee := fileSystem.ReadDir(rootPath)
if len(ee) == 0 {
@@ -267,10 +278,16 @@ func (tst *tsTable) Close() error {
defer tst.Unlock()
tst.deleteMetrics()
if tst.snapshot == nil {
+ if tst.sidx != nil {
+ return tst.sidx.Close()
+ }
return nil
}
tst.snapshot.decRef()
tst.snapshot = nil
+ if tst.sidx != nil {
+ return tst.sidx.Close()
+ }
return nil
}
diff --git a/banyand/trace/write_liaison.go b/banyand/trace/write_liaison.go
index 10869540..18e95e66 100644
--- a/banyand/trace/write_liaison.go
+++ b/banyand/trace/write_liaison.go
@@ -87,7 +87,7 @@ func (w *writeQueueCallback) handle(dst map[string]*tracesInQueue, writeEvent *t
if err != nil {
return nil, err
}
- err = processTraces(w.schemaRepo, et.traces, writeEvent)
+ err = processTraces(w.schemaRepo, et, writeEvent)
if err != nil {
return nil, err
}
diff --git a/banyand/trace/write_standalone.go b/banyand/trace/write_standalone.go
index 1ed6f464..918654b9 100644
--- a/banyand/trace/write_standalone.go
+++ b/banyand/trace/write_standalone.go
@@ -29,6 +29,7 @@ import (
databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
tracev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/trace/v1"
+ "github.com/apache/skywalking-banyandb/banyand/internal/sidx"
"github.com/apache/skywalking-banyandb/banyand/internal/storage"
"github.com/apache/skywalking-banyandb/banyand/observability"
"github.com/apache/skywalking-banyandb/pkg/bus"
@@ -90,7 +91,7 @@ func (w *writeCallback) handle(dst map[string]*tracesInGroup, writeEvent *tracev
if err != nil {
return nil, err
}
- err = processTraces(w.schemaRepo, et.traces, writeEvent)
+ err = processTraces(w.schemaRepo, et, writeEvent)
if err != nil {
return nil, err
}
@@ -163,7 +164,7 @@ func (w *writeCallback) prepareTracesInTable(eg *tracesInGroup, writeEvent *trac
return et, nil
}
-func processTraces(schemaRepo *schemaRepo, traces *traces, writeEvent *tracev1.InternalWriteRequest) error {
+func processTraces(schemaRepo *schemaRepo, tracesInTable *tracesInTable, writeEvent *tracev1.InternalWriteRequest) error {
req := writeEvent.Request
stm, ok := schemaRepo.loadTrace(req.GetMetadata())
if !ok {
@@ -174,8 +175,9 @@ func processTraces(schemaRepo *schemaRepo, traces *traces, writeEvent *tracev1.I
if err != nil {
return err
}
- traces.traceIDs = append(traces.traceIDs, req.Tags[idx].GetStr().GetValue())
- traces.spans = append(traces.spans, req.Span)
+ traceID := req.Tags[idx].GetStr().GetValue()
+ tracesInTable.traces.traceIDs = append(tracesInTable.traces.traceIDs, traceID)
+ tracesInTable.traces.spans = append(tracesInTable.traces.spans, req.Span)
tLen := len(req.GetTags())
if tLen < 1 {
@@ -199,7 +201,7 @@ func processTraces(schemaRepo *schemaRepo, traces *traces, writeEvent *tracev1.I
continue
}
if tagSpec.Name == stm.schema.TimestampTagName {
- traces.timestamps = append(traces.timestamps, req.Tags[i].GetTimestamp().AsTime().UnixNano())
+ tracesInTable.traces.timestamps = append(tracesInTable.traces.timestamps, req.Tags[i].GetTimestamp().AsTime().UnixNano())
}
var tagValue *modelv1.TagValue
@@ -211,11 +213,61 @@ func processTraces(schemaRepo *schemaRepo, traces *traces, writeEvent *tracev1.I
tv := encodeTagValue(tagSpec.Name, tagSpec.Type, tagValue)
tags = append(tags, tv)
}
- traces.tags = append(traces.tags, tags)
+ tracesInTable.traces.tags = append(tracesInTable.traces.tags, tags)
+
+ sidxTags := make([]sidx.Tag, 0, len(tags))
+ // TODO: set Indexed
+ for _, tag := range tags {
+ sidxTags = append(sidxTags, sidx.Tag{
+ Name: tag.tag,
+ Value: tag.value,
+ ValueType: tag.valueType,
+ })
+ }
+
+ indexRules := stm.GetIndexRules()
+ for _, indexRule := range indexRules {
+ tagName := indexRule.Tags[len(indexRule.Tags)-1]
+ tagIdx, err := getTagIndex(stm, tagName)
+ if err != nil || tagIdx >= len(req.Tags) {
+ continue
+ }
+ tv := tags[tagIdx]
+ if tv.valueType != pbv1.ValueTypeInt64 {
+ return fmt.Errorf("unsupported tag value type: %s", tv.tag)
+ }
+ key := req.Tags[tagIdx].GetInt().GetValue()
+
+ entityValues := make([]*modelv1.TagValue, 0, len(indexRule.Tags))
+ for _, tagName := range indexRule.Tags {
+ tagIdx, err := getTagIndex(stm, tagName)
+ if err != nil || tagIdx >= len(req.Tags) {
+ continue
+ }
+ entityValues = append(entityValues, req.Tags[tagIdx])
+ }
+
+ series := &pbv1.Series{
+ Subject: req.Metadata.Name,
+ EntityValues: entityValues,
+ }
+ if err := series.Marshal(); err != nil {
+ return fmt.Errorf("cannot marshal series: %w", err)
+ }
+
+ writeReq := sidx.WriteRequest{
+ Data: []byte(traceID),
+ Tags: sidxTags,
+ SeriesID: series.ID,
+ Key: key,
+ }
+ tracesInTable.sidxReqs = append(tracesInTable.sidxReqs, writeReq)
+ }
+
return nil
}
-func (w *writeCallback) Rev(_ context.Context, message bus.Message) (resp bus.Message) {
+func (w *writeCallback) Rev(ctx context.Context, message bus.Message) (resp bus.Message) {
events, ok := message.Data().([]any)
if !ok {
w.l.Warn().Msg("invalid event data type")
@@ -253,6 +305,11 @@ func (w *writeCallback) Rev(_ context.Context, message bus.Message) (resp bus.Me
for j := range g.tables {
es := g.tables[j]
es.tsTable.mustAddTraces(es.traces)
+ if len(es.sidxReqs) > 0 {
+ if err := es.tsTable.sidx.Write(ctx, es.sidxReqs); err != nil {
+ w.l.Error().Err(err).Msg("cannot write to secondary index")
+ }
+ }
releaseTraces(es.traces)
}
if len(g.segments) > 0 {