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 2024/01/03 12:45:57 UTC
(skywalking-banyandb) 01/03: Add the measure merger loop
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch measure-merger
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit c1d936f839a56bc1c6e099f7d5c5f23dfb510012
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Mon Jan 1 06:09:44 2024 +0800
Add the measure merger loop
Signed-off-by: Gao Hongtao <ha...@gmail.com>
---
banyand/measure/block.go | 192 ++++++++++++++++---
banyand/measure/block_metadata.go | 25 ++-
banyand/measure/block_reader.go | 98 +++++++++-
banyand/measure/block_reader_test.go | 164 ++++++++++++++++
banyand/measure/block_test.go | 358 ++++++++++++++++++++++++++++++++++-
banyand/measure/block_writer.go | 32 +++-
banyand/measure/flusher.go | 119 ++++++++++--
banyand/measure/gc.go | 98 +++++-----
banyand/measure/gc_test.go | 94 +++------
banyand/measure/introducer.go | 53 +++++-
banyand/measure/merger.go | 308 ++++++++++++++++++++++++++++++
banyand/measure/merger_test.go | 356 ++++++++++++++++++++++++++++++++++
banyand/measure/part.go | 20 +-
banyand/measure/part_iter.go | 125 ++++++++++++
banyand/measure/part_iter_test.go | 64 ++++++-
banyand/measure/part_test.go | 2 +-
banyand/measure/query.go | 2 +-
banyand/measure/query_test.go | 38 +++-
banyand/measure/snapshot.go | 27 ++-
banyand/measure/snapshot_test.go | 142 ++++++++++++++
banyand/measure/tstable.go | 41 ++--
banyand/measure/tstable_test.go | 267 ++++++++++++++++----------
pkg/fs/file_system.go | 11 ++
pkg/fs/local_file_system.go | 11 ++
pkg/fs/local_file_system_nix.go | 5 +
pkg/watcher/watcher.go | 5 +
26 files changed, 2356 insertions(+), 301 deletions(-)
diff --git a/banyand/measure/block.go b/banyand/measure/block.go
index d3f10459..6b97c465 100644
--- a/banyand/measure/block.go
+++ b/banyand/measure/block.go
@@ -134,7 +134,7 @@ func (b *block) mustWriteTo(sid common.SeriesID, bm *blockMetadata, ww *writers)
bm.uncompressedSizeBytes = b.uncompressedSizeBytes()
bm.count = uint64(b.Len())
- mustWriteTimestampsTo(&bm.timestamps, b.timestamps, ww.timestampsWriter)
+ mustWriteTimestampsTo(&bm.timestamps, b.timestamps, &ww.timestampsWriter)
for ti := range b.tagFamilies {
b.marshalTagFamily(b.tagFamilies[ti], bm, ww)
@@ -196,9 +196,9 @@ func (b *block) marshalTagFamily(tf columnFamily, bm *blockMetadata, ww *writers
}
func (b *block) unmarshalTagFamily(decoder *encoding.BytesBlockDecoder, tfIndex int, name string,
- columnFamilyMetadataBlock *dataBlock, tagProjection []string, metaReader, valueReader fs.Reader,
+ columnFamilyMetadataBlock *dataBlock, tagProjection []string, metaReader, valueReader fs.Reader, readTagByProjection bool,
) {
- if len(tagProjection) < 1 {
+ if readTagByProjection && len(tagProjection) < 1 {
return
}
bb := bigValuePool.Generate()
@@ -212,15 +212,25 @@ func (b *block) unmarshalTagFamily(decoder *encoding.BytesBlockDecoder, tfIndex
}
bigValuePool.Release(bb)
b.tagFamilies[tfIndex].name = name
- cc := b.tagFamilies[tfIndex].resizeColumns(len(tagProjection))
- for j := range tagProjection {
- for i := range cfm.columnMetadata {
- if tagProjection[j] == cfm.columnMetadata[i].name {
- cc[j].mustReadValues(decoder, valueReader, cfm.columnMetadata[i], uint64(b.Len()))
- break
+ if readTagByProjection {
+ if len(tagProjection) < 1 {
+ return
+ }
+ cc := b.tagFamilies[tfIndex].resizeColumns(len(tagProjection))
+ for j := range tagProjection {
+ for i := range cfm.columnMetadata {
+ if tagProjection[j] == cfm.columnMetadata[i].name {
+ cc[j].mustReadValues(decoder, valueReader, cfm.columnMetadata[i], uint64(b.Len()))
+ break
+ }
}
}
+ return
+ }
+ cc := b.tagFamilies[tfIndex].resizeColumns(len(cfm.columnMetadata))
+ for i := range cfm.columnMetadata {
+ cc[i].mustReadValues(decoder, valueReader, cfm.columnMetadata[i], uint64(b.Len()))
}
}
@@ -256,24 +266,42 @@ func (b *block) uncompressedSizeBytes() uint64 {
return n
}
-func (b *block) mustReadFrom(decoder *encoding.BytesBlockDecoder, p *part, bm blockMetadata) {
+func (b *block) mustReadFrom(decoder *encoding.BytesBlockDecoder, p *part, bm blockMetadata, readTagByProjection bool) {
b.reset()
b.timestamps = mustReadTimestampsFrom(b.timestamps, &bm.timestamps, int(bm.count), p.timestamps)
- _ = b.resizeTagFamilies(len(bm.tagProjection))
- for i := range bm.tagProjection {
- name := bm.tagProjection[i].Family
- block, ok := bm.tagFamilies[name]
- if !ok {
- continue
- }
- b.unmarshalTagFamily(decoder, i, name, block, bm.tagProjection[i].Names, p.tagFamilyMetadata[name], p.tagFamilies[name])
- }
cc := b.field.resizeColumns(len(bm.field.columnMetadata))
for i := range cc {
cc[i].mustReadValues(decoder, p.fieldValues, bm.field.columnMetadata[i], bm.count)
}
+
+ if readTagByProjection {
+ _ = b.resizeTagFamilies(len(bm.tagProjection))
+ for i := range bm.tagProjection {
+ name := bm.tagProjection[i].Family
+ block, ok := bm.tagFamilies[name]
+ if !ok {
+ continue
+ }
+ b.unmarshalTagFamily(decoder, i, name, block,
+ bm.tagProjection[i].Names, p.tagFamilyMetadata[name],
+ p.tagFamilies[name], readTagByProjection)
+ }
+ return
+ }
+
+ _ = b.resizeTagFamilies(len(bm.tagFamilies))
+ keys := make([]string, 0, len(bm.tagFamilies))
+ for k := range bm.tagFamilies {
+ keys = append(keys, k)
+ }
+ sort.Strings(keys)
+ for i, name := range keys {
+ block := bm.tagFamilies[name]
+ b.unmarshalTagFamily(decoder, i, name, block,
+ nil, p.tagFamilyMetadata[name], p.tagFamilies[name], readTagByProjection)
+ }
}
// For testing purpose only.
@@ -283,7 +311,7 @@ func (b *block) sortTagFamilies() {
})
}
-func mustWriteTimestampsTo(tm *timestampsMetadata, timestamps []int64, timestampsWriter writer) {
+func mustWriteTimestampsTo(tm *timestampsMetadata, timestamps []int64, timestampsWriter *writer) {
tm.reset()
bb := bigValuePool.Generate()
@@ -369,9 +397,20 @@ func (bc *blockCursor) init(p *part, bm blockMetadata, queryOpts queryOptions) {
bc.fieldProjection = queryOpts.FieldProjection
}
-func (bc *blockCursor) copyAllTo(r *pbv1.Result) {
+func (bc *blockCursor) copyAllTo(r *pbv1.Result, desc bool) {
+ var idx, offset int
+ if desc {
+ idx = 0
+ offset = bc.idx + 1
+ } else {
+ idx = bc.idx
+ offset = len(bc.timestamps)
+ }
+ if offset <= idx {
+ return
+ }
r.SID = bc.bm.seriesID
- r.Timestamps = append(r.Timestamps, bc.timestamps...)
+ r.Timestamps = append(r.Timestamps, bc.timestamps[idx:offset]...)
for _, cf := range bc.tagFamilies {
tf := pbv1.TagFamily{
Name: cf.name,
@@ -380,7 +419,7 @@ func (bc *blockCursor) copyAllTo(r *pbv1.Result) {
t := pbv1.Tag{
Name: c.name,
}
- for _, v := range c.values {
+ for _, v := range c.values[idx:offset] {
t.Values = append(t.Values, mustDecodeTagValue(c.valueType, v))
}
tf.Tags = append(tf.Tags, t)
@@ -391,7 +430,7 @@ func (bc *blockCursor) copyAllTo(r *pbv1.Result) {
f := pbv1.Field{
Name: c.name,
}
- for _, v := range c.values {
+ for _, v := range c.values[idx:offset] {
f.Values = append(f.Values, mustDecodeFieldValue(c.valueType, v))
}
r.Fields = append(r.Fields, f)
@@ -461,7 +500,7 @@ func (bc *blockCursor) loadData(tmpBlock *block) bool {
}
}
bc.bm.tagFamilies = tf
- tmpBlock.mustReadFrom(&bc.columnValuesDecoder, bc.p, bc.bm)
+ tmpBlock.mustReadFrom(&bc.columnValuesDecoder, bc.p, bc.bm, true)
start, end, ok := findRange(tmpBlock.timestamps, bc.minTimestamp, bc.maxTimestamp)
if !ok {
@@ -548,3 +587,106 @@ func releaseBlockCursor(bc *blockCursor) {
bc.reset()
blockCursorPool.Put(bc)
}
+
+type blockPointer struct {
+ block
+
+ idx int
+ bm blockMetadata
+ // from partMetadata.ID
+ partID uint64
+}
+
+func (bi *blockPointer) copyFrom(src *blockPointer) {
+ bi.idx = 0
+ bi.bm.copyFrom(&src.bm)
+ bi.appendAll(src)
+}
+
+func (bi *blockPointer) appendAll(b *blockPointer) {
+ if len(b.timestamps) == 0 {
+ return
+ }
+ bi.append(b, len(b.timestamps))
+}
+
+func (bi *blockPointer) append(b *blockPointer, offset int) {
+ if offset <= b.idx {
+ return
+ }
+ if len(bi.tagFamilies) == 0 && len(b.tagFamilies) > 0 {
+ for _, tf := range b.tagFamilies {
+ tagFamily := columnFamily{name: tf.name}
+ for _, c := range tf.columns {
+ col := column{name: c.name, valueType: c.valueType}
+ assertIdxAndOffset(col.name, len(c.values), b.idx, offset)
+ col.values = append(col.values, c.values[b.idx:offset]...)
+ tagFamily.columns = append(tagFamily.columns, col)
+ }
+ bi.tagFamilies = append(bi.tagFamilies, tagFamily)
+ }
+ } else {
+ if len(bi.tagFamilies) != len(b.tagFamilies) {
+ logger.Panicf("unexpected number of tag families: got %d; want %d", len(bi.tagFamilies), len(b.tagFamilies))
+ }
+ for i := range bi.tagFamilies {
+ if bi.tagFamilies[i].name != b.tagFamilies[i].name {
+ logger.Panicf("unexpected tag family name: got %q; want %q", bi.tagFamilies[i].name, b.tagFamilies[i].name)
+ }
+ if len(bi.tagFamilies[i].columns) != len(b.tagFamilies[i].columns) {
+ logger.Panicf("unexpected number of tags for tag family %q: got %d; want %d", bi.tagFamilies[i].name, len(bi.tagFamilies[i].columns), len(b.tagFamilies[i].columns))
+ }
+ for j := range bi.tagFamilies[i].columns {
+ if bi.tagFamilies[i].columns[j].name != b.tagFamilies[i].columns[j].name {
+ logger.Panicf("unexpected tag name for tag family %q: got %q; want %q", bi.tagFamilies[i].name, bi.tagFamilies[i].columns[j].name, b.tagFamilies[i].columns[j].name)
+ }
+ assertIdxAndOffset(b.tagFamilies[i].columns[j].name, len(b.tagFamilies[i].columns[j].values), b.idx, offset)
+ bi.tagFamilies[i].columns[j].values = append(bi.tagFamilies[i].columns[j].values, b.tagFamilies[i].columns[j].values[b.idx:offset]...)
+ }
+ }
+ }
+
+ if len(bi.field.columns) == 0 && len(b.field.columns) > 0 {
+ for _, c := range b.field.columns {
+ col := column{name: c.name, valueType: c.valueType}
+ assertIdxAndOffset(col.name, len(c.values), b.idx, offset)
+ col.values = append(col.values, c.values[b.idx:offset]...)
+ bi.field.columns = append(bi.field.columns, col)
+ }
+ } else {
+ if len(bi.field.columns) != len(b.field.columns) {
+ logger.Panicf("unexpected number of fields: got %d; want %d", len(bi.field.columns), len(b.field.columns))
+ }
+ for i := range bi.field.columns {
+ assertIdxAndOffset(b.field.columns[i].name, len(b.field.columns[i].values), b.idx, offset)
+ bi.field.columns[i].values = append(bi.field.columns[i].values, b.field.columns[i].values[b.idx:offset]...)
+ }
+ }
+
+ assertIdxAndOffset("timestamps", len(b.timestamps), bi.idx, offset)
+ bi.timestamps = append(bi.timestamps, b.timestamps[b.idx:offset]...)
+
+ if bi.partID < b.partID {
+ bi.partID = b.partID
+ }
+}
+
+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.uncompressedSizeBytes >= maxUncompressedBlockSize
+}
+
+func (bi *blockPointer) reset() {
+ bi.idx = 0
+ bi.partID = 0
+ bi.block.reset()
+ bi.bm = blockMetadata{}
+}
diff --git a/banyand/measure/block_metadata.go b/banyand/measure/block_metadata.go
index e239162c..6202c9eb 100644
--- a/banyand/measure/block_metadata.go
+++ b/banyand/measure/block_metadata.go
@@ -20,6 +20,7 @@ package measure
import (
"errors"
"fmt"
+ "sort"
"sync"
"github.com/apache/skywalking-banyandb/api/common"
@@ -74,6 +75,21 @@ type blockMetadata struct {
count uint64
}
+func (bh *blockMetadata) copyFrom(src *blockMetadata) {
+ bh.seriesID = src.seriesID
+ bh.uncompressedSizeBytes = src.uncompressedSizeBytes
+ bh.count = src.count
+ bh.timestamps.copyFrom(&src.timestamps)
+ for k, db := range src.tagFamilies {
+ if bh.tagFamilies == nil {
+ bh.tagFamilies = make(map[string]*dataBlock)
+ }
+ bh.tagFamilies[k] = &dataBlock{}
+ bh.tagFamilies[k].copyFrom(db)
+ }
+ bh.field.copyFrom(&src.field)
+}
+
func (bh *blockMetadata) getTagFamilyMetadata(name string) *dataBlock {
if bh.tagFamilies == nil {
bh.tagFamilies = make(map[string]*dataBlock)
@@ -105,7 +121,14 @@ func (bh *blockMetadata) marshal(dst []byte) []byte {
dst = encoding.VarUint64ToBytes(dst, bh.count)
dst = bh.timestamps.marshal(dst)
dst = encoding.VarUint64ToBytes(dst, uint64(len(bh.tagFamilies)))
- for name, cf := range bh.tagFamilies {
+ // make sure the order of tagFamilies is stable
+ keys := make([]string, 0, len(bh.tagFamilies))
+ for k := range bh.tagFamilies {
+ keys = append(keys, k)
+ }
+ sort.Strings(keys)
+ for _, name := range keys {
+ cf := bh.tagFamilies[name]
dst = encoding.EncodeBytes(dst, convert.StringToBytes(name))
dst = cf.marshal(dst)
}
diff --git a/banyand/measure/block_reader.go b/banyand/measure/block_reader.go
index de5c5c2f..88f7fb06 100644
--- a/banyand/measure/block_reader.go
+++ b/banyand/measure/block_reader.go
@@ -17,7 +17,13 @@
package measure
-import "github.com/apache/skywalking-banyandb/pkg/fs"
+import (
+ "container/heap"
+ "fmt"
+ "io"
+
+ "github.com/apache/skywalking-banyandb/pkg/fs"
+)
type reader struct {
r fs.Reader
@@ -37,3 +43,93 @@ func (r *reader) Read(p []byte) (int, error) {
r.bytesRead += uint64(n)
return n, err
}
+
+type blockReader struct {
+ block *blockPointer
+ pih partMergeIterHeap
+ nextBlockNoop bool
+ err error
+}
+
+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.nextBlock() {
+ 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) nextBlock() bool {
+ if br.err != nil {
+ return false
+ }
+ if br.nextBlockNoop {
+ br.nextBlockNoop = false
+ return true
+ }
+
+ br.err = br.next()
+ switch br.err {
+ case nil:
+ return true
+ case io.EOF:
+ return false
+ default:
+ br.err = fmt.Errorf("can't get the block to merge: %w", br.err)
+ return false
+ }
+}
+
+func (br *blockReader) next() error {
+ head := br.pih[0]
+ if head.nextBlock() {
+ 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) error() error {
+ if br.err == io.EOF {
+ return nil
+ }
+ return br.err
+}
diff --git a/banyand/measure/block_reader_test.go b/banyand/measure/block_reader_test.go
new file mode 100644
index 00000000..fc187f4e
--- /dev/null
+++ b/banyand/measure/block_reader_test.go
@@ -0,0 +1,164 @@
+// 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 measure
+
+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"
+ "github.com/apache/skywalking-banyandb/pkg/test"
+)
+
+func Test_blockReader_nextBlock(t *testing.T) {
+ tests := []struct {
+ wantErr error
+ name string
+ dpsList []*dataPoints
+ want []blockMetadata
+ }{
+ {
+ name: "Test with no data points",
+ dpsList: []*dataPoints{},
+ },
+ {
+ name: "Test with single part",
+ dpsList: []*dataPoints{dpsTS1},
+ want: []blockMetadata{
+ {seriesID: 1, count: 1, uncompressedSizeBytes: 1676},
+ {seriesID: 2, count: 1, uncompressedSizeBytes: 55},
+ {seriesID: 3, count: 1, uncompressedSizeBytes: 24},
+ },
+ },
+ {
+ name: "Test with multiple parts with different ts",
+ dpsList: []*dataPoints{dpsTS1, dpsTS2},
+ want: []blockMetadata{
+ {seriesID: 1, count: 1, uncompressedSizeBytes: 1676},
+ {seriesID: 1, count: 1, uncompressedSizeBytes: 1676},
+ {seriesID: 2, count: 1, uncompressedSizeBytes: 55},
+ {seriesID: 2, count: 1, uncompressedSizeBytes: 55},
+ {seriesID: 3, count: 1, uncompressedSizeBytes: 24},
+ {seriesID: 3, count: 1, uncompressedSizeBytes: 24},
+ },
+ },
+ {
+ name: "Test with multiple parts with same ts",
+ dpsList: []*dataPoints{dpsTS1, dpsTS1},
+ want: []blockMetadata{
+ {seriesID: 1, count: 1, uncompressedSizeBytes: 1676},
+ {seriesID: 1, count: 1, uncompressedSizeBytes: 1676},
+ {seriesID: 2, count: 1, uncompressedSizeBytes: 55},
+ {seriesID: 2, count: 1, uncompressedSizeBytes: 55},
+ {seriesID: 3, count: 1, uncompressedSizeBytes: 24},
+ {seriesID: 3, count: 1, uncompressedSizeBytes: 24},
+ },
+ },
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ verify := func(pp []*part) {
+ var pii []*partMergeIter
+ for _, p := range pp {
+ pmi := &partMergeIter{}
+ pmi.mustInitFromPart(p)
+ pii = append(pii, pmi)
+ }
+ reader := &blockReader{}
+ reader.init(pii)
+ var got []blockMetadata
+ for reader.nextBlock() {
+ if reader.block.bm.seriesID == 0 {
+ t.Errorf("Expected curBlock to be initialized, but it was nil")
+ }
+ var tagFamilyNames []string
+ for _, cf := range reader.block.block.tagFamilies {
+ tagFamilyNames = append(tagFamilyNames, cf.name)
+ }
+ if !sort.IsSorted(sort.StringSlice(tagFamilyNames)) {
+ t.Errorf("Expected tagFamilyNames %q to be sorted, but it was not", tagFamilyNames)
+ }
+ 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{}, "timestamps"),
+ cmpopts.IgnoreFields(blockMetadata{}, "field"),
+ cmpopts.IgnoreFields(blockMetadata{}, "tagFamilies"),
+ 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 _, dps := range tt.dpsList {
+ mp := generateMemPart()
+ mpp = append(mpp, mp)
+ mp.mustInitFromDataPoints(dps)
+ pp = append(pp, openMemPart(mp))
+ }
+ verify(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.decRef()
+ }
+ defFn()
+ }()
+ var pp []*part
+ fileSystem := fs.NewLocalFileSystem()
+ for i, dps := range tt.dpsList {
+ mp := generateMemPart()
+ mpp = append(mpp, mp)
+ mp.mustInitFromDataPoints(dps)
+ mp.mustFlush(fileSystem, partPath(tmpPath, uint64(i)))
+ filePW := newPartWrapper(nil, mustOpenFilePart(uint64(i), tmpPath, fileSystem))
+ filePW.p.partMetadata.ID = uint64(i)
+ fpp = append(fpp, filePW)
+ pp = append(pp, filePW.p)
+ }
+ verify(pp)
+ })
+ })
+ }
+}
diff --git a/banyand/measure/block_test.go b/banyand/measure/block_test.go
index e7b776c1..ff480210 100644
--- a/banyand/measure/block_test.go
+++ b/banyand/measure/block_test.go
@@ -264,7 +264,7 @@ func Test_mustWriteAndReadTimestamps(t *testing.T) {
}()
tm := ×tampsMetadata{}
b := &bytes.Buffer{}
- mustWriteTimestampsTo(tm, tt.args, writer{w: b})
+ mustWriteTimestampsTo(tm, tt.args, &writer{w: b})
timestamps := mustReadTimestampsFrom(nil, tm, len(tt.args), b)
if !reflect.DeepEqual(timestamps, tt.args) {
t.Errorf("mustReadTimestampsFrom() = %v, want %v", timestamps, tt.args)
@@ -318,7 +318,7 @@ func Test_marshalAndUnmarshalTagFamily(t *testing.T) {
unmarshaled.timestamps = make([]int64, len(b.timestamps))
unmarshaled.resizeTagFamilies(1)
- unmarshaled.unmarshalTagFamily(decoder, tfIndex, name, bm.getTagFamilyMetadata(name), tagProjection[name], metaBuffer, dataBuffer)
+ unmarshaled.unmarshalTagFamily(decoder, tfIndex, name, bm.getTagFamilyMetadata(name), tagProjection[name], metaBuffer, dataBuffer, true)
if diff := cmp.Diff(unmarshaled.tagFamilies[0], b.tagFamilies[0],
cmp.AllowUnexported(columnFamily{}, column{}),
@@ -371,7 +371,7 @@ func Test_marshalAndUnmarshalBlock(t *testing.T) {
})
}
bm.tagProjection = tp
- unmarshaled.mustReadFrom(decoder, p, bm)
+ unmarshaled.mustReadFrom(decoder, p, bm, true)
// blockMetadata is using a map, so the order of tag families is not guaranteed
unmarshaled.sortTagFamilies()
@@ -464,3 +464,355 @@ func Test_findRange(t *testing.T) {
})
}
}
+
+func Test_blockPointer_append(t *testing.T) {
+ type fields struct {
+ timestamps []int64
+ tagFamilies []columnFamily
+ field columnFamily
+ partID uint64
+ }
+ type args struct {
+ b *blockPointer
+ offset int
+ }
+ tests := []struct {
+ name string
+ fields fields
+ args args
+ want *blockPointer
+ wantPanic bool
+ }{
+ {
+ name: "Test append with empty block",
+ fields: fields{
+ timestamps: []int64{1, 2},
+ tagFamilies: []columnFamily{
+ {
+ name: "arrTag",
+ columns: []column{
+ {
+ name: "strArrTag", valueType: pbv1.ValueTypeStrArr,
+ values: [][]byte{marshalStrArr([][]byte{[]byte("value5"), []byte("value6")}), marshalStrArr([][]byte{[]byte("value7"), []byte("value8")})},
+ },
+ },
+ },
+ },
+ field: columnFamily{
+ columns: []column{
+ {name: "strField", valueType: pbv1.ValueTypeStr, values: [][]byte{[]byte("field3"), []byte("field4")}},
+ },
+ },
+ },
+ args: args{
+ b: &blockPointer{
+ block: block{
+ timestamps: []int64{},
+ tagFamilies: []columnFamily{},
+ field: columnFamily{},
+ },
+ idx: 0,
+ },
+ offset: 0,
+ },
+ want: &blockPointer{
+ block: block{
+ timestamps: []int64{1, 2},
+ tagFamilies: []columnFamily{
+ {
+ name: "arrTag",
+ columns: []column{
+ {
+ name: "strArrTag", valueType: pbv1.ValueTypeStrArr,
+ values: [][]byte{marshalStrArr([][]byte{[]byte("value5"), []byte("value6")}), marshalStrArr([][]byte{[]byte("value7"), []byte("value8")})},
+ },
+ },
+ },
+ },
+ field: columnFamily{
+ columns: []column{
+ {name: "strField", valueType: pbv1.ValueTypeStr, values: [][]byte{[]byte("field3"), []byte("field4")}},
+ },
+ },
+ },
+ idx: 0,
+ },
+ },
+ {
+ name: "Test append to a empty block",
+ fields: fields{
+ timestamps: nil,
+ tagFamilies: nil,
+ field: columnFamily{
+ columns: nil,
+ },
+ partID: 0,
+ },
+ args: args{
+ b: &blockPointer{
+ partID: 2,
+ block: block{
+ timestamps: []int64{4, 5},
+ tagFamilies: []columnFamily{
+ {
+ name: "arrTag",
+ columns: []column{
+ {
+ name: "strArrTag", valueType: pbv1.ValueTypeStrArr,
+ values: [][]byte{marshalStrArr([][]byte{[]byte("value9"), []byte("value10")}), marshalStrArr([][]byte{[]byte("value11"), []byte("value12")})},
+ },
+ },
+ },
+ },
+ field: columnFamily{
+ columns: []column{
+ {name: "strField", valueType: pbv1.ValueTypeStr, values: [][]byte{[]byte("field5"), []byte("field6")}},
+ },
+ },
+ },
+ idx: 0,
+ },
+ offset: 2,
+ },
+ want: &blockPointer{
+ partID: 2,
+ block: block{
+ timestamps: []int64{4, 5},
+ tagFamilies: []columnFamily{
+ {
+ name: "arrTag",
+ columns: []column{
+ {
+ name: "strArrTag", valueType: pbv1.ValueTypeStrArr,
+ values: [][]byte{marshalStrArr([][]byte{[]byte("value9"), []byte("value10")}), marshalStrArr([][]byte{[]byte("value11"), []byte("value12")})},
+ },
+ },
+ },
+ },
+ field: columnFamily{
+ columns: []column{
+ {name: "strField", valueType: pbv1.ValueTypeStr, values: [][]byte{[]byte("field5"), []byte("field6")}},
+ },
+ },
+ },
+ idx: 0,
+ },
+ },
+ {
+ name: "Test append with offset equals to the data size. All data",
+ fields: fields{
+ timestamps: []int64{1, 2},
+ tagFamilies: []columnFamily{
+ {
+ name: "arrTag",
+ columns: []column{
+ {
+ name: "strArrTag", valueType: pbv1.ValueTypeStrArr,
+ values: [][]byte{marshalStrArr([][]byte{[]byte("value5"), []byte("value6")}), marshalStrArr([][]byte{[]byte("value7"), []byte("value8")})},
+ },
+ },
+ },
+ },
+ field: columnFamily{
+ columns: []column{
+ {name: "strField", valueType: pbv1.ValueTypeStr, values: [][]byte{[]byte("field3"), []byte("field4")}},
+ },
+ },
+ partID: 1,
+ },
+ args: args{
+ b: &blockPointer{
+ partID: 2,
+ block: block{
+ timestamps: []int64{4, 5},
+ tagFamilies: []columnFamily{
+ {
+ name: "arrTag",
+ columns: []column{
+ {
+ name: "strArrTag", valueType: pbv1.ValueTypeStrArr,
+ values: [][]byte{marshalStrArr([][]byte{[]byte("value9"), []byte("value10")}), marshalStrArr([][]byte{[]byte("value11"), []byte("value12")})},
+ },
+ },
+ },
+ },
+ field: columnFamily{
+ columns: []column{
+ {name: "strField", valueType: pbv1.ValueTypeStr, values: [][]byte{[]byte("field5"), []byte("field6")}},
+ },
+ },
+ },
+ idx: 0,
+ },
+ offset: 2,
+ },
+ want: &blockPointer{
+ partID: 2,
+ block: block{
+ timestamps: []int64{1, 2, 4, 5},
+ tagFamilies: []columnFamily{
+ {
+ name: "arrTag",
+ columns: []column{
+ {
+ name: "strArrTag", valueType: pbv1.ValueTypeStrArr,
+ values: [][]byte{marshalStrArr([][]byte{[]byte("value5"), []byte("value6")}), marshalStrArr([][]byte{[]byte("value7"), []byte("value8")}), marshalStrArr([][]byte{[]byte("value9"), []byte("value10")}), marshalStrArr([][]byte{[]byte("value11"), []byte("value12")})},
+ },
+ },
+ },
+ },
+ field: columnFamily{
+ columns: []column{
+ {name: "strField", valueType: pbv1.ValueTypeStr, values: [][]byte{[]byte("field3"), []byte("field4"), []byte("field5"), []byte("field6")}},
+ },
+ },
+ },
+ idx: 0,
+ },
+ },
+ {
+ name: "Test append with non-empty block and offset less than timestamps",
+ fields: fields{
+ timestamps: []int64{1, 2},
+ tagFamilies: []columnFamily{
+ {
+ name: "arrTag",
+ columns: []column{
+ {
+ name: "strArrTag", valueType: pbv1.ValueTypeStrArr,
+ values: [][]byte{marshalStrArr([][]byte{[]byte("value5"), []byte("value6")}), marshalStrArr([][]byte{[]byte("value7"), []byte("value8")})},
+ },
+ },
+ },
+ },
+ field: columnFamily{
+ columns: []column{
+ {name: "strField", valueType: pbv1.ValueTypeStr, values: [][]byte{[]byte("field3"), []byte("field4")}},
+ },
+ },
+ partID: 3,
+ },
+ args: args{
+ b: &blockPointer{
+ partID: 2,
+ block: block{
+ timestamps: []int64{4, 5},
+ tagFamilies: []columnFamily{
+ {
+ name: "arrTag",
+ columns: []column{
+ {
+ name: "strArrTag", valueType: pbv1.ValueTypeStrArr,
+ values: [][]byte{marshalStrArr([][]byte{[]byte("value9"), []byte("value10")}), marshalStrArr([][]byte{[]byte("value11"), []byte("value12")})},
+ },
+ },
+ },
+ },
+ field: columnFamily{
+ columns: []column{
+ {name: "strField", valueType: pbv1.ValueTypeStr, values: [][]byte{[]byte("field5"), []byte("field6")}},
+ },
+ },
+ },
+ idx: 0,
+ },
+ offset: 1,
+ },
+ want: &blockPointer{
+ partID: 3,
+ block: block{
+ timestamps: []int64{1, 2, 4},
+ tagFamilies: []columnFamily{
+ {
+ name: "arrTag",
+ columns: []column{
+ {
+ name: "strArrTag", valueType: pbv1.ValueTypeStrArr,
+ values: [][]byte{marshalStrArr([][]byte{[]byte("value5"), []byte("value6")}), marshalStrArr([][]byte{[]byte("value7"), []byte("value8")}), marshalStrArr([][]byte{[]byte("value9"), []byte("value10")})},
+ },
+ },
+ },
+ },
+ field: columnFamily{
+ columns: []column{
+ {name: "strField", valueType: pbv1.ValueTypeStr, values: [][]byte{[]byte("field3"), []byte("field4"), []byte("field5")}},
+ },
+ },
+ },
+ idx: 0,
+ },
+ },
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ defer func() {
+ r := recover()
+ if (r != nil) != tt.wantPanic {
+ t.Errorf("blockPointer.append() recover = %v, wantPanic = %v", r, tt.wantPanic)
+ }
+ }()
+ bi := &blockPointer{
+ block: block{
+ timestamps: tt.fields.timestamps,
+ tagFamilies: tt.fields.tagFamilies,
+ field: tt.fields.field,
+ },
+ partID: tt.fields.partID,
+ }
+ bi.append(tt.args.b, tt.args.offset)
+ if !reflect.DeepEqual(bi, tt.want) {
+ t.Errorf("blockPointer.append() = %+v, want %+v", bi, tt.want)
+ }
+ })
+ }
+}
+
+func Test_blockPointer_copyFrom(t *testing.T) {
+ type fields struct {
+ bm blockMetadata
+ idx int
+ }
+ type args struct {
+ src *blockPointer
+ }
+ tests := []struct {
+ name string
+ fields fields
+ args args
+ want *blockPointer
+ }{
+ {
+ name: "Test copyFrom",
+ fields: fields{
+ bm: blockMetadata{},
+ idx: 0,
+ },
+ args: args{
+ src: &blockPointer{
+ bm: blockMetadata{count: 1},
+ idx: 0,
+ block: conventionalBlock,
+ partID: 2,
+ },
+ },
+ want: &blockPointer{
+ bm: blockMetadata{count: 1},
+ idx: 0,
+ block: conventionalBlock,
+ partID: 2,
+ },
+ },
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ bi := &blockPointer{
+ bm: tt.fields.bm,
+ idx: tt.fields.idx,
+ }
+ bi.copyFrom(tt.args.src)
+ if !reflect.DeepEqual(bi, tt.want) {
+ t.Errorf("blockPointer.copyFrom() = %+v, want %+v", bi, tt.want)
+ }
+ })
+ }
+}
diff --git a/banyand/measure/block_writer.go b/banyand/measure/block_writer.go
index fc840084..15b9bd96 100644
--- a/banyand/measure/block_writer.go
+++ b/banyand/measure/block_writer.go
@@ -18,6 +18,7 @@
package measure
import (
+ "path/filepath"
"sync"
"github.com/apache/skywalking-banyandb/api/common"
@@ -170,6 +171,19 @@ func (bw *blockWriter) MustInitForMemPart(mp *memPart) {
bw.writers.fieldValuesWriter.init(&mp.fieldValues)
}
+func (bw *blockWriter) mustInitForFilePart(fileSystem fs.FileSystem, path string) {
+ bw.reset()
+ fileSystem.MkdirPanicIfExist(path, dirPermission)
+ bw.writers.mustCreateTagFamilyWriters = func(name string) (fs.Writer, fs.Writer) {
+ return fs.MustCreateFile(fileSystem, filepath.Join(path, name+tagFamiliesMetadataFilenameExt), filePermission),
+ fs.MustCreateFile(fileSystem, filepath.Join(path, name+tagFamiliesFilenameExt), filePermission)
+ }
+ bw.writers.metaWriter.init(fs.MustCreateFile(fileSystem, filepath.Join(path, metaFilename), filePermission))
+ bw.writers.primaryWriter.init(fs.MustCreateFile(fileSystem, filepath.Join(path, primaryFilename), filePermission))
+ bw.writers.timestampsWriter.init(fs.MustCreateFile(fileSystem, filepath.Join(path, timestampsFilename), filePermission))
+ bw.writers.fieldValuesWriter.init(fs.MustCreateFile(fileSystem, filepath.Join(path, fieldValuesFilename), filePermission))
+}
+
func (bw *blockWriter) MustWriteDataPoints(sid common.SeriesID, timestamps []int64, tagFamilies [][]nameValues, fields []nameValues) {
if len(timestamps) == 0 {
return
@@ -178,6 +192,10 @@ func (bw *blockWriter) MustWriteDataPoints(sid common.SeriesID, timestamps []int
b := generateBlock()
defer releaseBlock(b)
b.mustInitFromDataPoints(timestamps, tagFamilies, fields)
+ bw.mustWriteBlock(sid, b)
+}
+
+func (bw *blockWriter) mustWriteBlock(sid common.SeriesID, b *block) {
if b.Len() == 0 {
return
}
@@ -235,12 +253,12 @@ func (bw *blockWriter) mustFlushPrimaryBlock(data []byte) {
bw.sidFirst = 0
}
-func (bw *blockWriter) Flush(ph *partMetadata) {
- ph.UncompressedSizeBytes = bw.totalUncompressedSizeBytes
- ph.TotalCount = bw.totalCount
- ph.BlocksCount = bw.totalBlocksCount
- ph.MinTimestamp = bw.totalMinTimestamp
- ph.MaxTimestamp = bw.totalMaxTimestamp
+func (bw *blockWriter) Flush(pm *partMetadata) {
+ pm.UncompressedSizeBytes = bw.totalUncompressedSizeBytes
+ pm.TotalCount = bw.totalCount
+ pm.BlocksCount = bw.totalBlocksCount
+ pm.MinTimestamp = bw.totalMinTimestamp
+ pm.MaxTimestamp = bw.totalMaxTimestamp
bw.mustFlushPrimaryBlock(bw.primaryBlockData)
@@ -249,7 +267,7 @@ func (bw *blockWriter) Flush(ph *partMetadata) {
bw.writers.metaWriter.MustWrite(bb.Buf)
bigValuePool.Release(bb)
- ph.CompressedSizeBytes = bw.writers.totalBytesWritten()
+ pm.CompressedSizeBytes = bw.writers.totalBytesWritten()
bw.writers.MustClose()
bw.reset()
diff --git a/banyand/measure/flusher.go b/banyand/measure/flusher.go
index 4fe26408..43b49e52 100644
--- a/banyand/measure/flusher.go
+++ b/banyand/measure/flusher.go
@@ -18,22 +18,36 @@
package measure
import (
+ "errors"
+ "math"
+ "time"
+
"github.com/apache/skywalking-banyandb/pkg/watcher"
)
-func (tst *tsTable) flusherLoop(flushCh chan *flusherIntroduction, introducerWatcher watcher.Channel, epoch uint64) {
+const flusherWaitTimeout = 5 * time.Second
+
+func (tst *tsTable) flusherLoop(flushCh chan *flusherIntroduction, mergeCh chan *mergerIntroduction, introducerWatcher, flusherWatcher watcher.Channel, epoch uint64) {
defer tst.loopCloser.Done()
epochWatcher := introducerWatcher.Add(0, tst.loopCloser.CloseNotify())
if epochWatcher == nil {
return
}
+ var flusherWatchers watcher.Epochs
for {
select {
case <-tst.loopCloser.CloseNotify():
return
+ case e := <-flusherWatcher:
+ flusherWatchers.Add(e)
case <-epochWatcher.Watch():
- var curSnapshot *snapshot
+ curSnapshot := tst.currentSnapshot()
+ if curSnapshot != nil {
+ epoch, flusherWatchers = tst.pauseFlusherToPileupMemParts(epoch, flusherWatcher, introducerWatcher, flusherWatchers)
+ curSnapshot.decRef()
+ curSnapshot = nil
+ }
tst.RLock()
if tst.snapshot != nil && tst.snapshot.epoch > epoch {
curSnapshot = tst.snapshot
@@ -41,11 +55,26 @@ func (tst *tsTable) flusherLoop(flushCh chan *flusherIntroduction, introducerWat
}
tst.RUnlock()
if curSnapshot != nil {
- epoch = tst.flush(curSnapshot, flushCh)
- if epoch == 0 {
- return
+ newSnapshot, err := tst.mergeMemParts(curSnapshot, mergeCh)
+ if err != nil {
+ tst.l.Logger.Warn().Err(err).Msgf("cannot merge snapshot: %d", curSnapshot.epoch)
+ curSnapshot.decRef()
+ continue
}
- tst.persistSnapshot(curSnapshot)
+ if newSnapshot != nil {
+ epoch = tst.persistSnapshot(newSnapshot)
+ } else {
+ epoch = tst.flush(curSnapshot, flushCh)
+ if epoch == 0 {
+ curSnapshot.decRef()
+ return
+ }
+ tst.persistSnapshot(curSnapshot)
+ }
+ // Notify merger to start a new round of merge.
+ // This round might have be triggered in pauseFlusherToPileupMemParts.
+ flusherWatchers.Notify(math.MaxUint64)
+ flusherWatcher = nil
curSnapshot.decRef()
if tst.currentEpoch() != epoch {
continue
@@ -60,6 +89,71 @@ func (tst *tsTable) flusherLoop(flushCh chan *flusherIntroduction, introducerWat
}
}
+// pauseFlusherToPileupMemParts takes a pause to wait for in-memory parts to pile up.
+// If there is no in-memory part, we can skip the pause.
+// When a merging is finished, we can skip the pause.
+func (tst *tsTable) pauseFlusherToPileupMemParts(epoch uint64, flushWatcher, introducerWatcher watcher.Channel, flusherWatchers watcher.Epochs) (uint64, watcher.Epochs) {
+ curSnapshot := tst.currentSnapshot()
+ if curSnapshot == nil {
+ return epoch, flusherWatchers
+ }
+ if curSnapshot.creator != snapshotCreatorMemPart {
+ curSnapshot.decRef()
+ return epoch, flusherWatchers
+ }
+ curSnapshot.decRef()
+ select {
+ case <-tst.loopCloser.CloseNotify():
+ case <-time.After(flusherWaitTimeout):
+ case e := <-flushWatcher:
+ flusherWatchers.Add(e)
+ epoch = e.Val()
+ flusherWatchers.Notify(epoch)
+ }
+ return epoch, flusherWatchers
+}
+
+func (tst *tsTable) mergeMemParts(snp *snapshot, mergeCh chan *mergerIntroduction) (*snapshot, error) {
+ var memParts []*partWrapper
+ var remainingParts []*partWrapper
+ remainingIDs := make(map[uint64]struct{})
+ for i := range snp.parts {
+ if snp.parts[i].mp != nil {
+ memParts = append(memParts, snp.parts[i])
+ continue
+ }
+ remainingParts = append(remainingParts, snp.parts[i])
+ remainingIDs[snp.parts[i].ID()] = struct{}{}
+ }
+ if len(memParts) < 2 {
+ return nil, nil
+ }
+ // merge memory must not be closed by the tsTable.close
+ closeCh := make(chan struct{})
+ newPart, err := tst.mergePartsThenSendIntroduction(memParts, remainingIDs, mergeCh, closeCh)
+ close(closeCh)
+ if err != nil {
+ if errors.Is(err, errClosed) {
+ return &snapshot{
+ epoch: snp.epoch,
+ parts: append(remainingParts, newPart),
+ creator: snapshotCreatorMergedFlusher,
+ ref: 1,
+ }, nil
+ }
+ return nil, err
+ }
+ if newPart == nil {
+ return nil, nil
+ }
+ return &snapshot{
+ epoch: snp.epoch,
+ parts: append(remainingParts, newPart),
+ creator: snapshotCreatorMergedFlusher,
+ ref: 1,
+ }, nil
+}
+
func (tst *tsTable) flush(snapshot *snapshot, flushCh chan *flusherIntroduction) uint64 {
ind := generateFlusherIntroduction()
defer releaseFlusherIntroduction(ind)
@@ -69,32 +163,33 @@ func (tst *tsTable) flush(snapshot *snapshot, flushCh chan *flusherIntroduction)
}
partPath := partPath(tst.root, pw.ID())
pw.mp.mustFlush(tst.fileSystem, partPath)
- newPW := newPartWrapper(nil, mustOpenFilePart(partPath, tst.fileSystem), tst.fileSystem)
+ newPW := newPartWrapper(nil, mustOpenFilePart(pw.ID(), tst.root, tst.fileSystem))
newPW.p.partMetadata.ID = pw.ID()
ind.flushed[newPW.ID()] = newPW
}
if len(ind.flushed) < 1 {
- return snapshot.epoch
+ return 0
}
ind.applied = make(chan struct{})
select {
case flushCh <- ind:
case <-tst.loopCloser.CloseNotify():
- return 0
+ return snapshot.epoch
}
select {
case <-ind.applied:
return snapshot.epoch
case <-tst.loopCloser.CloseNotify():
- return 0
+ return snapshot.epoch
}
}
-func (tst *tsTable) persistSnapshot(snapshot *snapshot) {
+func (tst *tsTable) persistSnapshot(snapshot *snapshot) uint64 {
var partNames []string
for i := range snapshot.parts {
partNames = append(partNames, partName(snapshot.parts[i].ID()))
}
tst.mustWriteSnapshot(snapshot.epoch, partNames)
- tst.gc.registerSnapshot(snapshot.epoch)
+ tst.gc.registerSnapshot(snapshot)
+ return snapshot.epoch
}
diff --git a/banyand/measure/gc.go b/banyand/measure/gc.go
index 8fb6f608..6c5c6b44 100644
--- a/banyand/measure/gc.go
+++ b/banyand/measure/gc.go
@@ -18,69 +18,73 @@
package measure
import (
- "errors"
- "path"
- "sort"
-
- "github.com/apache/skywalking-banyandb/pkg/fs"
+ "path/filepath"
)
type garbageCleaner struct {
- parent *tsTable
- snapshots []uint64
- parts []uint64
+ parent *tsTable
+ liveEpoch uint64
+ deletableEpochs []uint64
+ liveParts map[uint64]map[uint64]struct{}
+ knownPartFiles map[uint64]struct{}
+}
+
+func (g *garbageCleaner) init(parent *tsTable) {
+ g.parent = parent
+ g.liveParts = make(map[uint64]map[uint64]struct{})
+ g.knownPartFiles = make(map[uint64]struct{})
}
-func (g *garbageCleaner) registerSnapshot(snapshot uint64) {
- g.snapshots = append(g.snapshots, snapshot)
+func (g *garbageCleaner) registerSnapshot(snapshot *snapshot) {
+ parts := make(map[uint64]struct{})
+ for _, part := range snapshot.parts {
+ parts[part.ID()] = struct{}{}
+ g.knownPartFiles[part.ID()] = struct{}{}
+ }
+ g.liveParts[snapshot.epoch] = parts
+
+ if g.liveEpoch > 0 {
+ g.deletableEpochs = append(g.deletableEpochs, g.liveEpoch)
+ }
+ g.liveEpoch = snapshot.epoch
}
-func (g *garbageCleaner) submitParts(parts ...uint64) {
- g.parts = append(g.parts, parts...)
+func (g *garbageCleaner) submitParts(partID uint64) {
+ g.knownPartFiles[partID] = struct{}{}
}
func (g garbageCleaner) clean() {
- if len(g.snapshots) > 1 {
- g.cleanSnapshots()
- }
- if len(g.parts) > 0 {
- g.cleanParts()
- }
+ g.cleanSnapshots()
+ g.cleanParts()
}
func (g *garbageCleaner) cleanSnapshots() {
- if len(g.snapshots) < 2 {
- return
- }
- if !sort.SliceIsSorted(g.snapshots, func(i, j int) bool {
- return g.snapshots[i] < g.snapshots[j]
- }) {
- sort.Slice(g.snapshots, func(i, j int) bool {
- return g.snapshots[i] < g.snapshots[j]
- })
- }
- // keep the latest snapshot
- var remainingSnapshots []uint64
- for i := 0; i < len(g.snapshots)-1; i++ {
- filePath := path.Join(g.parent.root, snapshotName(g.snapshots[i]))
- if err := g.parent.fileSystem.DeleteFile(filePath); err != nil {
- var notExistErr *fs.FileSystemError
- if errors.As(err, ¬ExistErr) && notExistErr.Code != fs.IsNotExistError {
- g.parent.l.Warn().Err(err).Str("path", filePath).Msg("failed to delete snapshot, will retry in next round. Please check manually")
- remainingSnapshots = append(remainingSnapshots, g.snapshots[i])
- }
+ 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 {
+ delete(g.liveParts, deletableEpoch)
+ continue
}
+ g.parent.l.Warn().Err(err).Msgf("cannot delete snapshot file: %s", path)
+ remainingEpochs = append(remainingEpochs, deletableEpoch)
}
- if remainingSnapshots == nil {
- g.snapshots = g.snapshots[len(g.snapshots)-1:]
- return
- }
- remained := g.snapshots[len(g.snapshots)-1]
- g.snapshots = g.snapshots[:0]
- g.snapshots = append(g.snapshots, remainingSnapshots...)
- g.snapshots = append(g.snapshots, remained)
+ g.deletableEpochs = remainingEpochs
}
func (g garbageCleaner) cleanParts() {
- panic("implement me")
+OUTER:
+ for partID := range g.knownPartFiles {
+ for _, partInSnapshot := range g.liveParts {
+ if _, ok := partInSnapshot[partID]; ok {
+ continue OUTER
+ }
+ }
+ err := g.parent.fileSystem.DeleteFile(partPath(g.parent.root, partID))
+ if err != nil {
+ continue
+ }
+ delete(g.knownPartFiles, partID)
+ }
}
diff --git a/banyand/measure/gc_test.go b/banyand/measure/gc_test.go
index 2d6d8901..e77aebdb 100644
--- a/banyand/measure/gc_test.go
+++ b/banyand/measure/gc_test.go
@@ -18,88 +18,54 @@
package measure
import (
- "path/filepath"
- "sort"
+ "fmt"
+ "reflect"
"testing"
- "github.com/stretchr/testify/assert"
- "github.com/stretchr/testify/require"
-
"github.com/apache/skywalking-banyandb/pkg/fs"
- "github.com/apache/skywalking-banyandb/pkg/logger"
- "github.com/apache/skywalking-banyandb/pkg/test"
)
-func TestGarbageCleanerCleanSnapshot(t *testing.T) {
+func TestDeletableEpochs(t *testing.T) {
tests := []struct {
name string
- snapshotsOnDisk []uint64
- snapshotsInGC []uint64
- wantOnDisk []uint64
- wantInGC []uint64
+ n int
+ knownEpochs []uint64
+ deletableEpochs []uint64
}{
{
- name: "Test with no snapshots on disk and no snapshots in GC",
- },
- {
- name: "Test with some snapshots on disk and no snapshots in GC",
- snapshotsInGC: nil,
- snapshotsOnDisk: []uint64{1, 2, 3},
- wantInGC: nil,
- wantOnDisk: []uint64{1, 2, 3},
- },
- {
- name: "Test with no snapshots on disk and some snapshots in GC",
- snapshotsOnDisk: nil,
- snapshotsInGC: []uint64{1, 2, 3},
- wantOnDisk: nil,
- // gc won't fix the miss match between inmemory and disk
- wantInGC: []uint64{3},
+ name: "empty",
+ n: 1,
+ knownEpochs: nil,
+ deletableEpochs: nil,
},
{
- name: "Test with some snapshots on disk and some snapshots in GC",
- snapshotsOnDisk: []uint64{1, 2, 3, 4, 5},
- snapshotsInGC: []uint64{1, 3, 5},
- wantOnDisk: []uint64{2, 4, 5},
- wantInGC: []uint64{5},
+ name: "one",
+ n: 1,
+ knownEpochs: []uint64{1},
+ deletableEpochs: nil,
},
{
- name: "Test with unsorted",
- snapshotsOnDisk: []uint64{1, 3, 2, 5, 4},
- snapshotsInGC: []uint64{5, 1, 3},
- wantOnDisk: []uint64{2, 4, 5},
- wantInGC: []uint64{5},
+ name: "many",
+ n: 1,
+ knownEpochs: []uint64{1, 2, 3, 4},
+ deletableEpochs: []uint64{1, 2, 3},
},
}
- for _, tt := range tests {
- t.Run(tt.name, func(t *testing.T) {
- root, defFn := test.Space(require.New(t))
- defer defFn()
- parent := &tsTable{
- root: root,
- l: logger.GetLogger("test"),
+
+ 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(&tsTable{
fileSystem: fs.NewLocalFileSystem(),
+ })
+
+ for _, epoch := range test.knownEpochs {
+ gc.registerSnapshot(&snapshot{epoch: epoch})
}
- for i := range tt.snapshotsOnDisk {
- filePath := filepath.Join(parent.root, snapshotName(tt.snapshotsOnDisk[i]))
- file, err := parent.fileSystem.CreateFile(filePath, filePermission)
- require.NoError(t, err)
- require.NoError(t, file.Close())
- }
- g := &garbageCleaner{
- parent: parent,
- snapshots: tt.snapshotsInGC,
- }
- g.cleanSnapshots()
- var got []uint64
- for _, de := range parent.fileSystem.ReadDir(parent.root) {
- s, err := parseSnapshot(de.Name())
- require.NoError(t, err, "failed to parse snapshot name:%s", de.Name())
- got = append(got, s)
+ if !reflect.DeepEqual(gc.deletableEpochs, test.deletableEpochs) {
+ t.Errorf("expected deletable: %#v, got %#v", test.deletableEpochs, gc.deletableEpochs)
}
- sort.Slice(got, func(i, j int) bool { return got[i] < got[j] })
- assert.Equal(t, tt.wantOnDisk, got)
- assert.Equal(t, tt.wantInGC, g.snapshots)
})
}
}
diff --git a/banyand/measure/introducer.go b/banyand/measure/introducer.go
index 17a4edef..88821649 100644
--- a/banyand/measure/introducer.go
+++ b/banyand/measure/introducer.go
@@ -19,6 +19,7 @@ package measure
import (
"sync"
+ "sync/atomic"
"github.com/apache/skywalking-banyandb/pkg/watcher"
)
@@ -77,7 +78,36 @@ func releaseFlusherIntroduction(i *flusherIntroduction) {
flusherIntroductionPool.Put(i)
}
-func (tst *tsTable) introducerLoop(flushCh chan *flusherIntroduction, watcherCh watcher.Channel, epoch uint64) {
+type mergerIntroduction struct {
+ remaining map[uint64]struct{}
+ newPart *partWrapper
+ applied chan struct{}
+}
+
+func (i *mergerIntroduction) reset() {
+ for k := range i.remaining {
+ delete(i.remaining, k)
+ }
+ i.newPart = nil
+ i.applied = nil
+}
+
+var mergerIntroductionPool = sync.Pool{}
+
+func generateMergerIntroduction() *mergerIntroduction {
+ v := mergerIntroductionPool.Get()
+ if v == nil {
+ return &mergerIntroduction{}
+ }
+ return v.(*mergerIntroduction)
+}
+
+func releaseMergerIntroduction(i *mergerIntroduction) {
+ i.reset()
+ mergerIntroductionPool.Put(i)
+}
+
+func (tst *tsTable) introducerLoop(flushCh chan *flusherIntroduction, mergeCh chan *mergerIntroduction, watcherCh watcher.Channel, epoch uint64) {
var introducerWatchers watcher.Epochs
defer tst.loopCloser.Done()
for {
@@ -90,6 +120,9 @@ func (tst *tsTable) introducerLoop(flushCh chan *flusherIntroduction, watcherCh
case next := <-flushCh:
tst.introduceFlushed(next, epoch)
epoch++
+ case next := <-mergeCh:
+ tst.introduceMerged(next, epoch)
+ epoch++
case epochWatcher := <-watcherCh:
introducerWatchers.Add(epochWatcher)
}
@@ -108,8 +141,9 @@ func (tst *tsTable) introduceMemPart(nextIntroduction *introduction, epoch uint6
next := nextIntroduction.memPart
nextSnp := cur.copyAllTo(epoch)
- next.p.partMetadata.ID = epoch
+ next.p.partMetadata.ID = atomic.AddUint64(&tst.curPartID, 1)
nextSnp.parts = append(nextSnp.parts, next)
+ nextSnp.creator = snapshotCreatorMemPart
tst.replaceSnapshot(&nextSnp)
if nextIntroduction.applied != nil {
close(nextIntroduction.applied)
@@ -123,12 +157,27 @@ func (tst *tsTable) introduceFlushed(nextIntroduction *flusherIntroduction, epoc
}
defer cur.decRef()
nextSnp := cur.merge(epoch, nextIntroduction.flushed)
+ nextSnp.creator = snapshotCreatorFlusher
tst.replaceSnapshot(&nextSnp)
if nextIntroduction.applied != nil {
close(nextIntroduction.applied)
}
}
+func (tst *tsTable) introduceMerged(nextIntroduction *mergerIntroduction, epoch uint64) {
+ cur := tst.currentSnapshot()
+ if cur == nil {
+ tst.l.Panic().Msg("current snapshot is nil")
+ }
+ defer cur.decRef()
+ nextSnp := cur.remove(epoch, nextIntroduction.remaining)
+ nextSnp.parts = append(nextSnp.parts, nextIntroduction.newPart)
+ nextSnp.creator = snapshotCreatorMerger
+ if nextIntroduction.applied != nil {
+ close(nextIntroduction.applied)
+ }
+}
+
func (tst *tsTable) replaceSnapshot(next *snapshot) {
tst.Lock()
defer tst.Unlock()
diff --git a/banyand/measure/merger.go b/banyand/measure/merger.go
new file mode 100644
index 00000000..24ee3c32
--- /dev/null
+++ b/banyand/measure/merger.go
@@ -0,0 +1,308 @@
+// 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 measure
+
+import (
+ "errors"
+ "fmt"
+ "sync/atomic"
+
+ "github.com/apache/skywalking-banyandb/pkg/fs"
+ "github.com/apache/skywalking-banyandb/pkg/watcher"
+)
+
+func (tst *tsTable) mergeLoop(merges chan *mergerIntroduction, flusherNotifier watcher.Channel) {
+ defer tst.loopCloser.Done()
+
+ var epoch uint64
+
+ ew := flusherNotifier.Add(0, tst.loopCloser.CloseNotify())
+ if ew == nil {
+ return
+ }
+
+ for {
+ select {
+ case <-tst.loopCloser.CloseNotify():
+ return
+ case <-ew.Watch():
+ curSnapshot := tst.currentSnapshot()
+ if curSnapshot.epoch != epoch {
+ if err := tst.mergeSnapshot(curSnapshot, epoch, merges); err != nil {
+ if errors.Is(err, errClosed) {
+ curSnapshot.decRef()
+ return
+ }
+ tst.l.Logger.Warn().Err(err).Msgf("cannot merge snapshot: %d", curSnapshot.epoch)
+ curSnapshot.decRef()
+ continue
+ }
+ epoch = curSnapshot.epoch
+ }
+ curSnapshot.decRef()
+ ew = flusherNotifier.Add(epoch, tst.loopCloser.CloseNotify())
+ if ew == nil {
+ break
+ }
+ }
+ }
+}
+
+func (tst *tsTable) mergeSnapshot(curSnapshot *snapshot, epoch uint64, merges chan *mergerIntroduction) error {
+ maxFanOut := tst.freeDiskSpace(tst.root) / 2
+ partsToMerge, remainingParts := tst.getPartsToMerge(curSnapshot, maxFanOut)
+ if len(partsToMerge) < 2 {
+ return nil
+ }
+ if _, err := tst.mergePartsThenSendIntroduction(partsToMerge, remainingParts, merges, tst.loopCloser.CloseNotify()); err != nil {
+ return err
+ }
+ return nil
+}
+
+func (tst *tsTable) mergePartsThenSendIntroduction(parts []*partWrapper, remaining map[uint64]struct{}, merges chan *mergerIntroduction, closeCh <-chan struct{}) (*partWrapper, error) {
+ reservedSpace := tst.reserveSpace(parts)
+ defer releaseDiskSpace(reservedSpace)
+ newPart, err := mergeParts(tst.fileSystem, closeCh, parts, atomic.AddUint64(&tst.curPartID, 1), tst.root)
+ if err != nil {
+ return nil, err
+ }
+
+ mi := generateMergerIntroduction()
+ defer releaseMergerIntroduction(mi)
+ mi.newPart = newPart
+ mi.remaining = remaining
+ mi.applied = make(chan struct{})
+ select {
+ case merges <- mi:
+ case <-tst.loopCloser.CloseNotify():
+ return newPart, errClosed
+ }
+ select {
+ case <-mi.applied:
+ case <-tst.loopCloser.CloseNotify():
+ return newPart, errClosed
+ }
+ return newPart, nil
+}
+
+func (tst *tsTable) freeDiskSpace(path string) uint64 {
+ free := tst.fileSystem.MustGetFreeSpace(path)
+ reserved := atomic.LoadUint64(&reservedDiskSpace)
+ if free < reserved {
+ return 0
+ }
+ return free - reserved
+}
+
+func (tst *tsTable) tryReserveDiskSpace(n uint64) bool {
+ available := tst.fileSystem.MustGetFreeSpace(tst.root)
+ reserved := reserveDiskSpace(n)
+ if available > reserved {
+ return true
+ }
+ releaseDiskSpace(n)
+ return false
+}
+
+func reserveDiskSpace(n uint64) uint64 {
+ return atomic.AddUint64(&reservedDiskSpace, n)
+}
+
+func releaseDiskSpace(n uint64) {
+ atomic.AddUint64(&reservedDiskSpace, ^(n - 1))
+}
+
+var reservedDiskSpace uint64
+
+func (tst *tsTable) getPartsToMerge(snapshot *snapshot, maxFanOut uint64) ([]*partWrapper, map[uint64]struct{}) {
+ var parts []*partWrapper
+ remaining := make(map[uint64]struct{})
+ for _, pw := range snapshot.parts {
+ if pw.p.partMetadata.TotalCount < 1 || pw.p.partMetadata.CompressedSizeBytes > maxFanOut {
+ remaining[pw.ID()] = struct{}{}
+ continue
+ }
+ parts = append(parts, pw)
+ }
+ // TODO: select proper parts to generate the lowest write amplification
+ return parts, remaining
+}
+
+func (tst *tsTable) reserveSpace(parts []*partWrapper) uint64 {
+ var needSize uint64
+ for i := range parts {
+ needSize = +parts[i].p.partMetadata.CompressedSizeBytes
+ }
+ if tst.tryReserveDiskSpace(needSize) {
+ return needSize
+ }
+ return 0
+}
+
+var errNoPartToMerge = fmt.Errorf("no part to merge")
+
+func 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)
+ pii := make([]*partMergeIter, 0, len(parts))
+ for i := range parts {
+ pmi := generatePartMergeIter()
+ pmi.mustInitFromPart(parts[i].p)
+ pii = append(pii, pmi)
+ }
+ br := new(blockReader)
+ br.init(pii)
+ bw := generateBlockWriter()
+ bw.mustInitForFilePart(fileSystem, dstPath)
+
+ pm, err := mergeBlocks(closeCh, bw, br)
+ releaseBlockWriter(bw)
+ for i := range pii {
+ releasePartMergeIter(pii[i])
+ }
+ if err != nil {
+ return nil, err
+ }
+ pm.mustWriteMetadata(fileSystem, dstPath)
+ fileSystem.SyncPath(dstPath)
+ p := mustOpenFilePart(partID, root, 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 := &blockPointer{}
+ var tmpBlock, tmpBlock2 *blockPointer
+ for br.nextBlock() {
+ select {
+ case <-closeCh:
+ return nil, errClosed
+ default:
+ }
+ b := br.block
+
+ if pendingBlockIsEmpty {
+ pendingBlock.copyFrom(b)
+ pendingBlockIsEmpty = false
+ continue
+ }
+
+ if pendingBlock.bm.seriesID != b.bm.seriesID ||
+ (pendingBlock.isFull() && pendingBlock.bm.timestamps.max <= b.bm.timestamps.min) {
+ bw.mustWriteBlock(pendingBlock.bm.seriesID, &pendingBlock.block)
+ pendingBlock.reset()
+ pendingBlock.copyFrom(b)
+ continue
+ }
+
+ if tmpBlock == nil {
+ tmpBlock = &blockPointer{}
+ }
+ tmpBlock.reset()
+ tmpBlock.bm.seriesID = b.bm.seriesID
+ mergeTwoBlocks(tmpBlock, pendingBlock, b)
+ if len(tmpBlock.timestamps) <= maxBlockLength && tmpBlock.uncompressedSizeBytes() <= maxUncompressedBlockSize {
+ if len(tmpBlock.timestamps) == 0 {
+ pendingBlockIsEmpty = true
+ }
+ pendingBlock, tmpBlock = tmpBlock, pendingBlock
+ continue
+ }
+
+ if len(tmpBlock.timestamps) <= maxBlockLength {
+ bw.mustWriteBlock(tmpBlock.bm.seriesID, &tmpBlock.block)
+ continue
+ }
+ tmpBlock.idx = maxBlockLength
+ pendingBlock.copyFrom(tmpBlock)
+ l := tmpBlock.idx
+ tmpBlock.idx = 0
+ if tmpBlock2 == nil {
+ tmpBlock2 = &blockPointer{}
+ }
+ tmpBlock2.reset()
+ tmpBlock2.append(tmpBlock, l)
+ bw.mustWriteBlock(tmpBlock.bm.seriesID, &tmpBlock2.block)
+ }
+ 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)
+ }
+ 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.timestamps) {
+ target.appendAll(ib2)
+ return true
+ }
+ return false
+ }
+
+ if left.bm.timestamps.max < right.bm.timestamps.min {
+ target.appendAll(left)
+ target.appendAll(right)
+ return
+ }
+ if right.bm.timestamps.max < left.bm.timestamps.min {
+ target.appendAll(right)
+ target.appendAll(left)
+ return
+ }
+ if appendIfEmpty(left, right) || appendIfEmpty(right, left) {
+ return
+ }
+
+ for {
+ i := left.idx
+ ts2 := right.timestamps[right.idx]
+ for i < len(left.timestamps) && left.timestamps[i] <= ts2 {
+ i++
+ }
+ if left.timestamps[i-1] == ts2 {
+ if left.partID >= right.partID {
+ target.append(left, i)
+ } else {
+ target.append(left, i-1) // skip left
+ target.append(right, right.idx+1)
+ }
+ left.idx = i
+ right.idx++ // skip right
+ if appendIfEmpty(right, left) {
+ return
+ }
+ } else {
+ target.append(left, i)
+ left.idx = i
+ }
+ if appendIfEmpty(left, right) {
+ return
+ }
+ left, right = right, left
+ }
+}
diff --git a/banyand/measure/merger_test.go b/banyand/measure/merger_test.go
new file mode 100644
index 00000000..c4eea86a
--- /dev/null
+++ b/banyand/measure/merger_test.go
@@ -0,0 +1,356 @@
+// 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 measure
+
+import (
+ "errors"
+ "reflect"
+ "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"
+)
+
+func Test_mergeTwoBlocks(t *testing.T) {
+ tests := []struct {
+ name string
+ left *blockPointer
+ right *blockPointer
+ want *blockPointer
+ }{
+ {
+ name: "Merge two empty blocks",
+ left: &blockPointer{},
+ right: &blockPointer{},
+ want: &blockPointer{},
+ },
+ {
+ name: "Merge left is non-empty right is empty",
+ left: &blockPointer{block: conventionalBlock},
+ right: &blockPointer{},
+ want: &blockPointer{block: conventionalBlock},
+ },
+ {
+ name: "Merge left is empty right is non-empty",
+ left: &blockPointer{},
+ right: &blockPointer{block: conventionalBlock},
+ want: &blockPointer{block: conventionalBlock},
+ },
+ {
+ name: "Merge two non-empty blocks without overlap",
+ left: &blockPointer{
+ block: block{
+ timestamps: []int64{1, 2},
+ tagFamilies: []columnFamily{
+ {
+ name: "arrTag",
+ columns: []column{
+ {
+ name: "strArrTag", valueType: pbv1.ValueTypeStrArr,
+ values: [][]byte{marshalStrArr([][]byte{[]byte("value1"), []byte("value2")}), marshalStrArr([][]byte{[]byte("value3"), []byte("value4")})},
+ },
+ },
+ },
+ },
+ field: columnFamily{
+ columns: []column{
+ {name: "strField", valueType: pbv1.ValueTypeStr, values: [][]byte{[]byte("field1"), []byte("field2")}},
+ },
+ },
+ },
+ },
+ right: &blockPointer{
+ block: block{
+ timestamps: []int64{3, 4},
+ tagFamilies: []columnFamily{
+ {
+ name: "arrTag",
+ columns: []column{
+ {
+ name: "strArrTag", valueType: pbv1.ValueTypeStrArr,
+ values: [][]byte{marshalStrArr([][]byte{[]byte("value5"), []byte("value6")}), marshalStrArr([][]byte{[]byte("value7"), []byte("value8")})},
+ },
+ },
+ },
+ },
+ field: columnFamily{
+ columns: []column{
+ {name: "strField", valueType: pbv1.ValueTypeStr, values: [][]byte{[]byte("field3"), []byte("field4")}},
+ },
+ },
+ },
+ },
+ want: &blockPointer{block: mergedBlock},
+ },
+ {
+ name: "Merge two non-empty blocks without duplicated timestamps",
+ left: &blockPointer{
+ block: block{
+ timestamps: []int64{1, 3},
+ tagFamilies: []columnFamily{
+ {
+ name: "arrTag",
+ columns: []column{
+ {
+ name: "strArrTag", valueType: pbv1.ValueTypeStrArr,
+ values: [][]byte{marshalStrArr([][]byte{[]byte("value1"), []byte("value2")}), marshalStrArr([][]byte{[]byte("value5"), []byte("value6")})},
+ },
+ },
+ },
+ },
+ field: columnFamily{
+ columns: []column{
+ {name: "strField", valueType: pbv1.ValueTypeStr, values: [][]byte{[]byte("field1"), []byte("field3")}},
+ },
+ },
+ },
+ },
+ right: &blockPointer{
+ block: block{
+ timestamps: []int64{2, 4},
+ tagFamilies: []columnFamily{
+ {
+ name: "arrTag",
+ columns: []column{
+ {
+ name: "strArrTag", valueType: pbv1.ValueTypeStrArr,
+ values: [][]byte{marshalStrArr([][]byte{[]byte("value3"), []byte("value4")}), marshalStrArr([][]byte{[]byte("value7"), []byte("value8")})},
+ },
+ },
+ },
+ },
+ field: columnFamily{
+ columns: []column{
+ {name: "strField", valueType: pbv1.ValueTypeStr, values: [][]byte{[]byte("field2"), []byte("field4")}},
+ },
+ },
+ },
+ },
+ want: &blockPointer{block: mergedBlock},
+ },
+ {
+ name: "Merge two non-empty blocks with duplicated timestamps",
+ left: &blockPointer{
+ partID: 1, // the less partID will be skipped
+ block: block{
+ timestamps: []int64{1, 2, 3},
+ tagFamilies: []columnFamily{
+ {
+ name: "arrTag",
+ columns: []column{
+ {
+ name: "strArrTag", valueType: pbv1.ValueTypeStrArr,
+ values: [][]byte{
+ marshalStrArr([][]byte{[]byte("value1"), []byte("value2")}),
+ marshalStrArr([][]byte{[]byte("duplicated1")}),
+ marshalStrArr([][]byte{[]byte("duplicated2")}),
+ },
+ },
+ },
+ },
+ },
+ field: columnFamily{
+ columns: []column{
+ {name: "strField", valueType: pbv1.ValueTypeStr, values: [][]byte{[]byte("field1"), []byte("duplicated1"), []byte("duplicated2")}},
+ },
+ },
+ },
+ },
+ right: &blockPointer{
+ partID: 2, // the greater partID will be appended
+ block: block{
+ timestamps: []int64{2, 3, 4},
+ tagFamilies: []columnFamily{
+ {
+ name: "arrTag",
+ columns: []column{
+ {
+ name: "strArrTag", valueType: pbv1.ValueTypeStrArr,
+ values: [][]byte{
+ marshalStrArr([][]byte{[]byte("value3"), []byte("value4")}), marshalStrArr([][]byte{[]byte("value5"), []byte("value6")}),
+ marshalStrArr([][]byte{[]byte("value7"), []byte("value8")}),
+ },
+ },
+ },
+ },
+ },
+ field: columnFamily{
+ columns: []column{
+ {name: "strField", valueType: pbv1.ValueTypeStr, values: [][]byte{[]byte("field2"), []byte("field3"), []byte("field4")}},
+ },
+ },
+ },
+ },
+ want: &blockPointer{block: mergedBlock, partID: 2},
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ target := &blockPointer{}
+ mergeTwoBlocks(target, tt.left, tt.right)
+ if !reflect.DeepEqual(target, tt.want) {
+ t.Errorf("mergeTwoBlocks() = %v, want %v", target, tt.want)
+ }
+ })
+ }
+}
+
+var mergedBlock = block{
+ timestamps: []int64{1, 2, 3, 4},
+ tagFamilies: []columnFamily{
+ {
+ name: "arrTag",
+ columns: []column{
+ {
+ name: "strArrTag", valueType: pbv1.ValueTypeStrArr,
+ values: [][]byte{
+ marshalStrArr([][]byte{[]byte("value1"), []byte("value2")}), marshalStrArr([][]byte{[]byte("value3"), []byte("value4")}),
+ marshalStrArr([][]byte{[]byte("value5"), []byte("value6")}), marshalStrArr([][]byte{[]byte("value7"), []byte("value8")}),
+ },
+ },
+ },
+ },
+ },
+ field: columnFamily{
+ columns: []column{
+ {name: "strField", valueType: pbv1.ValueTypeStr, values: [][]byte{[]byte("field1"), []byte("field2"), []byte("field3"), []byte("field4")}},
+ },
+ },
+}
+
+func Test_mergeParts(t *testing.T) {
+ tests := []struct {
+ wantErr error
+ name string
+ dpsList []*dataPoints
+ want []blockMetadata
+ }{
+ {
+ name: "Test with no data points",
+ dpsList: []*dataPoints{},
+ wantErr: errNoPartToMerge,
+ },
+ {
+ name: "Test with single part",
+ dpsList: []*dataPoints{dpsTS1},
+ want: []blockMetadata{
+ {seriesID: 1, count: 1, uncompressedSizeBytes: 1676},
+ {seriesID: 2, count: 1, uncompressedSizeBytes: 55},
+ {seriesID: 3, count: 1, uncompressedSizeBytes: 24},
+ },
+ },
+ {
+ name: "Test with multiple parts with different ts",
+ dpsList: []*dataPoints{dpsTS1, dpsTS2},
+ want: []blockMetadata{
+ {seriesID: 1, count: 2, uncompressedSizeBytes: 3352},
+ {seriesID: 2, count: 2, uncompressedSizeBytes: 110},
+ {seriesID: 3, count: 2, uncompressedSizeBytes: 48},
+ },
+ },
+ {
+ name: "Test with multiple parts with same ts",
+ dpsList: []*dataPoints{dpsTS1, dpsTS1},
+ want: []blockMetadata{
+ {seriesID: 1, count: 1, uncompressedSizeBytes: 1676},
+ {seriesID: 2, count: 1, uncompressedSizeBytes: 55},
+ {seriesID: 3, count: 1, uncompressedSizeBytes: 24},
+ },
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ verify := func(t *testing.T, pp []*partWrapper, fileSystem fs.FileSystem, root string, partID uint64) {
+ closeCh := make(chan struct{})
+ defer close(closeCh)
+ p, err := mergeParts(fileSystem, closeCh, pp, partID, root)
+ if tt.wantErr != nil {
+ if !errors.Is(err, tt.wantErr) {
+ t.Fatalf("Unexpected error: got %v, want %v", err, tt.wantErr)
+ }
+ return
+ }
+ defer p.decRef()
+ pmi := &partMergeIter{}
+ pmi.mustInitFromPart(p.p)
+ reader := &blockReader{}
+ reader.init([]*partMergeIter{pmi})
+ var got []blockMetadata
+ for reader.nextBlock() {
+ got = append(got, reader.block.bm)
+ }
+ require.NoError(t, reader.error())
+
+ if diff := cmp.Diff(got, tt.want,
+ cmpopts.IgnoreFields(blockMetadata{}, "timestamps"),
+ cmpopts.IgnoreFields(blockMetadata{}, "field"),
+ cmpopts.IgnoreFields(blockMetadata{}, "tagFamilies"),
+ cmp.AllowUnexported(blockMetadata{}),
+ ); diff != "" {
+ t.Errorf("Unexpected blockMetadata (-got +want):\n%s", diff)
+ }
+ }
+
+ t.Run("memory parts", func(t *testing.T) {
+ var pp []*partWrapper
+ tmpPath, defFn := test.Space(require.New(t))
+ defer func() {
+ for _, pw := range pp {
+ pw.decRef()
+ }
+ defFn()
+ }()
+ for _, dps := range tt.dpsList {
+ mp := generateMemPart()
+ mp.mustInitFromDataPoints(dps)
+ pp = append(pp, newPartWrapper(mp, openMemPart(mp)))
+ }
+ verify(t, pp, fs.NewLocalFileSystem(), tmpPath, 1)
+ })
+
+ t.Run("file parts", func(t *testing.T) {
+ var fpp []*partWrapper
+ tmpPath, defFn := test.Space(require.New(t))
+ defer func() {
+ for _, pw := range fpp {
+ pw.decRef()
+ }
+ defFn()
+ }()
+ fileSystem := fs.NewLocalFileSystem()
+ for i, dps := range tt.dpsList {
+ mp := generateMemPart()
+ mp.mustInitFromDataPoints(dps)
+ mp.mustFlush(fileSystem, partPath(tmpPath, uint64(i)))
+ filePW := newPartWrapper(nil, mustOpenFilePart(uint64(i), tmpPath, fileSystem))
+ filePW.p.partMetadata.ID = uint64(i)
+ fpp = append(fpp, filePW)
+ releaseMemPart(mp)
+ }
+ verify(t, fpp, fileSystem, tmpPath, uint64(len(tt.dpsList)))
+ })
+ })
+ }
+}
diff --git a/banyand/measure/part.go b/banyand/measure/part.go
index 81fcce0e..6545220b 100644
--- a/banyand/measure/part.go
+++ b/banyand/measure/part.go
@@ -221,15 +221,13 @@ func releaseMemPart(mp *memPart) {
var memPartPool sync.Pool
type partWrapper struct {
- fileSystem fs.FileSystem
- mp *memPart
- p *part
- ref int32
- mustBeDeleted uint32
+ mp *memPart
+ p *part
+ ref int32
}
-func newPartWrapper(mp *memPart, p *part, fileSystem fs.FileSystem) *partWrapper {
- return &partWrapper{mp: mp, p: p, fileSystem: fileSystem, ref: 1}
+func newPartWrapper(mp *memPart, p *part) *partWrapper {
+ return &partWrapper{mp: mp, p: p, ref: 1}
}
func (pw *partWrapper) incRef() {
@@ -248,20 +246,18 @@ func (pw *partWrapper) decRef() {
return
}
pw.p.close()
- if atomic.LoadUint32(&pw.mustBeDeleted) == 0 {
- return
- }
- pw.fileSystem.MustRMAll(pw.p.path)
}
func (pw *partWrapper) ID() uint64 {
return pw.p.partMetadata.ID
}
-func mustOpenFilePart(partPath string, fileSystem fs.FileSystem) *part {
+func mustOpenFilePart(id uint64, root string, fileSystem fs.FileSystem) *part {
var p part
+ partPath := partPath(root, id)
p.path = partPath
p.partMetadata.mustReadMetadata(fileSystem, partPath)
+ p.partMetadata.ID = id
metaPath := path.Join(partPath, metaFilename)
pr := mustOpenReader(metaPath, fileSystem)
diff --git a/banyand/measure/part_iter.go b/banyand/measure/part_iter.go
index 14b4c8c3..43fdd5df 100644
--- a/banyand/measure/part_iter.go
+++ b/banyand/measure/part_iter.go
@@ -22,10 +22,12 @@ import (
"fmt"
"io"
"sort"
+ "sync"
"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/logger"
)
@@ -233,3 +235,126 @@ func (pi *partIter) findBlock() bool {
pi.bms = nil
return false
}
+
+type partMergeIter struct {
+ err error
+ p *part
+ primaryMetadataIdx int
+
+ compressedPrimaryBuf []byte
+ primaryBuf []byte
+ block blockPointer
+ columnValuesDecoder encoding.BytesBlockDecoder
+}
+
+func (pmi *partMergeIter) reset() {
+ pmi.err = nil
+ pmi.p = nil
+ pmi.primaryMetadataIdx = 0
+ pmi.primaryBuf = pmi.primaryBuf[:0]
+ pmi.compressedPrimaryBuf = pmi.compressedPrimaryBuf[:0]
+ pmi.block.reset()
+ pmi.columnValuesDecoder.Reset()
+}
+
+func (pmi *partMergeIter) mustInitFromPart(p *part) {
+ pmi.reset()
+ pmi.p = p
+}
+
+func (pmi *partMergeIter) error() error {
+ if pmi.err == nil || pmi.err == io.EOF {
+ return nil
+ }
+ return pmi.err
+}
+
+func (pmi *partMergeIter) nextBlock() 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.loadBlock(); err != nil {
+ pmi.err = err
+ return false
+ }
+ return true
+}
+
+func (pmi *partMergeIter) loadPrimaryBuf() error {
+ if pmi.primaryMetadataIdx >= len(pmi.p.primaryBlockMetadata) {
+ return io.EOF
+ }
+ pm := pmi.p.primaryBlockMetadata[pmi.primaryMetadataIdx]
+ pmi.compressedPrimaryBuf = bytes.ResizeOver(pmi.compressedPrimaryBuf, int(pm.size))
+ fs.MustReadData(pmi.p.primary, int64(pm.offset), 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) loadBlock() error {
+ pmi.block.reset()
+ var err error
+ pmi.primaryBuf, err = pmi.block.bm.unmarshal(pmi.primaryBuf)
+ if err != nil {
+ pm := pmi.p.primaryBlockMetadata[pmi.primaryMetadataIdx-1]
+ return fmt.Errorf("can't read block metadata from primary at %d: %w", pm.offset, err)
+ }
+
+ pmi.block.block.mustReadFrom(&pmi.columnValuesDecoder, pmi.p, pmi.block.bm, false)
+ pmi.block.partID = pmi.p.partMetadata.ID
+ return nil
+}
+
+func generatePartMergeIter() *partMergeIter {
+ v := pmiPool.Get()
+ if v == nil {
+ return &partMergeIter{}
+ }
+ return v.(*partMergeIter)
+}
+
+func releasePartMergeIter(pmi *partMergeIter) {
+ pmi.reset()
+ pmiPool.Put(pmi)
+}
+
+var pmiPool sync.Pool
+
+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
+}
diff --git a/banyand/measure/part_iter_test.go b/banyand/measure/part_iter_test.go
index 4aa6d7a9..ae25f9a4 100644
--- a/banyand/measure/part_iter_test.go
+++ b/banyand/measure/part_iter_test.go
@@ -150,7 +150,69 @@ func Test_partIter_nextBlock(t *testing.T) {
partPath := partPath(tmpDir, epoch)
fileSystem := fs.NewLocalFileSystem()
mp.mustFlush(fileSystem, partPath)
- p = mustOpenFilePart(partPath, fileSystem)
+ p = mustOpenFilePart(epoch, tmpDir, fileSystem)
+ verifyPart(p)
+ })
+ }
+}
+
+func Test_partMergeIter_nextBlock(t *testing.T) {
+ tests := []struct {
+ wantErr error
+ name string
+ dps *dataPoints
+ want []blockMetadata
+ }{
+ {
+ name: "Test with all data",
+ dps: dps,
+ want: []blockMetadata{
+ {seriesID: 1, count: 2}, {seriesID: 2, count: 2}, {seriesID: 3, count: 2},
+ },
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ verifyPart := func(p *part) {
+ defer p.close()
+ pi := partMergeIter{}
+ pi.mustInitFromPart(p)
+ var got []blockMetadata
+ for pi.nextBlock() {
+ got = append(got, pi.block.bm)
+ require.Nil(t, pi.block.bm.tagProjection)
+ require.Equal(t, len(pi.block.bm.tagFamilies), len(pi.block.tagFamilies))
+ require.Equal(t, len(pi.block.bm.field.columnMetadata), len(pi.block.field.columns))
+ }
+
+ if !errors.Is(pi.error(), tt.wantErr) {
+ t.Errorf("Unexpected error: got %v, want %v", pi.err, tt.wantErr)
+ }
+
+ if diff := cmp.Diff(got, tt.want,
+ cmpopts.IgnoreFields(blockMetadata{}, "uncompressedSizeBytes"),
+ cmpopts.IgnoreFields(blockMetadata{}, "timestamps"),
+ cmpopts.IgnoreFields(blockMetadata{}, "field"),
+ cmpopts.IgnoreFields(blockMetadata{}, "tagFamilies"),
+ cmp.AllowUnexported(blockMetadata{}),
+ ); diff != "" {
+ t.Errorf("Unexpected blockMetadata (-got +want):\n%s", diff)
+ }
+ }
+ mp := generateMemPart()
+ releaseMemPart(mp)
+ mp.mustInitFromDataPoints(tt.dps)
+
+ p := openMemPart(mp)
+ verifyPart(p)
+ tmpDir, defFn := test.Space(require.New(t))
+ defer defFn()
+ epoch := uint64(1)
+ partPath := partPath(tmpDir, epoch)
+ fileSystem := fs.NewLocalFileSystem()
+ mp.mustFlush(fileSystem, partPath)
+ p = mustOpenFilePart(epoch, tmpDir, fileSystem)
verifyPart(p)
})
}
diff --git a/banyand/measure/part_test.go b/banyand/measure/part_test.go
index 5473fd76..1044a46c 100644
--- a/banyand/measure/part_test.go
+++ b/banyand/measure/part_test.go
@@ -103,7 +103,7 @@ func TestMustInitFromDataPoints(t *testing.T) {
fileSystem := fs.NewLocalFileSystem()
mp.mustFlush(fileSystem, path)
- p := mustOpenFilePart(path, fileSystem)
+ p := mustOpenFilePart(epoch, tmpPath, fileSystem)
defer p.close()
assert.Equal(t, tt.want.BlocksCount, p.partMetadata.BlocksCount)
assert.Equal(t, tt.want.MinTimestamp, p.partMetadata.MinTimestamp)
diff --git a/banyand/measure/query.go b/banyand/measure/query.go
index d0b17d7f..54024a94 100644
--- a/banyand/measure/query.go
+++ b/banyand/measure/query.go
@@ -349,7 +349,7 @@ func (qr *queryResult) Pull() *pbv1.Result {
if len(qr.data) == 1 {
r := &pbv1.Result{}
bc := qr.data[0]
- bc.copyAllTo(r)
+ bc.copyAllTo(r, qr.orderByTimestampDesc())
qr.data = qr.data[:0]
return r
}
diff --git a/banyand/measure/query_test.go b/banyand/measure/query_test.go
index 4226dfc7..4cf25b07 100644
--- a/banyand/measure/query_test.go
+++ b/banyand/measure/query_test.go
@@ -24,7 +24,6 @@ import (
"time"
"github.com/google/go-cmp/cmp"
- "github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"google.golang.org/protobuf/testing/protocmp"
@@ -371,7 +370,7 @@ func TestQueryResult(t *testing.T) {
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
- verify := func(tst *tsTable) uint64 {
+ verify := func(t *testing.T, tst *tsTable) {
defer tst.Close()
queryOpts := queryOptions{
minTimestamp: tt.minTimestamp,
@@ -380,8 +379,7 @@ func TestQueryResult(t *testing.T) {
s := tst.currentSnapshot()
require.NotNil(t, s)
defer s.decRef()
- pp, n := s.getParts(nil, queryOpts)
- require.Equal(t, len(tt.dpsList), n, "parts: %+v", pp)
+ pp, _ := s.getParts(nil, queryOpts)
sids := make([]common.SeriesID, len(tt.sids))
copy(sids, tt.sids)
sort.Slice(sids, func(i, j int) bool {
@@ -430,22 +428,27 @@ func TestQueryResult(t *testing.T) {
protocmp.IgnoreUnknown(), protocmp.Transform()); diff != "" {
t.Errorf("Unexpected []pbv1.Result (-got +want):\n%s", diff)
}
- return s.epoch
}
t.Run("memory snapshot", func(t *testing.T) {
+ tmpPath, defFn := test.Space(require.New(t))
+ defer defFn()
tst := &tsTable{
loopCloser: run.NewCloser(2),
introductions: make(chan *introduction),
+ fileSystem: fs.NewLocalFileSystem(),
+ root: tmpPath,
}
+ tst.gc.init(tst)
flushCh := make(chan *flusherIntroduction)
+ mergeCh := make(chan *mergerIntroduction)
introducerWatcher := make(watcher.Channel, 1)
- go tst.introducerLoop(flushCh, introducerWatcher, 1)
+ go tst.introducerLoop(flushCh, mergeCh, introducerWatcher, 1)
for _, dps := range tt.dpsList {
tst.mustAddDataPoints(dps)
time.Sleep(100 * time.Millisecond)
}
- _ = verify(tst)
+ verify(t, tst)
})
t.Run("file snapshot", func(t *testing.T) {
@@ -459,12 +462,29 @@ func TestQueryResult(t *testing.T) {
tst.mustAddDataPoints(dps)
time.Sleep(100 * time.Millisecond)
}
- epoch := verify(tst)
+ // wait until the introducer is done
+ if len(tt.dpsList) > 0 {
+ for {
+ snp := tst.currentSnapshot()
+ if snp == nil {
+ time.Sleep(100 * time.Millisecond)
+ continue
+ }
+ if len(snp.parts) == len(tt.dpsList) {
+ snp.decRef()
+ tst.Close()
+ break
+ }
+ snp.decRef()
+ time.Sleep(100 * time.Millisecond)
+ }
+ }
// reopen the table
tst, err = newTSTable(fileSystem, tmpPath, common.Position{}, logger.GetLogger("test"), timestamp.TimeRange{})
require.NoError(t, err)
- assert.Equal(t, epoch, verify(tst))
+
+ verify(t, tst)
})
})
}
diff --git a/banyand/measure/snapshot.go b/banyand/measure/snapshot.go
index ddc995dc..3c5f2943 100644
--- a/banyand/measure/snapshot.go
+++ b/banyand/measure/snapshot.go
@@ -35,9 +35,19 @@ func (tst *tsTable) currentSnapshot() *snapshot {
return s
}
+type snapshotCreator rune
+
+const (
+ snapshotCreatorMemPart = iota
+ snapshotCreatorFlusher
+ snapshotCreatorMerger
+ snapshotCreatorMergedFlusher
+)
+
type snapshot struct {
- parts []*partWrapper
- epoch uint64
+ parts []*partWrapper
+ epoch uint64
+ creator snapshotCreator
ref int32
}
@@ -96,6 +106,19 @@ func (s *snapshot) merge(nextEpoch uint64, nextParts map[uint64]*partWrapper) sn
return result
}
+func (s *snapshot) remove(nextEpoch uint64, remaining map[uint64]struct{}) snapshot {
+ var result snapshot
+ result.epoch = nextEpoch
+ result.ref = 1
+ for i := 0; i < len(s.parts); i++ {
+ if _, ok := remaining[s.parts[i].ID()]; ok {
+ s.parts[i].incRef()
+ result.parts = append(result.parts, s.parts[i])
+ }
+ }
+ return result
+}
+
func snapshotName(snapshot uint64) string {
return fmt.Sprintf("%016x%s", snapshot, snapshotSuffix)
}
diff --git a/banyand/measure/snapshot_test.go b/banyand/measure/snapshot_test.go
index f571d4c9..018557f2 100644
--- a/banyand/measure/snapshot_test.go
+++ b/banyand/measure/snapshot_test.go
@@ -21,6 +21,8 @@ import (
"testing"
"github.com/stretchr/testify/assert"
+
+ "github.com/apache/skywalking-banyandb/pkg/bytes"
)
func TestSnapshotGetParts(t *testing.T) {
@@ -285,3 +287,143 @@ func TestSnapshotMerge(t *testing.T) {
})
}
}
+
+func TestSnapshotRemove(t *testing.T) {
+ tests := []struct {
+ name string
+ snapshot *snapshot
+ nextEpoch uint64
+ remainingParts map[uint64]struct{}
+ expected snapshot
+ closePrev bool
+ }{
+ {
+ name: "Test with empty snapshot and no parts to remove",
+ snapshot: &snapshot{
+ parts: []*partWrapper{},
+ },
+ nextEpoch: 1,
+ remainingParts: map[uint64]struct{}{},
+ expected: snapshot{
+ epoch: 1,
+ ref: 1,
+ parts: nil,
+ },
+ },
+ {
+ name: "Test with non-empty snapshot and no parts to remove",
+ snapshot: &snapshot{
+ parts: []*partWrapper{
+ {p: &part{partMetadata: partMetadata{ID: 1}}, ref: 1},
+ {p: &part{partMetadata: partMetadata{ID: 2}}, ref: 2},
+ },
+ },
+ nextEpoch: 1,
+ remainingParts: map[uint64]struct{}{
+ 1: {},
+ 2: {},
+ },
+ expected: snapshot{
+ epoch: 1,
+ ref: 1,
+ parts: []*partWrapper{
+ {p: &part{partMetadata: partMetadata{ID: 1}}, ref: 2},
+ {p: &part{partMetadata: partMetadata{ID: 2}}, ref: 3},
+ },
+ },
+ },
+ {
+ name: "Test with non-empty snapshot and some parts to remove",
+ snapshot: &snapshot{
+ parts: []*partWrapper{
+ {p: &part{partMetadata: partMetadata{ID: 1}}, ref: 1},
+ {p: &part{partMetadata: partMetadata{ID: 2}}, ref: 2},
+ },
+ },
+ nextEpoch: 1,
+ remainingParts: map[uint64]struct{}{
+ 2: {},
+ },
+ expected: snapshot{
+ epoch: 1,
+ ref: 1,
+ parts: []*partWrapper{
+ {p: &part{partMetadata: partMetadata{ID: 2}}, ref: 3},
+ },
+ },
+ },
+ {
+ name: "Test with empty snapshot, no parts to remove, and closePrev=true",
+ snapshot: &snapshot{
+ parts: []*partWrapper{},
+ },
+ nextEpoch: 1,
+ remainingParts: map[uint64]struct{}{},
+ expected: snapshot{
+ epoch: 1,
+ ref: 1,
+ parts: nil,
+ },
+ closePrev: true,
+ },
+ {
+ name: "Test with non-empty snapshot, no parts to remove, and closePrev=true",
+ snapshot: &snapshot{
+ parts: []*partWrapper{
+ {p: &part{partMetadata: partMetadata{ID: 1}}, ref: 1},
+ {p: &part{partMetadata: partMetadata{ID: 2}}, ref: 2},
+ },
+ },
+ nextEpoch: 1,
+ remainingParts: map[uint64]struct{}{
+ 1: {},
+ 2: {},
+ },
+ expected: snapshot{
+ epoch: 1,
+ ref: 1,
+ parts: []*partWrapper{
+ {p: &part{partMetadata: partMetadata{ID: 1}}, ref: 1},
+ {p: &part{partMetadata: partMetadata{ID: 2}}, ref: 2},
+ },
+ },
+ closePrev: true,
+ },
+ {
+ name: "Test with non-empty snapshot, some parts to remove, and closePrev=true",
+ snapshot: &snapshot{
+ parts: []*partWrapper{
+ {p: &part{
+ partMetadata: partMetadata{ID: 1},
+ timestamps: &bytes.Buffer{},
+ primary: &bytes.Buffer{},
+ fieldValues: &bytes.Buffer{},
+ }, ref: 1},
+ {p: &part{partMetadata: partMetadata{ID: 2}}, ref: 2},
+ },
+ },
+ nextEpoch: 1,
+ remainingParts: map[uint64]struct{}{
+ 2: {},
+ },
+ expected: snapshot{
+ epoch: 1,
+ ref: 1,
+ parts: []*partWrapper{
+ {p: &part{partMetadata: partMetadata{ID: 2}}, ref: 2},
+ },
+ },
+ closePrev: true,
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ result := tt.snapshot.remove(tt.nextEpoch, tt.remainingParts)
+ if tt.closePrev {
+ tt.snapshot.decRef()
+ }
+ assert.Equal(t, tt.expected, result)
+ })
+ }
+}
diff --git a/banyand/measure/tstable.go b/banyand/measure/tstable.go
index f3c445dc..24031221 100644
--- a/banyand/measure/tstable.go
+++ b/banyand/measure/tstable.go
@@ -44,11 +44,12 @@ const (
)
func newTSTable(fileSystem fs.FileSystem, rootPath string, _ common.Position, l *logger.Logger, _ timestamp.TimeRange) (*tsTable, error) {
- var tst tsTable
- tst.fileSystem = fileSystem
- tst.root = rootPath
- tst.l = l
- tst.gc.parent = &tst
+ tst := tsTable{
+ fileSystem: fileSystem,
+ root: rootPath,
+ l: l,
+ }
+ tst.gc.init(&tst)
ee := fileSystem.ReadDir(rootPath)
if len(ee) == 0 {
t := &tst
@@ -79,7 +80,6 @@ func newTSTable(fileSystem fs.FileSystem, rootPath string, _ common.Position, l
continue
}
loadedSnapshots = append(loadedSnapshots, snapshot)
- tst.gc.registerSnapshot(snapshot)
}
for i := range needToDelete {
if err := fileSystem.DeleteFile(filepath.Join(rootPath, needToDelete[i])); err != nil {
@@ -109,6 +109,7 @@ type tsTable struct {
loopCloser *run.Closer
root string
gc garbageCleaner
+ curPartID uint64
sync.RWMutex
}
@@ -117,21 +118,25 @@ func (tst *tsTable) loadSnapshot(epoch uint64, loadedParts []uint64) {
snp := snapshot{
epoch: epoch,
}
- for _, partName := range loadedParts {
+ for _, id := range loadedParts {
var find bool
for j := range parts {
- if partName == parts[j] {
+ if id == parts[j] {
find = true
break
}
}
if !find {
- tst.gc.submitParts(partName)
+ tst.gc.submitParts(id)
+ }
+ p := mustOpenFilePart(id, tst.root, tst.fileSystem)
+ p.partMetadata.ID = id
+ snp.parts = append(snp.parts, newPartWrapper(nil, p))
+ if tst.curPartID < id {
+ tst.curPartID = id
}
- p := mustOpenFilePart(partPath(tst.root, partName), tst.fileSystem)
- p.partMetadata.ID = partName
- snp.parts = append(snp.parts, newPartWrapper(nil, p, tst.fileSystem))
}
+ tst.gc.registerSnapshot(&snp)
tst.gc.clean()
if len(snp.parts) < 1 {
return
@@ -141,12 +146,15 @@ func (tst *tsTable) loadSnapshot(epoch uint64, loadedParts []uint64) {
}
func (tst *tsTable) startLoop(cur uint64) {
- tst.loopCloser = run.NewCloser(3)
+ tst.loopCloser = run.NewCloser(1 + 3)
tst.introductions = make(chan *introduction)
flushCh := make(chan *flusherIntroduction)
+ mergeCh := make(chan *mergerIntroduction)
introducerWatcher := make(watcher.Channel, 1)
- go tst.introducerLoop(flushCh, introducerWatcher, cur+1)
- go tst.flusherLoop(flushCh, introducerWatcher, cur)
+ flusherWatcher := make(watcher.Channel, 1)
+ go tst.introducerLoop(flushCh, mergeCh, introducerWatcher, cur+1)
+ go tst.flusherLoop(flushCh, mergeCh, introducerWatcher, flusherWatcher, cur)
+ go tst.mergeLoop(mergeCh, flusherWatcher)
}
func parseEpoch(epochStr string) (uint64, error) {
@@ -206,6 +214,7 @@ func (tst *tsTable) Close() error {
defer tst.RUnlock()
if tst.snapshot != nil {
tst.snapshot.decRef()
+ tst.snapshot = nil
}
return nil
}
@@ -222,7 +231,7 @@ func (tst *tsTable) mustAddDataPoints(dps *dataPoints) {
ind := generateIntroduction()
defer releaseIntroduction(ind)
ind.applied = make(chan struct{})
- ind.memPart = newPartWrapper(mp, p, tst.fileSystem)
+ ind.memPart = newPartWrapper(mp, p)
select {
case tst.introductions <- ind:
diff --git a/banyand/measure/tstable_test.go b/banyand/measure/tstable_test.go
index ca231e4f..a101aad5 100644
--- a/banyand/measure/tstable_test.go
+++ b/banyand/measure/tstable_test.go
@@ -99,8 +99,9 @@ func Test_tsTable_mustAddDataPoints(t *testing.T) {
introductions: make(chan *introduction),
}
flushCh := make(chan *flusherIntroduction)
+ mergeCh := make(chan *mergerIntroduction)
introducerWatcher := make(watcher.Channel, 1)
- go tst.introducerLoop(flushCh, introducerWatcher, 1)
+ go tst.introducerLoop(flushCh, mergeCh, introducerWatcher, 1)
defer tst.Close()
for _, dps := range tt.dpsList {
tst.mustAddDataPoints(dps)
@@ -126,7 +127,7 @@ func Test_tsTable_mustAddDataPoints(t *testing.T) {
}
func Test_tstIter(t *testing.T) {
- tests := []struct {
+ type testCtx struct {
wantErr error
name string
dpsList []*dataPoints
@@ -134,112 +135,169 @@ func Test_tstIter(t *testing.T) {
want []blockMetadata
minTimestamp int64
maxTimestamp int64
- }{
- {
- name: "Test with no data points",
- dpsList: []*dataPoints{},
- sids: []common.SeriesID{1, 2, 3},
- minTimestamp: 1,
- maxTimestamp: 1,
- },
- {
- name: "Test with single part",
- dpsList: []*dataPoints{dpsTS1},
- sids: []common.SeriesID{1, 2, 3},
- minTimestamp: 1,
- maxTimestamp: 1,
- want: []blockMetadata{
- {seriesID: 1, count: 1, uncompressedSizeBytes: 1676},
- {seriesID: 2, count: 1, uncompressedSizeBytes: 55},
- {seriesID: 3, count: 1, uncompressedSizeBytes: 24},
- },
- },
- {
- name: "Test with multiple parts with different ts",
- dpsList: []*dataPoints{dpsTS1, dpsTS2},
- sids: []common.SeriesID{1, 2, 3},
- minTimestamp: 1,
- maxTimestamp: 2,
- want: []blockMetadata{
- {seriesID: 1, count: 1, uncompressedSizeBytes: 1676},
- {seriesID: 1, count: 1, uncompressedSizeBytes: 1676},
- {seriesID: 2, count: 1, uncompressedSizeBytes: 55},
- {seriesID: 2, count: 1, uncompressedSizeBytes: 55},
- {seriesID: 3, count: 1, uncompressedSizeBytes: 24},
- {seriesID: 3, count: 1, uncompressedSizeBytes: 24},
- },
- },
- {
- name: "Test with multiple parts with same ts",
- dpsList: []*dataPoints{dpsTS1, dpsTS1},
- sids: []common.SeriesID{1, 2, 3},
- minTimestamp: 1,
- maxTimestamp: 2,
- want: []blockMetadata{
- {seriesID: 1, count: 1, uncompressedSizeBytes: 1676},
- {seriesID: 1, count: 1, uncompressedSizeBytes: 1676},
- {seriesID: 2, count: 1, uncompressedSizeBytes: 55},
- {seriesID: 2, count: 1, uncompressedSizeBytes: 55},
- {seriesID: 3, count: 1, uncompressedSizeBytes: 24},
- {seriesID: 3, count: 1, uncompressedSizeBytes: 24},
- },
- },
}
- for _, tt := range tests {
- t.Run(tt.name, func(t *testing.T) {
- verify := func(tst *tsTable) uint64 {
- defer tst.Close()
- s := tst.currentSnapshot()
- if s == nil {
- s = new(snapshot)
- }
- defer s.decRef()
- pp, n := s.getParts(nil, queryOptions{
- minTimestamp: tt.minTimestamp,
- maxTimestamp: tt.maxTimestamp,
- })
- require.Equal(t, len(s.parts), n)
- ti := &tstIter{}
- ti.init(pp, tt.sids, tt.minTimestamp, tt.maxTimestamp)
- var got []blockMetadata
- for ti.nextBlock() {
- if ti.piHeap[0].curBlock.seriesID == 0 {
- t.Errorf("Expected curBlock to be initialized, but it was nil")
- }
- got = append(got, ti.piHeap[0].curBlock)
- }
- if !errors.Is(ti.Error(), tt.wantErr) {
- t.Errorf("Unexpected error: got %v, want %v", ti.err, tt.wantErr)
- }
-
- if diff := cmp.Diff(got, tt.want,
- cmpopts.IgnoreFields(blockMetadata{}, "timestamps"),
- cmpopts.IgnoreFields(blockMetadata{}, "field"),
- cmpopts.IgnoreFields(blockMetadata{}, "tagFamilies"),
- cmp.AllowUnexported(blockMetadata{}),
- ); diff != "" {
- t.Errorf("Unexpected blockMetadata (-got +want):\n%s", diff)
- }
- return s.epoch
+ verify := func(t *testing.T, tt testCtx, tst *tsTable) uint64 {
+ defer tst.Close()
+ s := tst.currentSnapshot()
+ if s == nil {
+ s = new(snapshot)
+ }
+ defer s.decRef()
+ pp, n := s.getParts(nil, queryOptions{
+ minTimestamp: tt.minTimestamp,
+ maxTimestamp: tt.maxTimestamp,
+ })
+ require.Equal(t, len(s.parts), n)
+ ti := &tstIter{}
+ ti.init(pp, tt.sids, tt.minTimestamp, tt.maxTimestamp)
+ var got []blockMetadata
+ for ti.nextBlock() {
+ if ti.piHeap[0].curBlock.seriesID == 0 {
+ t.Errorf("Expected curBlock to be initialized, but it was nil")
}
+ got = append(got, ti.piHeap[0].curBlock)
+ }
+
+ if !errors.Is(ti.Error(), tt.wantErr) {
+ t.Errorf("Unexpected error: got %v, want %v", ti.err, tt.wantErr)
+ }
+
+ if diff := cmp.Diff(got, tt.want,
+ cmpopts.IgnoreFields(blockMetadata{}, "timestamps"),
+ cmpopts.IgnoreFields(blockMetadata{}, "field"),
+ cmpopts.IgnoreFields(blockMetadata{}, "tagFamilies"),
+ cmp.AllowUnexported(blockMetadata{}),
+ ); diff != "" {
+ t.Errorf("Unexpected blockMetadata (-got +want):\n%s", diff)
+ }
+ return s.epoch
+ }
- t.Run("memory snapshot", func(t *testing.T) {
+ t.Run("memory snapshot", func(t *testing.T) {
+ tests := []testCtx{
+ {
+ name: "Test with no data points",
+ dpsList: []*dataPoints{},
+ sids: []common.SeriesID{1, 2, 3},
+ minTimestamp: 1,
+ maxTimestamp: 1,
+ },
+ {
+ name: "Test with single part",
+ dpsList: []*dataPoints{dpsTS1},
+ sids: []common.SeriesID{1, 2, 3},
+ minTimestamp: 1,
+ maxTimestamp: 1,
+ want: []blockMetadata{
+ {seriesID: 1, count: 1, uncompressedSizeBytes: 1676},
+ {seriesID: 2, count: 1, uncompressedSizeBytes: 55},
+ {seriesID: 3, count: 1, uncompressedSizeBytes: 24},
+ },
+ },
+ {
+ name: "Test with multiple parts with different ts",
+ dpsList: []*dataPoints{dpsTS1, dpsTS2},
+ sids: []common.SeriesID{1, 2, 3},
+ minTimestamp: 1,
+ maxTimestamp: 2,
+ want: []blockMetadata{
+ {seriesID: 1, count: 1, uncompressedSizeBytes: 1676},
+ {seriesID: 1, count: 1, uncompressedSizeBytes: 1676},
+ {seriesID: 2, count: 1, uncompressedSizeBytes: 55},
+ {seriesID: 2, count: 1, uncompressedSizeBytes: 55},
+ {seriesID: 3, count: 1, uncompressedSizeBytes: 24},
+ {seriesID: 3, count: 1, uncompressedSizeBytes: 24},
+ },
+ },
+ {
+ name: "Test with multiple parts with same ts",
+ dpsList: []*dataPoints{dpsTS1, dpsTS1},
+ sids: []common.SeriesID{1, 2, 3},
+ minTimestamp: 1,
+ maxTimestamp: 2,
+ want: []blockMetadata{
+ {seriesID: 1, count: 1, uncompressedSizeBytes: 1676},
+ {seriesID: 1, count: 1, uncompressedSizeBytes: 1676},
+ {seriesID: 2, count: 1, uncompressedSizeBytes: 55},
+ {seriesID: 2, count: 1, uncompressedSizeBytes: 55},
+ {seriesID: 3, count: 1, uncompressedSizeBytes: 24},
+ {seriesID: 3, count: 1, uncompressedSizeBytes: 24},
+ },
+ },
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ tmpPath, defFn := test.Space(require.New(t))
+ defer defFn()
tst := &tsTable{
loopCloser: run.NewCloser(2),
introductions: make(chan *introduction),
+ fileSystem: fs.NewLocalFileSystem(),
+ root: tmpPath,
}
+ tst.gc.init(tst)
flushCh := make(chan *flusherIntroduction)
+ mergeCh := make(chan *mergerIntroduction)
introducerWatcher := make(watcher.Channel, 1)
- go tst.introducerLoop(flushCh, introducerWatcher, 1)
+ go tst.introducerLoop(flushCh, mergeCh, introducerWatcher, 1)
for _, dps := range tt.dpsList {
tst.mustAddDataPoints(dps)
time.Sleep(100 * time.Millisecond)
}
- verify(tst)
+ verify(t, tt, tst)
})
+ }
+ })
- t.Run("file snapshot", func(t *testing.T) {
+ t.Run("file snapshot", func(t *testing.T) {
+ tests := []testCtx{
+ {
+ name: "Test with no data points",
+ dpsList: []*dataPoints{},
+ sids: []common.SeriesID{1, 2, 3},
+ minTimestamp: 1,
+ maxTimestamp: 1,
+ },
+ {
+ name: "Test with single part",
+ dpsList: []*dataPoints{dpsTS1},
+ sids: []common.SeriesID{1, 2, 3},
+ minTimestamp: 1,
+ maxTimestamp: 1,
+ want: []blockMetadata{
+ {seriesID: 1, count: 1, uncompressedSizeBytes: 1676},
+ {seriesID: 2, count: 1, uncompressedSizeBytes: 55},
+ {seriesID: 3, count: 1, uncompressedSizeBytes: 24},
+ },
+ },
+ {
+ name: "Test with multiple parts with different ts, the block will be merged",
+ dpsList: []*dataPoints{dpsTS1, dpsTS2},
+ sids: []common.SeriesID{1, 2, 3},
+ minTimestamp: 1,
+ maxTimestamp: 2,
+ want: []blockMetadata{
+ {seriesID: 1, count: 2, uncompressedSizeBytes: 3352},
+ {seriesID: 2, count: 2, uncompressedSizeBytes: 110},
+ {seriesID: 3, count: 2, uncompressedSizeBytes: 48},
+ },
+ },
+ {
+ name: "Test with multiple parts with same ts, duplicated blocks will be merged",
+ dpsList: []*dataPoints{dpsTS1, dpsTS1},
+ sids: []common.SeriesID{1, 2, 3},
+ minTimestamp: 1,
+ maxTimestamp: 2,
+ want: []blockMetadata{
+ {seriesID: 1, count: 1, uncompressedSizeBytes: 1676},
+ {seriesID: 2, count: 1, uncompressedSizeBytes: 55},
+ {seriesID: 3, count: 1, uncompressedSizeBytes: 24},
+ },
+ },
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
tmpPath, defFn := test.Space(require.New(t))
fileSystem := fs.NewLocalFileSystem()
defer defFn()
@@ -250,15 +308,30 @@ func Test_tstIter(t *testing.T) {
tst.mustAddDataPoints(dps)
time.Sleep(100 * time.Millisecond)
}
- epoch := verify(tst)
-
+ // wait until the introducer is done
+ if len(tt.dpsList) > 0 {
+ for {
+ snp := tst.currentSnapshot()
+ if snp == nil {
+ time.Sleep(100 * time.Millisecond)
+ continue
+ }
+ if len(snp.parts) == len(tt.dpsList) {
+ snp.decRef()
+ tst.Close()
+ break
+ }
+ snp.decRef()
+ time.Sleep(100 * time.Millisecond)
+ }
+ }
// reopen the table
tst, err = newTSTable(fileSystem, tmpPath, common.Position{}, logger.GetLogger("test"), timestamp.TimeRange{})
require.NoError(t, err)
- assert.Equal(t, epoch, verify(tst))
+ verify(t, tt, tst)
})
- })
- }
+ }
+ })
}
var tagProjections = map[int][]pbv1.TagProjection{
diff --git a/pkg/fs/file_system.go b/pkg/fs/file_system.go
index da483846..1e18163a 100644
--- a/pkg/fs/file_system.go
+++ b/pkg/fs/file_system.go
@@ -108,6 +108,8 @@ type FileSystem interface {
MustRMAll(path string)
// SyncPath the directory of file.
SyncPath(path string)
+ // MustGetFreeSpace returns the free space of the file system.
+ MustGetFreeSpace(path string) uint64
}
// DirEntry is the interface that wraps the basic information about a file or directory.
@@ -119,6 +121,15 @@ type DirEntry interface {
IsDir() bool
}
+// MustCreateFile creates a new file with the specified name and permission.
+func MustCreateFile(fs FileSystem, path string, permission Mode) File {
+ f, err := fs.CreateFile(path, permission)
+ if err != nil {
+ logger.GetLogger().Panic().Err(err).Str("path", path).Msg("cannot create file")
+ }
+ return f
+}
+
// MustFlush flushes all data to one file and panics if it cannot write all data.
func MustFlush(fs FileSystem, buffer []byte, name string, permission Mode) {
n, err := fs.Write(buffer, name, permission)
diff --git a/pkg/fs/local_file_system.go b/pkg/fs/local_file_system.go
index 006d7245..929f1896 100644
--- a/pkg/fs/local_file_system.go
+++ b/pkg/fs/local_file_system.go
@@ -27,6 +27,8 @@ import (
"path/filepath"
"time"
+ "github.com/shirou/gopsutil/v3/disk"
+
"github.com/apache/skywalking-banyandb/pkg/logger"
)
@@ -281,8 +283,17 @@ func (fs *localFileSystem) MustRMAll(path string) {
fs.logger.Panic().Str("path", path).Msg("failed to remove all files under path")
}
+func (fs *localFileSystem) MustGetFreeSpace(path string) uint64 {
+ usage, err := disk.Usage(path)
+ if err != nil {
+ fs.logger.Panic().Str("path", path).Err(err).Msg("failed to get disk usage")
+ }
+ return usage.Free
+}
+
// Write adds new data to the end of a file.
func (file *LocalFile) Write(buffer []byte) (int, error) {
+ // TODO: use bufio.Writer to optimize performance.
size, err := file.file.Write(buffer)
switch {
case err == nil:
diff --git a/pkg/fs/local_file_system_nix.go b/pkg/fs/local_file_system_nix.go
index 75db2649..f8aef20d 100644
--- a/pkg/fs/local_file_system_nix.go
+++ b/pkg/fs/local_file_system_nix.go
@@ -33,6 +33,11 @@ func (*localFileSystem) CreateLockFile(name string, permission Mode) (File, erro
switch {
case err == nil:
if err = unix.Flock(int(file.Fd()), unix.LOCK_EX|unix.LOCK_NB); err != nil {
+ // if errors.Is(err, syscall.EAGAIN) {
+ // return &LocalFile{
+ // file: file,
+ // }, nil
+ // }
return nil, &FileSystemError{
Code: lockError,
Message: fmt.Sprintf("Cannot lock file, file name: %s, error message: %s", name, err),
diff --git a/pkg/watcher/watcher.go b/pkg/watcher/watcher.go
index eaedaa38..f525d840 100644
--- a/pkg/watcher/watcher.go
+++ b/pkg/watcher/watcher.go
@@ -30,6 +30,11 @@ func (e *Epoch) Watch() <-chan struct{} {
return e.ch
}
+// Val returns the epoch value.
+func (e *Epoch) Val() uint64 {
+ return e.epoch
+}
+
// Epochs is a list of epoch watchers.
type Epochs []*Epoch