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/31 16:27:20 UTC
[PR] Implement sidx flushing and merging [skywalking-banyandb]
ButterBright opened a new pull request, #745:
URL: https://github.com/apache/skywalking-banyandb/pull/745
<!--
⚠️ 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 👆 ==== -->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement sidx flushing and merging [skywalking-banyandb]
Posted by "hanahmily (via GitHub)" <gi...@apache.org>.
hanahmily commented on code in PR #745:
URL: https://github.com/apache/skywalking-banyandb/pull/745#discussion_r2312923230
##########
banyand/internal/sidx/block_reader.go:
##########
@@ -0,0 +1,235 @@
+// 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 sidx
+
+import (
+ "container/heap"
+ "errors"
+ "fmt"
+ "io"
+
+ "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"
+)
+
+type seqReader struct {
+ sr fs.SeqReader
+ r fs.Reader
+ bytesRead uint64
+}
+
+func (sr *seqReader) reset() {
+ sr.r = nil
+ if sr.sr != nil {
+ fs.MustClose(sr.sr)
+ }
+ sr.sr = nil
+ sr.bytesRead = 0
+}
+
+func (sr *seqReader) Path() string {
+ return sr.r.Path()
+}
+
+func (sr *seqReader) init(r fs.Reader) {
+ sr.reset()
+ sr.sr = r.SequentialRead()
+ sr.r = r
+}
+
+func (sr *seqReader) mustReadFull(data []byte) {
+ n, err := io.ReadFull(sr.sr, data)
+ if err != nil {
+ if errors.Is(err, io.EOF) {
+ return
+ }
+ logger.Panicf("cannot read data: %v", err)
+ }
+ if n != len(data) {
+ logger.Panicf("cannot read full data: %d/%d", n, len(data))
+ }
+ sr.bytesRead += uint64(n)
+}
+
+func generateSeqReader() *seqReader {
+ if v := seqReaderPool.Get(); v != nil {
+ return v
+ }
+ return &seqReader{}
+}
+
+func releaseSeqReader(sr *seqReader) {
+ sr.reset()
+ seqReaderPool.Put(sr)
+}
+
+var seqReaderPool = pool.Register[*seqReader]("sidx-seqReader")
+
+type seqReaders struct {
+ tagMetadata map[string]*seqReader
+ tagData map[string]*seqReader
+ tagFilters map[string]*seqReader
+ primary seqReader
+ keys seqReader
+ data seqReader
+}
+
+func (sr *seqReaders) reset() {
+ sr.primary.reset()
+ sr.keys.reset()
+ sr.data.reset()
+ if sr.tagMetadata != nil {
+ for k, r := range sr.tagMetadata {
+ releaseSeqReader(r)
+ delete(sr.tagMetadata, k)
+ releaseSeqReader(r)
+ delete(sr.tagData, k)
+ releaseSeqReader(r)
+ delete(sr.tagFilters, k)
Review Comment:
This recommendation should be followed because it is more reliable.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement sidx flushing and merging [skywalking-banyandb]
Posted by "hanahmily (via GitHub)" <gi...@apache.org>.
hanahmily commented on code in PR #745:
URL: https://github.com/apache/skywalking-banyandb/pull/745#discussion_r2312914235
##########
banyand/internal/sidx/sidx.go:
##########
@@ -19,29 +19,43 @@ package sidx
import (
"context"
+ "encoding/json"
+ "fmt"
"math"
+ "path/filepath"
+ "slices"
"sort"
"sync"
"sync/atomic"
"github.com/apache/skywalking-banyandb/api/common"
modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
+ "github.com/apache/skywalking-banyandb/banyand/internal/storage"
"github.com/apache/skywalking-banyandb/banyand/protector"
+ "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/run"
"github.com/apache/skywalking-banyandb/pkg/watcher"
)
+const (
+ maxUncompressedBlockSize = 2 * 1024 * 1024
Review Comment:
```suggestion
```
The maximum size of a block is controlled by its length.
##########
banyand/internal/sidx/sidx.go:
##########
Review Comment:
Extract the flush and merge logic to independent files to make this file tidy.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement sidx flushing and merging [skywalking-banyandb]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #745:
URL: https://github.com/apache/skywalking-banyandb/pull/745#discussion_r2312539553
##########
banyand/internal/sidx/sidx.go:
##########
@@ -375,9 +393,172 @@ func (s *sidx) currentSnapshot() *snapshot {
return nil
}
-// decRef decrements the snapshot reference count (helper for snapshot interface).
-func (s *snapshot) decRef() {
- s.release()
+var errNoPartToMerge = fmt.Errorf("no part to merge")
+
+func (s *sidx) mergeParts(fileSystem fs.FileSystem, closeCh <-chan struct{}, parts []*partWrapper, partID uint64, root string) (*partWrapper, error) {
+ if len(parts) == 0 {
+ return nil, errNoPartToMerge
+ }
+ dstPath := partPath(root, partID)
+ var totalSize int64
+ pii := make([]*partMergeIter, 0, len(parts))
+ for i := range parts {
+ pmi := generatePartMergeIter()
+ pmi.mustInitFromPart(parts[i].p)
+ pii = append(pii, pmi)
+ totalSize += int64(parts[i].p.partMetadata.CompressedSizeBytes)
+ }
+ shouldCache := s.pm.ShouldCache(totalSize)
+ br := generateBlockReader()
+ br.init(pii)
+ bw := generateBlockWriter()
+ bw.mustInitForFilePart(fileSystem, dstPath, shouldCache)
+
+ pm, err := mergeBlocks(closeCh, bw, br)
+ releaseBlockWriter(bw)
+ releaseBlockReader(br)
+ for i := range pii {
+ releasePartMergeIter(pii[i])
+ }
+ if err != nil {
+ return nil, err
+ }
+ pm.mustWriteMetadata(fileSystem, dstPath)
+ fileSystem.SyncPath(dstPath)
+ p := mustOpenPart(dstPath, fileSystem)
+ return newPartWrapper(nil, p), nil
+}
+
+var errClosed = fmt.Errorf("the merger is closed")
+
+func mergeBlocks(closeCh <-chan struct{}, bw *blockWriter, br *blockReader) (*partMetadata, error) {
+ pendingBlockIsEmpty := true
+ pendingBlock := generateBlockPointer()
+ defer releaseBlockPointer(pendingBlock)
+ var tmpBlock, tmpBlock2 *blockPointer
+ var decoder *encoding.BytesBlockDecoder
+ getDecoder := func() *encoding.BytesBlockDecoder {
+ if decoder == nil {
+ decoder = generateColumnValuesDecoder()
+ }
+ return decoder
+ }
+ releaseDecoder := func() {
+ if decoder != nil {
+ releaseColumnValuesDecoder(decoder)
+ decoder = nil
+ }
+ }
+ for br.nextBlockMetadata() {
+ select {
+ case <-closeCh:
+ return nil, errClosed
+ default:
+ }
+ b := br.block
+
+ if pendingBlockIsEmpty {
+ br.loadBlockData(getDecoder())
+ pendingBlock.copyFrom(b)
+ pendingBlockIsEmpty = false
+ continue
+ }
+
+ if pendingBlock.bm.seriesID != b.bm.seriesID ||
+ (pendingBlock.isFull() && pendingBlock.bm.maxKey <= b.bm.minKey) {
+ bw.mustWriteBlock(pendingBlock.bm.seriesID, &pendingBlock.block)
+ releaseDecoder()
+ br.loadBlockData(getDecoder())
+ pendingBlock.copyFrom(b)
+ continue
+ }
+
+ if tmpBlock == nil {
+ tmpBlock = generateBlockPointer()
+ defer releaseBlockPointer(tmpBlock)
+ }
+ tmpBlock.reset()
+ tmpBlock.bm.seriesID = b.bm.seriesID
+ br.loadBlockData(getDecoder())
+ mergeTwoBlocks(tmpBlock, pendingBlock, b)
+ if len(tmpBlock.userKeys) <= maxBlockLength && tmpBlock.uncompressedSizeBytes() <= maxUncompressedBlockSize {
Review Comment:
The method `uncompressedSizeBytes()` is called on tmpBlock but this method doesn't exist in the blockPointer struct. Based on the context, it should likely be checking the block size using a different approach or the method needs to be implemented.
##########
banyand/internal/sidx/sidx.go:
##########
@@ -375,9 +393,172 @@ func (s *sidx) currentSnapshot() *snapshot {
return nil
}
-// decRef decrements the snapshot reference count (helper for snapshot interface).
-func (s *snapshot) decRef() {
- s.release()
+var errNoPartToMerge = fmt.Errorf("no part to merge")
+
+func (s *sidx) mergeParts(fileSystem fs.FileSystem, closeCh <-chan struct{}, parts []*partWrapper, partID uint64, root string) (*partWrapper, error) {
+ if len(parts) == 0 {
+ return nil, errNoPartToMerge
+ }
+ dstPath := partPath(root, partID)
+ var totalSize int64
+ pii := make([]*partMergeIter, 0, len(parts))
+ for i := range parts {
+ pmi := generatePartMergeIter()
+ pmi.mustInitFromPart(parts[i].p)
+ pii = append(pii, pmi)
+ totalSize += int64(parts[i].p.partMetadata.CompressedSizeBytes)
+ }
+ shouldCache := s.pm.ShouldCache(totalSize)
+ br := generateBlockReader()
+ br.init(pii)
+ bw := generateBlockWriter()
+ bw.mustInitForFilePart(fileSystem, dstPath, shouldCache)
+
+ pm, err := mergeBlocks(closeCh, bw, br)
+ releaseBlockWriter(bw)
+ releaseBlockReader(br)
+ for i := range pii {
+ releasePartMergeIter(pii[i])
+ }
+ if err != nil {
+ return nil, err
+ }
+ pm.mustWriteMetadata(fileSystem, dstPath)
+ fileSystem.SyncPath(dstPath)
+ p := mustOpenPart(dstPath, fileSystem)
+ return newPartWrapper(nil, p), nil
+}
+
+var errClosed = fmt.Errorf("the merger is closed")
+
+func mergeBlocks(closeCh <-chan struct{}, bw *blockWriter, br *blockReader) (*partMetadata, error) {
Review Comment:
The mergeBlocks function signature differs from the context example which returns (*partMetadata, *traceIDFilter, *tagType, error). This mismatch could cause compilation errors or incorrect behavior when integrating with existing code.
##########
banyand/internal/sidx/sidx.go:
##########
@@ -375,9 +393,172 @@ func (s *sidx) currentSnapshot() *snapshot {
return nil
}
-// decRef decrements the snapshot reference count (helper for snapshot interface).
-func (s *snapshot) decRef() {
- s.release()
+var errNoPartToMerge = fmt.Errorf("no part to merge")
+
+func (s *sidx) mergeParts(fileSystem fs.FileSystem, closeCh <-chan struct{}, parts []*partWrapper, partID uint64, root string) (*partWrapper, error) {
+ if len(parts) == 0 {
+ return nil, errNoPartToMerge
+ }
+ dstPath := partPath(root, partID)
+ var totalSize int64
+ pii := make([]*partMergeIter, 0, len(parts))
+ for i := range parts {
+ pmi := generatePartMergeIter()
+ pmi.mustInitFromPart(parts[i].p)
+ pii = append(pii, pmi)
+ totalSize += int64(parts[i].p.partMetadata.CompressedSizeBytes)
+ }
+ shouldCache := s.pm.ShouldCache(totalSize)
+ br := generateBlockReader()
+ br.init(pii)
+ bw := generateBlockWriter()
+ bw.mustInitForFilePart(fileSystem, dstPath, shouldCache)
+
+ pm, err := mergeBlocks(closeCh, bw, br)
+ releaseBlockWriter(bw)
+ releaseBlockReader(br)
+ for i := range pii {
+ releasePartMergeIter(pii[i])
+ }
+ if err != nil {
+ return nil, err
+ }
+ pm.mustWriteMetadata(fileSystem, dstPath)
+ fileSystem.SyncPath(dstPath)
+ p := mustOpenPart(dstPath, fileSystem)
+ return newPartWrapper(nil, p), nil
+}
+
+var errClosed = fmt.Errorf("the merger is closed")
+
+func mergeBlocks(closeCh <-chan struct{}, bw *blockWriter, br *blockReader) (*partMetadata, error) {
+ pendingBlockIsEmpty := true
+ pendingBlock := generateBlockPointer()
+ defer releaseBlockPointer(pendingBlock)
+ var tmpBlock, tmpBlock2 *blockPointer
+ var decoder *encoding.BytesBlockDecoder
+ getDecoder := func() *encoding.BytesBlockDecoder {
+ if decoder == nil {
+ decoder = generateColumnValuesDecoder()
+ }
+ return decoder
+ }
+ releaseDecoder := func() {
+ if decoder != nil {
+ releaseColumnValuesDecoder(decoder)
+ decoder = nil
+ }
+ }
+ for br.nextBlockMetadata() {
+ select {
+ case <-closeCh:
+ return nil, errClosed
+ default:
+ }
+ b := br.block
+
+ if pendingBlockIsEmpty {
+ br.loadBlockData(getDecoder())
+ pendingBlock.copyFrom(b)
+ pendingBlockIsEmpty = false
+ continue
+ }
+
+ if pendingBlock.bm.seriesID != b.bm.seriesID ||
+ (pendingBlock.isFull() && pendingBlock.bm.maxKey <= b.bm.minKey) {
+ bw.mustWriteBlock(pendingBlock.bm.seriesID, &pendingBlock.block)
+ releaseDecoder()
+ br.loadBlockData(getDecoder())
+ pendingBlock.copyFrom(b)
+ continue
+ }
+
+ if tmpBlock == nil {
+ tmpBlock = generateBlockPointer()
+ defer releaseBlockPointer(tmpBlock)
+ }
+ tmpBlock.reset()
+ tmpBlock.bm.seriesID = b.bm.seriesID
+ br.loadBlockData(getDecoder())
+ mergeTwoBlocks(tmpBlock, pendingBlock, b)
+ if len(tmpBlock.userKeys) <= maxBlockLength && tmpBlock.uncompressedSizeBytes() <= maxUncompressedBlockSize {
+ if len(tmpBlock.userKeys) == 0 {
+ pendingBlockIsEmpty = true
+ }
+ pendingBlock, tmpBlock = tmpBlock, pendingBlock
+ continue
+ }
+
+ if len(tmpBlock.userKeys) <= maxBlockLength {
+ bw.mustWriteBlock(tmpBlock.bm.seriesID, &tmpBlock.block)
+ pendingBlock.reset()
+ pendingBlockIsEmpty = true
+ releaseDecoder()
+ continue
+ }
+ tmpBlock.idx = maxBlockLength
+ pendingBlock.copyFrom(tmpBlock)
+ l := tmpBlock.idx
+ tmpBlock.idx = 0
+ if tmpBlock2 == nil {
+ tmpBlock2 = generateBlockPointer()
+ defer releaseBlockPointer(tmpBlock2)
+ }
+ tmpBlock2.reset()
+ tmpBlock2.append(tmpBlock, l)
+ bw.mustWriteBlock(tmpBlock.bm.seriesID, &tmpBlock2.block)
+ releaseDecoder()
+ }
+ if err := br.error(); err != nil {
+ return nil, fmt.Errorf("cannot read block to merge: %w", err)
+ }
+ if !pendingBlockIsEmpty {
+ bw.mustWriteBlock(pendingBlock.bm.seriesID, &pendingBlock.block)
+ }
+ releaseDecoder()
+ var result partMetadata
+ bw.Flush(&result)
+ return &result, nil
+}
+
+func mergeTwoBlocks(target, left, right *blockPointer) {
+ appendIfEmpty := func(ib1, ib2 *blockPointer) bool {
+ if ib1.idx >= len(ib1.userKeys) {
+ target.appendAll(ib2)
+ return true
+ }
+ return false
+ }
+
+ defer target.updateMetadata()
+
+ if left.bm.maxKey < right.bm.minKey {
+ target.appendAll(left)
+ target.appendAll(right)
+ return
+ }
+ if right.bm.maxKey < left.bm.minKey {
+ target.appendAll(right)
+ target.appendAll(left)
+ return
+ }
+ if appendIfEmpty(left, right) || appendIfEmpty(right, left) {
+ return
+ }
+
+ for {
+ i := left.idx
+ uk2 := right.userKeys[right.idx]
+ for i < len(left.userKeys) && left.userKeys[i] <= uk2 {
+ i++
+ }
+ target.append(left, i)
+ left.idx = i
+ if appendIfEmpty(left, right) {
+ return
+ }
+ left, right = right, left
+ }
Review Comment:
[nitpick] The mergeTwoBlocks implementation differs significantly from the context example, which was much simpler (just `target.appendAll(left); target.appendAll(right)`). The current complex implementation with multiple conditional branches makes it harder to understand and maintain. Consider if this complexity is necessary or if the simpler approach would suffice.
```suggestion
target.appendAll(left)
target.appendAll(right)
target.updateMetadata()
```
##########
banyand/internal/sidx/block.go:
##########
@@ -358,3 +359,276 @@ func mustWriteDataTo(db *dataBlock, data [][]byte, dataWriter *writer) {
db.size = uint64(len(compressedData))
dataWriter.MustWrite(compressedData)
}
+
+type blockPointer struct {
+ block
+ bm blockMetadata
+ idx int
+}
+
+func (bi *blockPointer) updateMetadata() {
+ if len(bi.block.userKeys) == 0 {
+ return
+ }
+ // only update minKey and maxKey since they are used for merging
+ // blockWriter will recompute all fields
+ bi.bm.minKey = bi.block.userKeys[0]
+ bi.bm.maxKey = bi.block.userKeys[len(bi.userKeys)-1]
Review Comment:
Incorrect field reference - should be `len(bi.block.userKeys)-1` instead of `len(bi.userKeys)-1` since userKeys belongs to the block, not the blockPointer directly.
```suggestion
bi.bm.maxKey = bi.block.userKeys[len(bi.block.userKeys)-1]
```
##########
banyand/internal/sidx/block_reader.go:
##########
@@ -0,0 +1,235 @@
+// 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 sidx
+
+import (
+ "container/heap"
+ "errors"
+ "fmt"
+ "io"
+
+ "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"
+)
+
+type seqReader struct {
+ sr fs.SeqReader
+ r fs.Reader
+ bytesRead uint64
+}
+
+func (sr *seqReader) reset() {
+ sr.r = nil
+ if sr.sr != nil {
+ fs.MustClose(sr.sr)
+ }
+ sr.sr = nil
+ sr.bytesRead = 0
+}
+
+func (sr *seqReader) Path() string {
+ return sr.r.Path()
+}
+
+func (sr *seqReader) init(r fs.Reader) {
+ sr.reset()
+ sr.sr = r.SequentialRead()
+ sr.r = r
+}
+
+func (sr *seqReader) mustReadFull(data []byte) {
+ n, err := io.ReadFull(sr.sr, data)
+ if err != nil {
+ if errors.Is(err, io.EOF) {
+ return
+ }
+ logger.Panicf("cannot read data: %v", err)
+ }
+ if n != len(data) {
+ logger.Panicf("cannot read full data: %d/%d", n, len(data))
+ }
+ sr.bytesRead += uint64(n)
+}
+
+func generateSeqReader() *seqReader {
+ if v := seqReaderPool.Get(); v != nil {
+ return v
+ }
+ return &seqReader{}
+}
+
+func releaseSeqReader(sr *seqReader) {
+ sr.reset()
+ seqReaderPool.Put(sr)
+}
+
+var seqReaderPool = pool.Register[*seqReader]("sidx-seqReader")
+
+type seqReaders struct {
+ tagMetadata map[string]*seqReader
+ tagData map[string]*seqReader
+ tagFilters map[string]*seqReader
+ primary seqReader
+ keys seqReader
+ data seqReader
+}
+
+func (sr *seqReaders) reset() {
+ sr.primary.reset()
+ sr.keys.reset()
+ sr.data.reset()
+ if sr.tagMetadata != nil {
+ for k, r := range sr.tagMetadata {
+ releaseSeqReader(r)
+ delete(sr.tagMetadata, k)
+ releaseSeqReader(r)
+ delete(sr.tagData, k)
+ releaseSeqReader(r)
+ delete(sr.tagFilters, k)
Review Comment:
The same reader `r` is being released twice in consecutive lines (102 and 104), which could cause issues with the pool management. The second release should use the corresponding reader from `sr.tagData[k]` and `sr.tagFilters[k]` respectively.
```suggestion
if r2, ok := sr.tagData[k]; ok {
releaseSeqReader(r2)
delete(sr.tagData, k)
}
if r3, ok := sr.tagFilters[k]; ok {
releaseSeqReader(r3)
delete(sr.tagFilters, k)
}
```
##########
banyand/internal/sidx/block.go:
##########
@@ -358,3 +359,276 @@ func mustWriteDataTo(db *dataBlock, data [][]byte, dataWriter *writer) {
db.size = uint64(len(compressedData))
dataWriter.MustWrite(compressedData)
}
+
+type blockPointer struct {
+ block
+ bm blockMetadata
+ idx int
+}
+
+func (bi *blockPointer) updateMetadata() {
+ if len(bi.block.userKeys) == 0 {
+ return
+ }
+ // only update minKey and maxKey since they are used for merging
+ // blockWriter will recompute all fields
+ bi.bm.minKey = bi.block.userKeys[0]
+ bi.bm.maxKey = bi.block.userKeys[len(bi.userKeys)-1]
+}
+
+func (bi *blockPointer) copyFrom(src *blockPointer) {
+ bi.reset()
+ bi.bm.copyFrom(&src.bm)
+ bi.appendAll(src)
+}
+
+func (bi *blockPointer) appendAll(b *blockPointer) {
+ if len(b.userKeys) == 0 {
+ return
+ }
+ bi.append(b, len(b.userKeys))
+}
+
+var log = logger.GetLogger("sidx").Named("block")
+
+func (bi *blockPointer) append(b *blockPointer, offset int) {
+ if offset <= b.idx {
+ return
+ }
+ if len(bi.tags) == 0 && len(b.tags) > 0 {
+ fullTagAppend(bi, b, offset)
+ } else {
+ if err := fastTagAppend(bi, b, offset); err != nil {
+ if log.Debug().Enabled() {
+ log.Debug().Msgf("fastTagMerge failed: %v; falling back to fullTagMerge", err)
+ }
+ fullTagAppend(bi, b, offset)
+ }
+ }
+
+ assertIdxAndOffset("userKeys", len(b.userKeys), bi.idx, offset)
+ bi.userKeys = append(bi.userKeys, b.userKeys[b.idx:offset]...)
+ assertIdxAndOffset("data", len(b.data), bi.idx, offset)
+ bi.data = append(bi.data, b.data[b.idx:offset]...)
+}
+
+func fastTagAppend(bi, b *blockPointer, offset int) error {
+ if len(bi.tags) != len(b.tags) {
+ return fmt.Errorf("unexpected number of tags: got %d; want %d", len(b.tags), len(bi.tags))
+ }
+ for i := range bi.tags {
+ if bi.tags[i].name != b.tags[i].name {
+ return fmt.Errorf("unexpected tag name for tag %q: got %q; want %q",
+ bi.tags[i].name, b.tags[i].name, bi.tags[i].name)
+ }
+ assertIdxAndOffset(b.tags[i].name, len(b.tags[i].values), b.idx, offset)
+ bi.tags[i].values = append(bi.tags[i].values, b.tags[i].values[b.idx:offset]...)
+ }
+ return nil
+}
+
+func fullTagAppend(bi, b *blockPointer, offset int) {
+ existDataSize := len(bi.userKeys)
+
+ if len(bi.tags) == 0 {
+ for _, t := range b.tags {
+ newTagData := tagData{name: t.name, valueType: t.valueType}
+ for j := 0; j < existDataSize; j++ {
+ newTagData.values = append(newTagData.values, nil)
+ }
+ assertIdxAndOffset(t.name, len(t.values), b.idx, offset)
+ newTagData.values = append(newTagData.values, t.values[b.idx:offset]...)
+ bi.tags[t.name] = &newTagData
+ }
+ return
+ }
+
+ for _, t := range b.tags {
+ if existingTag, exists := bi.tags[t.name]; exists {
+ assertIdxAndOffset(t.name, len(t.values), b.idx, offset)
+ existingTag.values = append(existingTag.values, t.values[b.idx:offset]...)
+ } else {
+ newTagData := tagData{name: t.name, valueType: t.valueType}
+ for j := 0; j < existDataSize; j++ {
+ newTagData.values = append(newTagData.values, nil)
+ }
+ assertIdxAndOffset(t.name, len(t.values), b.idx, offset)
+ newTagData.values = append(newTagData.values, t.values[b.idx:offset]...)
+ bi.tags[t.name] = &newTagData
+ }
+ }
+
+ sourceTags := make(map[string]struct{})
+ for _, t := range b.tags {
+ sourceTags[t.name] = struct{}{}
+ }
+
+ emptySize := offset - b.idx
+ for i := range bi.tags {
+ if _, exists := sourceTags[bi.tags[i].name]; !exists {
+ for j := 0; j < emptySize; j++ {
+ bi.tags[i].values = append(bi.tags[i].values, nil)
+ }
+ }
+ }
+}
+
+func assertIdxAndOffset(name string, length int, idx int, offset int) {
+ if idx >= offset {
+ logger.Panicf("%q idx %d must be less than offset %d", name, idx, offset)
+ }
+ if offset > length {
+ logger.Panicf("%q offset %d must be less than or equal to length %d", name, offset, length)
+ }
+}
+
+func (bi *blockPointer) isFull() bool {
+ return bi.bm.count >= maxBlockLength || bi.bm.uncompressedSize >= maxUncompressedBlockSize
Review Comment:
The method references `bi.bm.uncompressedSize` but the blockMetadata struct uses `uncompressedSizeBytes` field name based on other parts of the code. This field name inconsistency will cause compilation errors.
```suggestion
return bi.bm.count >= maxBlockLength || bi.bm.uncompressedSizeBytes >= maxUncompressedBlockSize
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement sidx flushing and merging [skywalking-banyandb]
Posted by "hanahmily (via GitHub)" <gi...@apache.org>.
hanahmily merged PR #745:
URL: https://github.com/apache/skywalking-banyandb/pull/745
--
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