Copilot commented on code in PR #1142: URL: https://github.com/apache/skywalking-banyandb/pull/1142#discussion_r3300961041
########## banyand/dump/helpers.go: ########## @@ -0,0 +1,128 @@ +// 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 dump + +import ( + "context" + "fmt" + "os" + "path/filepath" + "sort" + "strconv" + + "github.com/apache/skywalking-banyandb/api/common" + "github.com/apache/skywalking-banyandb/pkg/convert" + pkgdump "github.com/apache/skywalking-banyandb/pkg/dump" + "github.com/apache/skywalking-banyandb/pkg/fs" + "github.com/apache/skywalking-banyandb/pkg/index/inverted" + "github.com/apache/skywalking-banyandb/pkg/logger" +) + +const ( + dirNameSidx = "sidx" + dirNameMeta = "meta" + logName = "dump" +) + +// DiscoverPartIDs returns the sorted part IDs found directly under shardPath. +func DiscoverPartIDs(shardPath string) ([]uint64, error) { + entries, err := os.ReadDir(shardPath) + if err != nil { + return nil, fmt.Errorf("failed to read shard directory: %w", err) + } + + var partIDs []uint64 + for _, entry := range entries { + if !entry.IsDir() { + continue + } + name := entry.Name() + if name == dirNameSidx || name == dirNameMeta { + continue + } + partID, parseErr := strconv.ParseUint(name, 16, 64) + if parseErr == nil { + partIDs = append(partIDs, partID) + } + } + + sort.Slice(partIDs, func(i, j int) bool { + return partIDs[i] < partIDs[j] + }) + + return partIDs, nil +} + +// LoadSegmentSeriesMap loads the segment-level SeriesID -> EntityValues map from +// the sidx inverted store under segmentPath. Requires a local filesystem path. +func LoadSegmentSeriesMap(segmentPath string) (map[common.SeriesID]string, error) { + seriesIndexPath := filepath.Join(segmentPath, dirNameSidx) + + store, err := inverted.NewStore(inverted.StoreOpts{ + Path: seriesIndexPath, + Logger: logger.GetLogger(logName), + }) + if err != nil { + return nil, fmt.Errorf("failed to open series index: %w", err) + } + defer store.Close() + + ctx := context.Background() + iter, err := store.SeriesIterator(ctx) + if err != nil { + return nil, fmt.Errorf("failed to create series iterator: %w", err) + } + defer iter.Close() + + seriesMap := make(map[common.SeriesID]string) + for iter.Next() { + series := iter.Val() + if len(series.EntityValues) > 0 { + seriesID := common.SeriesID(convert.Hash(series.EntityValues)) + seriesMap[seriesID] = string(series.EntityValues) + } + } + + return seriesMap, nil +} + +// LoadPartSeriesMap reads the optional part-level smeta.bin under partPath and +// returns its SeriesID -> EntityValues map, or nil when smeta.bin is absent or +// cannot be parsed (the file is optional for backward compatibility). +func LoadPartSeriesMap(fileSystem fs.FileSystem, partPath string, id uint64) map[common.SeriesID][]byte { + sm := pkgdump.TryOpenSeriesMetadata(fileSystem, partPath) + if sm == nil { Review Comment: LoadPartSeriesMap calls pkg/dump.TryOpenSeriesMetadata, which prints warnings directly to os.Stderr on non-ENOENT errors. Since banyand/dump is intended as a reusable parser library, emitting directly to stderr is an unexpected side-effect for callers. Consider opening smeta.bin directly here and routing non-ENOENT errors through the dump package logger (and returning nil for optional/absent metadata) instead of using TryOpenSeriesMetadata. ########## banyand/dump/trace/iterator.go: ########## @@ -0,0 +1,124 @@ +// Licensed to Apache Software Foundation (ASF) under one or more contributor +// license agreements. See the NOTICE file distributed with +// this work for additional information regarding copyright +// ownership. Apache Software Foundation (ASF) licenses this file to you under +// the Apache License, Version 2.0 (the "License"); you may +// not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package trace + +import ( + "fmt" + + "github.com/apache/skywalking-banyandb/banyand/dump" + "github.com/apache/skywalking-banyandb/pkg/compress/zstd" + "github.com/apache/skywalking-banyandb/pkg/convert" + "github.com/apache/skywalking-banyandb/pkg/encoding" + "github.com/apache/skywalking-banyandb/pkg/fs" + pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1" +) + +// Iterator returns a new iterator over the part. All block metadata is parsed +// eagerly (cheap); span/tag data is decoded lazily one block at a time. +func (p *PartReader) Iterator() *dump.Iterator[Row] { + var blocks []*blockMetadata + for i := range p.primaryBlockMetadata { + pbm := p.primaryBlockMetadata[i] + primaryData := make([]byte, pbm.size) + fs.MustReadData(p.primary, int64(pbm.offset), primaryData) + decompressed, err := zstd.Decompress(nil, primaryData) + if err != nil { + return dump.NewErrIterator[Row](fmt.Errorf("cannot decompress primary block: %w", err)) + } + bms, err := parseAllBlockMetadata(decompressed, p.tagType) + if err != nil { + return dump.NewErrIterator[Row](fmt.Errorf("cannot parse block metadata: %w", err)) + } + blocks = append(blocks, bms...) + } + var decoder encoding.BytesBlockDecoder + return dump.NewIterator(len(blocks), func(blockIdx int) ([]Row, error) { + return p.decodeBlock(&decoder, blocks[blockIdx]) + }) +} + +func (p *PartReader) decodeBlock(decoder *encoding.BytesBlockDecoder, bm *blockMetadata) ([]Row, error) { + count := int(bm.count) + spans, spanIDs, err := readSpans(decoder, bm.spans, count, p.spans) + if err != nil { + return nil, fmt.Errorf("cannot read spans for trace %s: %w", bm.traceID, err) + } + + tagsBySpan := make(map[string][][]byte, len(bm.tags)) + for tagName, tagBlock := range bm.tags { + metaReader := p.tagMetadata[tagName] + valueReader := p.tags[tagName] + // A tag file may have failed to open (missing/corrupt); skip it rather + // than dereferencing a nil reader. + if metaReader == nil || valueReader == nil { + continue + } + values, tagErr := readTagValues(decoder, tagBlock, tagName, count, metaReader, valueReader, p.tagType[tagName]) + if tagErr != nil { + return nil, fmt.Errorf("cannot read tag %s for trace %s: %w", tagName, bm.traceID, tagErr) + } + tagsBySpan[tagName] = values + } + + rows := make([]Row, 0, len(spans)) + for i := range spans { + spanTags := make(map[string][]byte, len(tagsBySpan)) + tagTypesForRow := make(map[string]pbv1.ValueType, len(tagsBySpan)) + for tagName, tagValues := range tagsBySpan { + if i < len(tagValues) { + spanTags[tagName] = tagValues[i] + tagTypesForRow[tagName] = bm.tagType[tagName] + } + } + seriesID := calculateSeriesIDFromTags(spanTags) + var entityValues []byte + if p.seriesMap != nil { + entityValues = p.seriesMap[seriesID] + } + rows = append(rows, Row{ + TraceID: bm.traceID, + SpanID: spanIDs[i], + Span: spans[i], + Tags: spanTags, + TagTypes: tagTypesForRow, + SeriesID: seriesID, + EntityValues: entityValues, Review Comment: Trace SeriesID is computed via calculateSeriesIDFromTags(spanTags) (hash of sorted "name=value|" bytes), but the series maps loaded from smeta.bin / segment sidx are keyed by hash(series.EntityValues) as stored by the writer (typically the marshaled series buffer, e.g. subject+entity values, per pkg/pb/v1/series.go). These hashes are not compatible, so `p.seriesMap[seriesID]` will almost always miss on real data and EntityValues will stay empty. Consider removing/renaming this SeriesID for trace (or documenting it as a tag-hash), or compute SeriesID using the same series-buffer algorithm used by storage (requires subject + entity tag schema/order). ########## banyand/stream/test_helper.go: ########## @@ -18,75 +18,162 @@ package stream import ( + "fmt" + "path/filepath" "time" "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" + "github.com/apache/skywalking-banyandb/banyand/internal/storage" "github.com/apache/skywalking-banyandb/pkg/convert" "github.com/apache/skywalking-banyandb/pkg/fs" - pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1" + "github.com/apache/skywalking-banyandb/pkg/index" ) -// CreateTestPartForDump creates a test stream part for testing the dump tool. -// It takes a temporary path and a file system as input, generates test elements with various tag types, -// creates a memory part, flushes it to disk, and returns the path to the created part directory. -// Parameters: -// -// tmpPath: the base directory where the part will be created. -// fileSystem: the file system to use for writing the part. -// -// Returns: -// -// The path to the created part directory and a cleanup function. -func CreateTestPartForDump(tmpPath string, fileSystem fs.FileSystem) (string, func()) { - now := time.Now().UnixNano() +// DumpTag describes one tag of a stream row for the dump test builder. Value is a +// Go-native value (string, int64, []string, []int64 or []byte) encoded through +// the production encodeTagValue path. +type DumpTag struct { + Value any + Family string + Name string +} - // Create test elements with various tag types - es := &elements{ - seriesIDs: []common.SeriesID{1, 2, 3}, - timestamps: []int64{now, now + 1000, now + 2000}, - elementIDs: []uint64{11, 21, 31}, - tagFamilies: [][]tagValues{ - { - { - tag: "arrTag", - values: []*tagValue{ - {tag: "strArrTag", valueType: pbv1.ValueTypeStrArr, value: nil, valueArr: [][]byte{[]byte("value1"), []byte("value2")}}, - {tag: "intArrTag", valueType: pbv1.ValueTypeInt64Arr, value: nil, valueArr: [][]byte{convert.Int64ToBytes(25), convert.Int64ToBytes(30)}}, - }, - }, - { - tag: "singleTag", - values: []*tagValue{ - {tag: "strTag", valueType: pbv1.ValueTypeStr, value: []byte("test-value"), valueArr: nil}, - {tag: "intTag", valueType: pbv1.ValueTypeInt64, value: convert.Int64ToBytes(100), valueArr: nil}, - }, - }, - }, - { - { - tag: "singleTag", - values: []*tagValue{ - {tag: "strTag1", valueType: pbv1.ValueTypeStr, value: []byte("tag1"), valueArr: nil}, - {tag: "strTag2", valueType: pbv1.ValueTypeStr, value: []byte("tag2"), valueArr: nil}, - }, - }, - }, - {}, // empty tagFamilies for seriesID 3 - }, +// DumpRow is one stream element for BuildPartForDump. When Entity is set the +// SeriesID is derived as hash(Entity) and (optionally) written to smeta.bin. +type DumpRow struct { + Entity string + Tags []DumpTag + SeriesID common.SeriesID + Timestamp int64 + ElementID uint64 +} + +// BuildPartForDump writes rows into a stream part at root/<partID>, returning the +// part directory, the rows (with resolved SeriesID) for verification and a +// cleanup func. When writeSeriesMeta is true a matching smeta.bin is written for +// every row that carries an Entity. +func BuildPartForDump(tmpPath string, fileSystem fs.FileSystem, partID uint64, rows []DumpRow) (string, []DumpRow, func()) { + es := &elements{} + for i := range rows { + r := &rows[i] + if r.Entity != "" { + r.SeriesID = common.SeriesID(convert.Hash([]byte(r.Entity))) + } + es.seriesIDs = append(es.seriesIDs, r.SeriesID) + es.timestamps = append(es.timestamps, r.Timestamp) + es.elementIDs = append(es.elementIDs, r.ElementID) + es.tagFamilies = append(es.tagFamilies, buildDumpTagFamilies(r.Tags)) } - // Create memPart and flush mp := generateMemPart() mp.mustInitFromElements(es) - - epoch := uint64(12345) - path := partPath(tmpPath, epoch) + path := partPath(tmpPath, partID) mp.mustFlush(fileSystem, path) - cleanup := func() { - // Cleanup is handled by the caller's test.Space cleanup + return path, rows, func() { releaseMemPart(mp) Review Comment: BuildPartForDump builds tagValues via encodeTagValue (pooled) and stores them under `es.tagFamilies`, but the cleanup only releases the memPart. Since `es` is not reset/released, pooled tagValues can leak between tests. Consider calling `es.reset()` (or using generateElements/releaseElements) in the cleanup function after flushing. ########## banyand/dump/trace/iterator.go: ########## @@ -0,0 +1,124 @@ +// Licensed to Apache Software Foundation (ASF) under one or more contributor +// license agreements. See the NOTICE file distributed with +// this work for additional information regarding copyright +// ownership. Apache Software Foundation (ASF) licenses this file to you under +// the Apache License, Version 2.0 (the "License"); you may +// not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package trace + +import ( + "fmt" + + "github.com/apache/skywalking-banyandb/banyand/dump" + "github.com/apache/skywalking-banyandb/pkg/compress/zstd" + "github.com/apache/skywalking-banyandb/pkg/convert" + "github.com/apache/skywalking-banyandb/pkg/encoding" + "github.com/apache/skywalking-banyandb/pkg/fs" + pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1" +) + +// Iterator returns a new iterator over the part. All block metadata is parsed +// eagerly (cheap); span/tag data is decoded lazily one block at a time. +func (p *PartReader) Iterator() *dump.Iterator[Row] { + var blocks []*blockMetadata + for i := range p.primaryBlockMetadata { + pbm := p.primaryBlockMetadata[i] + primaryData := make([]byte, pbm.size) + fs.MustReadData(p.primary, int64(pbm.offset), primaryData) + decompressed, err := zstd.Decompress(nil, primaryData) + if err != nil { + return dump.NewErrIterator[Row](fmt.Errorf("cannot decompress primary block: %w", err)) + } Review Comment: Iterator() uses fs.MustReadData to load primary blocks (and trace.go uses MustReadData for tag metadata/values). MustReadData panics on short reads/I/O errors, so a corrupt or truncated part can crash the dump tool/library instead of returning an iterator error. Since this is a reusable parser library, prefer using reader.ReadAt/ReadFull with explicit error propagation so callers can handle corruption gracefully. ########## banyand/trace/test_helper.go: ########## @@ -18,64 +18,148 @@ package trace import ( + "fmt" "time" - "github.com/apache/skywalking-banyandb/pkg/convert" + "google.golang.org/protobuf/types/known/timestamppb" + + databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1" + modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1" "github.com/apache/skywalking-banyandb/pkg/fs" - pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1" ) -// CreateTestPartForDump creates a test trace part for testing the dump tool. -// It returns the part path (directory) and cleanup function. -func CreateTestPartForDump(tmpPath string, fileSystem fs.FileSystem) (string, func()) { - now := time.Now().UnixNano() +// Timestamp marks a DumpTag value as a ValueType=Timestamp tag carrying nanos +// since the epoch (trace stores its per-span time in such a tag). +type Timestamp int64 - // Create test traces with various tag types - traces := &traces{ - traceIDs: []string{"test-trace-1", "test-trace-1", "test-trace-2"}, - timestamps: []int64{now, now + 1000, now + 2000}, - spanIDs: []string{"span-1", "span-2", "span-3"}, - spans: [][]byte{ - []byte("span-data-1-with-content"), - []byte("span-data-2-with-content"), - []byte("span-data-3-with-content"), - }, - tags: [][]*tagValue{ - { - {tag: "service.name", valueType: pbv1.ValueTypeStr, value: []byte("test-service")}, - {tag: "http.status", valueType: pbv1.ValueTypeInt64, value: convert.Int64ToBytes(200)}, - {tag: "timestamp", valueType: pbv1.ValueTypeTimestamp, value: convert.Int64ToBytes(now)}, - {tag: "tags", valueType: pbv1.ValueTypeStrArr, valueArr: [][]byte{[]byte("tag1"), []byte("tag2")}}, - {tag: "duration", valueType: pbv1.ValueTypeInt64, value: convert.Int64ToBytes(1234567)}, +// DumpTag describes one (flat) tag of a trace span for the dump test builder. +// Value is a Go-native value: string, int64, []string, []int64, []byte or +// Timestamp. +type DumpTag struct { + Value any + Name string +} + +// DumpRow is one trace span for BuildPartForDump. SeriesID is derived from the +// tags by the parser, so it is not specified here. +type DumpRow struct { + TraceID string + SpanID string + Span []byte + Tags []DumpTag + Timestamp int64 +} + +// BuildPartForDump writes spans into a trace part at root/<partID>, returning the +// part directory, the rows for verification and a cleanup func. +func BuildPartForDump(tmpPath string, fileSystem fs.FileSystem, partID uint64, rows []DumpRow) (string, []DumpRow, func()) { + ts := &traces{} + for i := range rows { + r := &rows[i] + ts.traceIDs = append(ts.traceIDs, r.TraceID) + ts.timestamps = append(ts.timestamps, r.Timestamp) + ts.spanIDs = append(ts.spanIDs, r.SpanID) + ts.spans = append(ts.spans, r.Span) + ts.tags = append(ts.tags, buildDumpTags(r.Tags)) + } + + mp := &memPart{} + mp.mustInitFromTraces(ts) + path := partPath(tmpPath, partID) + mp.mustFlush(fileSystem, path) + + return path, rows, func() {} Review Comment: BuildPartForDump allocates tagValue objects via encodeTagValue (generateTagValue uses a pool), but the returned cleanup func is a no-op and `ts` is never reset. This leaks pooled tagValues across tests and can skew memory usage. Consider resetting/releasing `ts` (e.g., ts.reset() / releaseTraces) in the cleanup function after flushing the part. ########## banyand/dump/decode.go: ########## @@ -0,0 +1,100 @@ +// 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 dump + +import ( + "time" + + "google.golang.org/protobuf/types/known/timestamppb" + + modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1" + "github.com/apache/skywalking-banyandb/pkg/convert" + pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1" +) + +// DecodeTagValue converts a byte-encoded tag value back to a typed +// modelv1.TagValue. valueType selects the decoding; the returned value is +// self-contained (string / binary contents are copied), so it may be retained +// past the next Iterator.Next()/Close(). +func DecodeTagValue(valueType pbv1.ValueType, value []byte, valueArr [][]byte) *modelv1.TagValue { + if value == nil && valueArr == nil { + return pbv1.NullTagValue + } + + switch valueType { + case pbv1.ValueTypeStr: + if value == nil { + return pbv1.NullTagValue + } + return &modelv1.TagValue{ + Value: &modelv1.TagValue_Str{ + Str: &modelv1.Str{ + Value: string(value), + }, + }, + } + case pbv1.ValueTypeInt64: + if value == nil { + return pbv1.NullTagValue + } + return &modelv1.TagValue{ + Value: &modelv1.TagValue_Int{ + Int: &modelv1.Int{ + Value: convert.BytesToInt64(value), + }, + }, + } + case pbv1.ValueTypeTimestamp: + if value == nil { + return pbv1.NullTagValue + } + return &modelv1.TagValue{ + Value: &modelv1.TagValue_Timestamp{ + Timestamp: timestamppb.New(time.Unix(0, convert.BytesToInt64(value))), + }, + } + case pbv1.ValueTypeStrArr: + values := make([]string, 0, len(valueArr)) + for _, item := range valueArr { + values = append(values, string(item)) + } + return &modelv1.TagValue{ + Value: &modelv1.TagValue_StrArray{StrArray: &modelv1.StrArray{Value: values}}, + } + case pbv1.ValueTypeBinaryData: + if value == nil { + return pbv1.NullTagValue + } + b := make([]byte, len(value)) + copy(b, value) + return &modelv1.TagValue{ + Value: &modelv1.TagValue_BinaryData{BinaryData: b}, + } + default: + if value != nil { + return &modelv1.TagValue{ + Value: &modelv1.TagValue_Str{ + Str: &modelv1.Str{ + Value: string(value), + }, + }, + } + } + return pbv1.NullTagValue + } Review Comment: DecodeTagValue claims to convert a byte-encoded tag value into a typed modelv1.TagValue based on valueType, but it currently has no cases for several defined types (e.g., pbv1.ValueTypeInt64Arr / pbv1.ValueTypeInt64Arr is decoded via the default branch into a string). This makes the helper incorrect for array/int-array tags and can lead to wrong display/criteria behavior if callers rely on it. Add explicit decoding branches for the missing pbv1.ValueType* variants (at least Int64Arr, and any other types the dump parsers can emit). ########## banyand/measure/test_helper.go: ########## @@ -18,95 +18,201 @@ package measure import ( + "fmt" + "path/filepath" "time" "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" + "github.com/apache/skywalking-banyandb/banyand/internal/storage" "github.com/apache/skywalking-banyandb/pkg/convert" "github.com/apache/skywalking-banyandb/pkg/fs" + "github.com/apache/skywalking-banyandb/pkg/index" pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1" ) -// CreateTestPartForDump creates a test measure part for testing the dump tool. -// It takes a temporary path and a file system as input, generates test data points with various tag and field types, -// creates a memory part, flushes it to disk, and returns the path to the created part directory. -// Parameters: -// -// tmpPath: the base directory where the part will be created. -// fileSystem: the file system to use for writing the part. -// -// Returns: -// -// The path to the created part directory and a cleanup function. -func CreateTestPartForDump(tmpPath string, fileSystem fs.FileSystem) (string, func()) { - now := time.Now().UnixNano() +// DumpTag describes one tag of a measure row for the dump test builder. Value is +// a Go-native value (string, int64, []string, []int64 or []byte) encoded through +// the production encodeTagValue path. +type DumpTag struct { + Value any + Family string + Name string +} - // Create test data points with various tag and field types - dps := &dataPoints{ - seriesIDs: []common.SeriesID{1, 2, 3}, - timestamps: []int64{now, now + 1000, now + 2000}, - versions: []int64{1, 2, 3}, - tagFamilies: [][]nameValues{ - { - { - name: "arrTag", - values: []*nameValue{ - {name: "strArrTag", valueType: pbv1.ValueTypeStrArr, value: nil, valueArr: [][]byte{[]byte("value1"), []byte("value2")}}, - {name: "intArrTag", valueType: pbv1.ValueTypeInt64Arr, value: nil, valueArr: [][]byte{convert.Int64ToBytes(25), convert.Int64ToBytes(30)}}, - }, - }, - { - name: "singleTag", - values: []*nameValue{ - {name: "strTag", valueType: pbv1.ValueTypeStr, value: []byte("test-value"), valueArr: nil}, - {name: "intTag", valueType: pbv1.ValueTypeInt64, value: convert.Int64ToBytes(100), valueArr: nil}, - }, - }, - }, - { - { - name: "singleTag", - values: []*nameValue{ - {name: "strTag1", valueType: pbv1.ValueTypeStr, value: []byte("tag1"), valueArr: nil}, - {name: "strTag2", valueType: pbv1.ValueTypeStr, value: []byte("tag2"), valueArr: nil}, - }, - }, +// DumpField describes one field of a measure row. Value is int64 or float64. +type DumpField struct { + Value any + Name string +} + +// DumpRow is one measure data point for BuildPartForDump. When Entity is set the +// SeriesID is derived as hash(Entity) and (optionally) written to smeta.bin. +type DumpRow struct { + Entity string + Tags []DumpTag + Fields []DumpField + SeriesID common.SeriesID + Timestamp int64 + Version int64 +} + +// BuildPartForDump writes rows into a measure part at root/<partID>, returning +// the part directory, the rows (with resolved SeriesID) for verification and a +// cleanup func. When writeSeriesMeta is true a matching smeta.bin is written for +// every row that carries an Entity. +func BuildPartForDump(tmpPath string, fileSystem fs.FileSystem, partID uint64, rows []DumpRow) (string, []DumpRow, func()) { + dps := &dataPoints{} + for i := range rows { + r := &rows[i] + if r.Entity != "" { + r.SeriesID = common.SeriesID(convert.Hash([]byte(r.Entity))) + } + dps.seriesIDs = append(dps.seriesIDs, r.SeriesID) + dps.timestamps = append(dps.timestamps, r.Timestamp) + dps.versions = append(dps.versions, r.Version) + dps.tagFamilies = append(dps.tagFamilies, buildDumpTagFamilies(r.Tags)) + dps.fields = append(dps.fields, buildDumpFields(r.Fields)) + } + + mp := generateMemPart() + mp.mustInitFromDataPoints(dps) + path := partPath(tmpPath, partID) + mp.mustFlush(fileSystem, path) + + return path, rows, func() { + releaseMemPart(mp) Review Comment: BuildPartForDump creates nameValue objects via the production encodeTagValue path (nameValue is pooled), but the cleanup only releases the memPart. Since `dps` is never reset/released, pooled nameValues and nested nameValues/tagFamilies can leak between tests. Consider calling `dps.reset()` (or using generateDataPoints/releaseDataPoints) in the cleanup function after flushing. ########## banyand/dump/trace/trace.go: ########## @@ -0,0 +1,501 @@ +// 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 decodes the on-disk byte format of a trace part into rows. +// It is a pure parser: no filtering, projection or output formatting (those are +// CLI concerns). +package trace + +import ( + "encoding/json" + "fmt" + "io" + "path/filepath" + "sort" + "strings" + + "github.com/apache/skywalking-banyandb/api/common" + "github.com/apache/skywalking-banyandb/banyand/dump" + internalencoding "github.com/apache/skywalking-banyandb/banyand/internal/encoding" + "github.com/apache/skywalking-banyandb/pkg/bytes" + "github.com/apache/skywalking-banyandb/pkg/compress/zstd" + "github.com/apache/skywalking-banyandb/pkg/convert" + "github.com/apache/skywalking-banyandb/pkg/encoding" + "github.com/apache/skywalking-banyandb/pkg/fs" + pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1" +) + +// PartMetadata is the on-disk metadata of a trace part. +type PartMetadata struct { + CompressedSizeBytes uint64 `json:"compressedSizeBytes"` + UncompressedSpanSizeBytes uint64 `json:"uncompressedSpanSizeBytes"` + TotalCount uint64 `json:"totalCount"` + BlocksCount uint64 `json:"blocksCount"` + MinTimestamp int64 `json:"minTimestamp"` + MaxTimestamp int64 `json:"maxTimestamp"` + ID uint64 `json:"-"` +} + +type primaryBlockMetadata struct { + traceID string + offset uint64 + size uint64 +} + +type dataBlock struct { + offset uint64 + size uint64 +} + +type blockMetadata struct { + tags map[string]*dataBlock + tagType map[string]pbv1.ValueType + spans *dataBlock + traceID string + uncompressedSpanSizeBytes uint64 + count uint64 +} + +// PartReader provides read-only access to a trace part directory. +// Not safe for concurrent use. +type PartReader struct { + primary fs.Reader + spans fs.Reader + fileSystem fs.FileSystem + tagMetadata map[string]fs.Reader + tags map[string]fs.Reader + tagType map[string]pbv1.ValueType + seriesMap map[common.SeriesID][]byte // part-level smeta.bin, nil if absent + path string + primaryBlockMetadata []primaryBlockMetadata + partMetadata PartMetadata +} + +// Row is one decoded span from a trace part. The []byte / map values alias the +// iterator's block decode buffer and are valid only until the next +// Iterator.Next()/Close(); copy to retain. Tag keys are full tag names. +// SeriesID is computed from the tags (calculateSeriesIDFromTags); Timestamp is +// read from the tag whose type is ValueTypeTimestamp. Trace has no version/fields. +type Row struct { + Tags map[string][]byte + TagTypes map[string]pbv1.ValueType + TraceID string + SpanID string + EntityValues []byte + Span []byte + Timestamp int64 + SeriesID common.SeriesID +} + +// OpenPart opens the trace part directory root/<id> for read-only access. +func OpenPart(id uint64, root string, fileSystem fs.FileSystem) (*PartReader, error) { + var p PartReader + partPath := filepath.Join(root, fmt.Sprintf("%016x", id)) + p.path = partPath + p.fileSystem = fileSystem + + // Read metadata.json + metadataPath := filepath.Join(partPath, "metadata.json") + metadata, err := fileSystem.Read(metadataPath) + if err != nil { + return nil, fmt.Errorf("cannot read metadata.json: %w", err) + } + if unmarshalErr := json.Unmarshal(metadata, &p.partMetadata); unmarshalErr != nil { + return nil, fmt.Errorf("cannot parse metadata.json: %w", unmarshalErr) + } + p.partMetadata.ID = id + + // Read tag types + p.tagType = make(map[string]pbv1.ValueType) + tagTypePath := filepath.Join(partPath, "tag.type") + if tagTypeData, readErr := fileSystem.Read(tagTypePath); readErr == nil && len(tagTypeData) > 0 { + if unmarshalErr := unmarshalTagType(tagTypeData, p.tagType); unmarshalErr != nil { + return nil, fmt.Errorf("cannot parse tag.type: %w", unmarshalErr) + } + } + + // Read primary block metadata + metaPath := filepath.Join(partPath, "meta.bin") + metaFile, err := fileSystem.OpenFile(metaPath) + if err != nil { + return nil, fmt.Errorf("cannot open meta.bin: %w", err) + } + p.primaryBlockMetadata, err = readPrimaryBlockMetadata(metaFile) + fs.MustClose(metaFile) + if err != nil { + return nil, fmt.Errorf("cannot read primary block metadata: %w", err) + } + + // Open data files + p.primary, err = fileSystem.OpenFile(filepath.Join(partPath, "primary.bin")) + if err != nil { + return nil, fmt.Errorf("cannot open primary.bin: %w", err) + } + + p.spans, err = fileSystem.OpenFile(filepath.Join(partPath, "spans.bin")) + if err != nil { + fs.MustClose(p.primary) + return nil, fmt.Errorf("cannot open spans.bin: %w", err) + } + + // Load the optional part-level series metadata (smeta.bin). + p.seriesMap = dump.LoadPartSeriesMap(fileSystem, partPath, id) + + // Open tag files + entries := fileSystem.ReadDir(partPath) + p.tags = make(map[string]fs.Reader) + p.tagMetadata = make(map[string]fs.Reader) + for _, e := range entries { + if e.IsDir() { + continue + } + if strings.HasSuffix(e.Name(), ".tm") { + name := e.Name()[:len(e.Name())-3] + reader, err := fileSystem.OpenFile(filepath.Join(partPath, e.Name())) + if err == nil { + p.tagMetadata[name] = reader + } + } + if strings.HasSuffix(e.Name(), ".t") { + name := e.Name()[:len(e.Name())-2] + reader, err := fileSystem.OpenFile(filepath.Join(partPath, e.Name())) + if err == nil { + p.tags[name] = reader + } + } + } + + return &p, nil +} + +func closePart(p *PartReader) { + if p.primary != nil { + fs.MustClose(p.primary) + p.primary = nil + } + if p.spans != nil { + fs.MustClose(p.spans) + p.spans = nil + } + for _, r := range p.tags { + fs.MustClose(r) + } + p.tags = nil + for _, r := range p.tagMetadata { + fs.MustClose(r) + } + p.tagMetadata = nil +} + +// Metadata returns the part's metadata.json contents. +func (p *PartReader) Metadata() PartMetadata { return p.partMetadata } + +// PartID returns the part's numeric ID. +func (p *PartReader) PartID() uint64 { return p.partMetadata.ID } + +// SeriesMap returns the part-level SeriesID -> EntityValues map parsed from +// smeta.bin, or nil if smeta.bin is absent. +func (p *PartReader) SeriesMap() map[common.SeriesID][]byte { return p.seriesMap } + +// Close releases all file handles. Safe to call multiple times. +func (p *PartReader) Close() error { + closePart(p) + return nil +} + +func readPrimaryBlockMetadata(r fs.Reader) ([]primaryBlockMetadata, error) { + sr := r.SequentialRead() + data, err := io.ReadAll(sr) + fs.MustClose(sr) + if err != nil { + return nil, fmt.Errorf("cannot read: %w", err) + } + + decompressed, err := zstd.Decompress(nil, data) + if err != nil { + return nil, fmt.Errorf("cannot decompress: %w", err) + } + + var result []primaryBlockMetadata + src := decompressed + for len(src) > 0 { + var pbm primaryBlockMetadata + src, err = unmarshalPrimaryBlockMetadata(&pbm, src) + if err != nil { + return nil, err + } + result = append(result, pbm) + } + return result, nil +} + +func unmarshalPrimaryBlockMetadata(pbm *primaryBlockMetadata, src []byte) ([]byte, error) { + if len(src) < 4 { + return nil, fmt.Errorf("insufficient data") + } + src, traceIDBytes, err := encoding.DecodeBytes(src) + if err != nil { + return nil, fmt.Errorf("cannot unmarshal traceID: %w", err) + } + pbm.traceID = string(traceIDBytes) + if len(src) < 16 { + return nil, fmt.Errorf("insufficient data for offset and size") + } + pbm.offset = encoding.BytesToUint64(src) + src = src[8:] + pbm.size = encoding.BytesToUint64(src) + return src[8:], nil +} + +func unmarshalTagType(src []byte, tagType map[string]pbv1.ValueType) error { + src, count := encoding.BytesToVarUint64(src) + for i := uint64(0); i < count; i++ { + var nameBytes []byte + var err error + src, nameBytes, err = encoding.DecodeBytes(src) + if err != nil { + return err + } + name := string(nameBytes) + if len(src) < 1 { + return fmt.Errorf("insufficient data for valueType") + } + valueType := pbv1.ValueType(src[0]) + src = src[1:] + tagType[name] = valueType + } + return nil +} + +func parseAllBlockMetadata(src []byte, tagType map[string]pbv1.ValueType) ([]*blockMetadata, error) { + var result []*blockMetadata + for len(src) > 0 { + bm, tail, err := parseBlockMetadata(src, tagType) + if err != nil { + return nil, err + } + result = append(result, bm) + src = tail + } + return result, nil +} + +func parseBlockMetadata(src []byte, tagType map[string]pbv1.ValueType) (*blockMetadata, []byte, error) { + var bm blockMetadata + bm.tagType = make(map[string]pbv1.ValueType) + for k, v := range tagType { + bm.tagType[k] = v + } + + src, traceIDBytes, err := encoding.DecodeBytes(src) + if err != nil { + return nil, nil, fmt.Errorf("cannot unmarshal traceID: %w", err) + } + bm.traceID = string(traceIDBytes) + + src, n := encoding.BytesToVarUint64(src) + bm.uncompressedSpanSizeBytes = n + + src, n = encoding.BytesToVarUint64(src) + bm.count = n + + bm.spans = &dataBlock{} + src, n = encoding.BytesToVarUint64(src) + bm.spans.offset = n + src, n = encoding.BytesToVarUint64(src) + bm.spans.size = n + + src, tagCount := encoding.BytesToVarUint64(src) + if tagCount > 0 { + bm.tags = make(map[string]*dataBlock) + for i := uint64(0); i < tagCount; i++ { + var nameBytes []byte + src, nameBytes, err = encoding.DecodeBytes(src) + if err != nil { + return nil, nil, fmt.Errorf("cannot unmarshal tag name: %w", err) + } + name := string(nameBytes) + + db := &dataBlock{} + src, n = encoding.BytesToVarUint64(src) + db.offset = n + src, n = encoding.BytesToVarUint64(src) + db.size = n + bm.tags[name] = db + } + } + + return &bm, src, nil +} + +func readSpans(decoder *encoding.BytesBlockDecoder, sm *dataBlock, count int, reader fs.Reader) ([][]byte, []string, error) { + data := make([]byte, sm.size) + fs.MustReadData(reader, int64(sm.offset), data) + + var spanIDBytes [][]byte + var tail []byte + var err error + spanIDBytes, tail, err = decoder.DecodeWithTail(spanIDBytes[:0], data, uint64(count)) + if err != nil { + return nil, nil, fmt.Errorf("cannot decode spanIDs: %w", err) + } + + spanIDs := make([]string, count) + for i, idBytes := range spanIDBytes { + spanIDs[i] = string(idBytes) + } + + var spans [][]byte + spans, err = decoder.Decode(spans[:0], tail, uint64(count)) + if err != nil { + return nil, nil, fmt.Errorf("cannot decode spans: %w", err) + } + + return spans, spanIDs, nil +} + +type tagMetadata struct { + min []byte + max []byte + offset uint64 + size uint64 + filterBlock dataBlock +} + +func readTagValues(decoder *encoding.BytesBlockDecoder, tagBlock *dataBlock, _ string, count int, + metaReader, valueReader fs.Reader, valueType pbv1.ValueType, +) ([][]byte, error) { + // Read tag metadata + metaData := make([]byte, tagBlock.size) + fs.MustReadData(metaReader, int64(tagBlock.offset), metaData) + + var tm tagMetadata + if err := unmarshalTagMetadata(&tm, metaData); err != nil { + return nil, fmt.Errorf("cannot unmarshal tag metadata: %w", err) + } + + // Check if there's actual data to read + if tm.size == 0 { + // Return nil values for all items + values := make([][]byte, count) + return values, nil + } + + // Read tag values + bb := &bytes.Buffer{} + bb.Buf = make([]byte, tm.size) + fs.MustReadData(valueReader, int64(tm.offset), bb.Buf) + + // Decode values using the internal encoding package + var err error + var values [][]byte + values, err = internalencoding.DecodeTagValues(values, decoder, bb, valueType, count) + if err != nil { + return nil, fmt.Errorf("cannot decode tag values: %w", err) + } + + return values, nil +} + +func unmarshalTagMetadata(tm *tagMetadata, src []byte) error { + var n uint64 + var err error + + // Unmarshal dataBlock (offset and size) + src, n = encoding.BytesToVarUint64(src) + tm.offset = n + src, n = encoding.BytesToVarUint64(src) + tm.size = n + + // Unmarshal min + src, tm.min, err = encoding.DecodeBytes(src) + if err != nil { + return fmt.Errorf("cannot unmarshal tagMetadata.min: %w", err) + } + + // Unmarshal max + src, tm.max, err = encoding.DecodeBytes(src) + if err != nil { + return fmt.Errorf("cannot unmarshal tagMetadata.max: %w", err) + } + + // Unmarshal filterBlock + src, n = encoding.BytesToVarUint64(src) + tm.filterBlock.offset = n + _, n = encoding.BytesToVarUint64(src) Review Comment: In unmarshalTagMetadata, filterBlock.size is decoded from the same slice position as filterBlock.offset because the second BytesToVarUint64 call uses `src` without advancing it (line 439 uses `_` for the returned slice). This will mis-parse the metadata (size will be wrong unless offset encodes to the same varint length as size) and can lead to incorrect reads/EOF/panics downstream. Advance `src` after decoding offset and decode size from the advanced slice (i.e., use `src, n = ...` for both offset and size). -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
