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/03 15:51:33 UTC
[PR] New storage engine for trace [skywalking-banyandb]
ButterBright opened a new pull request, #713:
URL: https://github.com/apache/skywalking-banyandb/pull/713
<!--
⚠️ 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 👆 ==== -->
- [ ] If this pull request closes/resolves/fixes an existing issue, replace the issue number. Fixes apache/skywalking#13329.
- [ ] 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] New storage engine for trace [skywalking-banyandb]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #713:
URL: https://github.com/apache/skywalking-banyandb/pull/713#issuecomment-3165011443
## [Codecov](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/713?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `50.65737%` with `2477 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 49.31%. Comparing base ([`cffa54c`](https://app.codecov.io/gh/apache/skywalking-banyandb/commit/cffa54c9f8eb937d1ca3596f74bcb1c6e28b0786?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`f41b962`](https://app.codecov.io/gh/apache/skywalking-banyandb/commit/f41b9622895a3f55e977fae284d30a9deef4839d?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 5 commits behind head on main.
| [Files with missing lines](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/713?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [banyand/trace/write\_standalone.go](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/713?src=pr&el=tree&filepath=banyand%2Ftrace%2Fwrite_standalone.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-YmFueWFuZC90cmFjZS93cml0ZV9zdGFuZGFsb25lLmdv) | 0.00% | [236 Missing :warning: ](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/713?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [banyand/trace/block.go](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/713?src=pr&el=tree&filepath=banyand%2Ftrace%2Fblock.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-YmFueWFuZC90cmFjZS9ibG9jay5nbw==) | 50.74% | [208 Missing and 23 partials :warning: ](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/713?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [banyand/trace/tstable.go](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/713?src=pr&el=tree&filepath=banyand%2Ftrace%2Ftstable.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-YmFueWFuZC90cmFjZS90c3RhYmxlLmdv) | 26.12% | [222 Missing and 7 partials :warning: ](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/713?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [banyand/trace/metrics.go](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/713?src=pr&el=tree&filepath=banyand%2Ftrace%2Fmetrics.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-YmFueWFuZC90cmFjZS9tZXRyaWNzLmdv) | 24.57% | [221 Missing :warning: ](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/713?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [banyand/trace/part\_iter.go](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/713?src=pr&el=tree&filepath=banyand%2Ftrace%2Fpart_iter.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-YmFueWFuZC90cmFjZS9wYXJ0X2l0ZXIuZ28=) | 28.78% | [184 Missing and 4 partials :warning: ](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/713?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [banyand/trace/merger.go](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/713?src=pr&el=tree&filepath=banyand%2Ftrace%2Fmerger.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-YmFueWFuZC90cmFjZS9tZXJnZXIuZ28=) | 53.35% | [119 Missing and 6 partials :warning: ](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/713?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [banyand/trace/write\_liaison.go](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/713?src=pr&el=tree&filepath=banyand%2Ftrace%2Fwrite_liaison.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-YmFueWFuZC90cmFjZS93cml0ZV9saWFpc29uLmdv) | 0.00% | [124 Missing :warning: ](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/713?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [banyand/trace/metadata.go](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/713?src=pr&el=tree&filepath=banyand%2Ftrace%2Fmetadata.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-YmFueWFuZC90cmFjZS9tZXRhZGF0YS5nbw==) | 0.00% | [121 Missing :warning: ](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/713?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [banyand/trace/tag.go](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/713?src=pr&el=tree&filepath=banyand%2Ftrace%2Ftag.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-YmFueWFuZC90cmFjZS90YWcuZ28=) | 67.44% | [70 Missing and 27 partials :warning: ](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/713?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [banyand/trace/query.go](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/713?src=pr&el=tree&filepath=banyand%2Ftrace%2Fquery.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-YmFueWFuZC90cmFjZS9xdWVyeS5nbw==) | 0.00% | [93 Missing :warning: ](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/713?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| ... and [25 more](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/713?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## main #713 +/- ##
===========================================
+ Coverage 35.97% 49.31% +13.34%
===========================================
Files 229 261 +32
Lines 34998 41076 +6078
===========================================
+ Hits 12589 20256 +7667
+ Misses 21353 19101 -2252
- Partials 1056 1719 +663
```
</details>
[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/713?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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] New storage engine for trace [skywalking-banyandb]
Posted by "hanahmily (via GitHub)" <gi...@apache.org>.
hanahmily merged PR #713:
URL: https://github.com/apache/skywalking-banyandb/pull/713
--
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] New storage engine for trace [skywalking-banyandb]
Posted by "ButterBright (via GitHub)" <gi...@apache.org>.
ButterBright commented on code in PR #713:
URL: https://github.com/apache/skywalking-banyandb/pull/713#discussion_r2278980567
##########
banyand/trace/block_metadata.go:
##########
@@ -0,0 +1,290 @@
+// 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 (
+ "bytes"
+ "fmt"
+ "sort"
+ "strings"
+
+ "github.com/apache/skywalking-banyandb/pkg/convert"
+ "github.com/apache/skywalking-banyandb/pkg/encoding"
+ pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
+ "github.com/apache/skywalking-banyandb/pkg/pool"
+ "github.com/apache/skywalking-banyandb/pkg/query/model"
+)
+
+type dataBlock struct {
+ offset uint64
+ size uint64
+}
+
+func (d *dataBlock) reset() {
+ d.offset = 0
+ d.size = 0
+}
+
+func (d *dataBlock) copyFrom(src *dataBlock) {
+ d.offset = src.offset
+ d.size = src.size
+}
+
+func (d *dataBlock) marshal(dst []byte) []byte {
+ dst = encoding.VarUint64ToBytes(dst, d.offset)
+ dst = encoding.VarUint64ToBytes(dst, d.size)
+ return dst
+}
+
+func (d *dataBlock) unmarshal(src []byte) []byte {
+ src, n := encoding.BytesToVarUint64(src)
+ d.offset = n
+
+ src, n = encoding.BytesToVarUint64(src)
+ d.size = n
+ return src
+}
+
+type blockMetadata struct {
+ tags map[string]*dataBlock
+ tagType map[string]pbv1.ValueType
+ spans *dataBlock
+ tagProjection *model.TagProjection
+ traceID string
+ timestamps timestampsMetadata
Review Comment:
We need it only for writing part metadata.
--
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] New storage engine for trace [skywalking-banyandb]
Posted by "hanahmily (via GitHub)" <gi...@apache.org>.
hanahmily commented on code in PR #713:
URL: https://github.com/apache/skywalking-banyandb/pull/713#discussion_r2280021624
##########
banyand/trace/part.go:
##########
@@ -0,0 +1,692 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package trace
+
+import (
+ "encoding/json"
+ "fmt"
+ "io"
+ "path"
+ "path/filepath"
+ "sync/atomic"
+
+ "github.com/apache/skywalking-banyandb/banyand/internal/storage"
+ "github.com/apache/skywalking-banyandb/pkg/bytes"
+ "github.com/apache/skywalking-banyandb/pkg/compress/zstd"
+ "github.com/apache/skywalking-banyandb/pkg/encoding"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ "github.com/apache/skywalking-banyandb/pkg/pool"
+)
+
+const (
+ // Streaming file names (without extensions).
+ tracePrimaryName = "primary"
+ traceMetaName = "meta"
+ traceSpansName = "spans"
+ traceTagsPrefix = "t:"
+ traceTagMetadataPrefix = "tm:"
+ traceTagFilterPrefix = "tf:"
+ metadataFilename = "metadata.json"
+ traceIDFilterFilename = "traceID.filter"
+ tagTypeFilename = "tag.type"
+ primaryFilename = tracePrimaryName + ".bin"
+ metaFilename = traceMetaName + ".bin"
+ spansFilename = traceSpansName + ".bin"
+ tagsMetadataFilenameExt = ".tm"
+ tagsFilenameExt = ".t"
+ tagsFilterFilenameExt = ".tf"
+)
+
+type part struct {
+ primary fs.Reader
+ spans fs.Reader
+ fileSystem fs.FileSystem
+ tagMetadata map[string]fs.Reader
+ tags map[string]fs.Reader
+ tagFilter map[string]fs.Reader
+ tagType tagType
+ traceIDFilter traceIDFilter
+ path string
+ primaryBlockMetadata []primaryBlockMetadata
+ partMetadata partMetadata
+}
+
+func (p *part) close() {
+ fs.MustClose(p.primary)
+ fs.MustClose(p.spans)
+ for _, t := range p.tags {
+ fs.MustClose(t)
+ }
+ for _, tm := range p.tagMetadata {
+ fs.MustClose(tm)
+ }
+ for _, tf := range p.tagFilter {
+ fs.MustClose(tf)
+ }
+}
+
+func (p *part) String() string {
+ return fmt.Sprintf("part %d", p.partMetadata.ID)
+}
+
+func openMemPart(mp *memPart) *part {
+ var p part
+ p.partMetadata = mp.partMetadata
+ p.tagType = mp.tagType
+ p.traceIDFilter = mp.traceIDFilter
+
+ p.primaryBlockMetadata = mustReadPrimaryBlockMetadata(p.primaryBlockMetadata[:0], &mp.meta)
+
+ // Open data files
+ p.primary = &mp.primary
+ p.spans = &mp.spans
+ if mp.tags != nil {
+ p.tags = make(map[string]fs.Reader)
+ p.tagMetadata = make(map[string]fs.Reader)
+ p.tagFilter = make(map[string]fs.Reader)
+ for name, t := range mp.tags {
+ p.tags[name] = t
+ p.tagMetadata[name] = mp.tagMetadata[name]
+ p.tagFilter[name] = mp.tagFilter[name]
+ }
+ }
+ return &p
+}
+
+type memPart struct {
+ tagMetadata map[string]*bytes.Buffer
+ tags map[string]*bytes.Buffer
+ tagFilter map[string]*bytes.Buffer
+ tagType tagType
+ traceIDFilter traceIDFilter
+ spans bytes.Buffer
+ meta bytes.Buffer
+ primary bytes.Buffer
+ partMetadata partMetadata
+}
+
+func (mp *memPart) mustCreateMemTagWriters(name string) (fs.Writer, fs.Writer, fs.Writer) {
+ if mp.tags == nil {
+ mp.tags = make(map[string]*bytes.Buffer)
+ mp.tagMetadata = make(map[string]*bytes.Buffer)
+ mp.tagFilter = make(map[string]*bytes.Buffer)
+ }
+ t, ok := mp.tags[name]
+ tm := mp.tagMetadata[name]
+ tf := mp.tagFilter[name]
+ if ok {
+ t.Reset()
+ tm.Reset()
+ tf.Reset()
+ return tm, t, tf
+ }
+ mp.tags[name] = &bytes.Buffer{}
+ mp.tagMetadata[name] = &bytes.Buffer{}
+ mp.tagFilter[name] = &bytes.Buffer{}
+ return mp.tagMetadata[name], mp.tags[name], mp.tagFilter[name]
+}
+
+func (mp *memPart) reset() {
+ mp.partMetadata.reset()
+ if mp.tagType == nil {
+ mp.tagType = make(tagType)
+ } else {
+ mp.tagType.reset()
+ }
+ mp.traceIDFilter.reset()
+ mp.meta.Reset()
+ mp.primary.Reset()
+ mp.spans.Reset()
+ if mp.tags != nil {
+ for k, t := range mp.tags {
+ t.Reset()
+ delete(mp.tags, k)
+ }
+ }
+ if mp.tagMetadata != nil {
+ for k, tm := range mp.tagMetadata {
+ tm.Reset()
+ delete(mp.tagMetadata, k)
+ }
+ }
+ if mp.tagFilter != nil {
+ for k, tf := range mp.tagFilter {
+ tf.Reset()
+ delete(mp.tagFilter, k)
+ }
+ }
+}
+
+// Marshal serializes the memPart to []byte for network transmission.
+// The format is:
+// - partMetadata (JSON)
+// - meta buffer length + data
+// - primary buffer length + data
+// - tags count + (name + buffer length + data) for each
+// - tagMetadata count + (name + buffer length + data) for each
+// - tagFilter count + (name + buffer length + data) for each.
+func (mp *memPart) Marshal() ([]byte, error) {
+ bb := bigValuePool.Generate()
+ defer bigValuePool.Release(bb)
+
+ // Marshal partMetadata as JSON
+ metadataBytes, err := json.Marshal(mp.partMetadata)
+ if err != nil {
+ return nil, fmt.Errorf("cannot marshal partMetadata: %w", err)
+ }
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(metadataBytes)))
+ bb.Buf = append(bb.Buf, metadataBytes...)
+
+ // Marshal tagType
+ var tagTypeBuf []byte
+ tagTypeBuf = mp.tagType.marshal(tagTypeBuf)
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(tagTypeBuf)))
+ bb.Buf = append(bb.Buf, tagTypeBuf...)
+
+ // Marshal traceIDFilter
+ if mp.traceIDFilter.filter != nil {
+ var filterBuf []byte
+ filterBuf = encodeBloomFilter(filterBuf, mp.traceIDFilter.filter)
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(filterBuf)))
+ bb.Buf = append(bb.Buf, filterBuf...)
+ } else {
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, 0)
+ }
+
+ // Marshal meta buffer
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(mp.meta.Buf)))
+ bb.Buf = append(bb.Buf, mp.meta.Buf...)
+
+ // Marshal primary buffer
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(mp.primary.Buf)))
+ bb.Buf = append(bb.Buf, mp.primary.Buf...)
+
+ // Marshal spans
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(mp.spans.Buf)))
+ bb.Buf = append(bb.Buf, mp.spans.Buf...)
+
+ // Marshal tags
+ if mp.tags == nil {
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, 0)
+ } else {
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(mp.tags)))
+ for name, t := range mp.tags {
+ // Write name length and name
+ nameBytes := []byte(name)
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(nameBytes)))
+ bb.Buf = append(bb.Buf, nameBytes...)
+ // Write buffer length and data
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(t.Buf)))
+ bb.Buf = append(bb.Buf, t.Buf...)
+ }
+ }
+
+ // Marshal tagMetadata
+ if mp.tagMetadata == nil {
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, 0)
+ } else {
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(mp.tagMetadata)))
+ for name, tm := range mp.tagMetadata {
+ // Write name length and name
+ nameBytes := []byte(name)
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(nameBytes)))
+ bb.Buf = append(bb.Buf, nameBytes...)
+ // Write buffer length and data
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(tm.Buf)))
+ bb.Buf = append(bb.Buf, tm.Buf...)
+ }
+ }
+
+ // Marshal tagFilter
+ if mp.tagFilter == nil {
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, 0)
+ } else {
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(mp.tagFilter)))
+ for name, tf := range mp.tagFilter {
+ // Write name length and name
+ nameBytes := []byte(name)
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(nameBytes)))
+ bb.Buf = append(bb.Buf, nameBytes...)
+ // Write buffer length and data
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(tf.Buf)))
+ bb.Buf = append(bb.Buf, tf.Buf...)
+ }
+ }
+
+ result := make([]byte, len(bb.Buf))
+ copy(result, bb.Buf)
+ return result, nil
+}
+
+// Unmarshal deserializes the memPart from []byte.
+func (mp *memPart) Unmarshal(data []byte) error {
+ mp.reset()
+
+ // Unmarshal partMetadata
+ tail, metadataLen := encoding.BytesToVarUint64(data)
+ if uint64(len(tail)) < metadataLen {
+ return fmt.Errorf("insufficient data for partMetadata: need %d bytes, have %d", metadataLen, len(tail))
+ }
+ metadataBytes := tail[:metadataLen]
+ tail = tail[metadataLen:]
+ if err := json.Unmarshal(metadataBytes, &mp.partMetadata); err != nil {
+ return fmt.Errorf("cannot unmarshal partMetadata: %w", err)
+ }
+
+ // Unmarshal tagType
+ tail, tagTypeLen := encoding.BytesToVarUint64(tail)
+ if uint64(len(tail)) < tagTypeLen {
+ return fmt.Errorf("insufficient data for tagType: need %d bytes, have %d", tagTypeLen, len(tail))
+ }
+ tagTypeBytes := tail[:tagTypeLen]
+ tail = tail[tagTypeLen:]
+ if err := mp.tagType.unmarshal(tagTypeBytes); err != nil {
+ return fmt.Errorf("cannot unmarshal tagType: %w", err)
+ }
+
+ // Unmarshal traceIDFilter
+ tail, filterLen := encoding.BytesToVarUint64(tail)
+ if filterLen > 0 {
+ if uint64(len(tail)) < filterLen {
+ return fmt.Errorf("insufficient data for traceIDFilter: need %d bytes, have %d", filterLen, len(tail))
+ }
+ filterBytes := tail[:filterLen]
+ tail = tail[filterLen:]
+ bf := generateBloomFilter()
+ defer releaseBloomFilter(bf)
+ mp.traceIDFilter.filter = decodeBloomFilter(filterBytes, bf)
+ } else {
+ mp.traceIDFilter.filter = nil
+ }
+
+ // Unmarshal meta buffer
+ tail, metaLen := encoding.BytesToVarUint64(tail)
+ if uint64(len(tail)) < metaLen {
+ return fmt.Errorf("insufficient data for meta buffer: need %d bytes, have %d", metaLen, len(tail))
+ }
+ mp.meta.Buf = append(mp.meta.Buf[:0], tail[:metaLen]...)
+ tail = tail[metaLen:]
+
+ // Unmarshal primary buffer
+ tail, primaryLen := encoding.BytesToVarUint64(tail)
+ if uint64(len(tail)) < primaryLen {
+ return fmt.Errorf("insufficient data for primary buffer: need %d bytes, have %d", primaryLen, len(tail))
+ }
+ mp.primary.Buf = append(mp.primary.Buf[:0], tail[:primaryLen]...)
+ tail = tail[primaryLen:]
+
+ // Unmarshal spans
+ tail, spansLen := encoding.BytesToVarUint64(tail)
+ if uint64(len(tail)) < spansLen {
+ return fmt.Errorf("insufficient data for spans buffer: need %d bytes, have %d", spansLen, len(tail))
+ }
+ mp.spans.Buf = append(mp.spans.Buf[:0], tail[:spansLen]...)
+ tail = tail[spansLen:]
+
+ var nameLen, bufLen uint64
+ // Unmarshal tags
+ tail, tagsCount := encoding.BytesToVarUint64(tail)
+ if tagsCount > 0 {
+ mp.tags = make(map[string]*bytes.Buffer)
+ for i := uint64(0); i < tagsCount; i++ {
+ // Read name length and name
+ tail, nameLen = encoding.BytesToVarUint64(tail)
+ if uint64(len(tail)) < nameLen {
+ return fmt.Errorf("insufficient data for tag name: need %d bytes, have %d", nameLen, len(tail))
+ }
+ name := string(tail[:nameLen])
+ tail = tail[nameLen:]
+
+ // Read buffer length and data
+ tail, bufLen = encoding.BytesToVarUint64(tail)
+ if uint64(len(tail)) < bufLen {
+ return fmt.Errorf("insufficient data for tag buffer: need %d bytes, have %d", bufLen, len(tail))
+ }
+ t := &bytes.Buffer{}
+ t.Buf = append(t.Buf[:0], tail[:bufLen]...)
+ mp.tags[name] = t
+ tail = tail[bufLen:]
+ }
+ }
+
+ // Unmarshal tagMetadata
+ tail, tagMetadataCount := encoding.BytesToVarUint64(tail)
+ if tagMetadataCount > 0 {
+ mp.tagMetadata = make(map[string]*bytes.Buffer)
+ for i := uint64(0); i < tagMetadataCount; i++ {
+ // Read name length and name
+ tail, nameLen = encoding.BytesToVarUint64(tail)
+ if uint64(len(tail)) < nameLen {
+ return fmt.Errorf("insufficient data for tag metadata name: need %d bytes, have %d", nameLen, len(tail))
+ }
+ name := string(tail[:nameLen])
+ tail = tail[nameLen:]
+
+ // Read buffer length and data
+ tail, bufLen = encoding.BytesToVarUint64(tail)
+ if uint64(len(tail)) < bufLen {
+ return fmt.Errorf("insufficient data for tag metadata buffer: need %d bytes, have %d", bufLen, len(tail))
+ }
+ tm := &bytes.Buffer{}
+ tm.Buf = append(tm.Buf[:0], tail[:bufLen]...)
+ mp.tagMetadata[name] = tm
+ tail = tail[bufLen:]
+ }
+ }
+
+ // Unmarshal tagFilter
+ tail, tagFilterCount := encoding.BytesToVarUint64(tail)
+ if tagFilterCount > 0 {
+ mp.tagFilter = make(map[string]*bytes.Buffer)
+ for i := uint64(0); i < tagFilterCount; i++ {
+ // Read name length and name
+ tail, nameLen = encoding.BytesToVarUint64(tail)
+ if uint64(len(tail)) < nameLen {
+ return fmt.Errorf("insufficient data for tag filter name: need %d bytes, have %d", nameLen, len(tail))
+ }
+ name := string(tail[:nameLen])
+ tail = tail[nameLen:]
+
+ // Read buffer length and data
+ tail, bufLen = encoding.BytesToVarUint64(tail)
+ if uint64(len(tail)) < bufLen {
+ return fmt.Errorf("insufficient data for tag filter buffer: need %d bytes, have %d", bufLen, len(tail))
+ }
+ tf := &bytes.Buffer{}
+ tf.Buf = append(tf.Buf[:0], tail[:bufLen]...)
+ mp.tagFilter[name] = tf
+ tail = tail[bufLen:]
+ }
+ }
+
+ if len(tail) > 0 {
+ return fmt.Errorf("unexpected trailing data: %d bytes", len(tail))
+ }
+
+ return nil
+}
+
+func (mp *memPart) mustInitFromPart(p *part) {
+ mp.reset()
+
+ // Copy part metadata, tagType and traceIDFilter
+ mp.partMetadata = p.partMetadata
+ mp.tagType = p.tagType
+ mp.traceIDFilter = p.traceIDFilter
+
+ // Read primary data
+ sr := p.primary.SequentialRead()
+ data, err := io.ReadAll(sr)
+ if err != nil {
+ logger.Panicf("cannot read primary data from %s: %s", p.primary.Path(), err)
+ }
+ fs.MustClose(sr)
+ mp.primary.Buf = append(mp.primary.Buf[:0], data...)
+
+ // Read spans data
+ sr = p.spans.SequentialRead()
+ data, err = io.ReadAll(sr)
+ if err != nil {
+ logger.Panicf("cannot read spans data from %s: %s", p.spans.Path(), err)
+ }
+ fs.MustClose(sr)
+ mp.spans.Buf = append(mp.spans.Buf[:0], data...)
+
+ // Read tag data
+ if p.tags != nil {
+ mp.tags = make(map[string]*bytes.Buffer)
+ mp.tagMetadata = make(map[string]*bytes.Buffer)
+ mp.tagFilter = make(map[string]*bytes.Buffer)
+
+ for name, reader := range p.tags {
+ sr = reader.SequentialRead()
+ data, err = io.ReadAll(sr)
+ if err != nil {
+ logger.Panicf("cannot read tag data from %s: %s", reader.Path(), err)
+ }
+ fs.MustClose(sr)
+
+ mp.tags[name] = &bytes.Buffer{}
+ mp.tags[name].Buf = append(mp.tags[name].Buf[:0], data...)
+ }
+
+ for name, reader := range p.tagMetadata {
+ sr = reader.SequentialRead()
+ data, err = io.ReadAll(sr)
+ if err != nil {
+ logger.Panicf("cannot read tag metadata from %s: %s", reader.Path(), err)
+ }
+ fs.MustClose(sr)
+
+ mp.tagMetadata[name] = &bytes.Buffer{}
+ mp.tagMetadata[name].Buf = append(mp.tagMetadata[name].Buf[:0], data...)
+ }
+
+ for name, reader := range p.tagFilter {
+ sr = reader.SequentialRead()
+ data, err = io.ReadAll(sr)
+ if err != nil {
+ logger.Panicf("cannot read tag filter from %s: %s", reader.Path(), err)
+ }
+ fs.MustClose(sr)
+
+ mp.tagFilter[name] = &bytes.Buffer{}
+ mp.tagFilter[name].Buf = append(mp.tagFilter[name].Buf[:0], data...)
+ }
+ }
+
+ // Encode primaryBlockMetadata to memPart.meta
+ // This is the reverse process of mustReadPrimaryBlockMetadata
+ bb := bigValuePool.Generate()
+ defer bigValuePool.Release(bb)
+
+ // Marshal all primaryBlockMetadata entries
+ for _, pbm := range p.primaryBlockMetadata {
+ bb.Buf = pbm.marshal(bb.Buf)
+ }
+
+ // Compress the marshaled data
+ mp.meta.Buf = zstd.Compress(mp.meta.Buf[:0], bb.Buf, 1)
+}
+
+func (mp *memPart) mustInitFromTraces(ts *traces) {
+ mp.reset()
+
+ if len(ts.traceIDs) == 0 {
+ return
+ }
+
+ bsw := generateBlockWriter()
+ bsw.MustInitForMemPart(mp)
+ for _, tid := range ts.traceIDs {
+ if len(tid) > int(bsw.traceIDLen) {
+ bsw.traceIDLen = uint32(len(tid))
+ }
+ }
+
+ var tidPrev string
+ uncompressedSpansSizeBytes := uint64(0)
+ var indexPrev int
+ for i := range ts.spans {
+ tid := ts.traceIDs[i]
+ if tidPrev == "" {
+ tidPrev = tid
+ }
+
+ if uncompressedSpansSizeBytes >= maxUncompressedSpanSize || tid != tidPrev {
Review Comment:
In some cases, the span volume may be significantly larger. For example, a loop with a high limit could generate over a hundred spans.
--
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] New storage engine for trace [skywalking-banyandb]
Posted by "ButterBright (via GitHub)" <gi...@apache.org>.
ButterBright commented on code in PR #713:
URL: https://github.com/apache/skywalking-banyandb/pull/713#discussion_r2279007938
##########
banyand/trace/part.go:
##########
@@ -0,0 +1,692 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package trace
+
+import (
+ "encoding/json"
+ "fmt"
+ "io"
+ "path"
+ "path/filepath"
+ "sync/atomic"
+
+ "github.com/apache/skywalking-banyandb/banyand/internal/storage"
+ "github.com/apache/skywalking-banyandb/pkg/bytes"
+ "github.com/apache/skywalking-banyandb/pkg/compress/zstd"
+ "github.com/apache/skywalking-banyandb/pkg/encoding"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ "github.com/apache/skywalking-banyandb/pkg/pool"
+)
+
+const (
+ // Streaming file names (without extensions).
+ tracePrimaryName = "primary"
+ traceMetaName = "meta"
+ traceSpansName = "spans"
+ traceTagsPrefix = "t:"
+ traceTagMetadataPrefix = "tm:"
+ traceTagFilterPrefix = "tf:"
+ metadataFilename = "metadata.json"
+ traceIDFilterFilename = "traceID.filter"
+ tagTypeFilename = "tag.type"
+ primaryFilename = tracePrimaryName + ".bin"
+ metaFilename = traceMetaName + ".bin"
+ spansFilename = traceSpansName + ".bin"
+ tagsMetadataFilenameExt = ".tm"
+ tagsFilenameExt = ".t"
+ tagsFilterFilenameExt = ".tf"
+)
+
+type part struct {
+ primary fs.Reader
+ spans fs.Reader
+ fileSystem fs.FileSystem
+ tagMetadata map[string]fs.Reader
+ tags map[string]fs.Reader
+ tagFilter map[string]fs.Reader
+ tagType tagType
+ traceIDFilter traceIDFilter
+ path string
+ primaryBlockMetadata []primaryBlockMetadata
+ partMetadata partMetadata
+}
+
+func (p *part) close() {
+ fs.MustClose(p.primary)
+ fs.MustClose(p.spans)
+ for _, t := range p.tags {
+ fs.MustClose(t)
+ }
+ for _, tm := range p.tagMetadata {
+ fs.MustClose(tm)
+ }
+ for _, tf := range p.tagFilter {
+ fs.MustClose(tf)
+ }
+}
+
+func (p *part) String() string {
+ return fmt.Sprintf("part %d", p.partMetadata.ID)
+}
+
+func openMemPart(mp *memPart) *part {
+ var p part
+ p.partMetadata = mp.partMetadata
+ p.tagType = mp.tagType
+ p.traceIDFilter = mp.traceIDFilter
+
+ p.primaryBlockMetadata = mustReadPrimaryBlockMetadata(p.primaryBlockMetadata[:0], &mp.meta)
+
+ // Open data files
+ p.primary = &mp.primary
+ p.spans = &mp.spans
+ if mp.tags != nil {
+ p.tags = make(map[string]fs.Reader)
+ p.tagMetadata = make(map[string]fs.Reader)
+ p.tagFilter = make(map[string]fs.Reader)
+ for name, t := range mp.tags {
+ p.tags[name] = t
+ p.tagMetadata[name] = mp.tagMetadata[name]
+ p.tagFilter[name] = mp.tagFilter[name]
+ }
+ }
+ return &p
+}
+
+type memPart struct {
+ tagMetadata map[string]*bytes.Buffer
+ tags map[string]*bytes.Buffer
+ tagFilter map[string]*bytes.Buffer
+ tagType tagType
+ traceIDFilter traceIDFilter
+ spans bytes.Buffer
+ meta bytes.Buffer
+ primary bytes.Buffer
+ partMetadata partMetadata
+}
+
+func (mp *memPart) mustCreateMemTagWriters(name string) (fs.Writer, fs.Writer, fs.Writer) {
+ if mp.tags == nil {
+ mp.tags = make(map[string]*bytes.Buffer)
+ mp.tagMetadata = make(map[string]*bytes.Buffer)
+ mp.tagFilter = make(map[string]*bytes.Buffer)
+ }
+ t, ok := mp.tags[name]
+ tm := mp.tagMetadata[name]
+ tf := mp.tagFilter[name]
+ if ok {
+ t.Reset()
+ tm.Reset()
+ tf.Reset()
+ return tm, t, tf
+ }
+ mp.tags[name] = &bytes.Buffer{}
+ mp.tagMetadata[name] = &bytes.Buffer{}
+ mp.tagFilter[name] = &bytes.Buffer{}
+ return mp.tagMetadata[name], mp.tags[name], mp.tagFilter[name]
+}
+
+func (mp *memPart) reset() {
+ mp.partMetadata.reset()
+ if mp.tagType == nil {
+ mp.tagType = make(tagType)
+ } else {
+ mp.tagType.reset()
+ }
+ mp.traceIDFilter.reset()
+ mp.meta.Reset()
+ mp.primary.Reset()
+ mp.spans.Reset()
+ if mp.tags != nil {
+ for k, t := range mp.tags {
+ t.Reset()
+ delete(mp.tags, k)
+ }
+ }
+ if mp.tagMetadata != nil {
+ for k, tm := range mp.tagMetadata {
+ tm.Reset()
+ delete(mp.tagMetadata, k)
+ }
+ }
+ if mp.tagFilter != nil {
+ for k, tf := range mp.tagFilter {
+ tf.Reset()
+ delete(mp.tagFilter, k)
+ }
+ }
+}
+
+// Marshal serializes the memPart to []byte for network transmission.
+// The format is:
+// - partMetadata (JSON)
+// - meta buffer length + data
+// - primary buffer length + data
+// - tags count + (name + buffer length + data) for each
+// - tagMetadata count + (name + buffer length + data) for each
+// - tagFilter count + (name + buffer length + data) for each.
+func (mp *memPart) Marshal() ([]byte, error) {
+ bb := bigValuePool.Generate()
+ defer bigValuePool.Release(bb)
+
+ // Marshal partMetadata as JSON
+ metadataBytes, err := json.Marshal(mp.partMetadata)
+ if err != nil {
+ return nil, fmt.Errorf("cannot marshal partMetadata: %w", err)
+ }
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(metadataBytes)))
+ bb.Buf = append(bb.Buf, metadataBytes...)
+
+ // Marshal tagType
+ var tagTypeBuf []byte
+ tagTypeBuf = mp.tagType.marshal(tagTypeBuf)
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(tagTypeBuf)))
+ bb.Buf = append(bb.Buf, tagTypeBuf...)
+
+ // Marshal traceIDFilter
+ if mp.traceIDFilter.filter != nil {
+ var filterBuf []byte
+ filterBuf = encodeBloomFilter(filterBuf, mp.traceIDFilter.filter)
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(filterBuf)))
+ bb.Buf = append(bb.Buf, filterBuf...)
+ } else {
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, 0)
+ }
+
+ // Marshal meta buffer
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(mp.meta.Buf)))
+ bb.Buf = append(bb.Buf, mp.meta.Buf...)
+
+ // Marshal primary buffer
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(mp.primary.Buf)))
+ bb.Buf = append(bb.Buf, mp.primary.Buf...)
+
+ // Marshal spans
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(mp.spans.Buf)))
+ bb.Buf = append(bb.Buf, mp.spans.Buf...)
+
+ // Marshal tags
+ if mp.tags == nil {
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, 0)
+ } else {
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(mp.tags)))
+ for name, t := range mp.tags {
+ // Write name length and name
+ nameBytes := []byte(name)
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(nameBytes)))
+ bb.Buf = append(bb.Buf, nameBytes...)
+ // Write buffer length and data
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(t.Buf)))
+ bb.Buf = append(bb.Buf, t.Buf...)
+ }
+ }
+
+ // Marshal tagMetadata
+ if mp.tagMetadata == nil {
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, 0)
+ } else {
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(mp.tagMetadata)))
+ for name, tm := range mp.tagMetadata {
+ // Write name length and name
+ nameBytes := []byte(name)
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(nameBytes)))
+ bb.Buf = append(bb.Buf, nameBytes...)
+ // Write buffer length and data
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(tm.Buf)))
+ bb.Buf = append(bb.Buf, tm.Buf...)
+ }
+ }
+
+ // Marshal tagFilter
+ if mp.tagFilter == nil {
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, 0)
+ } else {
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(mp.tagFilter)))
+ for name, tf := range mp.tagFilter {
+ // Write name length and name
+ nameBytes := []byte(name)
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(nameBytes)))
+ bb.Buf = append(bb.Buf, nameBytes...)
+ // Write buffer length and data
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(tf.Buf)))
+ bb.Buf = append(bb.Buf, tf.Buf...)
+ }
+ }
+
+ result := make([]byte, len(bb.Buf))
+ copy(result, bb.Buf)
+ return result, nil
+}
+
+// Unmarshal deserializes the memPart from []byte.
+func (mp *memPart) Unmarshal(data []byte) error {
+ mp.reset()
+
+ // Unmarshal partMetadata
+ tail, metadataLen := encoding.BytesToVarUint64(data)
+ if uint64(len(tail)) < metadataLen {
+ return fmt.Errorf("insufficient data for partMetadata: need %d bytes, have %d", metadataLen, len(tail))
+ }
+ metadataBytes := tail[:metadataLen]
+ tail = tail[metadataLen:]
+ if err := json.Unmarshal(metadataBytes, &mp.partMetadata); err != nil {
+ return fmt.Errorf("cannot unmarshal partMetadata: %w", err)
+ }
+
+ // Unmarshal tagType
+ tail, tagTypeLen := encoding.BytesToVarUint64(tail)
+ if uint64(len(tail)) < tagTypeLen {
+ return fmt.Errorf("insufficient data for tagType: need %d bytes, have %d", tagTypeLen, len(tail))
+ }
+ tagTypeBytes := tail[:tagTypeLen]
+ tail = tail[tagTypeLen:]
+ if err := mp.tagType.unmarshal(tagTypeBytes); err != nil {
+ return fmt.Errorf("cannot unmarshal tagType: %w", err)
+ }
+
+ // Unmarshal traceIDFilter
+ tail, filterLen := encoding.BytesToVarUint64(tail)
+ if filterLen > 0 {
+ if uint64(len(tail)) < filterLen {
+ return fmt.Errorf("insufficient data for traceIDFilter: need %d bytes, have %d", filterLen, len(tail))
+ }
+ filterBytes := tail[:filterLen]
+ tail = tail[filterLen:]
+ bf := generateBloomFilter()
+ defer releaseBloomFilter(bf)
+ mp.traceIDFilter.filter = decodeBloomFilter(filterBytes, bf)
+ } else {
+ mp.traceIDFilter.filter = nil
+ }
+
+ // Unmarshal meta buffer
+ tail, metaLen := encoding.BytesToVarUint64(tail)
+ if uint64(len(tail)) < metaLen {
+ return fmt.Errorf("insufficient data for meta buffer: need %d bytes, have %d", metaLen, len(tail))
+ }
+ mp.meta.Buf = append(mp.meta.Buf[:0], tail[:metaLen]...)
+ tail = tail[metaLen:]
+
+ // Unmarshal primary buffer
+ tail, primaryLen := encoding.BytesToVarUint64(tail)
+ if uint64(len(tail)) < primaryLen {
+ return fmt.Errorf("insufficient data for primary buffer: need %d bytes, have %d", primaryLen, len(tail))
+ }
+ mp.primary.Buf = append(mp.primary.Buf[:0], tail[:primaryLen]...)
+ tail = tail[primaryLen:]
+
+ // Unmarshal spans
+ tail, spansLen := encoding.BytesToVarUint64(tail)
+ if uint64(len(tail)) < spansLen {
+ return fmt.Errorf("insufficient data for spans buffer: need %d bytes, have %d", spansLen, len(tail))
+ }
+ mp.spans.Buf = append(mp.spans.Buf[:0], tail[:spansLen]...)
+ tail = tail[spansLen:]
+
+ var nameLen, bufLen uint64
+ // Unmarshal tags
+ tail, tagsCount := encoding.BytesToVarUint64(tail)
+ if tagsCount > 0 {
+ mp.tags = make(map[string]*bytes.Buffer)
+ for i := uint64(0); i < tagsCount; i++ {
+ // Read name length and name
+ tail, nameLen = encoding.BytesToVarUint64(tail)
+ if uint64(len(tail)) < nameLen {
+ return fmt.Errorf("insufficient data for tag name: need %d bytes, have %d", nameLen, len(tail))
+ }
+ name := string(tail[:nameLen])
+ tail = tail[nameLen:]
+
+ // Read buffer length and data
+ tail, bufLen = encoding.BytesToVarUint64(tail)
+ if uint64(len(tail)) < bufLen {
+ return fmt.Errorf("insufficient data for tag buffer: need %d bytes, have %d", bufLen, len(tail))
+ }
+ t := &bytes.Buffer{}
+ t.Buf = append(t.Buf[:0], tail[:bufLen]...)
+ mp.tags[name] = t
+ tail = tail[bufLen:]
+ }
+ }
+
+ // Unmarshal tagMetadata
+ tail, tagMetadataCount := encoding.BytesToVarUint64(tail)
+ if tagMetadataCount > 0 {
+ mp.tagMetadata = make(map[string]*bytes.Buffer)
+ for i := uint64(0); i < tagMetadataCount; i++ {
+ // Read name length and name
+ tail, nameLen = encoding.BytesToVarUint64(tail)
+ if uint64(len(tail)) < nameLen {
+ return fmt.Errorf("insufficient data for tag metadata name: need %d bytes, have %d", nameLen, len(tail))
+ }
+ name := string(tail[:nameLen])
+ tail = tail[nameLen:]
+
+ // Read buffer length and data
+ tail, bufLen = encoding.BytesToVarUint64(tail)
+ if uint64(len(tail)) < bufLen {
+ return fmt.Errorf("insufficient data for tag metadata buffer: need %d bytes, have %d", bufLen, len(tail))
+ }
+ tm := &bytes.Buffer{}
+ tm.Buf = append(tm.Buf[:0], tail[:bufLen]...)
+ mp.tagMetadata[name] = tm
+ tail = tail[bufLen:]
+ }
+ }
+
+ // Unmarshal tagFilter
+ tail, tagFilterCount := encoding.BytesToVarUint64(tail)
+ if tagFilterCount > 0 {
+ mp.tagFilter = make(map[string]*bytes.Buffer)
+ for i := uint64(0); i < tagFilterCount; i++ {
+ // Read name length and name
+ tail, nameLen = encoding.BytesToVarUint64(tail)
+ if uint64(len(tail)) < nameLen {
+ return fmt.Errorf("insufficient data for tag filter name: need %d bytes, have %d", nameLen, len(tail))
+ }
+ name := string(tail[:nameLen])
+ tail = tail[nameLen:]
+
+ // Read buffer length and data
+ tail, bufLen = encoding.BytesToVarUint64(tail)
+ if uint64(len(tail)) < bufLen {
+ return fmt.Errorf("insufficient data for tag filter buffer: need %d bytes, have %d", bufLen, len(tail))
+ }
+ tf := &bytes.Buffer{}
+ tf.Buf = append(tf.Buf[:0], tail[:bufLen]...)
+ mp.tagFilter[name] = tf
+ tail = tail[bufLen:]
+ }
+ }
+
+ if len(tail) > 0 {
+ return fmt.Errorf("unexpected trailing data: %d bytes", len(tail))
+ }
+
+ return nil
+}
+
+func (mp *memPart) mustInitFromPart(p *part) {
+ mp.reset()
+
+ // Copy part metadata, tagType and traceIDFilter
+ mp.partMetadata = p.partMetadata
+ mp.tagType = p.tagType
+ mp.traceIDFilter = p.traceIDFilter
+
+ // Read primary data
+ sr := p.primary.SequentialRead()
+ data, err := io.ReadAll(sr)
+ if err != nil {
+ logger.Panicf("cannot read primary data from %s: %s", p.primary.Path(), err)
+ }
+ fs.MustClose(sr)
+ mp.primary.Buf = append(mp.primary.Buf[:0], data...)
+
+ // Read spans data
+ sr = p.spans.SequentialRead()
+ data, err = io.ReadAll(sr)
+ if err != nil {
+ logger.Panicf("cannot read spans data from %s: %s", p.spans.Path(), err)
+ }
+ fs.MustClose(sr)
+ mp.spans.Buf = append(mp.spans.Buf[:0], data...)
+
+ // Read tag data
+ if p.tags != nil {
+ mp.tags = make(map[string]*bytes.Buffer)
+ mp.tagMetadata = make(map[string]*bytes.Buffer)
+ mp.tagFilter = make(map[string]*bytes.Buffer)
+
+ for name, reader := range p.tags {
+ sr = reader.SequentialRead()
+ data, err = io.ReadAll(sr)
+ if err != nil {
+ logger.Panicf("cannot read tag data from %s: %s", reader.Path(), err)
+ }
+ fs.MustClose(sr)
+
+ mp.tags[name] = &bytes.Buffer{}
+ mp.tags[name].Buf = append(mp.tags[name].Buf[:0], data...)
+ }
+
+ for name, reader := range p.tagMetadata {
+ sr = reader.SequentialRead()
+ data, err = io.ReadAll(sr)
+ if err != nil {
+ logger.Panicf("cannot read tag metadata from %s: %s", reader.Path(), err)
+ }
+ fs.MustClose(sr)
+
+ mp.tagMetadata[name] = &bytes.Buffer{}
+ mp.tagMetadata[name].Buf = append(mp.tagMetadata[name].Buf[:0], data...)
+ }
+
+ for name, reader := range p.tagFilter {
+ sr = reader.SequentialRead()
+ data, err = io.ReadAll(sr)
+ if err != nil {
+ logger.Panicf("cannot read tag filter from %s: %s", reader.Path(), err)
+ }
+ fs.MustClose(sr)
+
+ mp.tagFilter[name] = &bytes.Buffer{}
+ mp.tagFilter[name].Buf = append(mp.tagFilter[name].Buf[:0], data...)
+ }
+ }
+
+ // Encode primaryBlockMetadata to memPart.meta
+ // This is the reverse process of mustReadPrimaryBlockMetadata
+ bb := bigValuePool.Generate()
+ defer bigValuePool.Release(bb)
+
+ // Marshal all primaryBlockMetadata entries
+ for _, pbm := range p.primaryBlockMetadata {
+ bb.Buf = pbm.marshal(bb.Buf)
+ }
+
+ // Compress the marshaled data
+ mp.meta.Buf = zstd.Compress(mp.meta.Buf[:0], bb.Buf, 1)
+}
+
+func (mp *memPart) mustInitFromTraces(ts *traces) {
+ mp.reset()
+
+ if len(ts.traceIDs) == 0 {
+ return
+ }
+
+ bsw := generateBlockWriter()
+ bsw.MustInitForMemPart(mp)
+ for _, tid := range ts.traceIDs {
+ if len(tid) > int(bsw.traceIDLen) {
+ bsw.traceIDLen = uint32(len(tid))
+ }
+ }
+
+ var tidPrev string
+ uncompressedSpansSizeBytes := uint64(0)
+ var indexPrev int
+ for i := range ts.spans {
+ tid := ts.traceIDs[i]
+ if tidPrev == "" {
+ tidPrev = tid
+ }
+
+ if uncompressedSpansSizeBytes >= maxUncompressedSpanSize || tid != tidPrev {
Review Comment:
Is it necessary to keep the first condition in case there are anomalous traces, considering that each block is expected to be relatively small?
--
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] New storage engine for trace [skywalking-banyandb]
Posted by "ButterBright (via GitHub)" <gi...@apache.org>.
ButterBright commented on code in PR #713:
URL: https://github.com/apache/skywalking-banyandb/pull/713#discussion_r2280745294
##########
banyand/trace/part.go:
##########
@@ -0,0 +1,692 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package trace
+
+import (
+ "encoding/json"
+ "fmt"
+ "io"
+ "path"
+ "path/filepath"
+ "sync/atomic"
+
+ "github.com/apache/skywalking-banyandb/banyand/internal/storage"
+ "github.com/apache/skywalking-banyandb/pkg/bytes"
+ "github.com/apache/skywalking-banyandb/pkg/compress/zstd"
+ "github.com/apache/skywalking-banyandb/pkg/encoding"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ "github.com/apache/skywalking-banyandb/pkg/pool"
+)
+
+const (
+ // Streaming file names (without extensions).
+ tracePrimaryName = "primary"
+ traceMetaName = "meta"
+ traceSpansName = "spans"
+ traceTagsPrefix = "t:"
+ traceTagMetadataPrefix = "tm:"
+ traceTagFilterPrefix = "tf:"
+ metadataFilename = "metadata.json"
+ traceIDFilterFilename = "traceID.filter"
+ tagTypeFilename = "tag.type"
+ primaryFilename = tracePrimaryName + ".bin"
+ metaFilename = traceMetaName + ".bin"
+ spansFilename = traceSpansName + ".bin"
+ tagsMetadataFilenameExt = ".tm"
+ tagsFilenameExt = ".t"
+ tagsFilterFilenameExt = ".tf"
+)
+
+type part struct {
+ primary fs.Reader
+ spans fs.Reader
+ fileSystem fs.FileSystem
+ tagMetadata map[string]fs.Reader
+ tags map[string]fs.Reader
+ tagFilter map[string]fs.Reader
+ tagType tagType
+ traceIDFilter traceIDFilter
+ path string
+ primaryBlockMetadata []primaryBlockMetadata
+ partMetadata partMetadata
+}
+
+func (p *part) close() {
+ fs.MustClose(p.primary)
+ fs.MustClose(p.spans)
+ for _, t := range p.tags {
+ fs.MustClose(t)
+ }
+ for _, tm := range p.tagMetadata {
+ fs.MustClose(tm)
+ }
+ for _, tf := range p.tagFilter {
+ fs.MustClose(tf)
+ }
+}
+
+func (p *part) String() string {
+ return fmt.Sprintf("part %d", p.partMetadata.ID)
+}
+
+func openMemPart(mp *memPart) *part {
+ var p part
+ p.partMetadata = mp.partMetadata
+ p.tagType = mp.tagType
+ p.traceIDFilter = mp.traceIDFilter
+
+ p.primaryBlockMetadata = mustReadPrimaryBlockMetadata(p.primaryBlockMetadata[:0], &mp.meta)
+
+ // Open data files
+ p.primary = &mp.primary
+ p.spans = &mp.spans
+ if mp.tags != nil {
+ p.tags = make(map[string]fs.Reader)
+ p.tagMetadata = make(map[string]fs.Reader)
+ p.tagFilter = make(map[string]fs.Reader)
+ for name, t := range mp.tags {
+ p.tags[name] = t
+ p.tagMetadata[name] = mp.tagMetadata[name]
+ p.tagFilter[name] = mp.tagFilter[name]
+ }
+ }
+ return &p
+}
+
+type memPart struct {
+ tagMetadata map[string]*bytes.Buffer
+ tags map[string]*bytes.Buffer
+ tagFilter map[string]*bytes.Buffer
+ tagType tagType
+ traceIDFilter traceIDFilter
+ spans bytes.Buffer
+ meta bytes.Buffer
+ primary bytes.Buffer
+ partMetadata partMetadata
+}
+
+func (mp *memPart) mustCreateMemTagWriters(name string) (fs.Writer, fs.Writer, fs.Writer) {
+ if mp.tags == nil {
+ mp.tags = make(map[string]*bytes.Buffer)
+ mp.tagMetadata = make(map[string]*bytes.Buffer)
+ mp.tagFilter = make(map[string]*bytes.Buffer)
+ }
+ t, ok := mp.tags[name]
+ tm := mp.tagMetadata[name]
+ tf := mp.tagFilter[name]
+ if ok {
+ t.Reset()
+ tm.Reset()
+ tf.Reset()
+ return tm, t, tf
+ }
+ mp.tags[name] = &bytes.Buffer{}
+ mp.tagMetadata[name] = &bytes.Buffer{}
+ mp.tagFilter[name] = &bytes.Buffer{}
+ return mp.tagMetadata[name], mp.tags[name], mp.tagFilter[name]
+}
+
+func (mp *memPart) reset() {
+ mp.partMetadata.reset()
+ if mp.tagType == nil {
+ mp.tagType = make(tagType)
+ } else {
+ mp.tagType.reset()
+ }
+ mp.traceIDFilter.reset()
+ mp.meta.Reset()
+ mp.primary.Reset()
+ mp.spans.Reset()
+ if mp.tags != nil {
+ for k, t := range mp.tags {
+ t.Reset()
+ delete(mp.tags, k)
+ }
+ }
+ if mp.tagMetadata != nil {
+ for k, tm := range mp.tagMetadata {
+ tm.Reset()
+ delete(mp.tagMetadata, k)
+ }
+ }
+ if mp.tagFilter != nil {
+ for k, tf := range mp.tagFilter {
+ tf.Reset()
+ delete(mp.tagFilter, k)
+ }
+ }
+}
+
+// Marshal serializes the memPart to []byte for network transmission.
+// The format is:
+// - partMetadata (JSON)
+// - meta buffer length + data
+// - primary buffer length + data
+// - tags count + (name + buffer length + data) for each
+// - tagMetadata count + (name + buffer length + data) for each
+// - tagFilter count + (name + buffer length + data) for each.
+func (mp *memPart) Marshal() ([]byte, error) {
+ bb := bigValuePool.Generate()
+ defer bigValuePool.Release(bb)
+
+ // Marshal partMetadata as JSON
+ metadataBytes, err := json.Marshal(mp.partMetadata)
+ if err != nil {
+ return nil, fmt.Errorf("cannot marshal partMetadata: %w", err)
+ }
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(metadataBytes)))
+ bb.Buf = append(bb.Buf, metadataBytes...)
+
+ // Marshal tagType
+ var tagTypeBuf []byte
+ tagTypeBuf = mp.tagType.marshal(tagTypeBuf)
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(tagTypeBuf)))
+ bb.Buf = append(bb.Buf, tagTypeBuf...)
+
+ // Marshal traceIDFilter
+ if mp.traceIDFilter.filter != nil {
+ var filterBuf []byte
+ filterBuf = encodeBloomFilter(filterBuf, mp.traceIDFilter.filter)
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(filterBuf)))
+ bb.Buf = append(bb.Buf, filterBuf...)
+ } else {
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, 0)
+ }
+
+ // Marshal meta buffer
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(mp.meta.Buf)))
+ bb.Buf = append(bb.Buf, mp.meta.Buf...)
+
+ // Marshal primary buffer
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(mp.primary.Buf)))
+ bb.Buf = append(bb.Buf, mp.primary.Buf...)
+
+ // Marshal spans
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(mp.spans.Buf)))
+ bb.Buf = append(bb.Buf, mp.spans.Buf...)
+
+ // Marshal tags
+ if mp.tags == nil {
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, 0)
+ } else {
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(mp.tags)))
+ for name, t := range mp.tags {
+ // Write name length and name
+ nameBytes := []byte(name)
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(nameBytes)))
+ bb.Buf = append(bb.Buf, nameBytes...)
+ // Write buffer length and data
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(t.Buf)))
+ bb.Buf = append(bb.Buf, t.Buf...)
+ }
+ }
+
+ // Marshal tagMetadata
+ if mp.tagMetadata == nil {
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, 0)
+ } else {
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(mp.tagMetadata)))
+ for name, tm := range mp.tagMetadata {
+ // Write name length and name
+ nameBytes := []byte(name)
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(nameBytes)))
+ bb.Buf = append(bb.Buf, nameBytes...)
+ // Write buffer length and data
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(tm.Buf)))
+ bb.Buf = append(bb.Buf, tm.Buf...)
+ }
+ }
+
+ // Marshal tagFilter
+ if mp.tagFilter == nil {
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, 0)
+ } else {
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(mp.tagFilter)))
+ for name, tf := range mp.tagFilter {
+ // Write name length and name
+ nameBytes := []byte(name)
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(nameBytes)))
+ bb.Buf = append(bb.Buf, nameBytes...)
+ // Write buffer length and data
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(tf.Buf)))
+ bb.Buf = append(bb.Buf, tf.Buf...)
+ }
+ }
+
+ result := make([]byte, len(bb.Buf))
+ copy(result, bb.Buf)
+ return result, nil
+}
+
+// Unmarshal deserializes the memPart from []byte.
+func (mp *memPart) Unmarshal(data []byte) error {
+ mp.reset()
+
+ // Unmarshal partMetadata
+ tail, metadataLen := encoding.BytesToVarUint64(data)
+ if uint64(len(tail)) < metadataLen {
+ return fmt.Errorf("insufficient data for partMetadata: need %d bytes, have %d", metadataLen, len(tail))
+ }
+ metadataBytes := tail[:metadataLen]
+ tail = tail[metadataLen:]
+ if err := json.Unmarshal(metadataBytes, &mp.partMetadata); err != nil {
+ return fmt.Errorf("cannot unmarshal partMetadata: %w", err)
+ }
+
+ // Unmarshal tagType
+ tail, tagTypeLen := encoding.BytesToVarUint64(tail)
+ if uint64(len(tail)) < tagTypeLen {
+ return fmt.Errorf("insufficient data for tagType: need %d bytes, have %d", tagTypeLen, len(tail))
+ }
+ tagTypeBytes := tail[:tagTypeLen]
+ tail = tail[tagTypeLen:]
+ if err := mp.tagType.unmarshal(tagTypeBytes); err != nil {
+ return fmt.Errorf("cannot unmarshal tagType: %w", err)
+ }
+
+ // Unmarshal traceIDFilter
+ tail, filterLen := encoding.BytesToVarUint64(tail)
+ if filterLen > 0 {
+ if uint64(len(tail)) < filterLen {
+ return fmt.Errorf("insufficient data for traceIDFilter: need %d bytes, have %d", filterLen, len(tail))
+ }
+ filterBytes := tail[:filterLen]
+ tail = tail[filterLen:]
+ bf := generateBloomFilter()
+ defer releaseBloomFilter(bf)
+ mp.traceIDFilter.filter = decodeBloomFilter(filterBytes, bf)
+ } else {
+ mp.traceIDFilter.filter = nil
+ }
+
+ // Unmarshal meta buffer
+ tail, metaLen := encoding.BytesToVarUint64(tail)
+ if uint64(len(tail)) < metaLen {
+ return fmt.Errorf("insufficient data for meta buffer: need %d bytes, have %d", metaLen, len(tail))
+ }
+ mp.meta.Buf = append(mp.meta.Buf[:0], tail[:metaLen]...)
+ tail = tail[metaLen:]
+
+ // Unmarshal primary buffer
+ tail, primaryLen := encoding.BytesToVarUint64(tail)
+ if uint64(len(tail)) < primaryLen {
+ return fmt.Errorf("insufficient data for primary buffer: need %d bytes, have %d", primaryLen, len(tail))
+ }
+ mp.primary.Buf = append(mp.primary.Buf[:0], tail[:primaryLen]...)
+ tail = tail[primaryLen:]
+
+ // Unmarshal spans
+ tail, spansLen := encoding.BytesToVarUint64(tail)
+ if uint64(len(tail)) < spansLen {
+ return fmt.Errorf("insufficient data for spans buffer: need %d bytes, have %d", spansLen, len(tail))
+ }
+ mp.spans.Buf = append(mp.spans.Buf[:0], tail[:spansLen]...)
+ tail = tail[spansLen:]
+
+ var nameLen, bufLen uint64
+ // Unmarshal tags
+ tail, tagsCount := encoding.BytesToVarUint64(tail)
+ if tagsCount > 0 {
+ mp.tags = make(map[string]*bytes.Buffer)
+ for i := uint64(0); i < tagsCount; i++ {
+ // Read name length and name
+ tail, nameLen = encoding.BytesToVarUint64(tail)
+ if uint64(len(tail)) < nameLen {
+ return fmt.Errorf("insufficient data for tag name: need %d bytes, have %d", nameLen, len(tail))
+ }
+ name := string(tail[:nameLen])
+ tail = tail[nameLen:]
+
+ // Read buffer length and data
+ tail, bufLen = encoding.BytesToVarUint64(tail)
+ if uint64(len(tail)) < bufLen {
+ return fmt.Errorf("insufficient data for tag buffer: need %d bytes, have %d", bufLen, len(tail))
+ }
+ t := &bytes.Buffer{}
+ t.Buf = append(t.Buf[:0], tail[:bufLen]...)
+ mp.tags[name] = t
+ tail = tail[bufLen:]
+ }
+ }
+
+ // Unmarshal tagMetadata
+ tail, tagMetadataCount := encoding.BytesToVarUint64(tail)
+ if tagMetadataCount > 0 {
+ mp.tagMetadata = make(map[string]*bytes.Buffer)
+ for i := uint64(0); i < tagMetadataCount; i++ {
+ // Read name length and name
+ tail, nameLen = encoding.BytesToVarUint64(tail)
+ if uint64(len(tail)) < nameLen {
+ return fmt.Errorf("insufficient data for tag metadata name: need %d bytes, have %d", nameLen, len(tail))
+ }
+ name := string(tail[:nameLen])
+ tail = tail[nameLen:]
+
+ // Read buffer length and data
+ tail, bufLen = encoding.BytesToVarUint64(tail)
+ if uint64(len(tail)) < bufLen {
+ return fmt.Errorf("insufficient data for tag metadata buffer: need %d bytes, have %d", bufLen, len(tail))
+ }
+ tm := &bytes.Buffer{}
+ tm.Buf = append(tm.Buf[:0], tail[:bufLen]...)
+ mp.tagMetadata[name] = tm
+ tail = tail[bufLen:]
+ }
+ }
+
+ // Unmarshal tagFilter
+ tail, tagFilterCount := encoding.BytesToVarUint64(tail)
+ if tagFilterCount > 0 {
+ mp.tagFilter = make(map[string]*bytes.Buffer)
+ for i := uint64(0); i < tagFilterCount; i++ {
+ // Read name length and name
+ tail, nameLen = encoding.BytesToVarUint64(tail)
+ if uint64(len(tail)) < nameLen {
+ return fmt.Errorf("insufficient data for tag filter name: need %d bytes, have %d", nameLen, len(tail))
+ }
+ name := string(tail[:nameLen])
+ tail = tail[nameLen:]
+
+ // Read buffer length and data
+ tail, bufLen = encoding.BytesToVarUint64(tail)
+ if uint64(len(tail)) < bufLen {
+ return fmt.Errorf("insufficient data for tag filter buffer: need %d bytes, have %d", bufLen, len(tail))
+ }
+ tf := &bytes.Buffer{}
+ tf.Buf = append(tf.Buf[:0], tail[:bufLen]...)
+ mp.tagFilter[name] = tf
+ tail = tail[bufLen:]
+ }
+ }
+
+ if len(tail) > 0 {
+ return fmt.Errorf("unexpected trailing data: %d bytes", len(tail))
+ }
+
+ return nil
+}
+
+func (mp *memPart) mustInitFromPart(p *part) {
+ mp.reset()
+
+ // Copy part metadata, tagType and traceIDFilter
+ mp.partMetadata = p.partMetadata
+ mp.tagType = p.tagType
+ mp.traceIDFilter = p.traceIDFilter
+
+ // Read primary data
+ sr := p.primary.SequentialRead()
+ data, err := io.ReadAll(sr)
+ if err != nil {
+ logger.Panicf("cannot read primary data from %s: %s", p.primary.Path(), err)
+ }
+ fs.MustClose(sr)
+ mp.primary.Buf = append(mp.primary.Buf[:0], data...)
+
+ // Read spans data
+ sr = p.spans.SequentialRead()
+ data, err = io.ReadAll(sr)
+ if err != nil {
+ logger.Panicf("cannot read spans data from %s: %s", p.spans.Path(), err)
+ }
+ fs.MustClose(sr)
+ mp.spans.Buf = append(mp.spans.Buf[:0], data...)
+
+ // Read tag data
+ if p.tags != nil {
+ mp.tags = make(map[string]*bytes.Buffer)
+ mp.tagMetadata = make(map[string]*bytes.Buffer)
+ mp.tagFilter = make(map[string]*bytes.Buffer)
+
+ for name, reader := range p.tags {
+ sr = reader.SequentialRead()
+ data, err = io.ReadAll(sr)
+ if err != nil {
+ logger.Panicf("cannot read tag data from %s: %s", reader.Path(), err)
+ }
+ fs.MustClose(sr)
+
+ mp.tags[name] = &bytes.Buffer{}
+ mp.tags[name].Buf = append(mp.tags[name].Buf[:0], data...)
+ }
+
+ for name, reader := range p.tagMetadata {
+ sr = reader.SequentialRead()
+ data, err = io.ReadAll(sr)
+ if err != nil {
+ logger.Panicf("cannot read tag metadata from %s: %s", reader.Path(), err)
+ }
+ fs.MustClose(sr)
+
+ mp.tagMetadata[name] = &bytes.Buffer{}
+ mp.tagMetadata[name].Buf = append(mp.tagMetadata[name].Buf[:0], data...)
+ }
+
+ for name, reader := range p.tagFilter {
+ sr = reader.SequentialRead()
+ data, err = io.ReadAll(sr)
+ if err != nil {
+ logger.Panicf("cannot read tag filter from %s: %s", reader.Path(), err)
+ }
+ fs.MustClose(sr)
+
+ mp.tagFilter[name] = &bytes.Buffer{}
+ mp.tagFilter[name].Buf = append(mp.tagFilter[name].Buf[:0], data...)
+ }
+ }
+
+ // Encode primaryBlockMetadata to memPart.meta
+ // This is the reverse process of mustReadPrimaryBlockMetadata
+ bb := bigValuePool.Generate()
+ defer bigValuePool.Release(bb)
+
+ // Marshal all primaryBlockMetadata entries
+ for _, pbm := range p.primaryBlockMetadata {
+ bb.Buf = pbm.marshal(bb.Buf)
+ }
+
+ // Compress the marshaled data
+ mp.meta.Buf = zstd.Compress(mp.meta.Buf[:0], bb.Buf, 1)
+}
+
+func (mp *memPart) mustInitFromTraces(ts *traces) {
+ mp.reset()
+
+ if len(ts.traceIDs) == 0 {
+ return
+ }
+
+ bsw := generateBlockWriter()
+ bsw.MustInitForMemPart(mp)
+ for _, tid := range ts.traceIDs {
+ if len(tid) > int(bsw.traceIDLen) {
+ bsw.traceIDLen = uint32(len(tid))
+ }
+ }
+
+ var tidPrev string
+ uncompressedSpansSizeBytes := uint64(0)
+ var indexPrev int
+ for i := range ts.spans {
+ tid := ts.traceIDs[i]
+ if tidPrev == "" {
+ tidPrev = tid
+ }
+
+ if uncompressedSpansSizeBytes >= maxUncompressedSpanSize || tid != tidPrev {
Review Comment:
Okay, I will keep it then.
--
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] New storage engine for trace [skywalking-banyandb]
Posted by "ButterBright (via GitHub)" <gi...@apache.org>.
ButterBright commented on PR #713:
URL: https://github.com/apache/skywalking-banyandb/pull/713#issuecomment-3148510195
I will continue to add unit tests in this PR and implement the traceID-based query in the next one.
--
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] New storage engine for trace [skywalking-banyandb]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #713:
URL: https://github.com/apache/skywalking-banyandb/pull/713#discussion_r2250028532
##########
pkg/query/model/model.go:
##########
@@ -314,3 +314,41 @@ type StreamQueryResult interface {
Pull(context.Context) *StreamResult
Release()
}
+
+// TraceQueryOptions is the options of a trace query.
+type TraceQueryOptions struct {
+ Name string
+ TimeRange *timestamp.TimeRange
+ SkippingFilter index.Filter
+ Order *index.OrderBy
+ TagProjection *TagProjection
+ MaxTraceSize int
+}
+
+// Reset resets the StreamQueryOptions.
+func (t *TraceQueryOptions) Reset() {
+ t.Name = ""
+ t.TimeRange = nil
+ t.SkippingFilter = nil
+ t.Order = nil
+ t.TagProjection = nil
+ t.MaxTraceSize = 0
+}
+
+// CopyFrom copies the StreamQueryOptions from other to s.
Review Comment:
The comment incorrectly references 'StreamQueryOptions' but should reference 'TraceQueryOptions' since this method belongs to the TraceQueryOptions struct.
```suggestion
// CopyFrom copies the TraceQueryOptions from other to t.
```
##########
banyand/trace/part_iter.go:
##########
@@ -0,0 +1,402 @@
+// 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 (
+ "errors"
+ "fmt"
+ "io"
+ "sort"
+
+ "github.com/apache/skywalking-banyandb/pkg/bytes"
+ "github.com/apache/skywalking-banyandb/pkg/compress/zstd"
+ "github.com/apache/skywalking-banyandb/pkg/encoding"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
+ "github.com/apache/skywalking-banyandb/pkg/index"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
+ "github.com/apache/skywalking-banyandb/pkg/pool"
+ logicaltrace "github.com/apache/skywalking-banyandb/pkg/query/logical/trace"
+)
+
+type partIter struct {
+ err error
+ p *part
+ curBlock *blockMetadata
+ tids []string
+ blockFilter index.Filter
+ primaryBlockMetadata []primaryBlockMetadata
+ bms []blockMetadata
+ compressedPrimaryBuf []byte
+ primaryBuf []byte
+ tidIdx int
+ minTimestamp int64
+ maxTimestamp int64
+}
+
+func (pi *partIter) reset() {
+ pi.curBlock = nil
+ pi.p = nil
+ pi.tids = nil
+ pi.blockFilter = nil
+ pi.tidIdx = 0
+ pi.primaryBlockMetadata = nil
+ pi.bms = nil
+ pi.compressedPrimaryBuf = pi.compressedPrimaryBuf[:0]
+ pi.primaryBuf = pi.primaryBuf[:0]
+ pi.err = nil
+}
+
+func (pi *partIter) init(bma *blockMetadataArray, p *part, tids []string, minTimestamp, maxTimestamp int64, blockFilter index.Filter) {
+ pi.reset()
+ pi.curBlock = &blockMetadata{}
+ pi.p = p
+
+ pi.bms = bma.arr
+ pi.tids = tids
+ pi.blockFilter = blockFilter
+ pi.minTimestamp = minTimestamp
+ pi.maxTimestamp = maxTimestamp
+
+ pi.primaryBlockMetadata = p.primaryBlockMetadata
+
+ pi.nextSeriesID()
+}
+
+func (pi *partIter) nextBlock() bool {
+ for {
+ if pi.err != nil {
+ return false
+ }
+ if len(pi.bms) == 0 {
+ if !pi.loadNextBlockMetadata() {
+ return false
+ }
+ }
+ if pi.findBlock() {
+ return true
+ }
+ }
+}
+
+func (pi *partIter) error() error {
+ if errors.Is(pi.err, io.EOF) {
+ return nil
+ }
+ return pi.err
+}
+
+func (pi *partIter) nextSeriesID() bool {
+ if pi.tidIdx >= len(pi.tids) {
+ pi.err = io.EOF
+ return false
+ }
+ pi.curBlock.traceID = pi.tids[pi.tidIdx]
+ pi.tidIdx++
+ return true
+}
+
+func (pi *partIter) searchTargetTraceID(tid string) bool {
+ if pi.curBlock.traceID >= tid {
+ return true
+ }
+ if !pi.nextSeriesID() {
+ return false
+ }
+ if pi.curBlock.traceID >= tid {
+ return true
+ }
+ tids := pi.tids[pi.tidIdx:]
+ pi.tidIdx += sort.Search(len(tids), func(i int) bool {
+ return tid <= tids[i]
+ })
+ if pi.tidIdx >= len(pi.tids) {
+ pi.tidIdx = len(pi.tids)
+ pi.err = io.EOF
+ return false
+ }
+ pi.curBlock.traceID = pi.tids[pi.tidIdx]
+ pi.tidIdx++
+ return true
+}
+
+func (pi *partIter) loadNextBlockMetadata() bool {
+ for len(pi.primaryBlockMetadata) > 0 {
+ if !pi.searchTargetTraceID(pi.primaryBlockMetadata[0].traceID) {
+ return false
+ }
+ pi.primaryBlockMetadata = searchPBM(pi.primaryBlockMetadata, pi.curBlock.traceID)
+
+ pbm := &pi.primaryBlockMetadata[0]
+ pi.primaryBlockMetadata = pi.primaryBlockMetadata[1:]
+ if pi.curBlock.traceID < pbm.traceID {
+ logger.Panicf("invariant violation: pi.curBlock.traceID cannot be smaller than pbm.traceID; got %+v vs %+v", &pi.curBlock.traceID, &pbm.traceID)
+ }
+
+ var err error
+ pi.bms, err = pi.readPrimaryBlock(pi.bms[:0], pbm)
+ if err != nil {
+ pi.err = fmt.Errorf("cannot read primary block for part %q at offset %d with size %d: %w",
+ &pi.p.partMetadata, pbm.offset, pbm.size, err)
+ return false
+ }
+ return true
+ }
+ pi.err = io.EOF
+ return false
+}
+
+func searchPBM(pbmIndex []primaryBlockMetadata, tid string) []primaryBlockMetadata {
+ if tid < pbmIndex[0].traceID {
+ logger.Panicf("invariant violation: tid cannot be smaller than pbmIndex[0]; got %d vs %d", tid, &pbmIndex[0].traceID)
+ }
+
+ if tid == pbmIndex[0].traceID {
+ return pbmIndex
+ }
+
+ n := sort.Search(len(pbmIndex), func(i int) bool {
+ return tid <= pbmIndex[i].traceID
+ })
+ if n == 0 {
+ logger.Panicf("invariant violation: sort.Search returned 0 for tid > pbmIndex[0].traceID; tid=%+v; pbmIndex[0].traceID=%+v",
+ tid, &pbmIndex[0].traceID)
Review Comment:
[nitpick] Inconsistent format specifiers. The message uses '%+v' which is typically for structs, but for string values '%s' would be more appropriate.
```suggestion
logger.Panicf("invariant violation: sort.Search returned 0 for tid > pbmIndex[0].traceID; tid=%s; pbmIndex[0].traceID=%s",
tid, pbmIndex[0].traceID)
```
##########
pkg/query/model/model.go:
##########
@@ -314,3 +314,41 @@ type StreamQueryResult interface {
Pull(context.Context) *StreamResult
Release()
}
+
+// TraceQueryOptions is the options of a trace query.
+type TraceQueryOptions struct {
+ Name string
+ TimeRange *timestamp.TimeRange
+ SkippingFilter index.Filter
+ Order *index.OrderBy
+ TagProjection *TagProjection
+ MaxTraceSize int
+}
+
+// Reset resets the StreamQueryOptions.
Review Comment:
The comment incorrectly references 'StreamQueryOptions' but should reference 'TraceQueryOptions' since this method belongs to the TraceQueryOptions struct.
```suggestion
// Reset resets the TraceQueryOptions.
```
##########
banyand/trace/tracing.go:
##########
@@ -0,0 +1,50 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package trace
+
+import (
+ "context"
+ "fmt"
+ "time"
+
+ "github.com/dustin/go-humanize"
+)
+
+const (
+ partMetadataHeader = "MinTimestamp, MaxTimestamp, CompressionSize, UncompressedSize, TotalCount, BlocksCount"
+ blockHeader = "PartID, SeriesID, MinTimestamp, MaxTimestamp, Count, UncompressedSize"
+)
+
+func (pm *partMetadata) String() string {
+ minTimestamp := time.Unix(0, pm.MinTimestamp).Format(time.Stamp)
+ maxTimestamp := time.Unix(0, pm.MaxTimestamp).Format(time.Stamp)
+
+ return fmt.Sprintf("%s, %s, %s, %s, %s, %s",
+ minTimestamp, maxTimestamp, humanize.Bytes(pm.CompressedSizeBytes),
+ humanize.Bytes(pm.UncompressedSpanSizeBytes), humanize.Comma(int64(pm.TotalCount)),
+ humanize.Comma(int64(pm.BlocksCount)))
+}
+
+func (bc *blockCursor) String() string {
+ return fmt.Sprintf("%d, %s, %d, %s",
+ bc.p.partMetadata.ID, bc.bm.traceID, bc.bm.count, humanize.Bytes(bc.bm.spanSize))
+}
+
+func startBlockScanSpan(ctx context.Context, sids int, parts []*part) func() {
+ panic("unimplemented")
Review Comment:
This function contains only a panic statement indicating it's unimplemented. Consider removing it or implementing the functionality before merging to production.
```suggestion
// No-op implementation; replace with real logic as needed.
return func() {}
```
##########
banyand/trace/metadata.go:
##########
@@ -106,6 +107,146 @@ func (sr *schemaRepo) GetRemovalSegmentsTimeRange(_ string) *timestamp.TimeRange
panic("not implemented")
}
+func (sr *schemaRepo) OnInit(kinds []schema.Kind) (bool, []int64) {
+ if len(kinds) != 4 {
+ logger.Panicf("invalid kinds: %v", kinds)
+ return false, nil
+ }
+ _, revs := sr.Repository.Init(schema.KindStream)
+ return true, revs
+}
+
+func (sr *schemaRepo) OnAddOrUpdate(metadata schema.Metadata) {
+ switch metadata.Kind {
+ case schema.KindGroup:
+ g := metadata.Spec.(*commonv1.Group)
+ if g.Catalog != commonv1.Catalog_CATALOG_STREAM {
Review Comment:
This condition checks for CATALOG_STREAM but should check for CATALOG_TRACE since this is in the trace package and should handle trace-related groups.
```suggestion
if g.Catalog != commonv1.Catalog_CATALOG_TRACE {
```
##########
banyand/trace/part_iter.go:
##########
@@ -0,0 +1,402 @@
+// 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 (
+ "errors"
+ "fmt"
+ "io"
+ "sort"
+
+ "github.com/apache/skywalking-banyandb/pkg/bytes"
+ "github.com/apache/skywalking-banyandb/pkg/compress/zstd"
+ "github.com/apache/skywalking-banyandb/pkg/encoding"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
+ "github.com/apache/skywalking-banyandb/pkg/index"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
+ "github.com/apache/skywalking-banyandb/pkg/pool"
+ logicaltrace "github.com/apache/skywalking-banyandb/pkg/query/logical/trace"
+)
+
+type partIter struct {
+ err error
+ p *part
+ curBlock *blockMetadata
+ tids []string
+ blockFilter index.Filter
+ primaryBlockMetadata []primaryBlockMetadata
+ bms []blockMetadata
+ compressedPrimaryBuf []byte
+ primaryBuf []byte
+ tidIdx int
+ minTimestamp int64
+ maxTimestamp int64
+}
+
+func (pi *partIter) reset() {
+ pi.curBlock = nil
+ pi.p = nil
+ pi.tids = nil
+ pi.blockFilter = nil
+ pi.tidIdx = 0
+ pi.primaryBlockMetadata = nil
+ pi.bms = nil
+ pi.compressedPrimaryBuf = pi.compressedPrimaryBuf[:0]
+ pi.primaryBuf = pi.primaryBuf[:0]
+ pi.err = nil
+}
+
+func (pi *partIter) init(bma *blockMetadataArray, p *part, tids []string, minTimestamp, maxTimestamp int64, blockFilter index.Filter) {
+ pi.reset()
+ pi.curBlock = &blockMetadata{}
+ pi.p = p
+
+ pi.bms = bma.arr
+ pi.tids = tids
+ pi.blockFilter = blockFilter
+ pi.minTimestamp = minTimestamp
+ pi.maxTimestamp = maxTimestamp
+
+ pi.primaryBlockMetadata = p.primaryBlockMetadata
+
+ pi.nextSeriesID()
+}
+
+func (pi *partIter) nextBlock() bool {
+ for {
+ if pi.err != nil {
+ return false
+ }
+ if len(pi.bms) == 0 {
+ if !pi.loadNextBlockMetadata() {
+ return false
+ }
+ }
+ if pi.findBlock() {
+ return true
+ }
+ }
+}
+
+func (pi *partIter) error() error {
+ if errors.Is(pi.err, io.EOF) {
+ return nil
+ }
+ return pi.err
+}
+
+func (pi *partIter) nextSeriesID() bool {
+ if pi.tidIdx >= len(pi.tids) {
+ pi.err = io.EOF
+ return false
+ }
+ pi.curBlock.traceID = pi.tids[pi.tidIdx]
+ pi.tidIdx++
+ return true
+}
+
+func (pi *partIter) searchTargetTraceID(tid string) bool {
+ if pi.curBlock.traceID >= tid {
+ return true
+ }
+ if !pi.nextSeriesID() {
+ return false
+ }
+ if pi.curBlock.traceID >= tid {
+ return true
+ }
+ tids := pi.tids[pi.tidIdx:]
+ pi.tidIdx += sort.Search(len(tids), func(i int) bool {
+ return tid <= tids[i]
+ })
+ if pi.tidIdx >= len(pi.tids) {
+ pi.tidIdx = len(pi.tids)
+ pi.err = io.EOF
+ return false
+ }
+ pi.curBlock.traceID = pi.tids[pi.tidIdx]
+ pi.tidIdx++
+ return true
+}
+
+func (pi *partIter) loadNextBlockMetadata() bool {
+ for len(pi.primaryBlockMetadata) > 0 {
+ if !pi.searchTargetTraceID(pi.primaryBlockMetadata[0].traceID) {
+ return false
+ }
+ pi.primaryBlockMetadata = searchPBM(pi.primaryBlockMetadata, pi.curBlock.traceID)
+
+ pbm := &pi.primaryBlockMetadata[0]
+ pi.primaryBlockMetadata = pi.primaryBlockMetadata[1:]
+ if pi.curBlock.traceID < pbm.traceID {
+ logger.Panicf("invariant violation: pi.curBlock.traceID cannot be smaller than pbm.traceID; got %+v vs %+v", &pi.curBlock.traceID, &pbm.traceID)
+ }
+
+ var err error
+ pi.bms, err = pi.readPrimaryBlock(pi.bms[:0], pbm)
+ if err != nil {
+ pi.err = fmt.Errorf("cannot read primary block for part %q at offset %d with size %d: %w",
+ &pi.p.partMetadata, pbm.offset, pbm.size, err)
+ return false
+ }
+ return true
+ }
+ pi.err = io.EOF
+ return false
+}
+
+func searchPBM(pbmIndex []primaryBlockMetadata, tid string) []primaryBlockMetadata {
+ if tid < pbmIndex[0].traceID {
+ logger.Panicf("invariant violation: tid cannot be smaller than pbmIndex[0]; got %d vs %d", tid, &pbmIndex[0].traceID)
Review Comment:
The format specifiers '%d' are used for string values. Since 'tid' and 'pbmIndex[0].traceID' are strings, the format should use '%s' instead of '%d'.
```suggestion
logger.Panicf("invariant violation: tid cannot be smaller than pbmIndex[0]; got %s vs %s", tid, pbmIndex[0].traceID)
```
##########
banyand/trace/write_standalone.go:
##########
@@ -0,0 +1,377 @@
+// 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 (
+ "bytes"
+ "context"
+ "fmt"
+ "time"
+
+ "google.golang.org/protobuf/proto"
+
+ "github.com/apache/skywalking-banyandb/api/common"
+ 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/storage"
+ "github.com/apache/skywalking-banyandb/banyand/observability"
+ "github.com/apache/skywalking-banyandb/pkg/bus"
+ "github.com/apache/skywalking-banyandb/pkg/convert"
+ "github.com/apache/skywalking-banyandb/pkg/index"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
+ "github.com/apache/skywalking-banyandb/pkg/timestamp"
+)
+
+type writeCallback struct {
+ l *logger.Logger
+ schemaRepo *schemaRepo
+ maxDiskUsagePercent int
+}
+
+func setUpWriteCallback(l *logger.Logger, schemaRepo *schemaRepo, maxDiskUsagePercent int) bus.MessageListener {
+ if maxDiskUsagePercent > 100 {
+ maxDiskUsagePercent = 100
+ }
+ return &writeCallback{
+ l: l,
+ schemaRepo: schemaRepo,
+ maxDiskUsagePercent: maxDiskUsagePercent,
+ }
+}
+
+func (w *writeCallback) CheckHealth() *common.Error {
+ if w.maxDiskUsagePercent < 1 {
+ return common.NewErrorWithStatus(modelv1.Status_STATUS_DISK_FULL, "trace is readonly because \"trace-max-disk-usage-percent\" is 0")
+ }
+ diskPercent := observability.GetPathUsedPercent(w.schemaRepo.path)
+ if diskPercent < w.maxDiskUsagePercent {
+ return nil
+ }
+ w.l.Warn().Int("maxPercent", w.maxDiskUsagePercent).Int("diskPercent", diskPercent).Msg("disk usage is too high, stop writing")
+ return common.NewErrorWithStatus(modelv1.Status_STATUS_DISK_FULL, "disk usage is too high, stop writing")
+}
+
+func (w *writeCallback) handle(dst map[string]*tracesInGroup, writeEvent *tracev1.InternalWriteRequest,
+) (map[string]*tracesInGroup, error) {
+ stm, ok := w.schemaRepo.loadTrace(writeEvent.GetRequest().GetMetadata())
+ if !ok {
+ return nil, fmt.Errorf("cannot find trace definition: %s", writeEvent.GetRequest().GetMetadata())
+ }
+ idx, err := getTagIndex(stm, stm.schema.TimestampTagName)
+ if err != nil {
+ return nil, err
+ }
+ t := writeEvent.Request.Tags[idx].GetTimestamp().AsTime().Local()
+ if err := timestamp.Check(t); err != nil {
+ return nil, fmt.Errorf("invalid timestamp: %w", err)
+ }
+ ts := t.UnixNano()
+ eg, err := w.prepareTracesInGroup(dst, writeEvent, ts)
+ if err != nil {
+ return nil, err
+ }
+ et, err := w.prepareTracesInTable(eg, writeEvent, ts)
+ if err != nil {
+ return nil, err
+ }
+ err = processTraces(w.schemaRepo, et.traces, writeEvent)
+ if err != nil {
+ return nil, err
+ }
+ return dst, nil
+}
+
+func (w *writeCallback) prepareTracesInGroup(dst map[string]*tracesInGroup, writeEvent *tracev1.InternalWriteRequest, ts int64) (*tracesInGroup, error) {
+ gn := writeEvent.Request.Metadata.Group
+ tsdb, err := w.schemaRepo.loadTSDB(gn)
+ if err != nil {
+ return nil, fmt.Errorf("cannot load tsdb for group %s: %w", gn, err)
+ }
+
+ eg, ok := dst[gn]
+ if !ok {
+ eg = &tracesInGroup{
+ tsdb: tsdb,
+ tables: make([]*tracesInTable, 0),
+ segments: make([]storage.Segment[*tsTable, option], 0),
+ }
+ dst[gn] = eg
+ }
+ if eg.latestTS < ts {
+ eg.latestTS = ts
+ }
+ return eg, nil
+}
+
+func (w *writeCallback) prepareTracesInTable(eg *tracesInGroup, writeEvent *tracev1.InternalWriteRequest, ts int64) (*tracesInTable, error) {
+ var et *tracesInTable
+ for i := range eg.tables {
+ if eg.tables[i].timeRange.Contains(ts) {
+ et = eg.tables[i]
+ break
+ }
+ }
+
+ if et == nil {
+ var segment storage.Segment[*tsTable, option]
+ for _, seg := range eg.segments {
+ if seg.GetTimeRange().Contains(ts) {
+ segment = seg
+ break
+ }
+ }
+ if segment == nil {
+ var err error
+ segment, err = eg.tsdb.CreateSegmentIfNotExist(time.Unix(0, ts))
+ if err != nil {
+ return nil, fmt.Errorf("cannot create segment: %w", err)
+ }
+ eg.segments = append(eg.segments, segment)
+ }
+
+ shardID := common.ShardID(writeEvent.ShardId)
+ tstb, err := segment.CreateTSTableIfNotExist(shardID)
+ if err != nil {
+ return nil, fmt.Errorf("cannot create ts table: %w", err)
+ }
+
+ et = &tracesInTable{
+ timeRange: segment.GetTimeRange(),
+ tsTable: tstb,
+ traces: generateTraces(),
+ segment: segment,
+ }
+ et.traces.reset()
+ eg.tables = append(eg.tables, et)
+ }
+ return et, nil
+}
+
+func processTraces(schemaRepo *schemaRepo, traces *traces, writeEvent *tracev1.InternalWriteRequest) error {
+ req := writeEvent.Request
+ stm, ok := schemaRepo.loadTrace(req.GetMetadata())
+ if !ok {
+ return fmt.Errorf("cannot find trace definition: %s", req.GetMetadata())
+ }
+
+ idx, err := getTagIndex(stm, stm.schema.TraceIdTagName)
+ if err != nil {
+ return err
+ }
+ traces.traceIDs = append(traces.traceIDs, req.Tags[idx].GetStr().GetValue())
+ traces.spans = append(traces.spans, req.Span)
+
+ tLen := len(req.GetTags())
+ if tLen < 1 {
+ return fmt.Errorf("%s has no tag family", req)
+ }
+ if tLen > len(stm.schema.GetTags()) {
+ return fmt.Errorf("%s has more tag than %s", req.Metadata, stm.schema)
+ }
+
+ is := stm.indexSchema.Load().(indexSchema)
+ if len(is.indexRuleLocators) != len(stm.GetSchema().GetTags()) {
+ return fmt.Errorf("metadata crashed, tag rule length %d, tag length %d",
+ len(is.indexRuleLocators), len(stm.GetSchema().GetTags()))
+ }
+
+ tags := make([]*tagValue, 0, len(stm.schema.Tags))
+ tagSpecs := stm.GetSchema().GetTags()
+ for i := range tagSpecs {
+ tagSpec := tagSpecs[i]
+ if tagSpec.Name == stm.schema.TraceIdTagName {
+ continue
+ }
+ if tagSpec.Name == stm.schema.TimestampTagName {
+ traces.timestamps = append(traces.timestamps, req.Tags[i].GetTimestamp().AsTime().UnixNano())
+ }
Review Comment:
[nitpick] Missing newline after the closing brace. This creates inconsistent formatting compared to other conditional blocks in the same function.
```suggestion
}
```
##########
banyand/trace/metadata.go:
##########
@@ -106,6 +107,146 @@ func (sr *schemaRepo) GetRemovalSegmentsTimeRange(_ string) *timestamp.TimeRange
panic("not implemented")
}
+func (sr *schemaRepo) OnInit(kinds []schema.Kind) (bool, []int64) {
+ if len(kinds) != 4 {
+ logger.Panicf("invalid kinds: %v", kinds)
+ return false, nil
+ }
+ _, revs := sr.Repository.Init(schema.KindStream)
+ return true, revs
+}
+
+func (sr *schemaRepo) OnAddOrUpdate(metadata schema.Metadata) {
+ switch metadata.Kind {
+ case schema.KindGroup:
+ g := metadata.Spec.(*commonv1.Group)
+ if g.Catalog != commonv1.Catalog_CATALOG_STREAM {
+ return
+ }
+ if err := validate.GroupForStreamOrMeasure(g); err != nil {
+ sr.l.Warn().Err(err).Msg("group is ignored")
+ return
+ }
+ sr.SendMetadataEvent(resourceSchema.MetadataEvent{
+ Typ: resourceSchema.EventAddOrUpdate,
+ Kind: resourceSchema.EventKindGroup,
+ Metadata: g,
+ })
+ case schema.KindStream:
+ if err := validate.Stream(metadata.Spec.(*databasev1.Stream)); err != nil {
+ sr.l.Warn().Err(err).Msg("stream is ignored")
+ return
+ }
+ sr.SendMetadataEvent(resourceSchema.MetadataEvent{
+ Typ: resourceSchema.EventAddOrUpdate,
+ Kind: resourceSchema.EventKindResource,
+ Metadata: metadata.Spec.(*databasev1.Stream),
+ })
+ case schema.KindIndexRuleBinding:
+ if irb, ok := metadata.Spec.(*databasev1.IndexRuleBinding); ok {
+ if err := validate.IndexRuleBinding(irb); err != nil {
+ sr.l.Warn().Err(err).Msg("index rule binding is ignored")
+ return
+ }
+ if irb.GetSubject().Catalog == commonv1.Catalog_CATALOG_STREAM {
+ sr.SendMetadataEvent(resourceSchema.MetadataEvent{
+ Typ: resourceSchema.EventAddOrUpdate,
+ Kind: resourceSchema.EventKindIndexRuleBinding,
+ Metadata: irb,
+ })
+ }
+ }
+ case schema.KindIndexRule:
+ if ir, ok := metadata.Spec.(*databasev1.IndexRule); ok {
+ if err := validate.IndexRule(metadata.Spec.(*databasev1.IndexRule)); err != nil {
+ sr.l.Warn().Err(err).Msg("index rule is ignored")
+ return
+ }
+ sr.SendMetadataEvent(resourceSchema.MetadataEvent{
+ Typ: resourceSchema.EventAddOrUpdate,
+ Kind: resourceSchema.EventKindIndexRule,
+ Metadata: ir,
+ })
+ }
+ default:
+ }
+}
+
+func (sr *schemaRepo) OnDelete(metadata schema.Metadata) {
+ switch metadata.Kind {
+ case schema.KindGroup:
+ g := metadata.Spec.(*commonv1.Group)
+ if g.Catalog != commonv1.Catalog_CATALOG_STREAM {
+ return
+ }
+ sr.SendMetadataEvent(resourceSchema.MetadataEvent{
+ Typ: resourceSchema.EventDelete,
+ Kind: resourceSchema.EventKindGroup,
+ Metadata: g,
+ })
+ case schema.KindStream:
+ sr.SendMetadataEvent(resourceSchema.MetadataEvent{
+ Typ: resourceSchema.EventDelete,
+ Kind: resourceSchema.EventKindResource,
+ Metadata: metadata.Spec.(*databasev1.Stream),
+ })
+ case schema.KindIndexRuleBinding:
+ if binding, ok := metadata.Spec.(*databasev1.IndexRuleBinding); ok {
+ if binding.GetSubject().Catalog == commonv1.Catalog_CATALOG_MEASURE {
Review Comment:
This condition checks for CATALOG_MEASURE but should check for CATALOG_TRACE since this is in the trace package and should handle trace-related index rule bindings.
```suggestion
if binding.GetSubject().Catalog == commonv1.Catalog_CATALOG_TRACE {
```
--
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] New storage engine for trace [skywalking-banyandb]
Posted by "wu-sheng (via GitHub)" <gi...@apache.org>.
wu-sheng commented on PR #713:
URL: https://github.com/apache/skywalking-banyandb/pull/713#issuecomment-3172995035
@ButterBright Please fix CI.
--
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] New storage engine for trace [skywalking-banyandb]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #713:
URL: https://github.com/apache/skywalking-banyandb/pull/713#discussion_r2257541536
##########
pkg/query/logical/trace/index_filter.go:
##########
@@ -0,0 +1,131 @@
+// 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 (
+ "github.com/apache/skywalking-banyandb/api/common"
+ "github.com/apache/skywalking-banyandb/pkg/index"
+ "github.com/apache/skywalking-banyandb/pkg/index/posting"
+)
+
+var (
+ // ENode is an empty node.
+ ENode = new(emptyNode)
+ bList = new(bypassList)
+)
+
+type emptyNode struct{}
+
+func (an emptyNode) Execute(_ index.GetSearcher, _ common.SeriesID, _ *index.RangeOpts) (posting.List, posting.List, error) {
+ return bList, bList, nil
+}
+
+func (an emptyNode) String() string {
+ return "empty"
+}
+
+func (an emptyNode) ShouldSkip(_ index.FilterOp) (bool, error) {
+ return false, nil
+}
+
+type bypassList struct{}
+
+func (bl bypassList) Contains(_ uint64) bool {
+ // all items should be fetched
+ return true
+}
+
+func (bl bypassList) IsEmpty() bool {
+ return false
+}
+
+func (bl bypassList) Min() (uint64, error) {
+ panic("not invoked")
Review Comment:
The panic message "not invoked" is not helpful for debugging. Consider providing a more descriptive message like "Min() method should not be called on bypassList".
```suggestion
panic("Min() method should not be called on bypassList")
```
##########
banyand/trace/tracing.go:
##########
@@ -0,0 +1,50 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package trace
+
+import (
+ "context"
+ "fmt"
+ "time"
+
+ "github.com/dustin/go-humanize"
+)
+
+const (
+ partMetadataHeader = "MinTimestamp, MaxTimestamp, CompressionSize, UncompressedSize, TotalCount, BlocksCount"
+ blockHeader = "PartID, SeriesID, MinTimestamp, MaxTimestamp, Count, UncompressedSize"
+)
+
+func (pm *partMetadata) String() string {
+ minTimestamp := time.Unix(0, pm.MinTimestamp).Format(time.Stamp)
+ maxTimestamp := time.Unix(0, pm.MaxTimestamp).Format(time.Stamp)
+
+ return fmt.Sprintf("%s, %s, %s, %s, %s, %s",
+ minTimestamp, maxTimestamp, humanize.Bytes(pm.CompressedSizeBytes),
+ humanize.Bytes(pm.UncompressedSpanSizeBytes), humanize.Comma(int64(pm.TotalCount)),
+ humanize.Comma(int64(pm.BlocksCount)))
+}
+
+func (bc *blockCursor) String() string {
+ return fmt.Sprintf("%d, %s, %d, %s",
+ bc.p.partMetadata.ID, bc.bm.traceID, bc.bm.count, humanize.Bytes(bc.bm.uncompressedSpanSizeBytes))
+}
+
+func startBlockScanSpan(ctx context.Context, sids int, parts []*part) func() {
+ panic("unimplemented")
Review Comment:
The panic message "unimplemented" should be more descriptive, such as "startBlockScanSpan is not yet implemented".
```suggestion
panic("startBlockScanSpan is not yet implemented")
```
##########
banyand/trace/write_standalone.go:
##########
@@ -0,0 +1,326 @@
+// 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 (
+ "bytes"
+ "context"
+ "fmt"
+ "time"
+
+ "google.golang.org/protobuf/proto"
+
+ "github.com/apache/skywalking-banyandb/api/common"
+ 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/storage"
+ "github.com/apache/skywalking-banyandb/banyand/observability"
+ "github.com/apache/skywalking-banyandb/pkg/bus"
+ "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/timestamp"
+)
+
+type writeCallback struct {
+ l *logger.Logger
+ schemaRepo *schemaRepo
+ maxDiskUsagePercent int
+}
+
+func setUpWriteCallback(l *logger.Logger, schemaRepo *schemaRepo, maxDiskUsagePercent int) bus.MessageListener {
+ if maxDiskUsagePercent > 100 {
+ maxDiskUsagePercent = 100
+ }
+ return &writeCallback{
+ l: l,
+ schemaRepo: schemaRepo,
+ maxDiskUsagePercent: maxDiskUsagePercent,
+ }
+}
+
+func (w *writeCallback) CheckHealth() *common.Error {
+ if w.maxDiskUsagePercent < 1 {
+ return common.NewErrorWithStatus(modelv1.Status_STATUS_DISK_FULL, "trace is readonly because \"trace-max-disk-usage-percent\" is 0")
+ }
+ diskPercent := observability.GetPathUsedPercent(w.schemaRepo.path)
+ if diskPercent < w.maxDiskUsagePercent {
+ return nil
+ }
+ w.l.Warn().Int("maxPercent", w.maxDiskUsagePercent).Int("diskPercent", diskPercent).Msg("disk usage is too high, stop writing")
+ return common.NewErrorWithStatus(modelv1.Status_STATUS_DISK_FULL, "disk usage is too high, stop writing")
+}
+
+func (w *writeCallback) handle(dst map[string]*tracesInGroup, writeEvent *tracev1.InternalWriteRequest,
+) (map[string]*tracesInGroup, error) {
+ stm, ok := w.schemaRepo.loadTrace(writeEvent.GetRequest().GetMetadata())
+ if !ok {
+ return nil, fmt.Errorf("cannot find trace definition: %s", writeEvent.GetRequest().GetMetadata())
+ }
+ idx, err := getTagIndex(stm, stm.schema.TimestampTagName)
+ if err != nil {
+ return nil, err
+ }
+ t := writeEvent.Request.Tags[idx].GetTimestamp().AsTime().Local()
+ if err := timestamp.Check(t); err != nil {
+ return nil, fmt.Errorf("invalid timestamp: %w", err)
+ }
+ ts := t.UnixNano()
+ eg, err := w.prepareTracesInGroup(dst, writeEvent, ts)
+ if err != nil {
+ return nil, err
+ }
+ et, err := w.prepareTracesInTable(eg, writeEvent, ts)
+ if err != nil {
+ return nil, err
+ }
+ err = processTraces(w.schemaRepo, et.traces, writeEvent)
+ if err != nil {
+ return nil, err
+ }
+ return dst, nil
+}
+
+func (w *writeCallback) prepareTracesInGroup(dst map[string]*tracesInGroup, writeEvent *tracev1.InternalWriteRequest, ts int64) (*tracesInGroup, error) {
+ gn := writeEvent.Request.Metadata.Group
+ tsdb, err := w.schemaRepo.loadTSDB(gn)
+ if err != nil {
+ return nil, fmt.Errorf("cannot load tsdb for group %s: %w", gn, err)
+ }
+
+ eg, ok := dst[gn]
+ if !ok {
+ eg = &tracesInGroup{
+ tsdb: tsdb,
+ tables: make([]*tracesInTable, 0),
+ segments: make([]storage.Segment[*tsTable, option], 0),
+ }
+ dst[gn] = eg
+ }
+ if eg.latestTS < ts {
+ eg.latestTS = ts
+ }
+ return eg, nil
+}
+
+func (w *writeCallback) prepareTracesInTable(eg *tracesInGroup, writeEvent *tracev1.InternalWriteRequest, ts int64) (*tracesInTable, error) {
+ var et *tracesInTable
+ for i := range eg.tables {
+ if eg.tables[i].timeRange.Contains(ts) {
+ et = eg.tables[i]
+ break
+ }
+ }
+
+ if et == nil {
+ var segment storage.Segment[*tsTable, option]
+ for _, seg := range eg.segments {
+ if seg.GetTimeRange().Contains(ts) {
+ segment = seg
+ break
+ }
+ }
+ if segment == nil {
+ var err error
+ segment, err = eg.tsdb.CreateSegmentIfNotExist(time.Unix(0, ts))
+ if err != nil {
+ return nil, fmt.Errorf("cannot create segment: %w", err)
+ }
+ eg.segments = append(eg.segments, segment)
+ }
+
+ shardID := common.ShardID(writeEvent.ShardId)
+ tstb, err := segment.CreateTSTableIfNotExist(shardID)
+ if err != nil {
+ return nil, fmt.Errorf("cannot create ts table: %w", err)
+ }
+
+ et = &tracesInTable{
+ timeRange: segment.GetTimeRange(),
+ tsTable: tstb,
+ traces: generateTraces(),
+ segment: segment,
+ }
+ et.traces.reset()
+ eg.tables = append(eg.tables, et)
+ }
+ return et, nil
+}
+
+func processTraces(schemaRepo *schemaRepo, traces *traces, writeEvent *tracev1.InternalWriteRequest) error {
+ req := writeEvent.Request
+ stm, ok := schemaRepo.loadTrace(req.GetMetadata())
+ if !ok {
+ return fmt.Errorf("cannot find trace definition: %s", req.GetMetadata())
+ }
+
+ idx, err := getTagIndex(stm, stm.schema.TraceIdTagName)
+ if err != nil {
+ return err
+ }
+ traces.traceIDs = append(traces.traceIDs, req.Tags[idx].GetStr().GetValue())
+ traces.spans = append(traces.spans, req.Span)
+
+ tLen := len(req.GetTags())
+ if tLen < 1 {
+ return fmt.Errorf("%s has no tag family", req)
+ }
+ if tLen > len(stm.schema.GetTags()) {
+ return fmt.Errorf("%s has more tag than %s", req.Metadata, stm.schema)
+ }
+
+ is := stm.indexSchema.Load().(indexSchema)
+ if len(is.indexRuleLocators) != len(stm.GetSchema().GetTags()) {
+ return fmt.Errorf("metadata crashed, tag rule length %d, tag length %d",
+ len(is.indexRuleLocators), len(stm.GetSchema().GetTags()))
+ }
+
+ tags := make([]*tagValue, 0, len(stm.schema.Tags))
+ tagSpecs := stm.GetSchema().GetTags()
+ for i := range tagSpecs {
+ tagSpec := tagSpecs[i]
+ if tagSpec.Name == stm.schema.TraceIdTagName {
+ continue
+ }
+ if tagSpec.Name == stm.schema.TimestampTagName {
+ traces.timestamps = append(traces.timestamps, req.Tags[i].GetTimestamp().AsTime().UnixNano())
+ }
+
+ var tagValue *modelv1.TagValue
+ if len(req.Tags) <= i {
+ tagValue = pbv1.NullTagValue
+ } else {
+ tagValue = req.Tags[i]
+ }
+
+ indexed := false
+ if r, ok := is.indexRuleLocators[tagSpec.Name]; ok && tagValue != pbv1.NullTagValue {
+ if r.GetType() == databasev1.IndexRule_TYPE_SKIPPING {
+ indexed = true
+ }
+ }
+ tv := encodeTagValue(tagSpec.Name, tagSpec.Type, tagValue)
+ tv.indexed = indexed
+ tags = append(tags, tv)
+ }
+ traces.tags = append(traces.tags, tags)
+ return nil
+}
+
+func (w *writeCallback) Rev(_ context.Context, message bus.Message) (resp bus.Message) {
+ events, ok := message.Data().([]any)
+ if !ok {
+ w.l.Warn().Msg("invalid event data type")
+ return
+ }
+ if len(events) < 1 {
+ w.l.Warn().Msg("empty event")
+ return
+ }
+ groups := make(map[string]*tracesInGroup)
+ for i := range events {
+ var writeEvent *tracev1.InternalWriteRequest
+ switch e := events[i].(type) {
+ case *tracev1.InternalWriteRequest:
+ writeEvent = e
+ case []byte:
+ writeEvent = &tracev1.InternalWriteRequest{}
+ if err := proto.Unmarshal(e, writeEvent); err != nil {
+ w.l.Error().Err(err).RawJSON("written", e).Msg("fail to unmarshal event")
+ continue
+ }
+ default:
+ w.l.Warn().Msg("invalid event data type")
+ continue
+ }
+ var err error
+ if groups, err = w.handle(groups, writeEvent); err != nil {
+ w.l.Error().Err(err).Msg("cannot handle write event")
+ groups = make(map[string]*tracesInGroup)
+ continue
+ }
+ }
+ for i := range groups {
+ g := groups[i]
+ for j := range g.tables {
+ es := g.tables[j]
+ es.tsTable.mustAddTraces(es.traces)
+ releaseTraces(es.traces)
+ }
+ if len(g.segments) > 0 {
+ for _, segment := range g.segments {
+ segment.DecRef()
+ }
+ }
+ g.tsdb.Tick(g.latestTS)
+ }
+ return
+}
+
+func encodeTagValue(name string, tagType databasev1.TagType, tagVal *modelv1.TagValue) *tagValue {
+ tv := generateTagValue()
+ tv.tag = name
+ switch tagType {
+ case databasev1.TagType_TAG_TYPE_INT:
+ tv.valueType = pbv1.ValueTypeInt64
+ if tagVal.GetInt() != nil {
+ tv.value = convert.Int64ToBytes(tagVal.GetInt().GetValue())
+ }
+ case databasev1.TagType_TAG_TYPE_STRING:
+ tv.valueType = pbv1.ValueTypeStr
+ if tagVal.GetStr() != nil {
+ tv.value = convert.StringToBytes(tagVal.GetStr().GetValue())
+ }
+ case databasev1.TagType_TAG_TYPE_DATA_BINARY:
+ tv.valueType = pbv1.ValueTypeBinaryData
+ if tagVal.GetBinaryData() != nil {
+ tv.value = bytes.Clone(tagVal.GetBinaryData())
+ }
+ case databasev1.TagType_TAG_TYPE_INT_ARRAY:
+ tv.valueType = pbv1.ValueTypeInt64Arr
+ if tagVal.GetIntArray() == nil {
+ return tv
+ }
+ tv.valueArr = make([][]byte, len(tagVal.GetIntArray().Value))
+ for i := range tagVal.GetIntArray().Value {
+ tv.valueArr[i] = convert.Int64ToBytes(tagVal.GetIntArray().Value[i])
+ }
+ case databasev1.TagType_TAG_TYPE_STRING_ARRAY:
+ tv.valueType = pbv1.ValueTypeStrArr
+ if tagVal.GetStrArray() == nil {
+ return tv
+ }
+ tv.valueArr = make([][]byte, len(tagVal.GetStrArray().Value))
+ for i := range tagVal.GetStrArray().Value {
+ tv.valueArr[i] = []byte(tagVal.GetStrArray().Value[i])
+ }
+ default:
+ logger.Panicf("unsupported tag value type: %T", tagVal.GetValue())
Review Comment:
The error message should include the actual value type instead of just the format specifier. Consider using %v or providing more context about the supported types.
```suggestion
logger.Panicf("unsupported tag value type: %T (%v)", tagVal.GetValue(), tagVal.GetValue())
```
##########
pkg/query/logical/trace/index_filter.go:
##########
@@ -0,0 +1,131 @@
+// 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 (
+ "github.com/apache/skywalking-banyandb/api/common"
+ "github.com/apache/skywalking-banyandb/pkg/index"
+ "github.com/apache/skywalking-banyandb/pkg/index/posting"
+)
+
+var (
+ // ENode is an empty node.
+ ENode = new(emptyNode)
+ bList = new(bypassList)
+)
+
+type emptyNode struct{}
+
+func (an emptyNode) Execute(_ index.GetSearcher, _ common.SeriesID, _ *index.RangeOpts) (posting.List, posting.List, error) {
+ return bList, bList, nil
+}
+
+func (an emptyNode) String() string {
+ return "empty"
+}
+
+func (an emptyNode) ShouldSkip(_ index.FilterOp) (bool, error) {
+ return false, nil
+}
+
+type bypassList struct{}
+
+func (bl bypassList) Contains(_ uint64) bool {
+ // all items should be fetched
+ return true
+}
+
+func (bl bypassList) IsEmpty() bool {
+ return false
+}
+
+func (bl bypassList) Min() (uint64, error) {
+ panic("not invoked")
+}
+
+func (bl bypassList) Max() (uint64, error) {
+ panic("not invoked")
Review Comment:
The panic message "not invoked" is not helpful for debugging. Consider providing a more descriptive message like "Max() method should not be called on bypassList".
```suggestion
panic("Max() method should not be called on bypassList")
```
--
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] New storage engine for trace [skywalking-banyandb]
Posted by "wu-sheng (via GitHub)" <gi...@apache.org>.
wu-sheng commented on code in PR #713:
URL: https://github.com/apache/skywalking-banyandb/pull/713#discussion_r2280069643
##########
banyand/trace/part.go:
##########
@@ -0,0 +1,692 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package trace
+
+import (
+ "encoding/json"
+ "fmt"
+ "io"
+ "path"
+ "path/filepath"
+ "sync/atomic"
+
+ "github.com/apache/skywalking-banyandb/banyand/internal/storage"
+ "github.com/apache/skywalking-banyandb/pkg/bytes"
+ "github.com/apache/skywalking-banyandb/pkg/compress/zstd"
+ "github.com/apache/skywalking-banyandb/pkg/encoding"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ "github.com/apache/skywalking-banyandb/pkg/pool"
+)
+
+const (
+ // Streaming file names (without extensions).
+ tracePrimaryName = "primary"
+ traceMetaName = "meta"
+ traceSpansName = "spans"
+ traceTagsPrefix = "t:"
+ traceTagMetadataPrefix = "tm:"
+ traceTagFilterPrefix = "tf:"
+ metadataFilename = "metadata.json"
+ traceIDFilterFilename = "traceID.filter"
+ tagTypeFilename = "tag.type"
+ primaryFilename = tracePrimaryName + ".bin"
+ metaFilename = traceMetaName + ".bin"
+ spansFilename = traceSpansName + ".bin"
+ tagsMetadataFilenameExt = ".tm"
+ tagsFilenameExt = ".t"
+ tagsFilterFilenameExt = ".tf"
+)
+
+type part struct {
+ primary fs.Reader
+ spans fs.Reader
+ fileSystem fs.FileSystem
+ tagMetadata map[string]fs.Reader
+ tags map[string]fs.Reader
+ tagFilter map[string]fs.Reader
+ tagType tagType
+ traceIDFilter traceIDFilter
+ path string
+ primaryBlockMetadata []primaryBlockMetadata
+ partMetadata partMetadata
+}
+
+func (p *part) close() {
+ fs.MustClose(p.primary)
+ fs.MustClose(p.spans)
+ for _, t := range p.tags {
+ fs.MustClose(t)
+ }
+ for _, tm := range p.tagMetadata {
+ fs.MustClose(tm)
+ }
+ for _, tf := range p.tagFilter {
+ fs.MustClose(tf)
+ }
+}
+
+func (p *part) String() string {
+ return fmt.Sprintf("part %d", p.partMetadata.ID)
+}
+
+func openMemPart(mp *memPart) *part {
+ var p part
+ p.partMetadata = mp.partMetadata
+ p.tagType = mp.tagType
+ p.traceIDFilter = mp.traceIDFilter
+
+ p.primaryBlockMetadata = mustReadPrimaryBlockMetadata(p.primaryBlockMetadata[:0], &mp.meta)
+
+ // Open data files
+ p.primary = &mp.primary
+ p.spans = &mp.spans
+ if mp.tags != nil {
+ p.tags = make(map[string]fs.Reader)
+ p.tagMetadata = make(map[string]fs.Reader)
+ p.tagFilter = make(map[string]fs.Reader)
+ for name, t := range mp.tags {
+ p.tags[name] = t
+ p.tagMetadata[name] = mp.tagMetadata[name]
+ p.tagFilter[name] = mp.tagFilter[name]
+ }
+ }
+ return &p
+}
+
+type memPart struct {
+ tagMetadata map[string]*bytes.Buffer
+ tags map[string]*bytes.Buffer
+ tagFilter map[string]*bytes.Buffer
+ tagType tagType
+ traceIDFilter traceIDFilter
+ spans bytes.Buffer
+ meta bytes.Buffer
+ primary bytes.Buffer
+ partMetadata partMetadata
+}
+
+func (mp *memPart) mustCreateMemTagWriters(name string) (fs.Writer, fs.Writer, fs.Writer) {
+ if mp.tags == nil {
+ mp.tags = make(map[string]*bytes.Buffer)
+ mp.tagMetadata = make(map[string]*bytes.Buffer)
+ mp.tagFilter = make(map[string]*bytes.Buffer)
+ }
+ t, ok := mp.tags[name]
+ tm := mp.tagMetadata[name]
+ tf := mp.tagFilter[name]
+ if ok {
+ t.Reset()
+ tm.Reset()
+ tf.Reset()
+ return tm, t, tf
+ }
+ mp.tags[name] = &bytes.Buffer{}
+ mp.tagMetadata[name] = &bytes.Buffer{}
+ mp.tagFilter[name] = &bytes.Buffer{}
+ return mp.tagMetadata[name], mp.tags[name], mp.tagFilter[name]
+}
+
+func (mp *memPart) reset() {
+ mp.partMetadata.reset()
+ if mp.tagType == nil {
+ mp.tagType = make(tagType)
+ } else {
+ mp.tagType.reset()
+ }
+ mp.traceIDFilter.reset()
+ mp.meta.Reset()
+ mp.primary.Reset()
+ mp.spans.Reset()
+ if mp.tags != nil {
+ for k, t := range mp.tags {
+ t.Reset()
+ delete(mp.tags, k)
+ }
+ }
+ if mp.tagMetadata != nil {
+ for k, tm := range mp.tagMetadata {
+ tm.Reset()
+ delete(mp.tagMetadata, k)
+ }
+ }
+ if mp.tagFilter != nil {
+ for k, tf := range mp.tagFilter {
+ tf.Reset()
+ delete(mp.tagFilter, k)
+ }
+ }
+}
+
+// Marshal serializes the memPart to []byte for network transmission.
+// The format is:
+// - partMetadata (JSON)
+// - meta buffer length + data
+// - primary buffer length + data
+// - tags count + (name + buffer length + data) for each
+// - tagMetadata count + (name + buffer length + data) for each
+// - tagFilter count + (name + buffer length + data) for each.
+func (mp *memPart) Marshal() ([]byte, error) {
+ bb := bigValuePool.Generate()
+ defer bigValuePool.Release(bb)
+
+ // Marshal partMetadata as JSON
+ metadataBytes, err := json.Marshal(mp.partMetadata)
+ if err != nil {
+ return nil, fmt.Errorf("cannot marshal partMetadata: %w", err)
+ }
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(metadataBytes)))
+ bb.Buf = append(bb.Buf, metadataBytes...)
+
+ // Marshal tagType
+ var tagTypeBuf []byte
+ tagTypeBuf = mp.tagType.marshal(tagTypeBuf)
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(tagTypeBuf)))
+ bb.Buf = append(bb.Buf, tagTypeBuf...)
+
+ // Marshal traceIDFilter
+ if mp.traceIDFilter.filter != nil {
+ var filterBuf []byte
+ filterBuf = encodeBloomFilter(filterBuf, mp.traceIDFilter.filter)
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(filterBuf)))
+ bb.Buf = append(bb.Buf, filterBuf...)
+ } else {
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, 0)
+ }
+
+ // Marshal meta buffer
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(mp.meta.Buf)))
+ bb.Buf = append(bb.Buf, mp.meta.Buf...)
+
+ // Marshal primary buffer
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(mp.primary.Buf)))
+ bb.Buf = append(bb.Buf, mp.primary.Buf...)
+
+ // Marshal spans
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(mp.spans.Buf)))
+ bb.Buf = append(bb.Buf, mp.spans.Buf...)
+
+ // Marshal tags
+ if mp.tags == nil {
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, 0)
+ } else {
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(mp.tags)))
+ for name, t := range mp.tags {
+ // Write name length and name
+ nameBytes := []byte(name)
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(nameBytes)))
+ bb.Buf = append(bb.Buf, nameBytes...)
+ // Write buffer length and data
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(t.Buf)))
+ bb.Buf = append(bb.Buf, t.Buf...)
+ }
+ }
+
+ // Marshal tagMetadata
+ if mp.tagMetadata == nil {
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, 0)
+ } else {
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(mp.tagMetadata)))
+ for name, tm := range mp.tagMetadata {
+ // Write name length and name
+ nameBytes := []byte(name)
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(nameBytes)))
+ bb.Buf = append(bb.Buf, nameBytes...)
+ // Write buffer length and data
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(tm.Buf)))
+ bb.Buf = append(bb.Buf, tm.Buf...)
+ }
+ }
+
+ // Marshal tagFilter
+ if mp.tagFilter == nil {
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, 0)
+ } else {
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(mp.tagFilter)))
+ for name, tf := range mp.tagFilter {
+ // Write name length and name
+ nameBytes := []byte(name)
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(nameBytes)))
+ bb.Buf = append(bb.Buf, nameBytes...)
+ // Write buffer length and data
+ bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(tf.Buf)))
+ bb.Buf = append(bb.Buf, tf.Buf...)
+ }
+ }
+
+ result := make([]byte, len(bb.Buf))
+ copy(result, bb.Buf)
+ return result, nil
+}
+
+// Unmarshal deserializes the memPart from []byte.
+func (mp *memPart) Unmarshal(data []byte) error {
+ mp.reset()
+
+ // Unmarshal partMetadata
+ tail, metadataLen := encoding.BytesToVarUint64(data)
+ if uint64(len(tail)) < metadataLen {
+ return fmt.Errorf("insufficient data for partMetadata: need %d bytes, have %d", metadataLen, len(tail))
+ }
+ metadataBytes := tail[:metadataLen]
+ tail = tail[metadataLen:]
+ if err := json.Unmarshal(metadataBytes, &mp.partMetadata); err != nil {
+ return fmt.Errorf("cannot unmarshal partMetadata: %w", err)
+ }
+
+ // Unmarshal tagType
+ tail, tagTypeLen := encoding.BytesToVarUint64(tail)
+ if uint64(len(tail)) < tagTypeLen {
+ return fmt.Errorf("insufficient data for tagType: need %d bytes, have %d", tagTypeLen, len(tail))
+ }
+ tagTypeBytes := tail[:tagTypeLen]
+ tail = tail[tagTypeLen:]
+ if err := mp.tagType.unmarshal(tagTypeBytes); err != nil {
+ return fmt.Errorf("cannot unmarshal tagType: %w", err)
+ }
+
+ // Unmarshal traceIDFilter
+ tail, filterLen := encoding.BytesToVarUint64(tail)
+ if filterLen > 0 {
+ if uint64(len(tail)) < filterLen {
+ return fmt.Errorf("insufficient data for traceIDFilter: need %d bytes, have %d", filterLen, len(tail))
+ }
+ filterBytes := tail[:filterLen]
+ tail = tail[filterLen:]
+ bf := generateBloomFilter()
+ defer releaseBloomFilter(bf)
+ mp.traceIDFilter.filter = decodeBloomFilter(filterBytes, bf)
+ } else {
+ mp.traceIDFilter.filter = nil
+ }
+
+ // Unmarshal meta buffer
+ tail, metaLen := encoding.BytesToVarUint64(tail)
+ if uint64(len(tail)) < metaLen {
+ return fmt.Errorf("insufficient data for meta buffer: need %d bytes, have %d", metaLen, len(tail))
+ }
+ mp.meta.Buf = append(mp.meta.Buf[:0], tail[:metaLen]...)
+ tail = tail[metaLen:]
+
+ // Unmarshal primary buffer
+ tail, primaryLen := encoding.BytesToVarUint64(tail)
+ if uint64(len(tail)) < primaryLen {
+ return fmt.Errorf("insufficient data for primary buffer: need %d bytes, have %d", primaryLen, len(tail))
+ }
+ mp.primary.Buf = append(mp.primary.Buf[:0], tail[:primaryLen]...)
+ tail = tail[primaryLen:]
+
+ // Unmarshal spans
+ tail, spansLen := encoding.BytesToVarUint64(tail)
+ if uint64(len(tail)) < spansLen {
+ return fmt.Errorf("insufficient data for spans buffer: need %d bytes, have %d", spansLen, len(tail))
+ }
+ mp.spans.Buf = append(mp.spans.Buf[:0], tail[:spansLen]...)
+ tail = tail[spansLen:]
+
+ var nameLen, bufLen uint64
+ // Unmarshal tags
+ tail, tagsCount := encoding.BytesToVarUint64(tail)
+ if tagsCount > 0 {
+ mp.tags = make(map[string]*bytes.Buffer)
+ for i := uint64(0); i < tagsCount; i++ {
+ // Read name length and name
+ tail, nameLen = encoding.BytesToVarUint64(tail)
+ if uint64(len(tail)) < nameLen {
+ return fmt.Errorf("insufficient data for tag name: need %d bytes, have %d", nameLen, len(tail))
+ }
+ name := string(tail[:nameLen])
+ tail = tail[nameLen:]
+
+ // Read buffer length and data
+ tail, bufLen = encoding.BytesToVarUint64(tail)
+ if uint64(len(tail)) < bufLen {
+ return fmt.Errorf("insufficient data for tag buffer: need %d bytes, have %d", bufLen, len(tail))
+ }
+ t := &bytes.Buffer{}
+ t.Buf = append(t.Buf[:0], tail[:bufLen]...)
+ mp.tags[name] = t
+ tail = tail[bufLen:]
+ }
+ }
+
+ // Unmarshal tagMetadata
+ tail, tagMetadataCount := encoding.BytesToVarUint64(tail)
+ if tagMetadataCount > 0 {
+ mp.tagMetadata = make(map[string]*bytes.Buffer)
+ for i := uint64(0); i < tagMetadataCount; i++ {
+ // Read name length and name
+ tail, nameLen = encoding.BytesToVarUint64(tail)
+ if uint64(len(tail)) < nameLen {
+ return fmt.Errorf("insufficient data for tag metadata name: need %d bytes, have %d", nameLen, len(tail))
+ }
+ name := string(tail[:nameLen])
+ tail = tail[nameLen:]
+
+ // Read buffer length and data
+ tail, bufLen = encoding.BytesToVarUint64(tail)
+ if uint64(len(tail)) < bufLen {
+ return fmt.Errorf("insufficient data for tag metadata buffer: need %d bytes, have %d", bufLen, len(tail))
+ }
+ tm := &bytes.Buffer{}
+ tm.Buf = append(tm.Buf[:0], tail[:bufLen]...)
+ mp.tagMetadata[name] = tm
+ tail = tail[bufLen:]
+ }
+ }
+
+ // Unmarshal tagFilter
+ tail, tagFilterCount := encoding.BytesToVarUint64(tail)
+ if tagFilterCount > 0 {
+ mp.tagFilter = make(map[string]*bytes.Buffer)
+ for i := uint64(0); i < tagFilterCount; i++ {
+ // Read name length and name
+ tail, nameLen = encoding.BytesToVarUint64(tail)
+ if uint64(len(tail)) < nameLen {
+ return fmt.Errorf("insufficient data for tag filter name: need %d bytes, have %d", nameLen, len(tail))
+ }
+ name := string(tail[:nameLen])
+ tail = tail[nameLen:]
+
+ // Read buffer length and data
+ tail, bufLen = encoding.BytesToVarUint64(tail)
+ if uint64(len(tail)) < bufLen {
+ return fmt.Errorf("insufficient data for tag filter buffer: need %d bytes, have %d", bufLen, len(tail))
+ }
+ tf := &bytes.Buffer{}
+ tf.Buf = append(tf.Buf[:0], tail[:bufLen]...)
+ mp.tagFilter[name] = tf
+ tail = tail[bufLen:]
+ }
+ }
+
+ if len(tail) > 0 {
+ return fmt.Errorf("unexpected trailing data: %d bytes", len(tail))
+ }
+
+ return nil
+}
+
+func (mp *memPart) mustInitFromPart(p *part) {
+ mp.reset()
+
+ // Copy part metadata, tagType and traceIDFilter
+ mp.partMetadata = p.partMetadata
+ mp.tagType = p.tagType
+ mp.traceIDFilter = p.traceIDFilter
+
+ // Read primary data
+ sr := p.primary.SequentialRead()
+ data, err := io.ReadAll(sr)
+ if err != nil {
+ logger.Panicf("cannot read primary data from %s: %s", p.primary.Path(), err)
+ }
+ fs.MustClose(sr)
+ mp.primary.Buf = append(mp.primary.Buf[:0], data...)
+
+ // Read spans data
+ sr = p.spans.SequentialRead()
+ data, err = io.ReadAll(sr)
+ if err != nil {
+ logger.Panicf("cannot read spans data from %s: %s", p.spans.Path(), err)
+ }
+ fs.MustClose(sr)
+ mp.spans.Buf = append(mp.spans.Buf[:0], data...)
+
+ // Read tag data
+ if p.tags != nil {
+ mp.tags = make(map[string]*bytes.Buffer)
+ mp.tagMetadata = make(map[string]*bytes.Buffer)
+ mp.tagFilter = make(map[string]*bytes.Buffer)
+
+ for name, reader := range p.tags {
+ sr = reader.SequentialRead()
+ data, err = io.ReadAll(sr)
+ if err != nil {
+ logger.Panicf("cannot read tag data from %s: %s", reader.Path(), err)
+ }
+ fs.MustClose(sr)
+
+ mp.tags[name] = &bytes.Buffer{}
+ mp.tags[name].Buf = append(mp.tags[name].Buf[:0], data...)
+ }
+
+ for name, reader := range p.tagMetadata {
+ sr = reader.SequentialRead()
+ data, err = io.ReadAll(sr)
+ if err != nil {
+ logger.Panicf("cannot read tag metadata from %s: %s", reader.Path(), err)
+ }
+ fs.MustClose(sr)
+
+ mp.tagMetadata[name] = &bytes.Buffer{}
+ mp.tagMetadata[name].Buf = append(mp.tagMetadata[name].Buf[:0], data...)
+ }
+
+ for name, reader := range p.tagFilter {
+ sr = reader.SequentialRead()
+ data, err = io.ReadAll(sr)
+ if err != nil {
+ logger.Panicf("cannot read tag filter from %s: %s", reader.Path(), err)
+ }
+ fs.MustClose(sr)
+
+ mp.tagFilter[name] = &bytes.Buffer{}
+ mp.tagFilter[name].Buf = append(mp.tagFilter[name].Buf[:0], data...)
+ }
+ }
+
+ // Encode primaryBlockMetadata to memPart.meta
+ // This is the reverse process of mustReadPrimaryBlockMetadata
+ bb := bigValuePool.Generate()
+ defer bigValuePool.Release(bb)
+
+ // Marshal all primaryBlockMetadata entries
+ for _, pbm := range p.primaryBlockMetadata {
+ bb.Buf = pbm.marshal(bb.Buf)
+ }
+
+ // Compress the marshaled data
+ mp.meta.Buf = zstd.Compress(mp.meta.Buf[:0], bb.Buf, 1)
+}
+
+func (mp *memPart) mustInitFromTraces(ts *traces) {
+ mp.reset()
+
+ if len(ts.traceIDs) == 0 {
+ return
+ }
+
+ bsw := generateBlockWriter()
+ bsw.MustInitForMemPart(mp)
+ for _, tid := range ts.traceIDs {
+ if len(tid) > int(bsw.traceIDLen) {
+ bsw.traceIDLen = uint32(len(tid))
+ }
+ }
+
+ var tidPrev string
+ uncompressedSpansSizeBytes := uint64(0)
+ var indexPrev int
+ for i := range ts.spans {
+ tid := ts.traceIDs[i]
+ if tidPrev == "" {
+ tidPrev = tid
+ }
+
+ if uncompressedSpansSizeBytes >= maxUncompressedSpanSize || tid != tidPrev {
Review Comment:
We saw 10k level spans in zipkin ecosystem.
--
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] New storage engine for trace [skywalking-banyandb]
Posted by "hanahmily (via GitHub)" <gi...@apache.org>.
hanahmily commented on code in PR #713:
URL: https://github.com/apache/skywalking-banyandb/pull/713#discussion_r2269450179
##########
banyand/trace/block.go:
##########
@@ -0,0 +1,659 @@
+// 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 (
+ "bytes"
+ "fmt"
+ "slices"
+ "sort"
+
+ modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
+ pkgbytes "github.com/apache/skywalking-banyandb/pkg/bytes"
+ "github.com/apache/skywalking-banyandb/pkg/encoding"
+ "github.com/apache/skywalking-banyandb/pkg/filter"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
+ "github.com/apache/skywalking-banyandb/pkg/pool"
+ "github.com/apache/skywalking-banyandb/pkg/query/model"
+)
+
+type block struct {
+ spans [][]byte
+ tags []tag
+ timestamps []int64
Review Comment:
```suggestion
minTS int64
maxTS int64
```
##########
banyand/trace/block_metadata.go:
##########
@@ -0,0 +1,290 @@
+// 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 (
+ "bytes"
+ "fmt"
+ "sort"
+ "strings"
+
+ "github.com/apache/skywalking-banyandb/pkg/convert"
+ "github.com/apache/skywalking-banyandb/pkg/encoding"
+ pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
+ "github.com/apache/skywalking-banyandb/pkg/pool"
+ "github.com/apache/skywalking-banyandb/pkg/query/model"
+)
+
+type dataBlock struct {
+ offset uint64
+ size uint64
+}
+
+func (d *dataBlock) reset() {
+ d.offset = 0
+ d.size = 0
+}
+
+func (d *dataBlock) copyFrom(src *dataBlock) {
+ d.offset = src.offset
+ d.size = src.size
+}
+
+func (d *dataBlock) marshal(dst []byte) []byte {
+ dst = encoding.VarUint64ToBytes(dst, d.offset)
+ dst = encoding.VarUint64ToBytes(dst, d.size)
+ return dst
+}
+
+func (d *dataBlock) unmarshal(src []byte) []byte {
+ src, n := encoding.BytesToVarUint64(src)
+ d.offset = n
+
+ src, n = encoding.BytesToVarUint64(src)
+ d.size = n
+ return src
+}
+
+type blockMetadata struct {
+ tags map[string]*dataBlock
+ tagType map[string]pbv1.ValueType
+ spans *dataBlock
+ tagProjection *model.TagProjection
+ traceID string
Review Comment:
The spans in a block could come from different traces. It can not contain a single traceID here.
##########
banyand/trace/tag.go:
##########
@@ -0,0 +1,392 @@
+// 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 (
+ "github.com/apache/skywalking-banyandb/pkg/bytes"
+ "github.com/apache/skywalking-banyandb/pkg/convert"
+ "github.com/apache/skywalking-banyandb/pkg/encoding"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
+ "github.com/apache/skywalking-banyandb/pkg/pool"
+)
+
+func generateInt64Slice(length int) *[]int64 {
+ v := int64SlicePool.Get()
+ if v == nil {
+ s := make([]int64, length)
+ return &s
+ }
+ if cap(*v) < length {
+ *v = make([]int64, length)
+ } else {
+ *v = (*v)[:length]
+ }
+ return v
+}
+
+func releaseInt64Slice(int64Slice *[]int64) {
+ *int64Slice = (*int64Slice)[:0]
+ int64SlicePool.Put(int64Slice)
+}
+
+func generateFloat64Slice(length int) *[]float64 {
+ v := float64SlicePool.Get()
+ if v == nil {
+ s := make([]float64, length)
+ return &s
+ }
+ if cap(*v) < length {
+ *v = make([]float64, length)
+ } else {
+ *v = (*v)[:length]
+ }
+ return v
+}
+
+func releaseFloat64Slice(float64Slice *[]float64) {
+ *float64Slice = (*float64Slice)[:0]
+ float64SlicePool.Put(float64Slice)
+}
+
+func generateDictionary() *encoding.Dictionary {
+ v := dictionaryPool.Get()
+ if v == nil {
+ return &encoding.Dictionary{}
+ }
+ return v
+}
+
+func releaseDictionary(d *encoding.Dictionary) {
+ d.Reset()
+ dictionaryPool.Put(d)
+}
+
+var (
+ int64SlicePool = pool.Register[*[]int64]("trace-int64Slice")
+ float64SlicePool = pool.Register[*[]float64]("trace-float64Slice")
+ dictionaryPool = pool.Register[*encoding.Dictionary]("trace-dictionary")
+)
+
+type tag struct {
+ tagFilter
Review Comment:
```suggestion
```
The traceID storage doesn't support the skipping index
##########
banyand/trace/block_metadata.go:
##########
@@ -0,0 +1,290 @@
+// 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 (
+ "bytes"
+ "fmt"
+ "sort"
+ "strings"
+
+ "github.com/apache/skywalking-banyandb/pkg/convert"
+ "github.com/apache/skywalking-banyandb/pkg/encoding"
+ pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
+ "github.com/apache/skywalking-banyandb/pkg/pool"
+ "github.com/apache/skywalking-banyandb/pkg/query/model"
+)
+
+type dataBlock struct {
+ offset uint64
+ size uint64
+}
+
+func (d *dataBlock) reset() {
+ d.offset = 0
+ d.size = 0
+}
+
+func (d *dataBlock) copyFrom(src *dataBlock) {
+ d.offset = src.offset
+ d.size = src.size
+}
+
+func (d *dataBlock) marshal(dst []byte) []byte {
+ dst = encoding.VarUint64ToBytes(dst, d.offset)
+ dst = encoding.VarUint64ToBytes(dst, d.size)
+ return dst
+}
+
+func (d *dataBlock) unmarshal(src []byte) []byte {
+ src, n := encoding.BytesToVarUint64(src)
+ d.offset = n
+
+ src, n = encoding.BytesToVarUint64(src)
+ d.size = n
+ return src
+}
+
+type blockMetadata struct {
+ tags map[string]*dataBlock
+ tagType map[string]pbv1.ValueType
+ spans *dataBlock
+ tagProjection *model.TagProjection
+ traceID string
+ timestamps timestampsMetadata
Review Comment:
```suggestion
```
traceID storage doesn't need timestamps
##########
banyand/trace/block_metadata.go:
##########
@@ -0,0 +1,290 @@
+// 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 (
+ "bytes"
+ "fmt"
+ "sort"
+ "strings"
+
+ "github.com/apache/skywalking-banyandb/pkg/convert"
+ "github.com/apache/skywalking-banyandb/pkg/encoding"
+ pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
+ "github.com/apache/skywalking-banyandb/pkg/pool"
+ "github.com/apache/skywalking-banyandb/pkg/query/model"
+)
+
+type dataBlock struct {
+ offset uint64
+ size uint64
+}
+
+func (d *dataBlock) reset() {
+ d.offset = 0
+ d.size = 0
+}
+
+func (d *dataBlock) copyFrom(src *dataBlock) {
+ d.offset = src.offset
+ d.size = src.size
+}
+
+func (d *dataBlock) marshal(dst []byte) []byte {
+ dst = encoding.VarUint64ToBytes(dst, d.offset)
+ dst = encoding.VarUint64ToBytes(dst, d.size)
+ return dst
+}
+
+func (d *dataBlock) unmarshal(src []byte) []byte {
+ src, n := encoding.BytesToVarUint64(src)
+ d.offset = n
+
+ src, n = encoding.BytesToVarUint64(src)
+ d.size = n
+ return src
+}
+
+type blockMetadata struct {
+ tags map[string]*dataBlock
+ tagType map[string]pbv1.ValueType
+ spans *dataBlock
+ tagProjection *model.TagProjection
+ traceID string
Review Comment:
it could be the maxTracID and minTraceID
--
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