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 2022/12/18 04:58:58 UTC
[skywalking-banyandb] branch encoding created (now d6421a4)
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a change to branch encoding
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
at d6421a4 Opt to a new encoding system
This branch includes the following new commits:
new d6421a4 Opt to a new encoding system
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[skywalking-banyandb] 01/01: Opt to a new encoding system
Posted by ha...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch encoding
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit d6421a44e6706b65bcc694610c7aa40b7817b7f3
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Sun Dec 18 04:56:17 2022 +0000
Opt to a new encoding system
Signed-off-by: Gao Hongtao <ha...@gmail.com>
---
.golangci.yml | 2 +
banyand/Makefile | 3 +
banyand/kv/badger.go | 82 +++++---
banyand/kv/kv.go | 28 ++-
banyand/measure/encode.go | 41 ++--
banyand/measure/measure_query.go | 8 -
banyand/measure/metadata.go | 9 +-
banyand/stream/metadata.go | 7 +-
banyand/tsdb/block.go | 75 ++++---
banyand/tsdb/segment.go | 1 -
banyand/tsdb/series.go | 7 +-
banyand/tsdb/series_seek.go | 63 ------
banyand/tsdb/series_seek_sort.go | 12 +-
banyand/tsdb/seriesdb.go | 4 +-
banyand/tsdb/tsdb.go | 20 +-
banyand/tsdb/tsdb_test.go | 9 +-
dist/LICENSE | 4 +-
go.mod | 7 +-
go.sum | 9 +-
pkg/bit/doc.go | 21 --
pkg/buffer/writer.go | 84 --------
pkg/encoding/encoding.go | 16 +-
pkg/encoding/int.go | 61 +++---
pkg/encoding/int_test.go | 21 +-
pkg/encoding/plain.go | 314 -----------------------------
pkg/{bit => encoding}/reader.go | 2 +-
pkg/{bit => encoding}/reader_test.go | 2 +-
pkg/{bit => encoding}/writer.go | 62 ++++--
pkg/{bit => encoding}/writer_test.go | 5 +-
pkg/encoding/xor.go | 10 +-
pkg/encoding/xor_test.go | 7 +-
pkg/index/lsm/lsm.go | 1 -
pkg/pb/v1/tsdb/{interval.go => options.go} | 6 +-
test/stress/env | 5 +-
test/stress/env.dev | 5 +-
35 files changed, 299 insertions(+), 714 deletions(-)
diff --git a/.golangci.yml b/.golangci.yml
index ae0cde1..aef9368 100644
--- a/.golangci.yml
+++ b/.golangci.yml
@@ -68,6 +68,8 @@ linters-settings:
alias: measurev1
- pkg: github.com/apache/skywalking-banyandb/api/proto/banyandb/stream/v1
alias: streamv1
+ - pkg: github.com/apache/skywalking-banyandb/pkg/pb/v1
+ alias: pbv1
lll:
line-length: 170
goconst:
diff --git a/banyand/Makefile b/banyand/Makefile
index 56dce25..376ac0c 100644
--- a/banyand/Makefile
+++ b/banyand/Makefile
@@ -21,6 +21,9 @@ SERVER := $(NAME)-server
BINARIES := $(SERVER)
DEBUG_BINARIES := $(SERVER)-debug
+# // TODO: add jemalloc installation on Linux
+# BUILD_TAGS := jemalloc
+
IMG_NAME := skywalking-banyandb
include ../scripts/build/version.mk
diff --git a/banyand/kv/badger.go b/banyand/kv/badger.go
index 6fb9c0b..f5a4489 100644
--- a/banyand/kv/badger.go
+++ b/banyand/kv/badger.go
@@ -25,7 +25,7 @@ import (
"time"
"github.com/dgraph-io/badger/v3"
- "github.com/dgraph-io/badger/v3/bydb"
+ "github.com/dgraph-io/badger/v3/banyandb"
"github.com/dgraph-io/badger/v3/y"
"github.com/apache/skywalking-banyandb/banyand/observability"
@@ -46,8 +46,7 @@ var (
type badgerTSS struct {
dbOpts badger.Options
badger.TSet
- db *badger.DB
- shardID int
+ db *badger.DB
}
func (b *badgerTSS) Context(key []byte, ts uint64, n int) (pre Iterator, next Iterator) {
@@ -73,6 +72,13 @@ func (b *badgerTSS) Stats() (s observability.Statistics) {
return badgerStats(b.db)
}
+func (b *badgerTSS) Close() error {
+ if b.db != nil && !b.db.IsClosed() {
+ return b.db.Close()
+ }
+ return nil
+}
+
func badgerStats(db *badger.DB) (s observability.Statistics) {
stat := db.Stats()
return observability.Statistics{
@@ -81,13 +87,6 @@ func badgerStats(db *badger.DB) (s observability.Statistics) {
}
}
-func (b *badgerTSS) Close() error {
- if b.db != nil && !b.db.IsClosed() {
- return b.db.Close()
- }
- return nil
-}
-
type mergedIter struct {
delegated Iterator
data []byte
@@ -139,9 +138,8 @@ func (i mergedIter) Value() y.ValueStruct {
}
type badgerDB struct {
- dbOpts badger.Options
- db *badger.DB
- shardID int
+ dbOpts badger.Options
+ db *badger.DB
}
func (b *badgerDB) Stats() observability.Statistics {
@@ -165,7 +163,7 @@ func (b *badgerDB) Scan(prefix, seekKey []byte, opt ScanOpts, f ScanFunc) error
if !bytes.Equal(prefix, k[0:len(prefix)]) {
continue
}
- err := f(b.shardID, k, func() ([]byte, error) {
+ err := f(k, func() ([]byte, error) {
return y.Copy(it.Value().Value), nil
})
if errors.Is(err, errStopScan) {
@@ -301,51 +299,75 @@ func (l *badgerLog) Debugf(f string, v ...interface{}) {
l.delegated.Debug().Msgf(f, v...)
}
-var _ bydb.TSetEncoderPool = (*encoderPoolDelegate)(nil)
+var _ banyandb.SeriesEncoderPool = (*encoderPoolDelegate)(nil)
type encoderPoolDelegate struct {
encoding.SeriesEncoderPool
}
-func (e *encoderPoolDelegate) Get(metadata []byte) bydb.TSetEncoder {
- return e.SeriesEncoderPool.Get(metadata)
+func (e *encoderPoolDelegate) Get(metadata []byte, buffer banyandb.Buffer) banyandb.SeriesEncoder {
+ encoder := e.SeriesEncoderPool.Get(metadata, buffer)
+ if encoder == nil {
+ return nil
+ }
+ return &encoderDelegate{SeriesEncoder: encoder}
+}
+
+func (e *encoderPoolDelegate) Put(encoder banyandb.SeriesEncoder) {
+ ee := encoder.(*encoderDelegate)
+ e.SeriesEncoderPool.Put(ee.SeriesEncoder)
+}
+
+var _ banyandb.SeriesEncoder = (*encoderDelegate)(nil)
+
+type encoderDelegate struct {
+ encoding.SeriesEncoder
}
-func (e *encoderPoolDelegate) Put(encoder bydb.TSetEncoder) {
- e.SeriesEncoderPool.Put(encoder)
+func (e *encoderDelegate) Reset(key []byte, buffer banyandb.Buffer) {
+ e.SeriesEncoder.Reset(key, &bufferDelegate{BufferWriter: buffer})
}
-var _ bydb.TSetDecoderPool = (*decoderPoolDelegate)(nil)
+var _ encoding.BufferWriter = (*bufferDelegate)(nil)
+
+type bufferDelegate struct {
+ encoding.BufferWriter
+}
+
+var _ banyandb.SeriesDecoderPool = (*decoderPoolDelegate)(nil)
type decoderPoolDelegate struct {
encoding.SeriesDecoderPool
}
-func (e *decoderPoolDelegate) Get(metadata []byte) bydb.TSetDecoder {
- return &decoderDelegate{
- e.SeriesDecoderPool.Get(metadata),
+func (e *decoderPoolDelegate) Get(metadata []byte) banyandb.SeriesDecoder {
+ if decoder := e.SeriesDecoderPool.Get(metadata); decoder != nil {
+ return &decoderDelegate{
+ e.SeriesDecoderPool.Get(metadata),
+ }
}
+ return nil
}
-func (e *decoderPoolDelegate) Put(decoder bydb.TSetDecoder) {
- dd := decoder.(*decoderDelegate)
- e.SeriesDecoderPool.Put(dd.SeriesDecoder)
+func (e *decoderPoolDelegate) Put(decoder banyandb.SeriesDecoder) {
+ if decoder != nil {
+ dd := decoder.(*decoderDelegate)
+ e.SeriesDecoderPool.Put(dd.SeriesDecoder)
+ }
}
-var _ bydb.TSetDecoder = (*decoderDelegate)(nil)
+var _ banyandb.SeriesDecoder = (*decoderDelegate)(nil)
type decoderDelegate struct {
encoding.SeriesDecoder
}
-func (d *decoderDelegate) Iterator() bydb.TSetIterator {
+func (d *decoderDelegate) Iterator() banyandb.SeriesIterator {
return &iterDelegate{
SeriesIterator: d.SeriesDecoder.Iterator(),
}
}
-var _ bydb.TSetDecoder = (*decoderDelegate)(nil)
-
type iterDelegate struct {
encoding.SeriesIterator
}
diff --git a/banyand/kv/kv.go b/banyand/kv/kv.go
index e8a3d78..e3a3f35 100644
--- a/banyand/kv/kv.go
+++ b/banyand/kv/kv.go
@@ -24,6 +24,7 @@ import (
"math"
"github.com/dgraph-io/badger/v3"
+ "github.com/dgraph-io/badger/v3/options"
"github.com/pkg/errors"
"github.com/apache/skywalking-banyandb/banyand/observability"
@@ -48,7 +49,7 @@ type writer interface {
}
// ScanFunc is the closure executed on scanning out a pair of key-value.
-type ScanFunc func(shardID int, key []byte, getVal func() ([]byte, error)) error
+type ScanFunc func(key []byte, getVal func() ([]byte, error)) error
// ScanOpts wraps options for scanning the kv storage.
type ScanOpts struct {
@@ -113,11 +114,11 @@ func TSSWithLogger(l *logger.Logger) TimeSeriesOptions {
}
}
-// TSSWithEncoding sets encoding and decoding pools for flushing and compacting.
+// TSSWithEncoding sets encoding and decoding pools for building chunks.
func TSSWithEncoding(encoderPool encoding.SeriesEncoderPool, decoderPool encoding.SeriesDecoderPool) TimeSeriesOptions {
return func(store TimeSeriesStore) {
if btss, ok := store.(*badgerTSS); ok {
- btss.dbOpts = btss.dbOpts.WithExternalCompactor(
+ btss.dbOpts = btss.dbOpts.WithKeyBasedEncoder(
&encoderPoolDelegate{
encoderPool,
}, &decoderPoolDelegate{
@@ -127,6 +128,18 @@ func TSSWithEncoding(encoderPool encoding.SeriesEncoderPool, decoderPool encodin
}
}
+// TSSWithZSTDCompression sets a ZSTD based compression method.
+func TSSWithZSTDCompression(chunkSize int) TimeSeriesOptions {
+ return func(store TimeSeriesStore) {
+ if btss, ok := store.(*badgerTSS); ok {
+ btss.dbOpts = btss.dbOpts.
+ WithCompression(options.ZSTD).
+ WithBlockSize(chunkSize).
+ WithSameKeyBlock()
+ }
+ }
+}
+
// TSSWithMemTableSize sets the size of memory table in bytes.
func TSSWithMemTableSize(sizeInBytes int64) TimeSeriesOptions {
return func(store TimeSeriesStore) {
@@ -167,9 +180,8 @@ type IndexStore interface {
// OpenTimeSeriesStore creates a new TimeSeriesStore.
// nolint: contextcheck
-func OpenTimeSeriesStore(shardID int, path string, options ...TimeSeriesOptions) (TimeSeriesStore, error) {
+func OpenTimeSeriesStore(path string, options ...TimeSeriesOptions) (TimeSeriesStore, error) {
btss := new(badgerTSS)
- btss.shardID = shardID
btss.dbOpts = badger.DefaultOptions(path)
for _, opt := range options {
opt(btss)
@@ -221,9 +233,8 @@ func StoreWithMemTableSize(size int64) StoreOptions {
// OpenStore creates a new Store.
// nolint: contextcheck
-func OpenStore(shardID int, path string, options ...StoreOptions) (Store, error) {
+func OpenStore(path string, options ...StoreOptions) (Store, error) {
bdb := new(badgerDB)
- bdb.shardID = shardID
bdb.dbOpts = badger.DefaultOptions(path)
for _, opt := range options {
opt(bdb)
@@ -256,9 +267,8 @@ func IndexWithLogger(l *logger.Logger) IndexOptions {
}
// OpenIndexStore creates a new IndexStore.
-func OpenIndexStore(shardID int, path string, options ...IndexOptions) (IndexStore, error) {
+func OpenIndexStore(path string, options ...IndexOptions) (IndexStore, error) {
bdb := new(badgerDB)
- bdb.shardID = shardID
bdb.dbOpts = badger.DefaultOptions(path)
for _, opt := range options {
opt(bdb)
diff --git a/banyand/measure/encode.go b/banyand/measure/encode.go
index 86a2bed..d446828 100644
--- a/banyand/measure/encode.go
+++ b/banyand/measure/encode.go
@@ -39,47 +39,41 @@ var (
)
type encoderPool struct {
- intPool encoding.SeriesEncoderPool
- defaultPool encoding.SeriesEncoderPool
- l *logger.Logger
+ intPool encoding.SeriesEncoderPool
+ l *logger.Logger
}
-func newEncoderPool(name string, plainSize, intSize int, l *logger.Logger) encoding.SeriesEncoderPool {
+func newEncoderPool(name string, size int, l *logger.Logger) encoding.SeriesEncoderPool {
return &encoderPool{
- intPool: encoding.NewIntEncoderPool(name, intSize, intervalFn),
- defaultPool: encoding.NewPlainEncoderPool(name, plainSize),
- l: l,
+ intPool: encoding.NewIntEncoderPool(name, size, intervalFn),
+ l: l,
}
}
-func (p *encoderPool) Get(metadata []byte) encoding.SeriesEncoder {
+func (p *encoderPool) Get(metadata []byte, buffer encoding.BufferWriter) encoding.SeriesEncoder {
fieldSpec, _, err := pbv1.DecodeFieldFlag(metadata)
if err != nil {
p.l.Err(err).Msg("failed to decode field flag")
- return p.defaultPool.Get(metadata)
}
if fieldSpec.EncodingMethod == databasev1.EncodingMethod_ENCODING_METHOD_GORILLA {
- return p.intPool.Get(metadata)
+ return p.intPool.Get(metadata, buffer)
}
- return p.defaultPool.Get(metadata)
+ return nil
}
func (p *encoderPool) Put(encoder encoding.SeriesEncoder) {
p.intPool.Put(encoder)
- p.defaultPool.Put(encoder)
}
type decoderPool struct {
- intPool encoding.SeriesDecoderPool
- defaultPool encoding.SeriesDecoderPool
- l *logger.Logger
+ intPool encoding.SeriesDecoderPool
+ l *logger.Logger
}
-func newDecoderPool(name string, plainSize, intSize int, l *logger.Logger) encoding.SeriesDecoderPool {
+func newDecoderPool(name string, size int, l *logger.Logger) encoding.SeriesDecoderPool {
return &decoderPool{
- intPool: encoding.NewIntDecoderPool(name, intSize, intervalFn),
- defaultPool: encoding.NewPlainDecoderPool(name, plainSize),
- l: l,
+ intPool: encoding.NewIntDecoderPool(name, size, intervalFn),
+ l: l,
}
}
@@ -87,15 +81,16 @@ func (p *decoderPool) Get(metadata []byte) encoding.SeriesDecoder {
fieldSpec, _, err := pbv1.DecodeFieldFlag(metadata)
if err != nil {
p.l.Err(err).Msg("failed to decode field flag")
- return p.defaultPool.Get(metadata)
+ return nil
}
if fieldSpec.EncodingMethod == databasev1.EncodingMethod_ENCODING_METHOD_GORILLA {
return p.intPool.Get(metadata)
}
- return p.defaultPool.Get(metadata)
+ return nil
}
func (p *decoderPool) Put(decoder encoding.SeriesDecoder) {
- p.intPool.Put(decoder)
- p.defaultPool.Put(decoder)
+ if decoder != nil {
+ p.intPool.Put(decoder)
+ }
}
diff --git a/banyand/measure/measure_query.go b/banyand/measure/measure_query.go
index 17ea882..5cf7700 100644
--- a/banyand/measure/measure_query.go
+++ b/banyand/measure/measure_query.go
@@ -118,12 +118,8 @@ func (s *measure) ParseTagFamily(family string, item tsdb.Item) (*modelv1.TagFam
fid := familyIdentity(family, pbv1.TagFlag)
familyRawBytes, err := item.Family(fid)
if err != nil {
- item.PrintContext(s.l.Named("tag-family"), fid, 10)
return nil, errors.Wrapf(err, "measure %s.%s parse family %s", s.name, s.group, family)
}
- if len(familyRawBytes) < 1 {
- item.PrintContext(s.l.Named("tag-family"), fid, 10)
- }
tagFamily := &modelv1.TagFamilyForWrite{}
err = proto.Unmarshal(familyRawBytes, tagFamily)
if err != nil {
@@ -164,12 +160,8 @@ func (s *measure) ParseField(name string, item tsdb.Item) (*measurev1.DataPoint_
fid := familyIdentity(name, pbv1.EncoderFieldFlag(fieldSpec, s.interval))
bytes, err := item.Family(fid)
if err != nil {
- item.PrintContext(s.l.Named("field"), fid, 10)
return nil, errors.Wrapf(err, "measure %s.%s parse field %s", s.name, s.group, name)
}
- if len(bytes) < 1 {
- item.PrintContext(s.l.Named("field"), fid, 10)
- }
fieldValue, err := pbv1.DecodeFieldValue(bytes, fieldSpec)
if err != nil {
return nil, err
diff --git a/banyand/measure/metadata.go b/banyand/measure/metadata.go
index 06b1f97..5361bee 100644
--- a/banyand/measure/metadata.go
+++ b/banyand/measure/metadata.go
@@ -221,9 +221,14 @@ func (s *supplier) OpenDB(groupSchema *commonv1.Group) (tsdb.Database, error) {
opts.Location = path.Join(s.path, groupSchema.Metadata.Name)
name := groupSchema.Metadata.Name
opts.EncodingMethod = tsdb.EncodingMethod{
- EncoderPool: newEncoderPool(name, plainChunkSize, intChunkSize, s.l),
- DecoderPool: newDecoderPool(name, plainChunkSize, intChunkSize, s.l),
+ EncoderPool: newEncoderPool(name, intChunkSize, s.l),
+ DecoderPool: newDecoderPool(name, intChunkSize, s.l),
}
+ opts.CompressionMethod = tsdb.CompressionMethod{
+ Type: tsdb.CompressionTypeZSTD,
+ ChunkSizeInBytes: plainChunkSize,
+ }
+
var err error
if opts.BlockInterval, err = pb_v1.ToIntervalRule(groupSchema.ResourceOpts.BlockInterval); err != nil {
return nil, err
diff --git a/banyand/stream/metadata.go b/banyand/stream/metadata.go
index 286a276..968fd7a 100644
--- a/banyand/stream/metadata.go
+++ b/banyand/stream/metadata.go
@@ -30,7 +30,6 @@ import (
"github.com/apache/skywalking-banyandb/banyand/metadata"
"github.com/apache/skywalking-banyandb/banyand/metadata/schema"
"github.com/apache/skywalking-banyandb/banyand/tsdb"
- "github.com/apache/skywalking-banyandb/pkg/encoding"
"github.com/apache/skywalking-banyandb/pkg/logger"
pb_v1 "github.com/apache/skywalking-banyandb/pkg/pb/v1/tsdb"
resourceSchema "github.com/apache/skywalking-banyandb/pkg/schema"
@@ -206,9 +205,9 @@ func (s *supplier) OpenDB(groupSchema *commonv1.Group) (tsdb.Database, error) {
opts := s.dbOpts
opts.ShardNum = groupSchema.ResourceOpts.ShardNum
opts.Location = path.Join(s.path, groupSchema.Metadata.Name)
- opts.EncodingMethod = tsdb.EncodingMethod{
- EncoderPool: encoding.NewPlainEncoderPool(name, chunkSize),
- DecoderPool: encoding.NewPlainDecoderPool(name, chunkSize),
+ opts.CompressionMethod = tsdb.CompressionMethod{
+ Type: tsdb.CompressionTypeZSTD,
+ ChunkSizeInBytes: chunkSize,
}
var err error
if opts.BlockInterval, err = pb_v1.ToIntervalRule(groupSchema.ResourceOpts.BlockInterval); err != nil {
diff --git a/banyand/tsdb/block.go b/banyand/tsdb/block.go
index 66251bf..0f1cd0a 100644
--- a/banyand/tsdb/block.go
+++ b/banyand/tsdb/block.go
@@ -36,7 +36,6 @@ import (
"github.com/apache/skywalking-banyandb/banyand/kv"
"github.com/apache/skywalking-banyandb/banyand/observability"
"github.com/apache/skywalking-banyandb/banyand/tsdb/bucket"
- "github.com/apache/skywalking-banyandb/pkg/encoding"
"github.com/apache/skywalking-banyandb/pkg/index"
"github.com/apache/skywalking-banyandb/pkg/index/inverted"
"github.com/apache/skywalking-banyandb/pkg/index/lsm"
@@ -56,9 +55,9 @@ const (
var errBlockClosingInterrupted = errors.New("interrupt to close the block")
type block struct {
- encodingMethod EncodingMethod
- store kv.TimeSeriesStore
- queue bucket.Queue
+ store kv.TimeSeriesStore
+ openOpts openOpts
+ queue bucket.Queue
bucket.Reporter
clock timestamp.Clock
lsmIndex index.Store
@@ -73,12 +72,16 @@ type block struct {
suffix string
path string
closableLst []io.Closer
- lsmMemSize int64
- memSize int64
lock sync.RWMutex
segID SectionID
blockID SectionID
- indexOpts InvertedIndexOpts
+}
+
+type openOpts struct {
+ store []kv.TimeSeriesOptions
+ storePath string
+ inverted inverted.StoreOpts
+ lsm lsm.StoreOpts
}
type blockOpts struct {
@@ -130,23 +133,34 @@ func (b *block) options(ctx context.Context) {
if o != nil {
options = o.(DatabaseOpts)
}
- if options.EncodingMethod.EncoderPool == nil {
- options.EncodingMethod.EncoderPool = encoding.NewPlainEncoderPool("tsdb", 0)
+ if options.EncodingMethod.EncoderPool != nil && options.EncodingMethod.DecoderPool != nil {
+ b.openOpts.store = append(b.openOpts.store, kv.TSSWithEncoding(options.EncodingMethod.EncoderPool, options.EncodingMethod.DecoderPool))
}
- if options.EncodingMethod.EncoderPool == nil {
- options.EncodingMethod.DecoderPool = encoding.NewPlainDecoderPool("tsdb", 0)
+ if options.CompressionMethod.Type == CompressionTypeZSTD {
+ b.openOpts.store = append(b.openOpts.store, kv.TSSWithZSTDCompression(options.CompressionMethod.ChunkSizeInBytes))
}
- b.encodingMethod = options.EncodingMethod
+ var memSize int64
if options.BlockMemSize < 1 {
- b.memSize = defaultMainMemorySize
+ memSize = defaultMainMemorySize
} else {
- b.memSize = options.BlockMemSize
+ memSize = options.BlockMemSize
+ }
+ b.openOpts.store = append(b.openOpts.store, kv.TSSWithMemTableSize(memSize), kv.TSSWithLogger(b.l.Named(componentMain)))
+ b.openOpts.storePath = path.Join(b.path, componentMain)
+ b.openOpts.inverted = inverted.StoreOpts{
+ Path: path.Join(b.path, componentSecondInvertedIdx),
+ Logger: b.l.Named(componentSecondInvertedIdx),
+ BatchWaitSec: options.BlockInvertedIndex.BatchWaitSec,
+ }
+ lsmMemSize := memSize / 8
+ if lsmMemSize < defaultKVMemorySize {
+ lsmMemSize = defaultKVMemorySize
}
- b.lsmMemSize = b.memSize / 8
- if b.lsmMemSize < defaultKVMemorySize {
- b.lsmMemSize = defaultKVMemorySize
+ b.openOpts.lsm = lsm.StoreOpts{
+ Path: path.Join(b.path, componentSecondLSMIdx),
+ Logger: b.l.Named(componentSecondLSMIdx),
+ MemTableSize: lsmMemSize,
}
- b.indexOpts = options.BlockInvertedIndex
}
func (b *block) openSafely() (err error) {
@@ -162,28 +176,14 @@ func (b *block) openSafely() (err error) {
}
func (b *block) open() (err error) {
- if b.store, err = kv.OpenTimeSeriesStore(
- 0,
- path.Join(b.path, componentMain),
- kv.TSSWithEncoding(b.encodingMethod.EncoderPool, b.encodingMethod.DecoderPool),
- kv.TSSWithLogger(b.l.Named(componentMain)),
- kv.TSSWithMemTableSize(b.memSize),
- ); err != nil {
+ if b.store, err = kv.OpenTimeSeriesStore(b.openOpts.storePath, b.openOpts.store...); err != nil {
return err
}
b.closableLst = append(b.closableLst, b.store)
- if b.invertedIndex, err = inverted.NewStore(inverted.StoreOpts{
- Path: path.Join(b.path, componentSecondInvertedIdx),
- Logger: b.l.Named(componentSecondInvertedIdx),
- BatchWaitSec: b.indexOpts.BatchWaitSec,
- }); err != nil {
+ if b.invertedIndex, err = inverted.NewStore(b.openOpts.inverted); err != nil {
return err
}
- if b.lsmIndex, err = lsm.NewStore(lsm.StoreOpts{
- Path: path.Join(b.path, componentSecondLSMIdx),
- Logger: b.l.Named(componentSecondLSMIdx),
- MemTableSize: b.lsmMemSize,
- }); err != nil {
+ if b.lsmIndex, err = lsm.NewStore(b.openOpts.lsm); err != nil {
return err
}
b.closableLst = append(b.closableLst, b.invertedIndex, b.lsmIndex)
@@ -324,7 +324,6 @@ type blockDelegate interface {
writeLSMIndex(fields []index.Field, id common.ItemID) error
writeInvertedIndex(fields []index.Field, id common.ItemID) error
dataReader() kv.TimeSeriesReader
- decoderPool() encoding.SeriesDecoderPool
lsmIndexReader() index.Searcher
invertedIndexReader() index.Searcher
primaryIndexReader() index.FieldIterable
@@ -339,10 +338,6 @@ type bDelegate struct {
delegate *block
}
-func (d *bDelegate) decoderPool() encoding.SeriesDecoderPool {
- return d.delegate.encodingMethod.DecoderPool
-}
-
func (d *bDelegate) dataReader() kv.TimeSeriesReader {
return d.delegate.store
}
diff --git a/banyand/tsdb/segment.go b/banyand/tsdb/segment.go
index c6582f2..e0cf74c 100644
--- a/banyand/tsdb/segment.go
+++ b/banyand/tsdb/segment.go
@@ -86,7 +86,6 @@ func openSegment(ctx context.Context, startTime, endTime time.Time, path, suffix
memSize = defaultKVMemorySize
}
if s.globalIndex, err = kv.OpenStore(
- 0,
indexPath,
kv.StoreWithLogger(s.l),
kv.StoreWithMemTableSize(memSize),
diff --git a/banyand/tsdb/series.go b/banyand/tsdb/series.go
index a872c82..0326bf6 100644
--- a/banyand/tsdb/series.go
+++ b/banyand/tsdb/series.go
@@ -110,10 +110,9 @@ func (s *series) Get(ctx context.Context, id GlobalItemID) (Item, io.Closer, err
return nil, nil, errors.WithMessagef(errBlockAbsent, "id: %v", id)
}
return &item{
- data: b.dataReader(),
- itemID: id.ID,
- seriesID: s.id,
- decoderPool: b.decoderPool(),
+ data: b.dataReader(),
+ itemID: id.ID,
+ seriesID: s.id,
}, b, nil
}
diff --git a/banyand/tsdb/series_seek.go b/banyand/tsdb/series_seek.go
index 28147f7..2534c15 100644
--- a/banyand/tsdb/series_seek.go
+++ b/banyand/tsdb/series_seek.go
@@ -18,18 +18,10 @@
package tsdb
import (
- "bytes"
- "encoding/hex"
- "time"
-
- "github.com/dgraph-io/badger/v3/y"
- "github.com/rs/zerolog"
-
"github.com/apache/skywalking-banyandb/api/common"
databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
"github.com/apache/skywalking-banyandb/banyand/kv"
- "github.com/apache/skywalking-banyandb/pkg/encoding"
"github.com/apache/skywalking-banyandb/pkg/index"
"github.com/apache/skywalking-banyandb/pkg/logger"
)
@@ -44,7 +36,6 @@ type Iterator interface {
// Item allows retrieving raw data from an item.
type Item interface {
Family(family []byte) ([]byte, error)
- PrintContext(l *logger.Logger, family []byte, n int)
Val() ([]byte, error)
ID() common.ItemID
SortedField() []byte
@@ -113,7 +104,6 @@ var _ Item = (*item)(nil)
type item struct {
data kv.TimeSeriesReader
- decoderPool encoding.SeriesDecoderPool
sortedField []byte
itemID common.ItemID
seriesID common.SeriesID
@@ -135,59 +125,6 @@ func (i *item) Family(family []byte) ([]byte, error) {
return i.data.Get(d.marshal(), uint64(i.itemID))
}
-func (i *item) PrintContext(l *logger.Logger, family []byte, n int) {
- decoder := i.decoderPool.Get(family)
- defer i.decoderPool.Put(decoder)
- d := dataBucket{
- seriesID: i.seriesID,
- family: family,
- }
- key := d.marshal()
- pre, next := i.data.Context(key, uint64(i.itemID), n)
- defer pre.Close()
- defer next.Close()
- j := 0
- each := func(iter kv.Iterator, logEvent *zerolog.Event) *zerolog.Event {
- if !bytes.Equal(key, iter.Key()) {
- return logEvent
- }
- j++
-
- logEvent = logEvent.Int("i", j).
- Time("ts", time.Unix(0, int64(y.ParseTs(iter.RawKey()))))
- locArr := zerolog.Arr()
- rangeArr := zerolog.Arr()
- decodedNumArr := zerolog.Arr()
- if err := decoder.Decode(family, iter.Val()); err != nil {
- locArr.Str("mem")
- } else {
- locArr.Str("table")
- start, end := decoder.Range()
- rangeArr.Time(time.Unix(0, int64(start)))
- rangeArr.Time(time.Unix(0, int64(end)))
- decodedNumArr.Int(decoder.Len())
- }
- logEvent = logEvent.Array("loc", locArr).Array("range", rangeArr).Array("decodedNum", decodedNumArr)
- return logEvent
- }
-
- s := hex.EncodeToString(key)
- if len(s) > 7 {
- s = s[:7]
- }
- event := l.Info().Str("prefix", s).Time("ts", time.Unix(0, int64(i.itemID)))
- for ; pre.Valid() && j < n; pre.Next() {
- event = each(pre, event)
- }
- event.Msg("print previous lines")
- j = 0
- event = l.Info().Str("prefix", s).Time("ts", time.Unix(0, int64(i.itemID)))
- for ; next.Valid() && j < n; next.Next() {
- event = each(next, event)
- }
- event.Msg("print next lines")
-}
-
func (i *item) Val() ([]byte, error) {
d := dataBucket{
seriesID: i.seriesID,
diff --git a/banyand/tsdb/series_seek_sort.go b/banyand/tsdb/series_seek_sort.go
index 40d05b2..f41e750 100644
--- a/banyand/tsdb/series_seek_sort.go
+++ b/banyand/tsdb/series_seek_sort.go
@@ -29,7 +29,6 @@ import (
modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
"github.com/apache/skywalking-banyandb/banyand/kv"
"github.com/apache/skywalking-banyandb/pkg/convert"
- "github.com/apache/skywalking-banyandb/pkg/encoding"
"github.com/apache/skywalking-banyandb/pkg/index"
"github.com/apache/skywalking-banyandb/pkg/index/posting"
"github.com/apache/skywalking-banyandb/pkg/logger"
@@ -95,7 +94,7 @@ func (s *seekerBuilder) buildSeriesByIndex() (series []Iterator, err error) {
return nil, err
}
if inner != nil {
- series = append(series, newSearcherIterator(s.seriesSpan.l, inner, b.dataReader(), b.decoderPool(),
+ series = append(series, newSearcherIterator(s.seriesSpan.l, inner, b.dataReader(),
s.seriesSpan.seriesID, filters))
}
}
@@ -142,10 +141,10 @@ func (s *seekerBuilder) buildSeriesByTime() ([]Iterator, error) {
return nil, err
}
if filter == nil {
- delegated = append(delegated, newSearcherIterator(s.seriesSpan.l, inner, b.dataReader(), b.decoderPool(),
+ delegated = append(delegated, newSearcherIterator(s.seriesSpan.l, inner, b.dataReader(),
s.seriesSpan.seriesID, emptyFilters))
} else {
- delegated = append(delegated, newSearcherIterator(s.seriesSpan.l, inner, b.dataReader(), b.decoderPool(),
+ delegated = append(delegated, newSearcherIterator(s.seriesSpan.l, inner, b.dataReader(),
s.seriesSpan.seriesID, []filterFn{filter}))
}
}
@@ -167,7 +166,6 @@ type searcherIterator struct {
fieldIterator index.FieldIterator
cur posting.Iterator
data kv.TimeSeriesReader
- decoderPool encoding.SeriesDecoderPool
l *logger.Logger
curKey []byte
filters []filterFn
@@ -205,7 +203,6 @@ func (s *searcherIterator) Val() Item {
itemID: s.cur.Current(),
data: s.data,
seriesID: s.seriesID,
- decoderPool: s.decoderPool,
}
}
@@ -214,7 +211,7 @@ func (s *searcherIterator) Close() error {
}
func newSearcherIterator(l *logger.Logger, fieldIterator index.FieldIterator, data kv.TimeSeriesReader,
- decoderPool encoding.SeriesDecoderPool, seriesID common.SeriesID, filters []filterFn,
+ seriesID common.SeriesID, filters []filterFn,
) Iterator {
return &searcherIterator{
fieldIterator: fieldIterator,
@@ -222,7 +219,6 @@ func newSearcherIterator(l *logger.Logger, fieldIterator index.FieldIterator, da
seriesID: seriesID,
filters: filters,
l: l,
- decoderPool: decoderPool,
}
}
diff --git a/banyand/tsdb/seriesdb.go b/banyand/tsdb/seriesdb.go
index 5b69639..33e1798 100644
--- a/banyand/tsdb/seriesdb.go
+++ b/banyand/tsdb/seriesdb.go
@@ -401,7 +401,7 @@ func (s *seriesDB) List(ctx context.Context, path Path) (SeriesList, error) {
}
result := make([]Series, 0)
var err error
- errScan := s.seriesMetadata.Scan(prefix, prepend(path.seekKey, entityPrefix), kv.DefaultScanOpts, func(_ int, key []byte, getVal func() ([]byte, error)) error {
+ errScan := s.seriesMetadata.Scan(prefix, prepend(path.seekKey, entityPrefix), kv.DefaultScanOpts, func(key []byte, getVal func() ([]byte, error)) error {
key = key[entityPrefixLen:]
comparableKey := make([]byte, len(key))
for i, b := range key {
@@ -512,7 +512,7 @@ func newSeriesDataBase(ctx context.Context, shardID common.ShardID, path string,
memSize = defaultKVMemorySize
}
var err error
- sdb.seriesMetadata, err = kv.OpenStore(0, path+"/md",
+ sdb.seriesMetadata, err = kv.OpenStore(path+"/md",
kv.StoreWithNamedLogger("metadata", sdb.l),
kv.StoreWithMemTableSize(memSize),
)
diff --git a/banyand/tsdb/tsdb.go b/banyand/tsdb/tsdb.go
index eb517d1..fc45804 100644
--- a/banyand/tsdb/tsdb.go
+++ b/banyand/tsdb/tsdb.go
@@ -98,6 +98,7 @@ var _ Database = (*database)(nil)
type DatabaseOpts struct {
EncodingMethod EncodingMethod
Location string
+ CompressionMethod CompressionMethod
SegmentInterval IntervalRule
BlockInterval IntervalRule
TTL IntervalRule
@@ -120,6 +121,22 @@ type EncodingMethod struct {
DecoderPool encoding.SeriesDecoderPool
}
+// CompressionMethod denotes how to compress a single chunk.
+type CompressionMethod struct {
+ Type CompressionType
+ ChunkSizeInBytes int
+}
+
+// CompressionType specifies how a chunk should be compressed.
+type CompressionType int
+
+const (
+ // CompressionTypeNone mode indicates that a chunk is not compressed.
+ CompressionTypeNone CompressionType = iota
+ // CompressionTypeZSTD mode indicates that a chunk is compressed using CompressionTypeZSTD algorithm.
+ CompressionTypeZSTD
+)
+
type (
// SectionID is the kind of a block/segment.
SectionID uint32
@@ -203,9 +220,6 @@ func (d *database) Close() error {
// OpenDatabase returns a new tsdb runtime. This constructor will create a new database if it's absent,
// or load an existing one.
func OpenDatabase(ctx context.Context, opts DatabaseOpts) (Database, error) {
- if opts.EncodingMethod.EncoderPool == nil || opts.EncodingMethod.DecoderPool == nil {
- return nil, errors.Wrap(errOpenDatabase, "encoding method is absent")
- }
if _, err := mkdir(opts.Location); err != nil {
return nil, err
}
diff --git a/banyand/tsdb/tsdb_test.go b/banyand/tsdb/tsdb_test.go
index 443ae6d..ffe4ac8 100644
--- a/banyand/tsdb/tsdb_test.go
+++ b/banyand/tsdb/tsdb_test.go
@@ -27,7 +27,6 @@ import (
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
- "github.com/apache/skywalking-banyandb/pkg/encoding"
"github.com/apache/skywalking-banyandb/pkg/logger"
"github.com/apache/skywalking-banyandb/pkg/test"
"github.com/apache/skywalking-banyandb/pkg/test/flags"
@@ -109,12 +108,8 @@ func openDatabase(ctx context.Context, t *require.Assertions, path string) (db D
db, err := OpenDatabase(
context.WithValue(ctx, logger.ContextKey, logger.GetLogger("test")),
DatabaseOpts{
- Location: path,
- ShardNum: 1,
- EncodingMethod: EncodingMethod{
- EncoderPool: encoding.NewPlainEncoderPool("tsdb", 0),
- DecoderPool: encoding.NewPlainDecoderPool("tsdb", 0),
- },
+ Location: path,
+ ShardNum: 1,
BlockInterval: IntervalRule{Num: 2},
SegmentInterval: IntervalRule{Num: 1, Unit: DAY},
TTL: IntervalRule{Num: 7, Unit: DAY},
diff --git a/dist/LICENSE b/dist/LICENSE
index fc75cee..fd077a7 100644
--- a/dist/LICENSE
+++ b/dist/LICENSE
@@ -178,12 +178,12 @@
Apache-2.0 licenses
========================================================================
- github.com/SkyAPM/badger/v3 v3.0.0-20220817114744-b3711444d876 Apache-2.0
+ github.com/SkyAPM/badger/v3 v3.0.0-20221218043700-57f894035264 Apache-2.0
github.com/blevesearch/segment v0.9.0 Apache-2.0
github.com/blevesearch/vellum v1.0.7 Apache-2.0
github.com/coreos/go-semver v0.3.0 Apache-2.0
github.com/coreos/go-systemd/v22 v22.5.0 Apache-2.0
- github.com/dgraph-io/ristretto v0.1.0 Apache-2.0
+ github.com/dgraph-io/ristretto v0.1.1 Apache-2.0
github.com/envoyproxy/protoc-gen-validate v0.1.0 Apache-2.0
github.com/go-logr/logr v1.2.3 Apache-2.0
github.com/go-logr/stdr v1.2.2 Apache-2.0
diff --git a/go.mod b/go.mod
index a74b376..4784f28 100644
--- a/go.mod
+++ b/go.mod
@@ -18,7 +18,6 @@ require (
github.com/grpc-ecosystem/go-grpc-middleware v1.3.0
github.com/grpc-ecosystem/grpc-gateway/v2 v2.14.0
github.com/hashicorp/golang-lru v0.5.4
- github.com/klauspost/compress v1.15.6
github.com/oklog/run v1.1.0
github.com/onsi/ginkgo/v2 v2.1.4
github.com/onsi/gomega v1.20.0
@@ -42,6 +41,8 @@ require (
sigs.k8s.io/yaml v1.3.0
)
+require github.com/klauspost/compress v1.15.6 // indirect
+
require (
github.com/axiomhq/hyperloglog v0.0.0-20191112132149-a4c4c47bc57f // indirect
github.com/beorn7/perks v1.0.1 // indirect
@@ -59,7 +60,7 @@ require (
github.com/coreos/go-semver v0.3.0 // indirect
github.com/coreos/go-systemd/v22 v22.5.0 // indirect
github.com/davecgh/go-spew v1.1.1 // indirect
- github.com/dgraph-io/ristretto v0.1.0 // indirect
+ github.com/dgraph-io/ristretto v0.1.1 // indirect
github.com/dgryski/go-metro v0.0.0-20180109044635-280f6062b5bc // indirect
github.com/dustin/go-humanize v1.0.0 // indirect
github.com/fsnotify/fsnotify v1.5.4 // indirect
@@ -135,5 +136,5 @@ replace (
github.com/blugelabs/bluge => github.com/zinclabs/bluge v1.1.5
github.com/blugelabs/bluge_segment_api => github.com/zinclabs/bluge_segment_api v1.0.0
github.com/blugelabs/ice => github.com/zinclabs/ice v1.1.3
- github.com/dgraph-io/badger/v3 v3.2011.1 => github.com/SkyAPM/badger/v3 v3.0.0-20220817114744-b3711444d876
+ github.com/dgraph-io/badger/v3 v3.2011.1 => github.com/SkyAPM/badger/v3 v3.0.0-20221218043700-57f894035264
)
diff --git a/go.sum b/go.sum
index 263b97d..91862cb 100644
--- a/go.sum
+++ b/go.sum
@@ -46,8 +46,8 @@ github.com/OneOfOne/xxhash v1.2.2 h1:KMrpdQIwFcEqXDklaen+P1axHaj9BSKzvpUUfnHldSE
github.com/OneOfOne/xxhash v1.2.2/go.mod h1:HSdplMjZKSmBqAxg5vPj2TmRDmfkzw+cTzAElWljhcU=
github.com/RoaringBitmap/roaring v0.9.4 h1:ckvZSX5gwCRaJYBNe7syNawCU5oruY9gQmjXlp4riwo=
github.com/RoaringBitmap/roaring v0.9.4/go.mod h1:icnadbWcNyfEHlYdr+tDlOTih1Bf/h+rzPpv4sbomAA=
-github.com/SkyAPM/badger/v3 v3.0.0-20220817114744-b3711444d876 h1:zH//2cmDpBla7rL9NzWr+vZ2UskMrvdnUvslz3KYTN0=
-github.com/SkyAPM/badger/v3 v3.0.0-20220817114744-b3711444d876/go.mod h1:TQb+QXS839JUIX5jEySXoAvAIDROfU22HqVHcBr9bRA=
+github.com/SkyAPM/badger/v3 v3.0.0-20221218043700-57f894035264 h1:AkLOytkAgB7NZJHydDqc6SkQMxLfaKggqdu3QOpDQsU=
+github.com/SkyAPM/badger/v3 v3.0.0-20221218043700-57f894035264/go.mod h1:4WETftF8A4mEeFgIsYB/VvGo5kfTVl/neYgEqlVW9Ag=
github.com/SkyAPM/clock v1.3.1-0.20220809233656-dc7607c94a97 h1:FKuhJ+6n/DHspGeLleeNbziWnKr9gHKYN4q7NcoCp4s=
github.com/SkyAPM/clock v1.3.1-0.20220809233656-dc7607c94a97/go.mod h1:2xGRl9H1pllhxTbEGO1W3gDkip8P9GQaHPni/wpdR44=
github.com/ajstarks/svgo v0.0.0-20180226025133-644b8db467af/go.mod h1:K08gAheRH3/J6wwsYMMT4xOr94bZjxIelGM0+d/wbFw=
@@ -134,8 +134,8 @@ github.com/creack/pty v1.1.11/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
-github.com/dgraph-io/ristretto v0.1.0 h1:Jv3CGQHp9OjuMBSne1485aDpUkTKEcUqF+jm/LuerPI=
-github.com/dgraph-io/ristretto v0.1.0/go.mod h1:fux0lOrBhrVCJd3lcTHsIJhq1T2rokOu6v9Vcb3Q9ug=
+github.com/dgraph-io/ristretto v0.1.1 h1:6CWw5tJNgpegArSHpNHJKldNeq03FQCwYvfMVWajOK8=
+github.com/dgraph-io/ristretto v0.1.1/go.mod h1:S1GPSBCYCIhmVNfcth17y2zZtQT6wzkzgwUve0VDWWA=
github.com/dgrijalva/jwt-go v3.2.0+incompatible/go.mod h1:E3ru+11k8xSBh+hMPgOLZmtrrCbhqsmaPHjLKYnJCaQ=
github.com/dgryski/go-farm v0.0.0-20190423205320-6a90982ecee2 h1:tdlZCpZ/P9DhczCTSixgIKmwPv6+wP5DGjqLYw5SUiA=
github.com/dgryski/go-farm v0.0.0-20190423205320-6a90982ecee2/go.mod h1:SqUrOPUnsFjfmXRMNPybcSiG0BgUW2AuFH8PAnS2iTw=
@@ -780,6 +780,7 @@ golang.org/x/sys v0.0.0-20220114195835-da31bd327af9/go.mod h1:oPkhp1MJrh7nUepCBc
golang.org/x/sys v0.0.0-20220412211240-33da011f77ad/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20221010170243-090e33056c14/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.2.0 h1:ljd4t30dBnAvMZaQCevtY0xLLD0A+bRZXbgLMLU1F/A=
golang.org/x/sys v0.2.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
diff --git a/pkg/bit/doc.go b/pkg/bit/doc.go
deleted file mode 100644
index 18bf1f6..0000000
--- a/pkg/bit/doc.go
+++ /dev/null
@@ -1,21 +0,0 @@
-// 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.
-
-// TODO:// move this into package encoding.
-
-// Package bit implement binary packer and unpacker to build custom binary stream.
-package bit
diff --git a/pkg/buffer/writer.go b/pkg/buffer/writer.go
deleted file mode 100644
index 0da0505..0000000
--- a/pkg/buffer/writer.go
+++ /dev/null
@@ -1,84 +0,0 @@
-// 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.
-
-// TODO:// move this into package encoding.
-
-// Package buffer implements a custom buffer based on a reusable binary array.
-package buffer
-
-import (
- "bytes"
- "encoding/binary"
-)
-
-// Writer writes data into a buffer.
-type Writer struct {
- buf *bytes.Buffer
-
- scratch [binary.MaxVarintLen64]byte
-}
-
-// NewBufferWriter returns a new Writer.
-func NewBufferWriter(buf *bytes.Buffer) *Writer {
- return &Writer{
- buf: buf,
- }
-}
-
-// Write binaries to the buffer.
-func (w *Writer) Write(p []byte) {
- _, _ = w.buf.Write(p)
-}
-
-// WriteTo writes data to another Writer.
-func (w *Writer) WriteTo(other *Writer) (n int64) {
- n, _ = w.buf.WriteTo(other.buf)
- return n
-}
-
-// PutUint16 puts uint16 data into the buffer.
-func (w *Writer) PutUint16(v uint16) {
- binary.LittleEndian.PutUint16(w.scratch[:], v)
- _, _ = w.buf.Write(w.scratch[:2])
-}
-
-// PutUint32 puts uint32 data into the buffer.
-func (w *Writer) PutUint32(v uint32) {
- binary.LittleEndian.PutUint32(w.scratch[:], v)
- _, _ = w.buf.Write(w.scratch[:4])
-}
-
-// PutUint64 puts uint64 data into the buffer.
-func (w *Writer) PutUint64(v uint64) {
- binary.LittleEndian.PutUint64(w.scratch[:], v)
- _, _ = w.buf.Write(w.scratch[:8])
-}
-
-// Reset cleans up the buffer.
-func (w *Writer) Reset() {
- w.buf.Reset()
-}
-
-// Len returns the length of the buffer.
-func (w *Writer) Len() int {
- return w.buf.Len()
-}
-
-// Bytes outputs the data in the buffer.
-func (w *Writer) Bytes() []byte {
- return w.buf.Bytes()
-}
diff --git a/pkg/encoding/encoding.go b/pkg/encoding/encoding.go
index d3566ec..6d23d37 100644
--- a/pkg/encoding/encoding.go
+++ b/pkg/encoding/encoding.go
@@ -19,13 +19,10 @@
package encoding
import (
- "github.com/pkg/errors"
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/client_golang/prometheus/promauto"
)
-var errEncodeEmpty = errors.New("encode an empty value")
-
var (
rawSize = promauto.NewCounterVec(prometheus.CounterOpts{
Name: "banyand_encoding_raw_size",
@@ -46,7 +43,7 @@ var (
// SeriesEncoderPool allows putting and getting SeriesEncoder.
type SeriesEncoderPool interface {
- Get(metadata []byte) SeriesEncoder
+ Get(metadata []byte, buffer BufferWriter) SeriesEncoder
Put(encoder SeriesEncoder)
}
@@ -57,9 +54,9 @@ type SeriesEncoder interface {
// IsFull returns whether the encoded data reached its capacity
IsFull() bool
// Reset the underlying buffer
- Reset(key []byte)
+ Reset(key []byte, buffer BufferWriter)
// Encode the time series data point to a binary
- Encode() ([]byte, error)
+ Encode() error
// StartTime indicates the first entry's time
StartTime() uint64
}
@@ -97,3 +94,10 @@ type SeriesIterator interface {
// Error might return an error indicates a decode failure
Error() error
}
+
+// BufferWriter allows writing a variable-sized buffer of bytes.
+type BufferWriter interface {
+ Write(data []byte) (n int, err error)
+ WriteByte(b byte) error
+ Bytes() []byte
+}
diff --git a/pkg/encoding/int.go b/pkg/encoding/int.go
index fc14d86..ed81b72 100644
--- a/pkg/encoding/int.go
+++ b/pkg/encoding/int.go
@@ -25,8 +25,6 @@ import (
"sync"
"time"
- "github.com/apache/skywalking-banyandb/pkg/bit"
- "github.com/apache/skywalking-banyandb/pkg/buffer"
"github.com/apache/skywalking-banyandb/pkg/convert"
)
@@ -39,6 +37,8 @@ var (
return &intDecoder{}
},
}
+
+ errInvalidValue = errors.New("invalid encoded value")
)
type intEncoderPoolDelegator struct {
@@ -58,12 +58,12 @@ func NewIntEncoderPool(name string, size int, fn ParseInterval) SeriesEncoderPoo
}
}
-func (b *intEncoderPoolDelegator) Get(metadata []byte) SeriesEncoder {
+func (b *intEncoderPoolDelegator) Get(metadata []byte, buffer BufferWriter) SeriesEncoder {
encoder := b.pool.Get().(*intEncoder)
encoder.name = b.name
encoder.size = b.size
encoder.fn = b.fn
- encoder.Reset(metadata)
+ encoder.Reset(metadata, buffer)
return encoder
}
@@ -112,8 +112,8 @@ var _ SeriesEncoder = (*intEncoder)(nil)
type ParseInterval = func(key []byte) time.Duration
type intEncoder struct {
- buff *bytes.Buffer
- bw *bit.Writer
+ buff BufferWriter
+ bw *Writer
values *XOREncoder
fn ParseInterval
name string
@@ -125,10 +125,8 @@ type intEncoder struct {
}
func newIntEncoder() interface{} {
- buff := &bytes.Buffer{}
- bw := bit.NewWriter(buff)
+ bw := NewWriter()
return &intEncoder{
- buff: buff,
bw: bw,
values: NewXOREncoder(bw),
}
@@ -166,8 +164,9 @@ func (ie *intEncoder) IsFull() bool {
return ie.num >= ie.size
}
-func (ie *intEncoder) Reset(key []byte) {
- ie.bw.Reset(nil)
+func (ie *intEncoder) Reset(key []byte, buffer BufferWriter) {
+ ie.buff = buffer
+ ie.bw.Reset(buffer)
ie.interval = ie.fn(key)
ie.startTime = 0
ie.prevTime = 0
@@ -175,14 +174,14 @@ func (ie *intEncoder) Reset(key []byte) {
ie.values = NewXOREncoder(ie.bw)
}
-func (ie *intEncoder) Encode() ([]byte, error) {
+func (ie *intEncoder) Encode() error {
ie.bw.Flush()
- buffWriter := buffer.NewBufferWriter(ie.buff)
+ buffWriter := NewPacker(ie.buff)
buffWriter.PutUint64(ie.startTime)
buffWriter.PutUint16(uint16(ie.num))
bb := buffWriter.Bytes()
encodedSize.WithLabelValues(ie.name, "int").Add(float64(len(bb)))
- return bb, nil
+ return nil
}
func (ie *intEncoder) StartTime() uint64 {
@@ -234,7 +233,7 @@ func (i intDecoder) Range() (start, end uint64) {
}
func (i intDecoder) Iterator() SeriesIterator {
- br := bit.NewReader(bytes.NewReader(i.area))
+ br := NewReader(bytes.NewReader(i.area))
return &intIterator{
endTime: i.startTime + uint64(i.num*int(i.interval)),
interval: int(i.interval),
@@ -252,7 +251,7 @@ var (
type intIterator struct {
err error
- br *bit.Reader
+ br *Reader
values *XORDecoder
endTime uint64
interval int
@@ -266,23 +265,23 @@ func (i *intIterator) Next() bool {
if i.index >= i.size {
return false
}
- b, err := i.br.ReadBool()
- if errors.Is(err, io.EOF) {
- return false
- }
- if err != nil {
- i.err = err
- return false
- }
- if b {
- if i.values.Next() {
- i.currVal = i.values.Value()
+ var b bool
+ var err error
+ for !b {
+ b, err = i.br.ReadBool()
+ if errors.Is(err, io.EOF) {
+ return false
}
- } else {
- i.currVal = zero
+ if err != nil {
+ i.err = err
+ return false
+ }
+ i.index++
+ i.currTime = i.endTime - uint64(i.interval*i.index)
+ }
+ if i.values.Next() {
+ i.currVal = i.values.Value()
}
- i.index++
- i.currTime = i.endTime - uint64(i.interval*i.index)
return true
}
diff --git a/pkg/encoding/int_test.go b/pkg/encoding/int_test.go
index 1f08278..82767ea 100644
--- a/pkg/encoding/int_test.go
+++ b/pkg/encoding/int_test.go
@@ -18,6 +18,7 @@
package encoding
import (
+ "bytes"
"testing"
"time"
@@ -84,8 +85,8 @@ func TestNewIntEncoderAndDecoder(t *testing.T) {
data: []int64{7, 9},
},
want: tsData{
- ts: []uint64{uint64(4 * time.Minute), uint64(3 * time.Minute), uint64(2 * time.Minute), uint64(1 * time.Minute)},
- data: []int64{7, 0, 0, 9},
+ ts: []uint64{uint64(4 * time.Minute), uint64(1 * time.Minute)},
+ data: []int64{7, 9},
start: uint64(time.Minute),
end: uint64(4 * time.Minute),
},
@@ -102,11 +103,11 @@ func TestNewIntEncoderAndDecoder(t *testing.T) {
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
at := assert.New(t)
- encoder := encoderPool.Get(key)
+ var buffer bytes.Buffer
+ encoder := encoderPool.Get(key, &buffer)
defer encoderPool.Put(encoder)
decoder := decoderPool.Get(key)
defer decoderPool.Put(decoder)
- encoder.Reset(key)
isFull := false
for i, v := range tt.args.ts {
encoder.Append(v, convert.Int64ToBytes(tt.args.data[i]))
@@ -115,11 +116,11 @@ func TestNewIntEncoderAndDecoder(t *testing.T) {
break
}
}
- bb, err := encoder.Encode()
+ err := encoder.Encode()
at.NoError(err)
at.Equal(tt.want.start, encoder.StartTime())
- at.NoError(decoder.Decode(key, bb))
+ at.NoError(decoder.Decode(key, buffer.Bytes()))
start, end := decoder.Range()
at.Equal(tt.want.start, start)
at.Equal(tt.want.end, end)
@@ -216,11 +217,11 @@ func TestNewIntDecoderGet(t *testing.T) {
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
at := assert.New(t)
- encoder := encoderPool.Get(key)
+ var buffer bytes.Buffer
+ encoder := encoderPool.Get(key, &buffer)
defer encoderPool.Put(encoder)
decoder := decoderPool.Get(key)
defer decoderPool.Put(decoder)
- encoder.Reset(key)
isFull := false
for i, v := range tt.args.ts {
encoder.Append(v, convert.Int64ToBytes(tt.args.data[i]))
@@ -229,11 +230,11 @@ func TestNewIntDecoderGet(t *testing.T) {
break
}
}
- bb, err := encoder.Encode()
+ err := encoder.Encode()
at.NoError(err)
at.Equal(tt.want.start, encoder.StartTime())
- at.NoError(decoder.Decode(key, bb))
+ at.NoError(decoder.Decode(key, buffer.Bytes()))
start, end := decoder.Range()
at.Equal(tt.want.start, start)
at.Equal(tt.want.end, end)
diff --git a/pkg/encoding/plain.go b/pkg/encoding/plain.go
deleted file mode 100644
index 504fc5a..0000000
--- a/pkg/encoding/plain.go
+++ /dev/null
@@ -1,314 +0,0 @@
-// 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 encoding
-
-import (
- "bytes"
- "encoding/binary"
- "fmt"
- "sort"
- "sync"
-
- "github.com/klauspost/compress/zstd"
- "github.com/pkg/errors"
-
- "github.com/apache/skywalking-banyandb/pkg/buffer"
-)
-
-var (
- plainEncoderPool = sync.Pool{
- New: newPlainEncoder,
- }
- plainDecoderPool = sync.Pool{
- New: func() interface{} {
- return &plainDecoder{}
- },
- }
-)
-
-type plainEncoderPoolDelegator struct {
- pool *sync.Pool
- name string
- size int
-}
-
-// NewPlainEncoderPool returns a SeriesEncoderPool which provides encoders encoding arbitrary data.
-func NewPlainEncoderPool(name string, size int) SeriesEncoderPool {
- return &plainEncoderPoolDelegator{
- name: name,
- pool: &plainEncoderPool,
- size: size,
- }
-}
-
-func (b *plainEncoderPoolDelegator) Get(metadata []byte) SeriesEncoder {
- encoder := b.pool.Get().(*plainEncoder)
- encoder.name = b.name
- encoder.Reset(metadata)
- encoder.valueSize = b.size
- return encoder
-}
-
-func (b *plainEncoderPoolDelegator) Put(encoder SeriesEncoder) {
- _, ok := encoder.(*plainEncoder)
- if ok {
- b.pool.Put(encoder)
- }
-}
-
-type plainDecoderPoolDelegator struct {
- pool *sync.Pool
- name string
- size int
-}
-
-// NewPlainDecoderPool returns a NewPlainDecoderPool which provides decoders decoding arbitrary data.
-func NewPlainDecoderPool(name string, size int) SeriesDecoderPool {
- return &plainDecoderPoolDelegator{
- name: name,
- pool: &plainDecoderPool,
- size: size,
- }
-}
-
-func (b *plainDecoderPoolDelegator) Get(_ []byte) SeriesDecoder {
- decoder := b.pool.Get().(*plainDecoder)
- decoder.name = b.name
- decoder.valueSize = b.size
- return decoder
-}
-
-func (b *plainDecoderPoolDelegator) Put(decoder SeriesDecoder) {
- _, ok := decoder.(*plainDecoder)
- if ok {
- b.pool.Put(decoder)
- }
-}
-
-var (
- zstdDecoder, _ = zstd.NewReader(nil)
- zstdEncoder, _ = zstd.NewWriter(nil, zstd.WithEncoderLevel(zstd.SpeedBetterCompression))
- _ SeriesEncoder = (*plainEncoder)(nil)
- _ SeriesDecoder = (*plainDecoder)(nil)
-)
-
-// plainEncoder backport to reduced value.
-type plainEncoder struct {
- tsBuff *buffer.Writer
- valBuff *buffer.Writer
- name string
- startTime uint64
- valueSize int
- len uint32
- num uint32
-}
-
-func newPlainEncoder() interface{} {
- return &plainEncoder{
- tsBuff: buffer.NewBufferWriter(&bytes.Buffer{}),
- valBuff: buffer.NewBufferWriter(&bytes.Buffer{}),
- }
-}
-
-func (t *plainEncoder) Append(ts uint64, value []byte) {
- if t.startTime == 0 {
- t.startTime = ts
- } else if t.startTime > ts {
- t.startTime = ts
- }
- vLen := len(value)
- offset := uint32(t.valBuff.Len())
- t.valBuff.PutUint32(uint32(vLen))
- t.valBuff.Write(value)
- t.tsBuff.PutUint64(ts)
- t.tsBuff.PutUint32(offset)
- t.num++
-}
-
-func (t *plainEncoder) IsFull() bool {
- return t.valBuff.Len() >= t.valueSize
-}
-
-func (t *plainEncoder) Reset(_ []byte) {
- t.tsBuff.Reset()
- t.valBuff.Reset()
- t.num = 0
- t.startTime = 0
-}
-
-func (t *plainEncoder) Encode() ([]byte, error) {
- if t.tsBuff.Len() < 1 {
- return nil, errEncodeEmpty
- }
- val := t.valBuff.Bytes()
- t.len = uint32(len(val))
- t.tsBuff.WriteTo(t.valBuff)
- t.valBuff.PutUint32(t.num)
- t.valBuff.PutUint32(t.len)
- data := t.valBuff.Bytes()
- l := len(data)
- dst := make([]byte, 0, compressBound(l))
- dst = zstdEncoder.EncodeAll(data, dst)
- result := buffer.NewBufferWriter(bytes.NewBuffer(make([]byte, 0, len(dst)+2)))
- result.Write(dst)
- result.PutUint16(uint16(l))
- dd := result.Bytes()
- itemsNum.WithLabelValues(t.name, "plain").Inc()
- rawSize.WithLabelValues(t.name, "plain").Add(float64(l))
- encodedSize.WithLabelValues(t.name, "plain").Add(float64(len(dd)))
- return dd, nil
-}
-
-func compressBound(srcSize int) int {
- return srcSize + (srcSize >> 8)
-}
-
-func (t *plainEncoder) StartTime() uint64 {
- return t.startTime
-}
-
-const (
- // TSLen equals ts(uint64) + data_offset(uint32).
- TSLen = 8 + 4
-)
-
-var errInvalidValue = errors.New("invalid encoded value")
-
-// plainDecoder decodes encoded time index.
-type plainDecoder struct {
- name string
- ts []byte
- val []byte
- len uint32
- num uint32
- valueSize int
-}
-
-func (t *plainDecoder) Len() int {
- return int(t.num)
-}
-
-func (t *plainDecoder) Decode(_, rawData []byte) (err error) {
- if len(rawData) < 2 {
- return errInvalidValue
- }
- var data []byte
- size := binary.LittleEndian.Uint16(rawData[len(rawData)-2:])
- if data, err = zstdDecoder.DecodeAll(rawData[:len(rawData)-2], make([]byte, 0, size)); err != nil {
- return errors.Wrap(err, "plain decoder fails to decode")
- }
- l := uint32(len(data))
- if l <= 8 {
- return errInvalidValue
- }
- lenOffset := len(data) - 4
- numOffset := lenOffset - 4
- t.num = binary.LittleEndian.Uint32(data[numOffset:lenOffset])
- t.len = binary.LittleEndian.Uint32(data[lenOffset:])
- if l <= t.len+8 {
- return errInvalidValue
- }
- t.val = data[:t.len]
- t.ts = data[t.len:numOffset]
- return nil
-}
-
-func (t *plainDecoder) IsFull() bool {
- return int(t.len) >= t.valueSize
-}
-
-func (t *plainDecoder) Get(ts uint64) ([]byte, error) {
- i := sort.Search(int(t.num), func(i int) bool {
- slot := getTSSlot(t.ts, i)
- return parseTS(slot) <= ts
- })
- if i >= int(t.num) {
- return nil, fmt.Errorf("%d doesn't exist", ts)
- }
- slot := getTSSlot(t.ts, i)
- if parseTS(slot) != ts {
- return nil, fmt.Errorf("%d is wrong", ts)
- }
- return getVal(t.val, parseOffset(slot))
-}
-
-func (t *plainDecoder) Range() (start, end uint64) {
- startSlot := getTSSlot(t.ts, int(t.num)-1)
- endSlot := getTSSlot(t.ts, 0)
- return parseTS(startSlot), parseTS(endSlot)
-}
-
-func (t *plainDecoder) Iterator() SeriesIterator {
- return newBlockItemIterator(t)
-}
-
-func getVal(buf []byte, offset uint32) ([]byte, error) {
- if uint32(len(buf)) <= offset+4 {
- return nil, errInvalidValue
- }
- dataLen := binary.LittleEndian.Uint32(buf[offset : offset+4])
- return buf[offset+4 : offset+4+dataLen], nil
-}
-
-func getTSSlot(data []byte, index int) []byte {
- return data[index*TSLen : (index+1)*TSLen]
-}
-
-func parseTS(tsSlot []byte) uint64 {
- return binary.LittleEndian.Uint64(tsSlot[:8])
-}
-
-func parseOffset(tsSlot []byte) uint32 {
- return binary.LittleEndian.Uint32(tsSlot[8:])
-}
-
-var _ SeriesIterator = (*plainIterator)(nil)
-
-type plainIterator struct {
- index []byte
- data []byte
- idx int
- num int
-}
-
-func newBlockItemIterator(decoder *plainDecoder) SeriesIterator {
- return &plainIterator{
- idx: -1,
- index: decoder.ts,
- data: decoder.val,
- num: int(decoder.num),
- }
-}
-
-func (b *plainIterator) Next() bool {
- b.idx++
- return b.idx >= 0 && b.idx < b.num
-}
-
-func (b *plainIterator) Val() []byte {
- v, _ := getVal(b.data, parseOffset(getTSSlot(b.index, b.idx)))
- return v
-}
-
-func (b *plainIterator) Time() uint64 {
- return parseTS(getTSSlot(b.index, b.idx))
-}
-
-func (b *plainIterator) Error() error {
- return nil
-}
diff --git a/pkg/bit/reader.go b/pkg/encoding/reader.go
similarity index 99%
rename from pkg/bit/reader.go
rename to pkg/encoding/reader.go
index 6174728..e77807b 100644
--- a/pkg/bit/reader.go
+++ b/pkg/encoding/reader.go
@@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-package bit
+package encoding
import (
"io"
diff --git a/pkg/bit/reader_test.go b/pkg/encoding/reader_test.go
similarity index 98%
rename from pkg/bit/reader_test.go
rename to pkg/encoding/reader_test.go
index 736e8c4..a334486 100644
--- a/pkg/bit/reader_test.go
+++ b/pkg/encoding/reader_test.go
@@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-package bit
+package encoding
import (
"bytes"
diff --git a/pkg/bit/writer.go b/pkg/encoding/writer.go
similarity index 64%
rename from pkg/bit/writer.go
rename to pkg/encoding/writer.go
index a7ed564..6099f0e 100644
--- a/pkg/bit/writer.go
+++ b/pkg/encoding/writer.go
@@ -15,33 +15,27 @@
// specific language governing permissions and limitations
// under the License.
-package bit
+package encoding
import (
- "bytes"
+ "encoding/binary"
)
// Writer writes bits to an io.BufferWriter.
type Writer struct {
- out *bytes.Buffer
+ out BufferWriter
cache byte
available byte
}
// NewWriter create bit writer.
-func NewWriter(buffer *bytes.Buffer) *Writer {
- bw := new(Writer)
- bw.Reset(buffer)
- return bw
+func NewWriter() *Writer {
+ return new(Writer)
}
// Reset writes to a new writer.
-func (w *Writer) Reset(buffer *bytes.Buffer) {
- if buffer == nil {
- w.out.Reset()
- } else {
- w.out = buffer
- }
+func (w *Writer) Reset(buffer BufferWriter) {
+ w.out = buffer
w.cache = 0
w.available = 8
}
@@ -97,3 +91,45 @@ func (w *Writer) Flush() {
}
w.Reset(w.out)
}
+
+// Packer writes data into a buffer.
+type Packer struct {
+ buf BufferWriter
+
+ scratch [binary.MaxVarintLen64]byte
+}
+
+// NewPacker returns a new Writer.
+func NewPacker(buf BufferWriter) *Packer {
+ return &Packer{
+ buf: buf,
+ }
+}
+
+// Write binaries to the buffer.
+func (w *Packer) Write(p []byte) {
+ _, _ = w.buf.Write(p)
+}
+
+// PutUint16 puts uint16 data into the buffer.
+func (w *Packer) PutUint16(v uint16) {
+ binary.LittleEndian.PutUint16(w.scratch[:], v)
+ _, _ = w.buf.Write(w.scratch[:2])
+}
+
+// PutUint32 puts uint32 data into the buffer.
+func (w *Packer) PutUint32(v uint32) {
+ binary.LittleEndian.PutUint32(w.scratch[:], v)
+ _, _ = w.buf.Write(w.scratch[:4])
+}
+
+// PutUint64 puts uint64 data into the buffer.
+func (w *Packer) PutUint64(v uint64) {
+ binary.LittleEndian.PutUint64(w.scratch[:], v)
+ _, _ = w.buf.Write(w.scratch[:8])
+}
+
+// Bytes outputs the data in the buffer.
+func (w *Packer) Bytes() []byte {
+ return w.buf.Bytes()
+}
diff --git a/pkg/bit/writer_test.go b/pkg/encoding/writer_test.go
similarity index 96%
rename from pkg/bit/writer_test.go
rename to pkg/encoding/writer_test.go
index 7624ff8..a93acf4 100644
--- a/pkg/bit/writer_test.go
+++ b/pkg/encoding/writer_test.go
@@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-package bit
+package encoding
import (
"bytes"
@@ -26,7 +26,8 @@ import (
func TestWriter(t *testing.T) {
out := &bytes.Buffer{}
- w := NewWriter(out)
+ w := NewWriter()
+ w.Reset(out)
a := assert.New(t)
diff --git a/pkg/encoding/xor.go b/pkg/encoding/xor.go
index 776722b..d80b09c 100644
--- a/pkg/encoding/xor.go
+++ b/pkg/encoding/xor.go
@@ -19,8 +19,6 @@ package encoding
import (
"math/bits"
-
- "github.com/apache/skywalking-banyandb/pkg/bit"
)
const (
@@ -31,7 +29,7 @@ const (
// XOREncoder intends to compress uint64 data
// https://www.vldb.org/pvldb/vol8/p1816-teller.pdf
type XOREncoder struct {
- bw *bit.Writer
+ bw *Writer
preVal uint64
leading int
trailing int
@@ -40,7 +38,7 @@ type XOREncoder struct {
}
// NewXOREncoder creates xor zstdEncoder for compressing uint64 data.
-func NewXOREncoder(bw *bit.Writer) *XOREncoder {
+func NewXOREncoder(bw *Writer) *XOREncoder {
return &XOREncoder{
bw: bw,
first: true,
@@ -85,7 +83,7 @@ func (e *XOREncoder) Write(val uint64) {
// XORDecoder decodes buffer to uint64 values using xor compress.
type XORDecoder struct {
err error
- br *bit.Reader
+ br *Reader
val uint64
leading uint64
trailing uint64
@@ -93,7 +91,7 @@ type XORDecoder struct {
}
// NewXORDecoder create zstdDecoder decompress buffer using xor.
-func NewXORDecoder(br *bit.Reader) *XORDecoder {
+func NewXORDecoder(br *Reader) *XORDecoder {
s := &XORDecoder{
br: br,
first: true,
diff --git a/pkg/encoding/xor_test.go b/pkg/encoding/xor_test.go
index 282ab57..b23a673 100644
--- a/pkg/encoding/xor_test.go
+++ b/pkg/encoding/xor_test.go
@@ -24,13 +24,12 @@ import (
"github.com/pkg/errors"
"github.com/stretchr/testify/assert"
-
- "github.com/apache/skywalking-banyandb/pkg/bit"
)
func TestXOR(t *testing.T) {
var buf bytes.Buffer
- bitWriter := bit.NewWriter(&buf)
+ bitWriter := NewWriter()
+ bitWriter.Reset(&buf)
e := NewXOREncoder(bitWriter)
e.Write(uint64(76))
e.Write(uint64(50))
@@ -41,7 +40,7 @@ func TestXOR(t *testing.T) {
bitWriter.Flush()
data := buf.Bytes()
- reader := bit.NewReader(bytes.NewReader(data))
+ reader := NewReader(bytes.NewReader(data))
d := NewXORDecoder(reader)
a := assert.New(t)
verify(d, a, uint64(76))
diff --git a/pkg/index/lsm/lsm.go b/pkg/index/lsm/lsm.go
index 55721f4..73adf6b 100644
--- a/pkg/index/lsm/lsm.go
+++ b/pkg/index/lsm/lsm.go
@@ -64,7 +64,6 @@ func NewStore(opts StoreOpts) (index.Store, error) {
var err error
var lsm kv.Store
if lsm, err = kv.OpenStore(
- 0,
opts.Path+"/lsm",
kv.StoreWithLogger(opts.Logger),
kv.StoreWithMemTableSize(opts.MemTableSize)); err != nil {
diff --git a/pkg/pb/v1/tsdb/interval.go b/pkg/pb/v1/tsdb/options.go
similarity index 91%
rename from pkg/pb/v1/tsdb/interval.go
rename to pkg/pb/v1/tsdb/options.go
index 75f9f83..6cd3f9c 100644
--- a/pkg/pb/v1/tsdb/interval.go
+++ b/pkg/pb/v1/tsdb/options.go
@@ -19,13 +19,13 @@
package tsdb
import (
- "errors"
+ "github.com/pkg/errors"
commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
"github.com/apache/skywalking-banyandb/banyand/tsdb"
)
-var errInvalidUnit = errors.New("invalid interval rule's unit")
+var errUnknown = errors.New("unknown")
// ToIntervalRule decodes commonv1.IntervalRule to tsdb.IntervalRule.
func ToIntervalRule(ir *commonv1.IntervalRule) (result tsdb.IntervalRule, err error) {
@@ -35,7 +35,7 @@ func ToIntervalRule(ir *commonv1.IntervalRule) (result tsdb.IntervalRule, err er
case commonv1.IntervalRule_UNIT_HOUR:
result.Unit = tsdb.HOUR
default:
- return result, errInvalidUnit
+ return result, errors.WithMessagef(errUnknown, "interval rule:%v", ir)
}
result.Num = int(ir.Num)
return result, err
diff --git a/test/stress/env b/test/stress/env
index 007f541..1ff3580 100644
--- a/test/stress/env
+++ b/test/stress/env
@@ -22,10 +22,11 @@ SW_AGENT_GO_COMMIT=4af380c2db6243106b0fc650b6003ce3b3eb82a0
SW_AGENT_PYTHON_COMMIT=c76a6ec51a478ac91abb20ec8f22a99b8d4d6a58
SW_AGENT_CLIENT_JS_COMMIT=af0565a67d382b683c1dbd94c379b7080db61449
SW_AGENT_CLIENT_JS_TEST_COMMIT=4f1eb1dcdbde3ec4a38534bf01dded4ab5d2f016
-SW_KUBERNETES_COMMIT_SHA=0f3ec68e5a7e1608cec8688716b848ed15e971e5
+SW_KUBERNETES_COMMIT_SHA=b670c41d94a82ddefcf466d54bab5c492d88d772
SW_ROVER_COMMIT=d956eaede57b62108b78bca48045bd09ba88e653
SW_CTL_COMMIT=e684fae0107045fc23799146d62f04cb68bd5a3b
-SW_OAP_COMMIT=d5388683322ee6a4aed2a3bc29d439aadfca9a04
+
+SW_OAP_COMMIT=1335a48f1c034abc1fe24f6197ee7acfa3118bf0
SW_AGENT_E2E_SERVICE_PROVIDER_COMMIT=828e6e2f2b57a0f06bb0d507e3296d2377943d9a
TARGET=test
diff --git a/test/stress/env.dev b/test/stress/env.dev
index add472c..604ded5 100644
--- a/test/stress/env.dev
+++ b/test/stress/env.dev
@@ -21,10 +21,11 @@ SW_AGENT_GO_COMMIT=4af380c2db6243106b0fc650b6003ce3b3eb82a0
SW_AGENT_PYTHON_COMMIT=c76a6ec51a478ac91abb20ec8f22a99b8d4d6a58
SW_AGENT_CLIENT_JS_COMMIT=af0565a67d382b683c1dbd94c379b7080db61449
SW_AGENT_CLIENT_JS_TEST_COMMIT=4f1eb1dcdbde3ec4a38534bf01dded4ab5d2f016
-SW_KUBERNETES_COMMIT_SHA=0f3ec68e5a7e1608cec8688716b848ed15e971e5
+SW_KUBERNETES_COMMIT_SHA=b670c41d94a82ddefcf466d54bab5c492d88d772
SW_ROVER_COMMIT=d956eaede57b62108b78bca48045bd09ba88e653
SW_CTL_COMMIT=e684fae0107045fc23799146d62f04cb68bd5a3b
-SW_OAP_COMMIT=d5388683322ee6a4aed2a3bc29d439aadfca9a04
+
+SW_OAP_COMMIT=1335a48f1c034abc1fe24f6197ee7acfa3118bf0
SW_AGENT_E2E_SERVICE_PROVIDER_COMMIT=828e6e2f2b57a0f06bb0d507e3296d2377943d9a
TARGET=dev