Copilot commented on code in PR #1184:
URL:
https://github.com/apache/skywalking-banyandb/pull/1184#discussion_r3452532600
##########
banyand/stream/part_metadata.go:
##########
@@ -107,6 +112,96 @@ func (pm *partMetadata) mustWriteMetadata(fileSystem
fs.FileSystem, partPath str
}
}
+type tagType map[string]map[string]pbv1.ValueType
+
+func (tt tagType) reset() {
+ clear(tt)
+}
+
+func (tt tagType) copyFrom(source tagType) {
+ for familyName, sourceTags := range source {
+ tags := tt[familyName]
+ if tags == nil {
+ tags = make(map[string]pbv1.ValueType, len(sourceTags))
+ tt[familyName] = tags
+ }
+ for tagName, valueType := range sourceTags {
+ tags[tagName] = valueType
+ }
+ }
+}
+
+func (tt tagType) marshal() []byte {
+ var dst []byte
+ dst = encoding.VarUint64ToBytes(dst, uint64(len(tt)))
+ familyNames := make([]string, 0, len(tt))
+ for familyName := range tt {
+ familyNames = append(familyNames, familyName)
+ }
+ sort.Strings(familyNames)
+ for _, familyName := range familyNames {
+ dst = encoding.EncodeBytes(dst,
convert.StringToBytes(familyName))
+ tags := tt[familyName]
+ dst = encoding.VarUint64ToBytes(dst, uint64(len(tags)))
+ tagNames := make([]string, 0, len(tags))
+ for tagName := range tags {
+ tagNames = append(tagNames, tagName)
+ }
+ sort.Strings(tagNames)
+ for _, tagName := range tagNames {
+ dst = encoding.EncodeBytes(dst,
convert.StringToBytes(tagName))
+ dst = append(dst, byte(tags[tagName]))
+ }
+ }
+ return dst
+}
+
+func (tt tagType) unmarshal(src []byte) error {
+ tt.reset()
+ remaining, familyCount := encoding.BytesToVarUint64(src)
+ for i := uint64(0); i < familyCount; i++ {
+ var familyNameBytes []byte
+ var decodeErr error
+ remaining, familyNameBytes, decodeErr =
encoding.DecodeBytes(remaining)
+ if decodeErr != nil {
+ return fmt.Errorf("cannot decode tag family name: %w",
decodeErr)
+ }
+ var tagCount uint64
+ remaining, tagCount = encoding.BytesToVarUint64(remaining)
+ tags := make(map[string]pbv1.ValueType, tagCount)
+ for j := uint64(0); j < tagCount; j++ {
Review Comment:
`tagCount` is a `uint64`, but `make(map[string]pbv1.ValueType, tagCount)`
requires an `int` capacity. As written this won't compile.
##########
banyand/stream/write_data.go:
##########
@@ -55,6 +57,9 @@ func (s *syncPartContext) FinishSync() error {
// Close writers first so file data is flushed before we write metadata.
s.releaseCoreWriters()
+ if len(s.tagTypeBuffer) > 0 {
+ fs.MustFlushAtomic(s.fileSystem, s.tagTypeBuffer,
filepath.Join(s.partPath, tagTypeFilename), storage.FilePerm)
+ }
Review Comment:
The new chunked-sync behavior for `tag.type` (buffering chunks and flushing
atomically in `FinishSync`) isn't covered by unit tests. Existing stream sync
tests (e.g., `file_part_sync_test.go` / `write_data_segmentref_test.go`) could
be extended to send a `tag.type` stream and assert the receiver writes it to
disk and cleans up the buffer on success/failure.
--
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]
