You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by "ButterBright (via GitHub)" <gi...@apache.org> on 2025/08/25 16:43:44 UTC
[PR] Implement traceID-based query [skywalking-banyandb]
ButterBright opened a new pull request, #735:
URL: https://github.com/apache/skywalking-banyandb/pull/735
<!--
⚠️ Please make sure to read this template first, pull requests that don't accord with this template
maybe closed without notice.
Texts surrounded by `<` and `>` are meant to be replaced by you, e.g. <framework name>, <issue number>.
Put an `x` in the `[ ]` to mark the item as CHECKED. `[x]`
-->
<!-- ==== 🐛 Remove this line WHEN AND ONLY WHEN you're fixing a bug, follow the checklist 👇 ====
### Fix <bug description or the bug issue number or bug issue link>
- [ ] Add a unit test to verify that the fix works.
- [ ] Explain briefly why the bug exists and how to fix it.
==== 🐛 Remove this line WHEN AND ONLY WHEN you're fixing a bug, follow the checklist 👆 ==== -->
<!-- ==== 🆕 Remove this line WHEN AND ONLY WHEN you're adding a new feature, follow the checklist 👇 ====
### <Feature description>
- [ ] If this is non-trivial feature, paste the links/URLs to the design doc.
- [ ] Update the documentation to include this new feature.
- [ ] Tests(including UT, IT, E2E) are added to verify the new feature.
- [ ] If it's UI related, attach the screenshots below.
==== 🆕 Remove this line WHEN AND ONLY WHEN you're adding a new feature, follow the checklist 👆 ==== -->
- [x] If this pull request closes/resolves/fixes an existing issue, replace the issue number. Fixes apache/skywalking#13329.
- [x] Update the [`CHANGES` log](https://github.com/apache/skywalking-banyandb/blob/main/CHANGES.md).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement traceID-based query [skywalking-banyandb]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #735:
URL: https://github.com/apache/skywalking-banyandb/pull/735#discussion_r2298596928
##########
banyand/trace/query.go:
##########
@@ -18,39 +18,298 @@
package trace
import (
- "github.com/apache/skywalking-banyandb/api/common"
+ "container/heap"
+ "context"
+ "fmt"
+ "sort"
+
+ "github.com/pkg/errors"
+
databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
+ "github.com/apache/skywalking-banyandb/banyand/internal/storage"
"github.com/apache/skywalking-banyandb/pkg/convert"
"github.com/apache/skywalking-banyandb/pkg/logger"
pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
"github.com/apache/skywalking-banyandb/pkg/query/model"
)
+const checkDoneEvery = 128
+
+var nilResult = model.TraceQueryResult(nil)
+
type queryOptions struct {
+ traceID string
model.TraceQueryOptions
- seriesToEntity map[common.SeriesID][]*modelv1.TagValue
- sortedSids []common.SeriesID
- minTimestamp int64
- maxTimestamp int64
+ minTimestamp int64
+ maxTimestamp int64
}
func (qo *queryOptions) reset() {
qo.TraceQueryOptions.Reset()
- qo.seriesToEntity = nil
- qo.sortedSids = nil
+ qo.traceID = ""
qo.minTimestamp = 0
qo.maxTimestamp = 0
}
func (qo *queryOptions) copyFrom(other *queryOptions) {
qo.TraceQueryOptions.CopyFrom(&other.TraceQueryOptions)
- qo.seriesToEntity = other.seriesToEntity
- qo.sortedSids = other.sortedSids
+ qo.traceID = other.traceID
qo.minTimestamp = other.minTimestamp
qo.maxTimestamp = other.maxTimestamp
}
+func (t *trace) Query(ctx context.Context, tqo model.TraceQueryOptions) (model.TraceQueryResult, error) {
+ if tqo.TimeRange == nil {
+ return nil, errors.New("invalid query options: timeRange are required")
+ }
+ if tqo.TagProjection == nil || len(tqo.TagProjection.Names) == 0 {
+ return nil, errors.New("invalid query options: tagProjection is required")
+ }
+ var tsdb storage.TSDB[*tsTable, option]
+ var err error
+ db := t.tsdb.Load()
+ if db == nil {
+ tsdb, err = t.schemaRepo.loadTSDB(t.group)
+ if err != nil {
+ return nil, err
+ }
+ t.tsdb.Store(tsdb)
+ } else {
+ tsdb = db.(storage.TSDB[*tsTable, option])
+ }
+
+ segments, err := tsdb.SelectSegments(*tqo.TimeRange)
+ if err != nil {
+ return nil, err
+ }
+ if len(segments) < 1 {
+ return nilResult, nil
+ }
+
+ result := queryResult{
+ ctx: ctx,
+ segments: segments,
+ tagProjection: tqo.TagProjection,
+ }
+ defer func() {
+ if err != nil {
+ result.Release()
+ }
+ }()
+ var parts []*part
+ qo := queryOptions{
+ TraceQueryOptions: tqo,
+ traceID: "",
+ minTimestamp: tqo.TimeRange.Start.UnixNano(),
+ maxTimestamp: tqo.TimeRange.End.UnixNano(),
+ }
+ var n int
+ tables := make([]*tsTable, 0)
+ for _, segment := range segments {
+ tt, _ := segment.Tables()
+ tables = append(tables, tt...)
+ }
+ for i := range tables {
+ s := tables[i].currentSnapshot()
+ if s == nil {
+ continue
+ }
+ parts, n = s.getParts(parts, qo.minTimestamp, qo.maxTimestamp)
+ if n < 1 {
+ s.decRef()
+ continue
+ }
+ result.snapshots = append(result.snapshots, s)
+ }
+
+ if err = t.searchBlocks(ctx, &result, parts, qo); err != nil {
+ return nil, err
+ }
+
+ return &result, nil
+}
+
+func (t *trace) searchBlocks(ctx context.Context, result *queryResult, parts []*part, qo queryOptions) error {
+ bma := generateBlockMetadataArray()
+ defer releaseBlockMetadataArray(bma)
+ defFn := startBlockScanSpan(ctx, qo.traceID, parts, result)
+ defer defFn()
+ tstIter := generateTstIter()
+ defer releaseTstIter(tstIter)
+ tstIter.init(bma, parts, qo.traceID)
+ if tstIter.Error() != nil {
+ return fmt.Errorf("cannot init tstIter: %w", tstIter.Error())
+ }
+ var hit int
+ var spanBlockBytes uint64
+ quota := t.pm.AvailableBytes()
+ for tstIter.nextBlock() {
+ if hit%checkDoneEvery == 0 {
+ select {
+ case <-ctx.Done():
+ return errors.WithMessagef(ctx.Err(), "interrupt: scanned %d blocks, remained %d/%d parts to scan",
+ len(result.data), len(tstIter.piPool)-tstIter.idx, len(tstIter.piPool))
+ default:
+ }
+ }
+ hit++
+ bc := generateBlockCursor()
+ p := tstIter.piPool[tstIter.idx]
+ bc.init(p.p, p.curBlock, qo)
+ result.data = append(result.data, bc)
+ spanBlockBytes += bc.bm.uncompressedSpanSizeBytes
+ if quota >= 0 && spanBlockBytes > uint64(quota) {
+ return fmt.Errorf("block scan quota exceeded: used %d bytes, quota is %d bytes", spanBlockBytes, quota)
+ }
+ }
+ if tstIter.Error() != nil {
+ return fmt.Errorf("cannot iterate tstIter: %w", tstIter.Error())
+ }
+ return t.pm.AcquireResource(ctx, spanBlockBytes)
+}
+
+type queryResult struct {
+ ctx context.Context
+ tagProjection *model.TagProjection
+ data []*blockCursor
+ snapshots []*snapshot
+ segments []storage.Segment[*tsTable, option]
+ hit int
+ loaded bool
+}
+
+func (qr *queryResult) Pull() *model.TraceResult {
+ select {
+ case <-qr.ctx.Done():
+ return &model.TraceResult{
+ Error: errors.WithMessagef(qr.ctx.Err(), "interrupt: hit %d", qr.hit),
+ }
+ default:
+ }
+ if !qr.loaded {
+ if len(qr.data) == 0 {
+ return nil
+ }
+
+ cursorChan := make(chan int, len(qr.data))
+ for i := 0; i < len(qr.data); i++ {
+ go func(i int) {
+ select {
+ case <-qr.ctx.Done():
+ cursorChan <- i
+ return
+ default:
+ }
+ tmpBlock := generateBlock()
+ defer releaseBlock(tmpBlock)
+ if !qr.data[i].loadData(tmpBlock) {
+ cursorChan <- i
+ return
+ }
+ cursorChan <- -1
+ }(i)
+ }
+
+ blankCursorList := []int{}
+ for completed := 0; completed < len(qr.data); completed++ {
+ result := <-cursorChan
+ if result != -1 {
+ blankCursorList = append(blankCursorList, result)
+ }
+ }
+ select {
+ case <-qr.ctx.Done():
+ return &model.TraceResult{
+ Error: errors.WithMessagef(qr.ctx.Err(), "interrupt: blank/total=%d/%d", len(blankCursorList), len(qr.data)),
+ }
+ default:
+ }
+ sort.Slice(blankCursorList, func(i, j int) bool {
+ return blankCursorList[i] > blankCursorList[j]
+ })
+ for _, index := range blankCursorList {
+ qr.data = append(qr.data[:index], qr.data[index+1:]...)
+ }
+ qr.loaded = true
+ heap.Init(qr)
+ }
+ if len(qr.data) == 0 {
+ return nil
+ }
+ if len(qr.data) == 1 {
+ r := &model.TraceResult{}
+ bc := qr.data[0]
+ bc.copyAllTo(r, false)
+ qr.data = qr.data[:0]
+ releaseBlockCursor(bc)
+ return r
+ }
+ return qr.merge()
+}
+
+func (qr *queryResult) Release() {
+ for i, v := range qr.data {
+ releaseBlockCursor(v)
+ qr.data[i] = nil
+ }
+ qr.data = qr.data[:0]
+ for i := range qr.snapshots {
+ qr.snapshots[i].decRef()
+ }
+ qr.snapshots = qr.snapshots[:0]
+ for i := range qr.segments {
+ qr.segments[i].DecRef()
+ }
+}
+
+func (qr queryResult) Len() int {
+ return len(qr.data)
+}
+
+func (qr queryResult) Less(i, j int) bool {
+ return qr.data[i].bm.traceID < qr.data[j].bm.traceID
+}
+
+func (qr queryResult) Swap(i, j int) {
+ qr.data[i], qr.data[j] = qr.data[j], qr.data[i]
+}
+
+func (qr *queryResult) Push(x interface{}) {
+ qr.data = append(qr.data, x.(*blockCursor))
Review Comment:
[nitpick] The Push and Pop methods use interface{} instead of type-safe generics. Consider using a type parameter or at least add type assertions with error handling to make the code more robust.
```suggestion
bc, ok := x.(*blockCursor)
if !ok {
panic(fmt.Sprintf("queryResult.Push: expected *blockCursor, got %T", x))
}
qr.data = append(qr.data, bc)
```
##########
banyand/trace/part_iter.go:
##########
@@ -193,24 +154,23 @@ func (pi *partIter) readPrimaryBlock(bms []blockMetadata, mr *primaryBlockMetada
func (pi *partIter) findBlock() bool {
bhs := pi.bms
- for len(bhs) > 0 {
+ if len(bhs) > 0 {
tid := pi.curBlock.traceID
if bhs[0].traceID < tid {
n := sort.Search(len(bhs), func(i int) bool {
return tid <= bhs[i].traceID
})
if n == len(bhs) {
- break
+ pi.bms = nil
+ return false
}
bhs = bhs[n:]
}
bm := &bhs[0]
- if bm.traceID != tid {
- if !pi.searchTargetTraceID(bm.traceID) {
- return false
- }
- continue
+ if bm.traceID > tid {
+ pi.bms = bhs[:0]
Review Comment:
Setting `pi.bms = bhs[:0]` creates a slice with zero length but retains the underlying array capacity. This should be `pi.bms = nil` to properly clear the reference and allow garbage collection.
```suggestion
pi.bms = nil
```
##########
banyand/trace/query.go:
##########
@@ -18,39 +18,298 @@
package trace
import (
- "github.com/apache/skywalking-banyandb/api/common"
+ "container/heap"
+ "context"
+ "fmt"
+ "sort"
+
+ "github.com/pkg/errors"
+
databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
+ "github.com/apache/skywalking-banyandb/banyand/internal/storage"
"github.com/apache/skywalking-banyandb/pkg/convert"
"github.com/apache/skywalking-banyandb/pkg/logger"
pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
"github.com/apache/skywalking-banyandb/pkg/query/model"
)
+const checkDoneEvery = 128
+
+var nilResult = model.TraceQueryResult(nil)
+
type queryOptions struct {
+ traceID string
model.TraceQueryOptions
- seriesToEntity map[common.SeriesID][]*modelv1.TagValue
- sortedSids []common.SeriesID
- minTimestamp int64
- maxTimestamp int64
+ minTimestamp int64
+ maxTimestamp int64
}
func (qo *queryOptions) reset() {
qo.TraceQueryOptions.Reset()
- qo.seriesToEntity = nil
- qo.sortedSids = nil
+ qo.traceID = ""
qo.minTimestamp = 0
qo.maxTimestamp = 0
}
func (qo *queryOptions) copyFrom(other *queryOptions) {
qo.TraceQueryOptions.CopyFrom(&other.TraceQueryOptions)
- qo.seriesToEntity = other.seriesToEntity
- qo.sortedSids = other.sortedSids
+ qo.traceID = other.traceID
qo.minTimestamp = other.minTimestamp
qo.maxTimestamp = other.maxTimestamp
}
+func (t *trace) Query(ctx context.Context, tqo model.TraceQueryOptions) (model.TraceQueryResult, error) {
+ if tqo.TimeRange == nil {
+ return nil, errors.New("invalid query options: timeRange are required")
+ }
+ if tqo.TagProjection == nil || len(tqo.TagProjection.Names) == 0 {
+ return nil, errors.New("invalid query options: tagProjection is required")
+ }
+ var tsdb storage.TSDB[*tsTable, option]
+ var err error
+ db := t.tsdb.Load()
+ if db == nil {
+ tsdb, err = t.schemaRepo.loadTSDB(t.group)
+ if err != nil {
+ return nil, err
+ }
+ t.tsdb.Store(tsdb)
+ } else {
+ tsdb = db.(storage.TSDB[*tsTable, option])
+ }
+
+ segments, err := tsdb.SelectSegments(*tqo.TimeRange)
+ if err != nil {
+ return nil, err
+ }
+ if len(segments) < 1 {
+ return nilResult, nil
+ }
+
+ result := queryResult{
+ ctx: ctx,
+ segments: segments,
+ tagProjection: tqo.TagProjection,
+ }
+ defer func() {
+ if err != nil {
+ result.Release()
+ }
+ }()
+ var parts []*part
+ qo := queryOptions{
+ TraceQueryOptions: tqo,
+ traceID: "",
+ minTimestamp: tqo.TimeRange.Start.UnixNano(),
+ maxTimestamp: tqo.TimeRange.End.UnixNano(),
+ }
+ var n int
+ tables := make([]*tsTable, 0)
+ for _, segment := range segments {
+ tt, _ := segment.Tables()
+ tables = append(tables, tt...)
+ }
+ for i := range tables {
+ s := tables[i].currentSnapshot()
+ if s == nil {
+ continue
+ }
+ parts, n = s.getParts(parts, qo.minTimestamp, qo.maxTimestamp)
+ if n < 1 {
+ s.decRef()
+ continue
+ }
+ result.snapshots = append(result.snapshots, s)
+ }
+
+ if err = t.searchBlocks(ctx, &result, parts, qo); err != nil {
+ return nil, err
+ }
+
+ return &result, nil
+}
+
+func (t *trace) searchBlocks(ctx context.Context, result *queryResult, parts []*part, qo queryOptions) error {
+ bma := generateBlockMetadataArray()
+ defer releaseBlockMetadataArray(bma)
+ defFn := startBlockScanSpan(ctx, qo.traceID, parts, result)
+ defer defFn()
+ tstIter := generateTstIter()
+ defer releaseTstIter(tstIter)
+ tstIter.init(bma, parts, qo.traceID)
+ if tstIter.Error() != nil {
+ return fmt.Errorf("cannot init tstIter: %w", tstIter.Error())
+ }
+ var hit int
+ var spanBlockBytes uint64
+ quota := t.pm.AvailableBytes()
+ for tstIter.nextBlock() {
+ if hit%checkDoneEvery == 0 {
+ select {
+ case <-ctx.Done():
+ return errors.WithMessagef(ctx.Err(), "interrupt: scanned %d blocks, remained %d/%d parts to scan",
+ len(result.data), len(tstIter.piPool)-tstIter.idx, len(tstIter.piPool))
+ default:
+ }
+ }
+ hit++
+ bc := generateBlockCursor()
+ p := tstIter.piPool[tstIter.idx]
+ bc.init(p.p, p.curBlock, qo)
+ result.data = append(result.data, bc)
+ spanBlockBytes += bc.bm.uncompressedSpanSizeBytes
+ if quota >= 0 && spanBlockBytes > uint64(quota) {
+ return fmt.Errorf("block scan quota exceeded: used %d bytes, quota is %d bytes", spanBlockBytes, quota)
+ }
+ }
+ if tstIter.Error() != nil {
+ return fmt.Errorf("cannot iterate tstIter: %w", tstIter.Error())
+ }
+ return t.pm.AcquireResource(ctx, spanBlockBytes)
+}
+
+type queryResult struct {
+ ctx context.Context
+ tagProjection *model.TagProjection
+ data []*blockCursor
+ snapshots []*snapshot
+ segments []storage.Segment[*tsTable, option]
+ hit int
+ loaded bool
+}
+
+func (qr *queryResult) Pull() *model.TraceResult {
+ select {
+ case <-qr.ctx.Done():
+ return &model.TraceResult{
+ Error: errors.WithMessagef(qr.ctx.Err(), "interrupt: hit %d", qr.hit),
+ }
+ default:
+ }
+ if !qr.loaded {
+ if len(qr.data) == 0 {
+ return nil
+ }
+
+ cursorChan := make(chan int, len(qr.data))
+ for i := 0; i < len(qr.data); i++ {
+ go func(i int) {
+ select {
+ case <-qr.ctx.Done():
+ cursorChan <- i
+ return
+ default:
+ }
+ tmpBlock := generateBlock()
+ defer releaseBlock(tmpBlock)
+ if !qr.data[i].loadData(tmpBlock) {
+ cursorChan <- i
+ return
+ }
+ cursorChan <- -1
+ }(i)
+ }
+
+ blankCursorList := []int{}
+ for completed := 0; completed < len(qr.data); completed++ {
+ result := <-cursorChan
+ if result != -1 {
+ blankCursorList = append(blankCursorList, result)
+ }
+ }
+ select {
+ case <-qr.ctx.Done():
+ return &model.TraceResult{
+ Error: errors.WithMessagef(qr.ctx.Err(), "interrupt: blank/total=%d/%d", len(blankCursorList), len(qr.data)),
+ }
+ default:
+ }
+ sort.Slice(blankCursorList, func(i, j int) bool {
+ return blankCursorList[i] > blankCursorList[j]
+ })
+ for _, index := range blankCursorList {
+ qr.data = append(qr.data[:index], qr.data[index+1:]...)
+ }
+ qr.loaded = true
+ heap.Init(qr)
+ }
+ if len(qr.data) == 0 {
+ return nil
+ }
+ if len(qr.data) == 1 {
+ r := &model.TraceResult{}
+ bc := qr.data[0]
+ bc.copyAllTo(r, false)
+ qr.data = qr.data[:0]
+ releaseBlockCursor(bc)
+ return r
+ }
+ return qr.merge()
+}
+
+func (qr *queryResult) Release() {
+ for i, v := range qr.data {
+ releaseBlockCursor(v)
+ qr.data[i] = nil
+ }
+ qr.data = qr.data[:0]
+ for i := range qr.snapshots {
+ qr.snapshots[i].decRef()
+ }
+ qr.snapshots = qr.snapshots[:0]
+ for i := range qr.segments {
+ qr.segments[i].DecRef()
+ }
+}
+
+func (qr queryResult) Len() int {
+ return len(qr.data)
+}
+
+func (qr queryResult) Less(i, j int) bool {
+ return qr.data[i].bm.traceID < qr.data[j].bm.traceID
+}
+
+func (qr queryResult) Swap(i, j int) {
+ qr.data[i], qr.data[j] = qr.data[j], qr.data[i]
+}
+
+func (qr *queryResult) Push(x interface{}) {
+ qr.data = append(qr.data, x.(*blockCursor))
+}
+
+func (qr *queryResult) Pop() interface{} {
+ old := qr.data
+ n := len(old)
+ x := old[n-1]
+ qr.data = old[0 : n-1]
+ releaseBlockCursor(x)
+ return x
+}
Review Comment:
[nitpick] The Push and Pop methods use interface{} instead of type-safe generics. Consider using a type parameter or at least add type assertions with error handling to make the code more robust.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement traceID-based query [skywalking-banyandb]
Posted by "hanahmily (via GitHub)" <gi...@apache.org>.
hanahmily merged PR #735:
URL: https://github.com/apache/skywalking-banyandb/pull/735
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org