You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by ha...@apache.org on 2025/09/02 23:28:58 UTC
(skywalking-banyandb) branch main updated: Implement sidx flushing and merging (#745)
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
The following commit(s) were added to refs/heads/main by this push:
new c4000531 Implement sidx flushing and merging (#745)
c4000531 is described below
commit c4000531c0db7f14393ed899f363e35d38e1c824
Author: Huang Youliang <bu...@gmail.com>
AuthorDate: Wed Sep 3 07:28:53 2025 +0800
Implement sidx flushing and merging (#745)
* Implement sidx flushing and merging
---
banyand/internal/sidx/block.go | 255 +++++++++++++++++
banyand/internal/sidx/block_reader.go | 243 ++++++++++++++++
banyand/internal/sidx/block_reader_test.go | 218 ++++++++++++++
banyand/internal/sidx/gc.go | 57 ++++
banyand/internal/sidx/gc_test.go | 71 +++++
banyand/internal/sidx/interfaces.go | 4 +-
banyand/internal/sidx/introducer.go | 4 +-
banyand/internal/sidx/merge.go | 228 +++++++++++++++
banyand/internal/sidx/metadata.go | 27 ++
banyand/internal/sidx/multi_sidx_query_test.go | 4 +-
banyand/internal/sidx/options.go | 63 +---
banyand/internal/sidx/options_test.go | 164 +----------
banyand/internal/sidx/part.go | 6 +-
banyand/internal/sidx/part_iter.go | 144 ++++++++++
banyand/internal/sidx/sidx.go | 317 +++++++++++++--------
banyand/internal/sidx/sidx_test.go | 32 ++-
banyand/internal/sidx/snapshot.go | 52 ++++
banyand/internal/sidx/snapshot_test.go | 23 +-
banyand/trace/flusher.go | 5 +-
banyand/trace/merger.go | 9 +-
banyand/trace/merger_test.go | 46 +--
banyand/trace/tstable.go | 45 ++-
banyand/trace/write_standalone.go | 2 +-
.../standalone/cold_query/query_suite_test.go | 5 +
.../query_ondisk/query_ondisk_suite_test.go | 5 +
25 files changed, 1629 insertions(+), 400 deletions(-)
diff --git a/banyand/internal/sidx/block.go b/banyand/internal/sidx/block.go
index 263515de..22473bad 100644
--- a/banyand/internal/sidx/block.go
+++ b/banyand/internal/sidx/block.go
@@ -27,6 +27,7 @@ import (
"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/logger"
pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
"github.com/apache/skywalking-banyandb/pkg/pool"
)
@@ -346,3 +347,257 @@ 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
+}
+
+func (bi *blockPointer) reset() {
+ bi.idx = 0
+ bi.block.reset()
+ bi.bm.reset()
+}
+
+func generateBlockPointer() *blockPointer {
+ v := blockPointerPool.Get()
+ if v == nil {
+ return &blockPointer{}
+ }
+ return v
+}
+
+func releaseBlockPointer(bi *blockPointer) {
+ bi.reset()
+ blockPointerPool.Put(bi)
+}
+
+func (b *block) mustSeqReadFrom(decoder *encoding.BytesBlockDecoder, sr *seqReaders, bm blockMetadata) {
+ b.reset()
+ if err := b.readUserKeys(sr, &bm); err != nil {
+ panic(fmt.Sprintf("failed to read user keys: %v", err))
+ }
+ if err := b.readData(decoder, sr, &bm); err != nil {
+ panic(fmt.Sprintf("failed to read data payloads: %v", err))
+ }
+ if err := b.readTagData(decoder, sr, &bm); err != nil {
+ panic(fmt.Sprintf("failed to read tag data: %v", err))
+ }
+}
+
+func (b *block) readUserKeys(sr *seqReaders, bm *blockMetadata) error {
+ bb := bigValuePool.Get()
+ if bb == nil {
+ bb = &bytes.Buffer{}
+ }
+ defer func() {
+ bb.Buf = bb.Buf[:0]
+ bigValuePool.Put(bb)
+ }()
+ bb.Buf = bytes.ResizeOver(bb.Buf[:0], int(bm.keysBlock.size))
+ sr.keys.mustReadFull(bb.Buf)
+ var err error
+ b.userKeys, err = encoding.BytesToInt64List(b.userKeys[:0], bb.Buf, bm.keysEncodeType, bm.minKey, int(bm.count))
+ if err != nil {
+ return fmt.Errorf("cannot decode user keys: %w", err)
+ }
+ return nil
+}
+
+func (b *block) readData(decoder *encoding.BytesBlockDecoder, sr *seqReaders, bm *blockMetadata) error {
+ bb := bigValuePool.Get()
+ if bb == nil {
+ bb = &bytes.Buffer{}
+ }
+ defer func() {
+ bb.Buf = bb.Buf[:0]
+ bigValuePool.Put(bb)
+ }()
+ bb.Buf = bytes.ResizeOver(bb.Buf[:0], int(bm.dataBlock.size))
+ sr.data.mustReadFull(bb.Buf)
+ dataBuf, err := zstd.Decompress(bb.Buf[:0], bb.Buf)
+ if err != nil {
+ return fmt.Errorf("cannot decompress data: %w", err)
+ }
+ b.data, err = decoder.Decode(b.data[:0], dataBuf, bm.count)
+ if err != nil {
+ return fmt.Errorf("cannot decode data payloads: %w", err)
+ }
+ return nil
+}
+
+func (b *block) readTagData(decoder *encoding.BytesBlockDecoder, sr *seqReaders, bm *blockMetadata) error {
+ if b.tags == nil {
+ b.tags = make(map[string]*tagData)
+ }
+ for tagName, tagBlock := range bm.tagsBlocks {
+ if err := b.readSingleTag(decoder, sr, tagName, &tagBlock, int(bm.count)); err != nil {
+ return fmt.Errorf("failed to read tag %s: %w", tagName, err)
+ }
+ }
+ return nil
+}
+
+func (b *block) readSingleTag(decoder *encoding.BytesBlockDecoder, sr *seqReaders, tagName string, tagBlock *dataBlock, count int) error {
+ tmReader, tmExists := sr.tagMetadata[tagName]
+ if !tmExists {
+ return fmt.Errorf("tag metadata reader not found for tag %s", tagName)
+ }
+ tdReader, tdExists := sr.tagData[tagName]
+ if !tdExists {
+ return fmt.Errorf("tag data reader not found for tag %s", tagName)
+ }
+
+ bb := bigValuePool.Get()
+ if bb == nil {
+ bb = &bytes.Buffer{}
+ }
+ defer func() {
+ bb.Buf = bb.Buf[:0]
+ bigValuePool.Put(bb)
+ }()
+
+ bb.Buf = bytes.ResizeOver(bb.Buf[:0], int(tagBlock.size))
+ tmReader.mustReadFull(bb.Buf)
+ tm, err := unmarshalTagMetadata(bb.Buf)
+ if err != nil {
+ return fmt.Errorf("cannot unmarshal tag metadata: %w", err)
+ }
+ defer releaseTagMetadata(tm)
+
+ bb.Buf = bytes.ResizeOver(bb.Buf[:0], int(tm.dataBlock.size))
+ tdReader.mustReadFull(bb.Buf)
+ td := generateTagData()
+ td.name = tagName
+ td.valueType = tm.valueType
+ td.values, err = internalencoding.DecodeTagValues(td.values[:0], decoder, bb, tm.valueType, count)
+ if err != nil {
+ releaseTagData(td)
+ return fmt.Errorf("cannot decode tag values: %w", err)
+ }
+ b.tags[tagName] = td
+ return nil
+}
+
+var blockPointerPool = pool.Register[*blockPointer]("sidx-blockPointer")
diff --git a/banyand/internal/sidx/block_reader.go b/banyand/internal/sidx/block_reader.go
new file mode 100644
index 00000000..64e4a747
--- /dev/null
+++ b/banyand/internal/sidx/block_reader.go
@@ -0,0 +1,243 @@
+// 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)
+ }
+ for k, r := range sr.tagData {
+ releaseSeqReader(r)
+ delete(sr.tagData, k)
+ }
+ for k, r := range sr.tagFilters {
+ releaseSeqReader(r)
+ delete(sr.tagFilters, k)
+ }
+ }
+}
+
+func (sr *seqReaders) init(p *part) {
+ sr.reset()
+ sr.primary.init(p.primary)
+ sr.keys.init(p.keys)
+ sr.data.init(p.data)
+ if sr.tagData == nil {
+ sr.tagData = make(map[string]*seqReader)
+ sr.tagMetadata = make(map[string]*seqReader)
+ sr.tagFilters = make(map[string]*seqReader)
+ }
+ for k, r := range p.tagData {
+ sr.tagData[k] = generateSeqReader()
+ sr.tagData[k].init(r)
+ }
+ for k, r := range p.tagMetadata {
+ sr.tagMetadata[k] = generateSeqReader()
+ sr.tagMetadata[k].init(r)
+ }
+ for k, r := range p.tagFilters {
+ sr.tagFilters[k] = generateSeqReader()
+ sr.tagFilters[k].init(r)
+ }
+}
+
+type blockReader struct {
+ err error
+ block *blockPointer
+ pih partMergeIterHeap
+ nextBlockNoop bool
+}
+
+func (br *blockReader) reset() {
+ br.block = nil
+ for i := range br.pih {
+ br.pih[i] = nil
+ }
+ br.pih = br.pih[:0]
+ br.nextBlockNoop = false
+ br.err = nil
+}
+
+func (br *blockReader) init(pii []*partMergeIter) {
+ br.reset()
+ for _, pi := range pii {
+ if pi.nextBlockMetadata() {
+ br.pih = append(br.pih, pi)
+ continue
+ }
+ if err := pi.error(); err != nil {
+ br.err = fmt.Errorf("can't get the block to merge: %w", err)
+ return
+ }
+ }
+ if len(br.pih) == 0 {
+ br.err = io.EOF
+ return
+ }
+ heap.Init(&br.pih)
+ br.block = &br.pih[0].block
+ br.nextBlockNoop = true
+}
+
+func (br *blockReader) nextBlockMetadata() bool {
+ if br.err != nil {
+ return false
+ }
+ if br.nextBlockNoop {
+ br.nextBlockNoop = false
+ return true
+ }
+
+ br.err = br.nextMetadata()
+ if br.err != nil {
+ if errors.Is(br.err, io.EOF) {
+ return false
+ }
+ br.err = fmt.Errorf("can't get the block to merge: %w", br.err)
+ return false
+ }
+ return true
+}
+
+func (br *blockReader) nextMetadata() error {
+ head := br.pih[0]
+ if head.nextBlockMetadata() {
+ heap.Fix(&br.pih, 0)
+ br.block = &br.pih[0].block
+ return nil
+ }
+
+ if err := head.error(); err != nil {
+ br.block = nil
+ return err
+ }
+
+ heap.Pop(&br.pih)
+
+ if len(br.pih) == 0 {
+ br.block = nil
+ return io.EOF
+ }
+
+ br.block = &br.pih[0].block
+ return nil
+}
+
+func (br *blockReader) loadBlockData(decoder *encoding.BytesBlockDecoder) {
+ br.pih[0].mustLoadBlockData(decoder, br.block)
+}
+
+func (br *blockReader) error() error {
+ if errors.Is(br.err, io.EOF) {
+ return nil
+ }
+ return br.err
+}
+
+var blockReaderPool = pool.Register[*blockReader]("sidx-blockReader")
+
+func generateBlockReader() *blockReader {
+ if v := blockReaderPool.Get(); v != nil {
+ return v
+ }
+ return &blockReader{}
+}
+
+func releaseBlockReader(br *blockReader) {
+ br.reset()
+ blockReaderPool.Put(br)
+}
diff --git a/banyand/internal/sidx/block_reader_test.go b/banyand/internal/sidx/block_reader_test.go
new file mode 100644
index 00000000..5d56a840
--- /dev/null
+++ b/banyand/internal/sidx/block_reader_test.go
@@ -0,0 +1,218 @@
+// 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 (
+ "errors"
+ "sort"
+ "testing"
+
+ "github.com/google/go-cmp/cmp"
+ "github.com/google/go-cmp/cmp/cmpopts"
+ "github.com/stretchr/testify/require"
+
+ "github.com/apache/skywalking-banyandb/pkg/fs"
+ pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
+ "github.com/apache/skywalking-banyandb/pkg/test"
+)
+
+var (
+ elementsSet1 = func() *elements {
+ es := generateElements()
+ es.mustAppend(1, 100, make([]byte, 1600), []Tag{
+ {Name: "service", Value: []byte("service1"), ValueType: pbv1.ValueTypeStr},
+ })
+ es.mustAppend(2, 200, make([]byte, 40), []Tag{
+ {Name: "env", Value: []byte("prod"), ValueType: pbv1.ValueTypeStr},
+ })
+ es.mustAppend(3, 300, make([]byte, 25), []Tag{
+ {Name: "region", Value: []byte("us"), ValueType: pbv1.ValueTypeStr},
+ })
+ return es
+ }()
+
+ elementsSet2 = func() *elements {
+ es := generateElements()
+ es.mustAppend(1, 150, make([]byte, 1600), []Tag{
+ {Name: "service", Value: []byte("service1"), ValueType: pbv1.ValueTypeStr},
+ })
+ es.mustAppend(2, 250, make([]byte, 40), []Tag{
+ {Name: "env", Value: []byte("prod"), ValueType: pbv1.ValueTypeStr},
+ })
+ es.mustAppend(3, 350, make([]byte, 25), []Tag{
+ {Name: "region", Value: []byte("us"), ValueType: pbv1.ValueTypeStr},
+ })
+ return es
+ }()
+
+ sameKeyElements = func() *elements {
+ es := generateElements()
+ es.mustAppend(1, 100, make([]byte, 25), []Tag{
+ {Name: "service", Value: []byte("service1"), ValueType: pbv1.ValueTypeStr},
+ })
+ es.mustAppend(1, 100, make([]byte, 35), []Tag{
+ {Name: "service", Value: []byte("service2"), ValueType: pbv1.ValueTypeStr},
+ })
+ return es
+ }()
+)
+
+func Test_blockReader_nextBlock(t *testing.T) {
+ tests := []struct {
+ wantErr error
+ name string
+ elemsList []*elements
+ want []blockMetadata
+ }{
+ {
+ name: "Test with no elements",
+ elemsList: []*elements{},
+ },
+ {
+ name: "Test with single part",
+ elemsList: []*elements{elementsSet1},
+ want: []blockMetadata{
+ {seriesID: 1, count: 1, uncompressedSize: 1623},
+ {seriesID: 2, count: 1, uncompressedSize: 55},
+ {seriesID: 3, count: 1, uncompressedSize: 41},
+ },
+ },
+ {
+ name: "Test with multiple parts with different keys",
+ elemsList: []*elements{elementsSet1, elementsSet2},
+ want: []blockMetadata{
+ {seriesID: 1, count: 1, uncompressedSize: 1623},
+ {seriesID: 1, count: 1, uncompressedSize: 1623},
+ {seriesID: 2, count: 1, uncompressedSize: 55},
+ {seriesID: 2, count: 1, uncompressedSize: 55},
+ {seriesID: 3, count: 1, uncompressedSize: 41},
+ {seriesID: 3, count: 1, uncompressedSize: 41},
+ },
+ },
+ {
+ name: "Test with single part with same key",
+ elemsList: []*elements{sameKeyElements},
+ want: []blockMetadata{
+ {seriesID: 1, count: 2, uncompressedSize: 106},
+ },
+ },
+ {
+ name: "Test with multiple parts with same keys",
+ elemsList: []*elements{elementsSet1, elementsSet1},
+ want: []blockMetadata{
+ {seriesID: 1, count: 1, uncompressedSize: 1623},
+ {seriesID: 1, count: 1, uncompressedSize: 1623},
+ {seriesID: 2, count: 1, uncompressedSize: 55},
+ {seriesID: 2, count: 1, uncompressedSize: 55},
+ {seriesID: 3, count: 1, uncompressedSize: 41},
+ {seriesID: 3, count: 1, uncompressedSize: 41},
+ },
+ },
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ verify := func(t *testing.T, pp []*part) {
+ var pii []*partMergeIter
+ for _, p := range pp {
+ pmi := &partMergeIter{}
+ pmi.mustInitFromPart(p)
+ pii = append(pii, pmi)
+ }
+ reader := generateBlockReader()
+ defer releaseBlockReader(reader)
+ reader.init(pii)
+ var got []blockMetadata
+ for reader.nextBlockMetadata() {
+ if reader.block.bm.seriesID == 0 {
+ t.Errorf("Expected curBlock to be initialized, but it was nil")
+ }
+ var tagNames []string
+ for tagName := range reader.block.bm.tagsBlocks {
+ tagNames = append(tagNames, tagName)
+ }
+ if !sort.IsSorted(sort.StringSlice(tagNames)) {
+ t.Errorf("Expected tagNames %q to be sorted, but it was not", tagNames)
+ }
+ got = append(got, reader.block.bm)
+ }
+
+ if !errors.Is(reader.error(), tt.wantErr) {
+ t.Errorf("Unexpected error: got %v, want %v", reader.err, tt.wantErr)
+ }
+
+ if diff := cmp.Diff(got, tt.want,
+ cmpopts.IgnoreFields(blockMetadata{}, "tagsBlocks"),
+ cmpopts.IgnoreFields(blockMetadata{}, "tagProjection"),
+ cmpopts.IgnoreFields(blockMetadata{}, "dataBlock"),
+ cmpopts.IgnoreFields(blockMetadata{}, "keysBlock"),
+ cmpopts.IgnoreFields(blockMetadata{}, "minKey"),
+ cmpopts.IgnoreFields(blockMetadata{}, "maxKey"),
+ cmpopts.IgnoreFields(blockMetadata{}, "keysEncodeType"),
+ cmp.AllowUnexported(blockMetadata{}),
+ ); diff != "" {
+ t.Errorf("Unexpected blockMetadata (-got +want):\n%s", diff)
+ }
+ }
+
+ t.Run("memory parts", func(t *testing.T) {
+ var mpp []*memPart
+ defer func() {
+ for _, mp := range mpp {
+ releaseMemPart(mp)
+ }
+ }()
+ var pp []*part
+ for _, elems := range tt.elemsList {
+ mp := generateMemPart()
+ mpp = append(mpp, mp)
+ mp.mustInitFromElements(elems)
+ pp = append(pp, openMemPart(mp))
+ }
+ verify(t, pp)
+ })
+
+ t.Run("file parts", func(t *testing.T) {
+ var mpp []*memPart
+ var fpp []*partWrapper
+ tmpPath, defFn := test.Space(require.New(t))
+ defer func() {
+ for _, mp := range mpp {
+ releaseMemPart(mp)
+ }
+ for _, pw := range fpp {
+ pw.release()
+ }
+ defFn()
+ }()
+ var pp []*part
+ fileSystem := fs.NewLocalFileSystem()
+ for i, elems := range tt.elemsList {
+ mp := generateMemPart()
+ mpp = append(mpp, mp)
+ mp.mustInitFromElements(elems)
+ mp.mustFlush(fileSystem, partPath(tmpPath, uint64(i)))
+ filePW := newPartWrapper(nil, mustOpenPart(partPath(tmpPath, uint64(i)), fileSystem))
+ filePW.p.partMetadata.ID = uint64(i)
+ fpp = append(fpp, filePW)
+ pp = append(pp, filePW.p)
+ }
+ verify(t, pp)
+ })
+ })
+ }
+}
diff --git a/banyand/internal/sidx/gc.go b/banyand/internal/sidx/gc.go
new file mode 100644
index 00000000..5b303029
--- /dev/null
+++ b/banyand/internal/sidx/gc.go
@@ -0,0 +1,57 @@
+// 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 (
+ "path/filepath"
+)
+
+type garbageCleaner struct {
+ parent *sidx
+ deletableEpochs []uint64
+ liveEpoch uint64
+}
+
+func (g *garbageCleaner) init(parent *sidx) {
+ g.parent = parent
+}
+
+func (g *garbageCleaner) registerSnapshot(snapshot *snapshot) {
+ if g.liveEpoch > 0 {
+ g.deletableEpochs = append(g.deletableEpochs, g.liveEpoch)
+ }
+ g.liveEpoch = snapshot.epoch
+}
+
+func (g *garbageCleaner) removePart(partID uint64) {
+ g.parent.fileSystem.MustRMAll(partPath(g.parent.root, partID))
+}
+
+func (g *garbageCleaner) clean() {
+ var remainingEpochs []uint64
+ for _, deletableEpoch := range g.deletableEpochs {
+ path := filepath.Join(g.parent.root, snapshotName(deletableEpoch))
+ err := g.parent.fileSystem.DeleteFile(path)
+ if err == nil {
+ continue
+ }
+ g.parent.l.Warn().Err(err).Msgf("cannot delete snapshot file: %s", path)
+ remainingEpochs = append(remainingEpochs, deletableEpoch)
+ }
+ g.deletableEpochs = remainingEpochs
+}
diff --git a/banyand/internal/sidx/gc_test.go b/banyand/internal/sidx/gc_test.go
new file mode 100644
index 00000000..4dea7d99
--- /dev/null
+++ b/banyand/internal/sidx/gc_test.go
@@ -0,0 +1,71 @@
+// 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 (
+ "fmt"
+ "reflect"
+ "testing"
+
+ "github.com/apache/skywalking-banyandb/pkg/fs"
+)
+
+func TestDeletableEpochs(t *testing.T) {
+ tests := []struct {
+ name string
+ knownEpochs []uint64
+ deletableEpochs []uint64
+ n int
+ }{
+ {
+ name: "empty",
+ n: 1,
+ knownEpochs: nil,
+ deletableEpochs: nil,
+ },
+ {
+ name: "one",
+ n: 1,
+ knownEpochs: []uint64{1},
+ deletableEpochs: nil,
+ },
+ {
+ name: "many",
+ n: 1,
+ knownEpochs: []uint64{1, 2, 3, 4},
+ deletableEpochs: []uint64{1, 2, 3},
+ },
+ }
+
+ for _, test := range tests {
+ test := test
+ t.Run(fmt.Sprintf("%s-%d", test.name, test.n), func(t *testing.T) {
+ var gc garbageCleaner
+ gc.init(&sidx{
+ fileSystem: fs.NewLocalFileSystem(),
+ })
+
+ for _, epoch := range test.knownEpochs {
+ gc.registerSnapshot(&snapshot{epoch: epoch})
+ }
+ if !reflect.DeepEqual(gc.deletableEpochs, test.deletableEpochs) {
+ t.Errorf("expected deletable: %#v, got %#v", test.deletableEpochs, gc.deletableEpochs)
+ }
+ })
+ }
+}
diff --git a/banyand/internal/sidx/interfaces.go b/banyand/internal/sidx/interfaces.go
index 9d5ccce1..d02708a2 100644
--- a/banyand/internal/sidx/interfaces.go
+++ b/banyand/internal/sidx/interfaces.go
@@ -37,7 +37,7 @@ import (
type SIDX interface {
// Write performs batch write operations. All writes must be submitted as batches.
// Elements within each batch should be pre-sorted by the caller for optimal performance.
- Write(ctx context.Context, reqs []WriteRequest) error
+ Write(ctx context.Context, reqs []WriteRequest, partID uint64) error
// Query executes a query with key range and tag filtering.
// Returns a QueryResponse directly with all results loaded.
@@ -76,7 +76,7 @@ type Merger interface {
// and coordinate with the introducer loop for snapshot updates.
// This operation is user-controlled and synchronous.
// Returns error if merge operation fails.
- Merge() error
+ Merge(partIDs []uint64, newPartID uint64, closeCh <-chan struct{}) error
}
// Writer handles write path operations for batch processing.
diff --git a/banyand/internal/sidx/introducer.go b/banyand/internal/sidx/introducer.go
index c6c0b6c9..c75700ae 100644
--- a/banyand/internal/sidx/introducer.go
+++ b/banyand/internal/sidx/introducer.go
@@ -199,8 +199,8 @@ func (s *sidx) introduceMerged(nextIntroduction *mergerIntroduction, epoch uint6
}
func (s *sidx) replaceSnapshot(next *snapshot) {
- s.Lock()
- defer s.Unlock()
+ s.mu.Lock()
+ defer s.mu.Unlock()
if s.snapshot != nil {
s.snapshot.decRef()
}
diff --git a/banyand/internal/sidx/merge.go b/banyand/internal/sidx/merge.go
new file mode 100644
index 00000000..eb397b64
--- /dev/null
+++ b/banyand/internal/sidx/merge.go
@@ -0,0 +1,228 @@
+// 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 (
+ "fmt"
+ "slices"
+
+ "github.com/apache/skywalking-banyandb/pkg/encoding"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
+)
+
+var (
+ errNoPartToMerge = fmt.Errorf("no part to merge")
+ errClosed = fmt.Errorf("the merger is closed")
+)
+
+// Merge implements Merger interface.
+func (s *sidx) Merge(partIDs []uint64, newPartID uint64, closeCh <-chan struct{}) error {
+ // Get current snapshot
+ snap := s.currentSnapshot()
+ if snap == nil {
+ return nil
+ }
+ defer snap.decRef()
+
+ // Create merge introduction
+ mergeIntro := generateMergerIntroduction()
+ defer releaseMergerIntroduction(mergeIntro)
+ mergeIntro.applied = make(chan struct{})
+
+ // Select parts to merge
+ var partsToMerge []*partWrapper
+ for _, pw := range snap.parts {
+ if pw.isActive() && !pw.isMemPart() && slices.Contains(partIDs, pw.ID()) {
+ partsToMerge = append(partsToMerge, pw)
+ }
+ }
+
+ if len(partsToMerge) < 2 {
+ return nil
+ }
+
+ // Mark parts for merging
+ for _, pw := range partsToMerge {
+ mergeIntro.merged[pw.ID()] = struct{}{}
+ }
+
+ // Create new merged part
+ newPart, err := s.mergeParts(s.fileSystem, closeCh, partsToMerge, newPartID, s.root)
+ if err != nil {
+ return err
+ }
+ mergeIntro.newPart = newPart.p
+
+ // Send to introducer loop
+ s.mergeCh <- mergeIntro
+
+ // Wait for merge to complete
+ <-mergeIntro.applied
+
+ return nil
+}
+
+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
+}
+
+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 {
+ if len(tmpBlock.userKeys) == 0 {
+ pendingBlockIsEmpty = true
+ }
+ pendingBlock, tmpBlock = tmpBlock, pendingBlock
+ continue
+ }
+ bw.mustWriteBlock(tmpBlock.bm.seriesID, &tmpBlock2.block)
+ releaseDecoder()
+ pendingBlock.reset()
+ tmpBlock.reset()
+ pendingBlockIsEmpty = true
+ }
+ 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
+ }
+}
diff --git a/banyand/internal/sidx/metadata.go b/banyand/internal/sidx/metadata.go
index f3fffbcc..c36b8a60 100644
--- a/banyand/internal/sidx/metadata.go
+++ b/banyand/internal/sidx/metadata.go
@@ -21,10 +21,16 @@ import (
"encoding/json"
"fmt"
"maps"
+ "path/filepath"
"sort"
+ "github.com/pkg/errors"
+
"github.com/apache/skywalking-banyandb/api/common"
+ "github.com/apache/skywalking-banyandb/banyand/internal/storage"
"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"
)
@@ -44,6 +50,19 @@ type partMetadata struct {
ID uint64 `json:"id"` // Unique part identifier
}
+func validatePartMetadata(fileSystem fs.FileSystem, partPath string) error {
+ metadataPath := filepath.Join(partPath, manifestFilename)
+ metadata, err := fileSystem.Read(metadataPath)
+ if err != nil {
+ return errors.WithMessage(err, "cannot read metadata.json")
+ }
+ var pm partMetadata
+ if err := json.Unmarshal(metadata, &pm); err != nil {
+ return errors.WithMessage(err, "cannot parse metadata.json")
+ }
+ return nil
+}
+
// blockMetadata contains metadata for a block within a part.
type blockMetadata struct {
tagsBlocks map[string]dataBlock
@@ -419,3 +438,11 @@ func (bm *blockMetadata) less(other *blockMetadata) bool {
}
return bm.seriesID < other.seriesID
}
+
+func (pm *partMetadata) mustWriteMetadata(fileSystem fs.FileSystem, partPath string) {
+ manifestData, err := pm.marshal()
+ if err != nil {
+ logger.GetLogger().Panic().Err(err).Str("path", partPath).Msg("failed to marshal part metadata")
+ }
+ fs.MustFlush(fileSystem, manifestData, filepath.Join(partPath, manifestFilename), storage.FilePerm)
+}
diff --git a/banyand/internal/sidx/multi_sidx_query_test.go b/banyand/internal/sidx/multi_sidx_query_test.go
index 93cafca6..2ff19b9c 100644
--- a/banyand/internal/sidx/multi_sidx_query_test.go
+++ b/banyand/internal/sidx/multi_sidx_query_test.go
@@ -37,7 +37,7 @@ type mockSIDX struct {
delay bool
}
-func (m *mockSIDX) Write(_ context.Context, _ []WriteRequest) error {
+func (m *mockSIDX) Write(_ context.Context, _ []WriteRequest, _ uint64) error {
return nil // Not implemented for tests
}
@@ -60,7 +60,7 @@ func (m *mockSIDX) Flush() error {
return nil
}
-func (m *mockSIDX) Merge() error {
+func (m *mockSIDX) Merge(_ []uint64, _ uint64, _ <-chan struct{}) error {
return nil
}
diff --git a/banyand/internal/sidx/options.go b/banyand/internal/sidx/options.go
index b1158049..58bbcc43 100644
--- a/banyand/internal/sidx/options.go
+++ b/banyand/internal/sidx/options.go
@@ -27,12 +27,6 @@ import (
// Options contains configuration options for the Secondary Index File System (SIDX).
// Path and Memory are mandatory fields that must be provided during construction.
type Options struct {
- // MergePolicy controls the merge behavior for parts.
- // It determines which parts should be merged together to optimize
- // storage efficiency and query performance.
- // If not set, a default policy will be used.
- MergePolicy *MergePolicy
-
// Memory is the memory protector that controls resource limits.
// It monitors memory usage and prevents OOM by applying backpressure
// when memory consumption exceeds configured thresholds.
@@ -76,9 +70,8 @@ func NewOptions(path string, memory protector.Memory) (*Options, error) {
}
return &Options{
- Path: path,
- Memory: memory,
- MergePolicy: NewDefaultMergePolicy(),
+ Path: path,
+ Memory: memory,
}, nil
}
@@ -87,28 +80,8 @@ func NewOptions(path string, memory protector.Memory) (*Options, error) {
// Memory must be set using WithMemory() before using the Options.
func NewDefaultOptions() *Options {
return &Options{
- Path: "/tmp/sidx",
- Memory: nil, // Must be provided by caller using WithMemory()
- MergePolicy: NewDefaultMergePolicy(),
- }
-}
-
-// NewDefaultMergePolicy creates a MergePolicy with default parameters optimized
-// for typical SIDX workloads.
-func NewDefaultMergePolicy() *MergePolicy {
- return &MergePolicy{
- MaxParts: 8,
- MinMergeMultiplier: 1.7,
- MaxFanOutSize: 1 << 30, // 1GB
- }
-}
-
-// NewMergePolicy creates a MergePolicy with custom parameters.
-func NewMergePolicy(maxParts int, minMergeMultiplier float64, maxFanOutSize uint64) *MergePolicy {
- return &MergePolicy{
- MaxParts: maxParts,
- MinMergeMultiplier: minMergeMultiplier,
- MaxFanOutSize: maxFanOutSize,
+ Path: "/tmp/sidx",
+ Memory: nil, // Must be provided by caller using WithMemory()
}
}
@@ -126,27 +99,6 @@ func (o *Options) Validate() error {
return fmt.Errorf("memory protector must not be nil")
}
- if o.MergePolicy == nil {
- return fmt.Errorf("merge policy must not be nil")
- }
-
- return o.MergePolicy.Validate()
-}
-
-// Validate validates the merge policy configuration.
-func (mp *MergePolicy) Validate() error {
- if mp.MaxParts < 2 {
- return fmt.Errorf("maxParts must be at least 2, got: %d", mp.MaxParts)
- }
-
- if mp.MinMergeMultiplier <= 1.0 {
- return fmt.Errorf("minMergeMultiplier must be greater than 1.0, got: %f", mp.MinMergeMultiplier)
- }
-
- if mp.MaxFanOutSize == 0 {
- return fmt.Errorf("maxFanOutSize must be greater than 0")
- }
-
return nil
}
@@ -163,10 +115,3 @@ func (o *Options) WithMemory(memory protector.Memory) *Options {
opts.Memory = memory
return &opts
}
-
-// WithMergePolicy returns a copy of the options with the specified merge policy.
-func (o *Options) WithMergePolicy(policy *MergePolicy) *Options {
- opts := *o
- opts.MergePolicy = policy
- return &opts
-}
diff --git a/banyand/internal/sidx/options_test.go b/banyand/internal/sidx/options_test.go
index 349371a0..9d74e534 100644
--- a/banyand/internal/sidx/options_test.go
+++ b/banyand/internal/sidx/options_test.go
@@ -37,12 +37,6 @@ func TestNewOptions(t *testing.T) {
assert.Equal(t, "/tmp/sidx", opts.Path)
assert.Equal(t, mockMemory, opts.Memory)
- assert.NotNil(t, opts.MergePolicy)
-
- // Verify default merge policy values
- assert.Equal(t, 8, opts.MergePolicy.MaxParts)
- assert.Equal(t, 1.7, opts.MergePolicy.MinMergeMultiplier)
- assert.Equal(t, uint64(1<<30), opts.MergePolicy.MaxFanOutSize)
// Should validate successfully
assert.NoError(t, opts.Validate())
@@ -75,38 +69,11 @@ func TestNewDefaultOptions(t *testing.T) {
assert.Equal(t, "/tmp/sidx", opts.Path)
assert.Nil(t, opts.Memory)
- assert.NotNil(t, opts.MergePolicy)
-
- // Verify default merge policy values
- assert.Equal(t, 8, opts.MergePolicy.MaxParts)
- assert.Equal(t, 1.7, opts.MergePolicy.MinMergeMultiplier)
- assert.Equal(t, uint64(1<<30), opts.MergePolicy.MaxFanOutSize)
// Should fail validation due to nil memory
assert.Error(t, opts.Validate())
}
-func TestNewDefaultMergePolicy(t *testing.T) {
- policy := NewDefaultMergePolicy()
-
- assert.Equal(t, 8, policy.MaxParts)
- assert.Equal(t, 1.7, policy.MinMergeMultiplier)
- assert.Equal(t, uint64(1<<30), policy.MaxFanOutSize)
-
- // Ensure it validates successfully
- assert.NoError(t, policy.Validate())
-}
-
-func TestNewMergePolicy(t *testing.T) {
- policy := NewMergePolicy(10, 2.0, 1<<20)
-
- assert.Equal(t, 10, policy.MaxParts)
- assert.Equal(t, 2.0, policy.MinMergeMultiplier)
- assert.Equal(t, uint64(1<<20), policy.MaxFanOutSize)
-
- assert.NoError(t, policy.Validate())
-}
-
func TestOptionsValidation(t *testing.T) {
mockMemory := protector.NewMemory(observability.NewBypassRegistry())
@@ -119,18 +86,16 @@ func TestOptionsValidation(t *testing.T) {
{
name: "valid options",
opts: &Options{
- Path: "/tmp/sidx",
- Memory: mockMemory,
- MergePolicy: NewDefaultMergePolicy(),
+ Path: "/tmp/sidx",
+ Memory: mockMemory,
},
expectError: false,
},
{
name: "empty path",
opts: &Options{
- Path: "",
- Memory: mockMemory,
- MergePolicy: NewDefaultMergePolicy(),
+ Path: "",
+ Memory: mockMemory,
},
expectError: true,
errorMsg: "path must not be empty",
@@ -138,9 +103,8 @@ func TestOptionsValidation(t *testing.T) {
{
name: "relative path",
opts: &Options{
- Path: "relative/path",
- Memory: mockMemory,
- MergePolicy: NewDefaultMergePolicy(),
+ Path: "relative/path",
+ Memory: mockMemory,
},
expectError: true,
errorMsg: "path must be absolute",
@@ -148,23 +112,12 @@ func TestOptionsValidation(t *testing.T) {
{
name: "nil memory protector",
opts: &Options{
- Path: "/tmp/sidx",
- Memory: nil,
- MergePolicy: NewDefaultMergePolicy(),
+ Path: "/tmp/sidx",
+ Memory: nil,
},
expectError: true,
errorMsg: "memory protector must not be nil",
},
- {
- name: "nil merge policy",
- opts: &Options{
- Path: "/tmp/sidx",
- Memory: mockMemory,
- MergePolicy: nil,
- },
- expectError: true,
- errorMsg: "merge policy must not be nil",
- },
}
for _, tt := range tests {
@@ -180,67 +133,6 @@ func TestOptionsValidation(t *testing.T) {
}
}
-func TestMergePolicyValidation(t *testing.T) {
- tests := []struct {
- policy *MergePolicy
- name string
- errorMsg string
- expectError bool
- }{
- {
- name: "valid merge policy",
- policy: &MergePolicy{
- MaxParts: 8,
- MinMergeMultiplier: 1.7,
- MaxFanOutSize: 1 << 30,
- },
- expectError: false,
- },
- {
- name: "maxParts too small",
- policy: &MergePolicy{
- MaxParts: 1,
- MinMergeMultiplier: 1.7,
- MaxFanOutSize: 1 << 30,
- },
- expectError: true,
- errorMsg: "maxParts must be at least 2",
- },
- {
- name: "minMergeMultiplier too small",
- policy: &MergePolicy{
- MaxParts: 8,
- MinMergeMultiplier: 1.0,
- MaxFanOutSize: 1 << 30,
- },
- expectError: true,
- errorMsg: "minMergeMultiplier must be greater than 1.0",
- },
- {
- name: "zero maxFanOutSize",
- policy: &MergePolicy{
- MaxParts: 8,
- MinMergeMultiplier: 1.7,
- MaxFanOutSize: 0,
- },
- expectError: true,
- errorMsg: "maxFanOutSize must be greater than 0",
- },
- }
-
- for _, tt := range tests {
- t.Run(tt.name, func(t *testing.T) {
- err := tt.policy.Validate()
- if tt.expectError {
- assert.Error(t, err)
- assert.Contains(t, err.Error(), tt.errorMsg)
- } else {
- assert.NoError(t, err)
- }
- })
- }
-}
-
func TestOptionsWithMethods(t *testing.T) {
mockMemory := protector.NewMemory(observability.NewBypassRegistry())
originalOpts := NewDefaultOptions()
@@ -250,8 +142,7 @@ func TestOptionsWithMethods(t *testing.T) {
newOpts := originalOpts.WithPath(newPath)
assert.Equal(t, newPath, newOpts.Path)
- assert.Equal(t, originalOpts.Path, "/tmp/sidx") // Original unchanged
- assert.Equal(t, originalOpts.MergePolicy, newOpts.MergePolicy) // Other fields copied
+ assert.Equal(t, originalOpts.Path, "/tmp/sidx") // Original unchanged
})
t.Run("WithMemory", func(t *testing.T) {
@@ -261,15 +152,6 @@ func TestOptionsWithMethods(t *testing.T) {
assert.Nil(t, originalOpts.Memory) // Original unchanged
assert.Equal(t, originalOpts.Path, newOpts.Path) // Other fields copied
})
-
- t.Run("WithMergePolicy", func(t *testing.T) {
- customPolicy := NewMergePolicy(10, 2.0, 1<<20)
- newOpts := originalOpts.WithMergePolicy(customPolicy)
-
- assert.Equal(t, customPolicy, newOpts.MergePolicy)
- assert.NotEqual(t, customPolicy, originalOpts.MergePolicy) // Original unchanged
- assert.Equal(t, originalOpts.Path, newOpts.Path) // Other fields copied
- })
}
func TestOptionsConfiguration(t *testing.T) {
@@ -278,25 +160,19 @@ func TestOptionsConfiguration(t *testing.T) {
// Test using new mandatory constructor
baseOpts, err := NewOptions("/base/sidx", mockMemory)
require.NoError(t, err)
- require.NotNil(t, baseOpts.MergePolicy)
// Test configuration can be merged and overridden
customOpts := baseOpts.
- WithPath("/custom/sidx").
- WithMergePolicy(NewMergePolicy(5, 1.5, 512<<20))
+ WithPath("/custom/sidx")
assert.Equal(t, "/custom/sidx", customOpts.Path)
assert.Equal(t, mockMemory, customOpts.Memory)
- assert.Equal(t, 5, customOpts.MergePolicy.MaxParts)
- assert.Equal(t, 1.5, customOpts.MergePolicy.MinMergeMultiplier)
- assert.Equal(t, uint64(512<<20), customOpts.MergePolicy.MaxFanOutSize)
// Ensure validation works with custom configuration
assert.NoError(t, customOpts.Validate())
// Test deprecated constructor still works but requires memory to be set
defaultOpts := NewDefaultOptions()
- require.NotNil(t, defaultOpts.MergePolicy)
// Should fail validation due to nil memory
assert.Error(t, defaultOpts.Validate())
@@ -305,23 +181,3 @@ func TestOptionsConfiguration(t *testing.T) {
optsWithMemory := defaultOpts.WithMemory(mockMemory)
assert.NoError(t, optsWithMemory.Validate())
}
-
-func TestOptionsEdgeCases(t *testing.T) {
- t.Run("minimum valid merge policy", func(t *testing.T) {
- policy := &MergePolicy{
- MaxParts: 2,
- MinMergeMultiplier: 1.001,
- MaxFanOutSize: 1,
- }
- assert.NoError(t, policy.Validate())
- })
-
- t.Run("large values", func(t *testing.T) {
- policy := &MergePolicy{
- MaxParts: 1000,
- MinMergeMultiplier: 100.0,
- MaxFanOutSize: 1 << 60,
- }
- assert.NoError(t, policy.Validate())
- })
-}
diff --git a/banyand/internal/sidx/part.go b/banyand/internal/sidx/part.go
index 2555a942..172ee9a5 100644
--- a/banyand/internal/sidx/part.go
+++ b/banyand/internal/sidx/part.go
@@ -648,11 +648,7 @@ func (mp *memPart) mustFlush(fileSystem fs.FileSystem, partPath string) {
// Write part metadata manifest
if mp.partMetadata != nil {
- manifestData, err := mp.partMetadata.marshal()
- if err != nil {
- logger.GetLogger().Panic().Err(err).Str("path", partPath).Msg("failed to marshal part metadata")
- }
- fs.MustFlush(fileSystem, manifestData, filepath.Join(partPath, manifestFilename), storage.FilePerm)
+ mp.partMetadata.mustWriteMetadata(fileSystem, partPath)
}
fileSystem.SyncPath(partPath)
diff --git a/banyand/internal/sidx/part_iter.go b/banyand/internal/sidx/part_iter.go
index 946a1a6c..87b9fbf4 100644
--- a/banyand/internal/sidx/part_iter.go
+++ b/banyand/internal/sidx/part_iter.go
@@ -26,9 +26,11 @@ import (
"github.com/apache/skywalking-banyandb/api/common"
"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"
+ "github.com/apache/skywalking-banyandb/pkg/pool"
)
type partIter struct {
@@ -258,3 +260,145 @@ func (pi *partIter) findBlock() bool {
pi.bms = nil
return false
}
+
+type partMergeIter struct {
+ seqReaders seqReaders
+ err error
+ primaryBlockMetadata []primaryBlockMetadata
+ compressedPrimaryBuf []byte
+ primaryBuf []byte
+ block blockPointer
+ partID uint64
+ primaryMetadataIdx int
+}
+
+func (pmi *partMergeIter) reset() {
+ pmi.err = nil
+ pmi.seqReaders.reset()
+ pmi.primaryBlockMetadata = nil
+ pmi.primaryMetadataIdx = 0
+ pmi.partID = 0
+ pmi.primaryBuf = pmi.primaryBuf[:0]
+ pmi.compressedPrimaryBuf = pmi.compressedPrimaryBuf[:0]
+ pmi.block.reset()
+}
+
+func (pmi *partMergeIter) mustInitFromPart(p *part) {
+ pmi.reset()
+ pmi.seqReaders.init(p)
+ pmi.primaryBlockMetadata = p.primaryBlockMetadata
+ pmi.partID = p.partMetadata.ID
+}
+
+func (pmi *partMergeIter) error() error {
+ if pmi.err == nil || errors.Is(pmi.err, io.EOF) {
+ return nil
+ }
+ return pmi.err
+}
+
+func (pmi *partMergeIter) nextBlockMetadata() bool {
+ if pmi.err != nil {
+ return false
+ }
+ pmi.block.reset()
+ if len(pmi.primaryBuf) == 0 {
+ if err := pmi.loadPrimaryBuf(); err != nil {
+ pmi.err = err
+ return false
+ }
+ }
+ if err := pmi.loadBlockMetadata(); err != nil {
+ pmi.err = err
+ return false
+ }
+ return true
+}
+
+func (pmi *partMergeIter) loadPrimaryBuf() error {
+ if pmi.primaryMetadataIdx >= len(pmi.primaryBlockMetadata) {
+ return io.EOF
+ }
+ pm := pmi.primaryBlockMetadata[pmi.primaryMetadataIdx]
+ pmi.compressedPrimaryBuf = bytes.ResizeOver(pmi.compressedPrimaryBuf, int(pm.size))
+ pmi.seqReaders.primary.mustReadFull(pmi.compressedPrimaryBuf)
+ var err error
+ pmi.primaryBuf, err = zstd.Decompress(pmi.primaryBuf[:0], pmi.compressedPrimaryBuf)
+ if err != nil {
+ return fmt.Errorf("cannot decompress primary block: %w", err)
+ }
+ pmi.primaryMetadataIdx++
+ return nil
+}
+
+func (pmi *partMergeIter) loadBlockMetadata() error {
+ pmi.block.reset()
+ var err error
+ pmi.primaryBuf, err = pmi.block.bm.unmarshal(pmi.primaryBuf)
+ if err != nil {
+ pm := pmi.primaryBlockMetadata[pmi.primaryMetadataIdx-1]
+ return fmt.Errorf("can't read block metadata from primary at %d: %w", pm.offset, err)
+ }
+ return nil
+}
+
+func (pmi *partMergeIter) mustLoadBlockData(decoder *encoding.BytesBlockDecoder, block *blockPointer) {
+ block.block.mustSeqReadFrom(decoder, &pmi.seqReaders, pmi.block.bm)
+}
+
+func generatePartMergeIter() *partMergeIter {
+ v := pmiPool.Get()
+ if v == nil {
+ return &partMergeIter{}
+ }
+ return v
+}
+
+func releasePartMergeIter(pmi *partMergeIter) {
+ pmi.reset()
+ pmiPool.Put(pmi)
+}
+
+var pmiPool = pool.Register[*partMergeIter]("sidx-partMergeIter")
+
+type partMergeIterHeap []*partMergeIter
+
+func (pih *partMergeIterHeap) Len() int {
+ return len(*pih)
+}
+
+func (pih *partMergeIterHeap) Less(i, j int) bool {
+ x := *pih
+ return x[i].block.bm.less(&x[j].block.bm)
+}
+
+func (pih *partMergeIterHeap) Swap(i, j int) {
+ x := *pih
+ x[i], x[j] = x[j], x[i]
+}
+
+func (pih *partMergeIterHeap) Push(x interface{}) {
+ *pih = append(*pih, x.(*partMergeIter))
+}
+
+func (pih *partMergeIterHeap) Pop() interface{} {
+ a := *pih
+ v := a[len(a)-1]
+ *pih = a[:len(a)-1]
+ return v
+}
+
+func generateColumnValuesDecoder() *encoding.BytesBlockDecoder {
+ v := columnValuesDecoderPool.Get()
+ if v == nil {
+ return &encoding.BytesBlockDecoder{}
+ }
+ return v
+}
+
+func releaseColumnValuesDecoder(d *encoding.BytesBlockDecoder) {
+ d.Reset()
+ columnValuesDecoderPool.Put(d)
+}
+
+var columnValuesDecoderPool = pool.Register[*encoding.BytesBlockDecoder]("sidx-columnValuesDecoder")
diff --git a/banyand/internal/sidx/sidx.go b/banyand/internal/sidx/sidx.go
index 27cd8b44..7507e531 100644
--- a/banyand/internal/sidx/sidx.go
+++ b/banyand/internal/sidx/sidx.go
@@ -20,7 +20,10 @@ package sidx
import (
"container/heap"
"context"
+ "encoding/json"
"math"
+ "os"
+ "path/filepath"
"sort"
"sync"
"sync/atomic"
@@ -29,24 +32,30 @@ import (
"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/cgroups"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
"github.com/apache/skywalking-banyandb/pkg/logger"
"github.com/apache/skywalking-banyandb/pkg/pool"
"github.com/apache/skywalking-banyandb/pkg/run"
"github.com/apache/skywalking-banyandb/pkg/watcher"
)
+const maxBlockLength = 8 * 1024
+
// sidx implements the SIDX interface with introduction channels for async operations.
type sidx struct {
+ fileSystem fs.FileSystem
snapshot *snapshot
introductions chan *introduction
flushCh chan *flusherIntroduction
mergeCh chan *mergerIntroduction
loopCloser *run.Closer
- gc *gc
l *logger.Logger
pm protector.Memory
+ root string
+ gc garbageCleaner
totalIntroduceLoopStarted atomic.Int64
totalIntroduceLoopFinished atomic.Int64
totalQueries atomic.Int64
@@ -55,7 +64,7 @@ type sidx struct {
}
// NewSIDX creates a new SIDX instance with introduction channels.
-func NewSIDX(opts *Options) (SIDX, error) {
+func NewSIDX(fileSystem fs.FileSystem, opts *Options) (SIDX, error) {
if opts == nil {
opts = NewDefaultOptions()
}
@@ -65,26 +74,31 @@ func NewSIDX(opts *Options) (SIDX, error) {
}
s := &sidx{
+ fileSystem: fileSystem,
introductions: make(chan *introduction),
flushCh: make(chan *flusherIntroduction),
mergeCh: make(chan *mergerIntroduction),
loopCloser: run.NewCloser(1),
l: logger.GetLogger().Named("sidx"),
pm: opts.Memory,
+ root: opts.Path,
}
// Initialize garbage collector
- s.gc = newGC(opts)
+ s.gc.init(s)
+
+ // Initialize sidx
+ epoch := s.init()
- // Start introducer loop
+ // Start introducer loop with loaded epoch
watcherCh := make(watcher.Channel, 10)
- go s.introducerLoop(s.flushCh, s.mergeCh, watcherCh, 0)
+ go s.introducerLoop(s.flushCh, s.mergeCh, watcherCh, epoch)
return s, nil
}
// Write implements SIDX interface.
-func (s *sidx) Write(ctx context.Context, reqs []WriteRequest) error {
+func (s *sidx) Write(ctx context.Context, reqs []WriteRequest, partID uint64) error {
// Validate requests
for _, req := range reqs {
if err := req.Validate(); err != nil {
@@ -110,6 +124,7 @@ func (s *sidx) Write(ctx context.Context, reqs []WriteRequest) error {
// Create introduction
intro := generateIntroduction()
intro.memPart = mp
+ intro.memPart.partMetadata.ID = partID
intro.applied = make(chan struct{})
// Send to introducer loop
@@ -126,42 +141,6 @@ func (s *sidx) Write(ctx context.Context, reqs []WriteRequest) error {
}
}
-// extractKeyRange extracts min/max key range from QueryRequest.
-func extractKeyRange(req QueryRequest) (int64, int64) {
- minKey := int64(math.MinInt64)
- maxKey := int64(math.MaxInt64)
-
- if req.MinKey != nil {
- minKey = *req.MinKey
- }
- if req.MaxKey != nil {
- maxKey = *req.MaxKey
- }
-
- return minKey, maxKey
-}
-
-// extractOrdering extracts ordering direction from QueryRequest.
-func extractOrdering(req QueryRequest) bool {
- if req.Order == nil {
- return true // Default ascending
- }
- return req.Order.Sort != modelv1.Sort_SORT_DESC
-}
-
-// selectPartsForQuery selects relevant parts from snapshot based on key range.
-func selectPartsForQuery(snap *snapshot, minKey, maxKey int64) []*part {
- var selectedParts []*part
-
- for _, pw := range snap.parts {
- if pw.isActive() && pw.overlapsKeyRange(minKey, maxKey) {
- selectedParts = append(selectedParts, pw.p)
- }
- }
-
- return selectedParts
-}
-
// Query implements SIDX interface.
func (s *sidx) Query(ctx context.Context, req QueryRequest) (*QueryResponse, error) {
if err := req.Validate(); err != nil {
@@ -245,6 +224,42 @@ func (s *sidx) Query(ctx context.Context, req QueryRequest) (*QueryResponse, err
return response, nil
}
+// extractKeyRange extracts min/max key range from QueryRequest.
+func extractKeyRange(req QueryRequest) (int64, int64) {
+ minKey := int64(math.MinInt64)
+ maxKey := int64(math.MaxInt64)
+
+ if req.MinKey != nil {
+ minKey = *req.MinKey
+ }
+ if req.MaxKey != nil {
+ maxKey = *req.MaxKey
+ }
+
+ return minKey, maxKey
+}
+
+// extractOrdering extracts ordering direction from QueryRequest.
+func extractOrdering(req QueryRequest) bool {
+ if req.Order == nil {
+ return true // Default ascending
+ }
+ return req.Order.Sort != modelv1.Sort_SORT_DESC
+}
+
+// selectPartsForQuery selects relevant parts from snapshot based on key range.
+func selectPartsForQuery(snap *snapshot, minKey, maxKey int64) []*part {
+ var selectedParts []*part
+
+ for _, pw := range snap.parts {
+ if pw.isActive() && pw.overlapsKeyRange(minKey, maxKey) {
+ selectedParts = append(selectedParts, pw.p)
+ }
+ }
+
+ return selectedParts
+}
+
// executeBlockScannerQuery coordinates the worker pool with block scanner following tsResult pattern.
func (s *sidx) executeBlockScannerQuery(ctx context.Context, bs *blockScanner, req QueryRequest) *QueryResponse {
workerSize := cgroups.CPUs()
@@ -392,17 +407,21 @@ func (s *sidx) Flush() error {
// Create flush introduction
flushIntro := generateFlusherIntroduction()
+ defer releaseFlusherIntroduction(flushIntro)
flushIntro.applied = make(chan struct{})
- // Select memory parts to flush (simplified logic)
+ // Select memory parts to flush
for _, pw := range snap.parts {
- if pw.isMemPart() && pw.isActive() {
- flushIntro.flushed[pw.ID()] = pw.p
+ if !pw.isMemPart() || !pw.isActive() {
+ continue
}
+ partPath := partPath(s.root, pw.ID())
+ pw.mp.mustFlush(s.fileSystem, partPath)
+ p := mustOpenPart(partPath, s.fileSystem)
+ flushIntro.flushed[p.partMetadata.ID] = p
}
if len(flushIntro.flushed) == 0 {
- releaseFlusherIntroduction(flushIntro)
return nil
}
@@ -411,54 +430,6 @@ func (s *sidx) Flush() error {
// Wait for flush to complete
<-flushIntro.applied
- releaseFlusherIntroduction(flushIntro)
-
- return nil
-}
-
-// Merge implements Merger interface.
-func (s *sidx) Merge() error {
- // Get current snapshot
- snap := s.currentSnapshot()
- if snap == nil {
- return nil
- }
- defer snap.decRef()
-
- // Create merge introduction
- mergeIntro := generateMergerIntroduction()
- mergeIntro.applied = make(chan struct{})
-
- // Select parts to merge (simplified logic - merge first 2 parts)
- var partsToMerge []*partWrapper
- for _, pw := range snap.parts {
- if pw.isActive() && !pw.isMemPart() {
- partsToMerge = append(partsToMerge, pw)
- if len(partsToMerge) >= 2 {
- break
- }
- }
- }
-
- if len(partsToMerge) < 2 {
- releaseMergerIntroduction(mergeIntro)
- return nil
- }
-
- // Mark parts for merging
- for _, pw := range partsToMerge {
- mergeIntro.merged[pw.ID()] = struct{}{}
- }
-
- // Create new merged part (simplified - would actually merge the parts)
- mergeIntro.newPart = partsToMerge[0].p
-
- // Send to introducer loop
- s.mergeCh <- mergeIntro
-
- // Wait for merge to complete
- <-mergeIntro.applied
- releaseMergerIntroduction(mergeIntro)
return nil
}
@@ -494,9 +465,33 @@ func (s *sidx) currentSnapshot() *snapshot {
return nil
}
-// decRef decrements the snapshot reference count (helper for snapshot interface).
-func (s *snapshot) decRef() {
- s.release()
+// persistSnapshot persists the snapshot to disk.
+func (s *sidx) persistSnapshot(snapshot *snapshot) {
+ var partNames []string
+ for i := range snapshot.parts {
+ partNames = append(partNames, partName(snapshot.parts[i].ID()))
+ }
+ s.mustWriteSnapshot(snapshot.epoch, partNames)
+ s.gc.registerSnapshot(snapshot)
+}
+
+func (s *sidx) mustWriteSnapshot(snapshot uint64, partNames []string) {
+ data, err := json.Marshal(partNames)
+ if err != nil {
+ logger.Panicf("cannot marshal partNames to JSON: %s", err)
+ }
+ snapshotPath := filepath.Join(s.root, snapshotName(snapshot))
+ lf, err := s.fileSystem.CreateLockFile(snapshotPath, storage.FilePerm)
+ if err != nil {
+ logger.Panicf("cannot create lock file %s: %s", snapshotPath, err)
+ }
+ n, err := lf.Write(data)
+ if err != nil {
+ logger.Panicf("cannot write snapshot %s: %s", snapshotPath, err)
+ }
+ if n != len(data) {
+ logger.Panicf("unexpected number of bytes written to %s; got %d; want %d", snapshotPath, n, len(data))
+ }
}
// Helper methods for metrics.
@@ -508,31 +503,6 @@ func (s *sidx) incTotalIntroduceLoopFinished(_ string) {
s.totalIntroduceLoopFinished.Add(1)
}
-// persistSnapshot persists the snapshot to disk (placeholder).
-func (s *sidx) persistSnapshot(_ *snapshot) {
- // TODO: Implement snapshot persistence
-}
-
-// Lock/Unlock methods for introducer loop.
-func (s *sidx) Lock() {
- s.mu.Lock()
-}
-
-func (s *sidx) Unlock() {
- s.mu.Unlock()
-}
-
-// gc represents garbage collector (placeholder).
-type gc struct{}
-
-func newGC(_ *Options) *gc {
- return &gc{}
-}
-
-func (g *gc) clean() {
- // TODO: Implement garbage collection
-}
-
// blockCursor represents a cursor for iterating through a loaded block, similar to query_by_ts.go.
type blockCursor struct {
p *part
@@ -766,3 +736,106 @@ func releaseBlockCursor(bc *blockCursor) {
bc.idx = 0
blockCursorPool.Put(bc)
}
+
+func (s *sidx) init() uint64 {
+ if _, err := os.Stat(s.root); os.IsNotExist(err) {
+ s.l.Debug().Str("path", s.root).Msg("sidx directory does not exist")
+ return 0
+ }
+
+ entries := s.fileSystem.ReadDir(s.root)
+ if len(entries) == 0 {
+ s.l.Debug().Str("path", s.root).Msg("no existing sidx data found")
+ return 0
+ }
+
+ var loadedParts []uint64
+ var loadedSnapshots []uint64
+ var needToDelete []string
+
+ for i := range entries {
+ if entries[i].IsDir() {
+ p, err := parseEpoch(entries[i].Name())
+ if err != nil {
+ s.l.Info().Err(err).Str("name", entries[i].Name()).Msg("cannot parse part directory name, skipping")
+ needToDelete = append(needToDelete, entries[i].Name())
+ continue
+ }
+ loadedParts = append(loadedParts, p)
+ continue
+ }
+
+ if filepath.Ext(entries[i].Name()) != snapshotSuffix {
+ continue
+ }
+ snapshot, err := parseSnapshot(entries[i].Name())
+ if err != nil {
+ s.l.Info().Err(err).Str("name", entries[i].Name()).Msg("cannot parse snapshot file name, skipping")
+ needToDelete = append(needToDelete, entries[i].Name())
+ continue
+ }
+ loadedSnapshots = append(loadedSnapshots, snapshot)
+ }
+
+ for i := range needToDelete {
+ s.l.Info().Str("path", filepath.Join(s.root, needToDelete[i])).Msg("deleting invalid directory or file")
+ s.fileSystem.MustRMAll(filepath.Join(s.root, needToDelete[i]))
+ }
+
+ if len(loadedParts) == 0 || len(loadedSnapshots) == 0 {
+ s.l.Debug().Msg("no valid parts or snapshots found")
+ return 0
+ }
+
+ sort.Slice(loadedSnapshots, func(i, j int) bool {
+ return loadedSnapshots[i] > loadedSnapshots[j]
+ })
+
+ latestEpoch := loadedSnapshots[0]
+ s.loadSnapshot(latestEpoch, loadedParts)
+
+ s.l.Info().Uint64("epoch", latestEpoch).Int("parts", len(loadedParts)).Msg("loaded existing sidx data")
+ return latestEpoch
+}
+
+func (s *sidx) loadSnapshot(epoch uint64, loadedParts []uint64) {
+ partIDs := mustReadSnapshot(s.fileSystem, s.root, epoch)
+ snp := newSnapshot(nil, epoch)
+ needToPersist := false
+ for _, id := range loadedParts {
+ var find bool
+ for j := range partIDs {
+ if id == partIDs[j] {
+ find = true
+ break
+ }
+ }
+ if !find {
+ s.gc.removePart(id)
+ continue
+ }
+ err := validatePartMetadata(s.fileSystem, partPath(s.root, id))
+ if err != nil {
+ s.l.Info().Err(err).Uint64("id", id).Msg("cannot validate part metadata. skip and delete it")
+ s.gc.removePart(id)
+ needToPersist = true
+ continue
+ }
+ partPath := partPath(s.root, id)
+ part := mustOpenPart(partPath, s.fileSystem)
+ part.partMetadata.ID = id
+ pw := newPartWrapper(nil, part)
+ snp.addPart(pw)
+ }
+ s.gc.registerSnapshot(snp)
+ s.gc.clean()
+ if snp.getPartCount() < 1 {
+ snp.release()
+ return
+ }
+ snp.acquire()
+ s.snapshot = snp
+ if needToPersist {
+ s.persistSnapshot(snp)
+ }
+}
diff --git a/banyand/internal/sidx/sidx_test.go b/banyand/internal/sidx/sidx_test.go
index e75a7e98..3dfd1f00 100644
--- a/banyand/internal/sidx/sidx_test.go
+++ b/banyand/internal/sidx/sidx_test.go
@@ -32,10 +32,13 @@ import (
modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
"github.com/apache/skywalking-banyandb/banyand/observability"
"github.com/apache/skywalking-banyandb/banyand/protector"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
"github.com/apache/skywalking-banyandb/pkg/index"
pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
)
+const partIDForTesting = 1
+
// Test helper functions.
func waitForIntroducerLoop() {
@@ -51,8 +54,9 @@ func createTestOptions(t *testing.T) *Options {
}
func createTestSIDX(t *testing.T) SIDX {
+ fileSystem := fs.NewLocalFileSystem()
opts := createTestOptions(t)
- sidx, err := NewSIDX(opts)
+ sidx, err := NewSIDX(fileSystem, opts)
require.NoError(t, err)
require.NotNil(t, sidx)
return sidx
@@ -96,7 +100,7 @@ func TestSIDX_Write_SingleRequest(t *testing.T) {
createTestWriteRequest(1, 100, "data1", createTestTag("tag1", "value1")),
}
- err := sidx.Write(ctx, reqs)
+ err := sidx.Write(ctx, reqs, partIDForTesting)
assert.NoError(t, err)
// Verify stats
@@ -120,7 +124,7 @@ func TestSIDX_Write_BatchRequest(t *testing.T) {
createTestWriteRequest(2, 200, "data3", createTestTag("tag2", "value3")),
}
- err := sidx.Write(ctx, reqs)
+ err := sidx.Write(ctx, reqs, partIDForTesting)
assert.NoError(t, err)
// Verify stats
@@ -166,7 +170,7 @@ func TestSIDX_Write_Validation(t *testing.T) {
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
- err := sidx.Write(ctx, []WriteRequest{tt.req})
+ err := sidx.Write(ctx, []WriteRequest{tt.req}, partIDForTesting)
if tt.expectErr {
assert.Error(t, err)
} else {
@@ -195,7 +199,7 @@ func TestSIDX_Write_WithTags(t *testing.T) {
createTestWriteRequest(1, 100, "trace-data", tags...),
}
- err := sidx.Write(ctx, reqs)
+ err := sidx.Write(ctx, reqs, partIDForTesting)
assert.NoError(t, err)
}
@@ -214,7 +218,7 @@ func TestSIDX_Query_BasicQuery(t *testing.T) {
createTestWriteRequest(1, 100, "data1"),
createTestWriteRequest(1, 101, "data2"),
}
- err := sidx.Write(ctx, reqs)
+ err := sidx.Write(ctx, reqs, partIDForTesting)
require.NoError(t, err)
// Wait for introducer loop to process
@@ -264,7 +268,7 @@ func TestSIDX_Query_KeyRangeFilter(t *testing.T) {
createTestWriteRequest(1, 150, "data150"),
createTestWriteRequest(1, 200, "data200"),
}
- err := sidx.Write(ctx, reqs)
+ err := sidx.Write(ctx, reqs, partIDForTesting)
require.NoError(t, err)
// Wait for introducer loop to process
@@ -318,7 +322,7 @@ func TestSIDX_Query_Ordering(t *testing.T) {
createTestWriteRequest(3, 75, "series3-data75"),
createTestWriteRequest(3, 175, "series3-data175"),
}
- err := sidx.Write(ctx, reqs)
+ err := sidx.Write(ctx, reqs, partIDForTesting)
require.NoError(t, err)
// Wait for introducer loop to process
@@ -475,7 +479,7 @@ func TestSIDX_WriteQueryIntegration(t *testing.T) {
createTestWriteRequest(2, 180, "series2-data2", createTestTag("env", "dev")),
}
- err := sidx.Write(ctx, reqs)
+ err := sidx.Write(ctx, reqs, partIDForTesting)
require.NoError(t, err)
// Test 1: Query single series
@@ -536,7 +540,7 @@ func TestSIDX_DataConsistency(t *testing.T) {
reqs = append(reqs, createTestWriteRequest(1, key, data))
}
- err := sidx.Write(ctx, reqs)
+ err := sidx.Write(ctx, reqs, partIDForTesting)
require.NoError(t, err)
// Query back and verify data integrity
@@ -583,7 +587,7 @@ func TestSIDX_LargeDataset(t *testing.T) {
))
}
- err := sidx.Write(ctx, reqs)
+ err := sidx.Write(ctx, reqs, partIDForTesting)
require.NoError(t, err)
// Query back and verify we can handle large result sets
@@ -630,7 +634,7 @@ func TestSIDX_ConcurrentWrites(t *testing.T) {
reqs = append(reqs, createTestWriteRequest(seriesID, key, data))
}
- if err := sidx.Write(ctx, reqs); err != nil {
+ if err := sidx.Write(ctx, reqs, partIDForTesting); err != nil {
errors <- err
}
}(g)
@@ -662,7 +666,7 @@ func TestSIDX_ConcurrentReadsWrites(t *testing.T) {
initialReqs := []WriteRequest{
createTestWriteRequest(1, 100, "initial-data"),
}
- err := sidx.Write(ctx, initialReqs)
+ err := sidx.Write(ctx, initialReqs, partIDForTesting)
require.NoError(t, err)
var wg sync.WaitGroup
@@ -704,7 +708,7 @@ func TestSIDX_ConcurrentReadsWrites(t *testing.T) {
int64(writeCount),
fmt.Sprintf("writer-%d-data-%d", writerID, writeCount),
)
- sidx.Write(ctx, []WriteRequest{req}) // Ignore errors during concurrent stress
+ sidx.Write(ctx, []WriteRequest{req}, partIDForTesting) // Ignore errors during concurrent stress
writeCount++
}
}(i)
diff --git a/banyand/internal/sidx/snapshot.go b/banyand/internal/sidx/snapshot.go
index e1fd1a28..2edfd3b9 100644
--- a/banyand/internal/sidx/snapshot.go
+++ b/banyand/internal/sidx/snapshot.go
@@ -18,14 +18,22 @@
package sidx
import (
+ "encoding/json"
"fmt"
+ "path/filepath"
"sort"
+ "strconv"
"sync/atomic"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
"github.com/apache/skywalking-banyandb/pkg/logger"
"github.com/apache/skywalking-banyandb/pkg/pool"
)
+const (
+ snapshotSuffix = ".snp"
+)
+
// snapshot represents an immutable collection of parts at a specific epoch.
// It provides safe concurrent access to parts through reference counting and
// enables queries to work with a consistent view of data.
@@ -90,6 +98,11 @@ func (s *snapshot) acquire() bool {
}
}
+// decRef decrements the snapshot reference count (helper for snapshot interface).
+func (s *snapshot) decRef() {
+ s.release()
+}
+
// release decrements the snapshot reference count.
// When the count reaches zero, all part references are released.
func (s *snapshot) release() {
@@ -277,6 +290,45 @@ func (s *snapshot) String() string {
s.epoch, activeCount, len(s.parts), s.refCount())
}
+func snapshotName(snapshot uint64) string {
+ return fmt.Sprintf("%016x%s", snapshot, snapshotSuffix)
+}
+
+func parseSnapshot(name string) (uint64, error) {
+ if filepath.Ext(name) != snapshotSuffix {
+ return 0, fmt.Errorf("invalid snapshot file ext")
+ }
+ if len(name) < 16 {
+ return 0, fmt.Errorf("invalid snapshot file name")
+ }
+ return parseEpoch(name[:16])
+}
+
+func parseEpoch(epochStr string) (uint64, error) {
+ return strconv.ParseUint(epochStr, 16, 64)
+}
+
+func mustReadSnapshot(fileSystem fs.FileSystem, root string, snapshot uint64) []uint64 {
+ snapshotPath := filepath.Join(root, snapshotName(snapshot))
+ data, err := fileSystem.Read(snapshotPath)
+ if err != nil {
+ logger.GetLogger().Panic().Err(err).Str("path", snapshotPath).Msg("cannot read snapshot")
+ }
+ var partNames []string
+ if err := json.Unmarshal(data, &partNames); err != nil {
+ logger.GetLogger().Panic().Err(err).Str("path", snapshotPath).Msg("cannot parse snapshot")
+ }
+ var result []uint64
+ for i := range partNames {
+ e, err := parseEpoch(partNames[i])
+ if err != nil {
+ logger.GetLogger().Panic().Err(err).Str("name", partNames[i]).Msg("cannot parse part name")
+ }
+ result = append(result, e)
+ }
+ return result
+}
+
// Pool for snapshot reuse.
var snapshotPool = pool.Register[*snapshot]("sidx-snapshot")
diff --git a/banyand/internal/sidx/snapshot_test.go b/banyand/internal/sidx/snapshot_test.go
index ce1bfc6e..dc94d291 100644
--- a/banyand/internal/sidx/snapshot_test.go
+++ b/banyand/internal/sidx/snapshot_test.go
@@ -26,6 +26,7 @@ import (
"github.com/apache/skywalking-banyandb/api/common"
"github.com/apache/skywalking-banyandb/banyand/protector"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
)
func TestSnapshot_Creation(t *testing.T) {
@@ -410,8 +411,9 @@ func anySubstring(s, substr string) bool {
func TestSnapshotReplacement_Basic(t *testing.T) {
// Test that snapshot replacement works correctly with basic operations
+ fileSystem := fs.NewLocalFileSystem()
opts := NewDefaultOptions().WithMemory(protector.Nop{})
- sidx, err := NewSIDX(opts)
+ sidx, err := NewSIDX(fileSystem, opts)
if err != nil {
t.Fatalf("failed to create SIDX: %v", err)
}
@@ -428,7 +430,7 @@ func TestSnapshotReplacement_Basic(t *testing.T) {
Tags: []Tag{{Name: "test", Value: []byte("snapshot-replacement")}},
}
- if err := sidx.Write(ctx, []WriteRequest{req}); err != nil {
+ if err := sidx.Write(ctx, []WriteRequest{req}, partIDForTesting); err != nil {
t.Errorf("write %d failed: %v", i, err)
}
@@ -448,8 +450,9 @@ func TestSnapshotReplacement_ConcurrentReadsConsistentData(t *testing.T) {
// Test that concurrent readers see consistent data during snapshot replacements
// This verifies that snapshot replacement doesn't cause readers to see inconsistent state
+ fileSystem := fs.NewLocalFileSystem()
opts := NewDefaultOptions().WithMemory(protector.Nop{})
- sidx, err := NewSIDX(opts)
+ sidx, err := NewSIDX(fileSystem, opts)
if err != nil {
t.Fatalf("failed to create SIDX: %v", err)
}
@@ -518,7 +521,7 @@ func TestSnapshotReplacement_ConcurrentReadsConsistentData(t *testing.T) {
},
}
- if err := sidx.Write(ctx, reqs); err != nil {
+ if err := sidx.Write(ctx, reqs, partIDForTesting); err != nil {
t.Errorf("write %d failed: %v", i, err)
}
@@ -572,8 +575,9 @@ func TestSnapshotReplacement_NoDataRacesDuringReplacement(t *testing.T) {
// This test should be run with -race flag to detect data races during snapshot replacement
// We test through concurrent write and read operations that trigger snapshot replacements
+ fileSystem := fs.NewLocalFileSystem()
opts := NewDefaultOptions().WithMemory(protector.Nop{})
- sidx, err := NewSIDX(opts)
+ sidx, err := NewSIDX(fileSystem, opts)
if err != nil {
t.Fatalf("failed to create SIDX: %v", err)
}
@@ -606,7 +610,7 @@ func TestSnapshotReplacement_NoDataRacesDuringReplacement(t *testing.T) {
},
},
}
- sidx.Write(ctx, reqs)
+ sidx.Write(ctx, reqs, partIDForTesting)
case 1:
// Stats operation - accesses current snapshot
sidx.Stats(ctx)
@@ -636,8 +640,9 @@ func TestSnapshotReplacement_MemoryLeaksPrevention(t *testing.T) {
// Test to ensure old snapshots are properly cleaned up during replacement operations
// This verifies that reference counting prevents memory leaks
+ fileSystem := fs.NewLocalFileSystem()
opts := NewDefaultOptions().WithMemory(protector.Nop{})
- sidx, err := NewSIDX(opts)
+ sidx, err := NewSIDX(fileSystem, opts)
if err != nil {
t.Fatalf("failed to create SIDX: %v", err)
}
@@ -665,7 +670,7 @@ func TestSnapshotReplacement_MemoryLeaksPrevention(t *testing.T) {
},
}
- if writeErr := sidx.Write(ctx, reqs); writeErr != nil {
+ if writeErr := sidx.Write(ctx, reqs, partIDForTesting); writeErr != nil {
t.Errorf("batch %d write %d failed: %v", i, j, writeErr)
}
}
@@ -726,7 +731,7 @@ func TestSnapshotReplacement_MemoryLeaksPrevention(t *testing.T) {
},
}
- if writeErr := sidx.Write(ctx, reqs); writeErr != nil {
+ if writeErr := sidx.Write(ctx, reqs, partIDForTesting); writeErr != nil {
t.Errorf("concurrent writer %d write %d failed: %v", writerID, j, writeErr)
}
diff --git a/banyand/trace/flusher.go b/banyand/trace/flusher.go
index 3e4c2d5a..da68d204 100644
--- a/banyand/trace/flusher.go
+++ b/banyand/trace/flusher.go
@@ -203,9 +203,10 @@ func (tst *tsTable) flush(snapshot *snapshot, flushCh chan *flusherIntroduction)
if len(ind.flushed) < 1 {
return
}
- for sidxName, sidxInstance := range tst.sidxMap {
+ allSidx := tst.getAllSidx()
+ for _, sidxInstance := range allSidx {
if err := sidxInstance.Flush(); err != nil {
- tst.l.Warn().Err(err).Str("sidx", sidxName).Msg("sidx flush failed")
+ tst.l.Warn().Err(err).Msg("sidx flush failed")
return
}
}
diff --git a/banyand/trace/merger.go b/banyand/trace/merger.go
index d391f9bd..095268cd 100644
--- a/banyand/trace/merger.go
+++ b/banyand/trace/merger.go
@@ -107,12 +107,17 @@ func (tst *tsTable) mergePartsThenSendIntroduction(creator snapshotCreator, part
reservedSpace := tst.reserveSpace(parts)
defer releaseDiskSpace(reservedSpace)
start := time.Now()
- newPart, err := tst.mergeParts(tst.fileSystem, closeCh, parts, atomic.AddUint64(&tst.curPartID, 1), tst.root)
+ newPartID := atomic.AddUint64(&tst.curPartID, 1)
+ newPart, err := tst.mergeParts(tst.fileSystem, closeCh, parts, newPartID, tst.root)
if err != nil {
return nil, err
}
+ partIDs := make([]uint64, 0, len(parts))
+ for _, pw := range parts {
+ partIDs = append(partIDs, pw.ID())
+ }
for sidxName, sidxInstance := range tst.sidxMap {
- if err := sidxInstance.Merge(); err != nil {
+ if err := sidxInstance.Merge(partIDs, newPartID, closeCh); err != nil {
tst.l.Warn().Err(err).Str("sidx", sidxName).Msg("sidx merge failed")
return nil, err
}
diff --git a/banyand/trace/merger_test.go b/banyand/trace/merger_test.go
index c855c5cb..a76b4486 100644
--- a/banyand/trace/merger_test.go
+++ b/banyand/trace/merger_test.go
@@ -116,29 +116,29 @@ func Test_mergeParts(t *testing.T) {
tsList []*traces
want []blockMetadata
}{
- // {
- // name: "Test with no trace",
- // tsList: []*traces{},
- // wantErr: errNoPartToMerge,
- // },
- // {
- // name: "Test with single part",
- // tsList: []*traces{tsTS1},
- // want: []blockMetadata{
- // {traceID: "trace1", count: 1, uncompressedSpanSizeBytes: 5},
- // {traceID: "trace2", count: 1, uncompressedSpanSizeBytes: 5},
- // {traceID: "trace3", count: 1, uncompressedSpanSizeBytes: 5},
- // },
- // },
- // {
- // name: "Test with multiple parts",
- // tsList: []*traces{tsTS1, tsTS2, tsTS2},
- // want: []blockMetadata{
- // {traceID: "trace1", count: 3, uncompressedSpanSizeBytes: 15},
- // {traceID: "trace2", count: 3, uncompressedSpanSizeBytes: 15},
- // {traceID: "trace3", count: 3, uncompressedSpanSizeBytes: 15},
- // },
- // },
+ {
+ name: "Test with no trace",
+ tsList: []*traces{},
+ wantErr: errNoPartToMerge,
+ },
+ {
+ name: "Test with single part",
+ tsList: []*traces{tsTS1},
+ want: []blockMetadata{
+ {traceID: "trace1", count: 1, uncompressedSpanSizeBytes: 5},
+ {traceID: "trace2", count: 1, uncompressedSpanSizeBytes: 5},
+ {traceID: "trace3", count: 1, uncompressedSpanSizeBytes: 5},
+ },
+ },
+ {
+ name: "Test with multiple parts",
+ tsList: []*traces{tsTS1, tsTS2, tsTS2},
+ want: []blockMetadata{
+ {traceID: "trace1", count: 3, uncompressedSpanSizeBytes: 15},
+ {traceID: "trace2", count: 3, uncompressedSpanSizeBytes: 15},
+ {traceID: "trace3", count: 3, uncompressedSpanSizeBytes: 15},
+ },
+ },
{
name: "Test with multiple parts with a large quantity of spans",
tsList: []*traces{generateHugeTraces(5000), generateHugeTraces(5000)},
diff --git a/banyand/trace/tstable.go b/banyand/trace/tstable.go
index 91a90cc1..785cf701 100644
--- a/banyand/trace/tstable.go
+++ b/banyand/trace/tstable.go
@@ -22,6 +22,7 @@ import (
"errors"
"fmt"
"io"
+ "os"
"path/filepath"
"sort"
"strconv"
@@ -43,6 +44,7 @@ import (
const (
snapshotSuffix = ".snp"
+ sidxDirName = "sidx"
)
type tsTable struct {
@@ -201,7 +203,7 @@ func initTSTable(fileSystem fs.FileSystem, rootPath string, p common.Position,
if m != nil {
tst.metrics = m.(*metrics)
}
- tst.sidxMap = make(map[string]sidx.SIDX)
+ tst.loadSidxMap()
tst.gc.init(&tst)
ee := fileSystem.ReadDir(rootPath)
if len(ee) == 0 {
@@ -212,6 +214,9 @@ func initTSTable(fileSystem fs.FileSystem, rootPath string, p common.Position,
var needToDelete []string
for i := range ee {
if ee[i].IsDir() {
+ if ee[i].Name() == sidxDirName {
+ continue
+ }
p, err := parseEpoch(ee[i].Name())
if err != nil {
l.Info().Err(err).Msg("cannot parse part file name. skip and delete it")
@@ -282,13 +287,13 @@ func (tst *tsTable) getOrCreateSidx(name string) (sidx.SIDX, error) {
}
// Create new sidx instance
- sidxPath := filepath.Join(tst.root, "sidx", name)
+ sidxPath := filepath.Join(tst.root, sidxDirName, name)
sidxOpts, err := sidx.NewOptions(sidxPath, tst.option.protector)
if err != nil {
return nil, fmt.Errorf("cannot create sidx options for %s: %w", name, err)
}
- newSidx, err := sidx.NewSIDX(sidxOpts)
+ newSidx, err := sidx.NewSIDX(tst.fileSystem, sidxOpts)
if err != nil {
return nil, fmt.Errorf("cannot create sidx for %s: %w", name, err)
}
@@ -309,6 +314,40 @@ func (tst *tsTable) getAllSidx() []sidx.SIDX {
return result
}
+func (tst *tsTable) loadSidxMap() {
+ tst.sidxMap = make(map[string]sidx.SIDX)
+ sidxRootPath := filepath.Join(tst.root, sidxDirName)
+ if _, err := os.Stat(sidxRootPath); os.IsNotExist(err) {
+ tst.l.Debug().Str("path", sidxRootPath).Msg("sidx directory does not exist")
+ return
+ }
+ entries := tst.fileSystem.ReadDir(sidxRootPath)
+ if len(entries) == 0 {
+ tst.l.Debug().Str("path", sidxRootPath).Msg("no sidx entries found")
+ return
+ }
+
+ for i := range entries {
+ if !entries[i].IsDir() {
+ continue
+ }
+ sidxName := entries[i].Name()
+ sidxPath := filepath.Join(sidxRootPath, sidxName)
+ sidxOpts, err := sidx.NewOptions(sidxPath, tst.option.protector)
+ if err != nil {
+ tst.l.Error().Err(err).Str("name", sidxName).Msg("failed to create sidx options, skipping")
+ continue
+ }
+ newSidx, err := sidx.NewSIDX(tst.fileSystem, sidxOpts)
+ if err != nil {
+ tst.l.Error().Err(err).Str("name", sidxName).Msg("failed to create sidx instance, skipping")
+ continue
+ }
+ tst.sidxMap[sidxName] = newSidx
+ tst.l.Debug().Str("name", sidxName).Str("path", sidxPath).Msg("loaded sidx instance")
+ }
+}
+
func (tst *tsTable) closeSidxMap() error {
var lastErr error
for name, sidxInstance := range tst.sidxMap {
diff --git a/banyand/trace/write_standalone.go b/banyand/trace/write_standalone.go
index ef089e2e..47c4b941 100644
--- a/banyand/trace/write_standalone.go
+++ b/banyand/trace/write_standalone.go
@@ -363,7 +363,7 @@ func (w *writeCallback) Rev(ctx context.Context, message bus.Message) (resp bus.
w.l.Error().Err(err).Str("sidx", sidxName).Msg("cannot get or create sidx instance")
continue
}
- if err := sidxInstance.Write(ctx, sidxReqs); err != nil {
+ if err := sidxInstance.Write(ctx, sidxReqs, es.tsTable.curPartID); err != nil {
w.l.Error().Err(err).Str("sidx", sidxName).Msg("cannot write to secondary index")
}
}
diff --git a/test/integration/standalone/cold_query/query_suite_test.go b/test/integration/standalone/cold_query/query_suite_test.go
index fb127bf0..db51df31 100644
--- a/test/integration/standalone/cold_query/query_suite_test.go
+++ b/test/integration/standalone/cold_query/query_suite_test.go
@@ -39,6 +39,7 @@ import (
casesmeasure "github.com/apache/skywalking-banyandb/test/cases/measure"
casesstream "github.com/apache/skywalking-banyandb/test/cases/stream"
casestopn "github.com/apache/skywalking-banyandb/test/cases/topn"
+ casestrace "github.com/apache/skywalking-banyandb/test/cases/trace"
integration_standalone "github.com/apache/skywalking-banyandb/test/integration/standalone"
)
@@ -81,6 +82,10 @@ var _ = SynchronizedBeforeSuite(func() []byte {
Connection: connection,
BaseTime: now,
}
+ casestrace.SharedContext = helpers.SharedContext{
+ Connection: connection,
+ BaseTime: now,
+ }
Expect(err).NotTo(HaveOccurred())
})
diff --git a/test/integration/standalone/query_ondisk/query_ondisk_suite_test.go b/test/integration/standalone/query_ondisk/query_ondisk_suite_test.go
index 99c0ec7e..5862a86e 100644
--- a/test/integration/standalone/query_ondisk/query_ondisk_suite_test.go
+++ b/test/integration/standalone/query_ondisk/query_ondisk_suite_test.go
@@ -40,6 +40,7 @@ import (
casesmeasure "github.com/apache/skywalking-banyandb/test/cases/measure"
casesstream "github.com/apache/skywalking-banyandb/test/cases/stream"
casestopn "github.com/apache/skywalking-banyandb/test/cases/topn"
+ casestrace "github.com/apache/skywalking-banyandb/test/cases/trace"
integration_standalone "github.com/apache/skywalking-banyandb/test/integration/standalone"
)
@@ -95,6 +96,10 @@ var _ = SynchronizedBeforeSuite(func() []byte {
Connection: connection,
BaseTime: now,
}
+ casestrace.SharedContext = helpers.SharedContext{
+ Connection: connection,
+ BaseTime: now,
+ }
Expect(err).NotTo(HaveOccurred())
})