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/28 13:46:20 UTC
[skywalking-banyandb] 01/01: Fix several flaws:
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch kv
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit 5de880c47e5ccaa6eba242c650a204bb332a64cf
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Wed Dec 28 13:36:53 2022 +0000
Fix several flaws:
* Fix memory overhead on building L0 and compacting.
* Increase the default limitation of querying a measure.
* Pass the raw key to the encoder pool.
* Parameterize the decoder's buffer size.
* Some chroes.
Signed-off-by: Gao Hongtao <ha...@gmail.com>
---
banyand/kv/badger.go | 25 +++----------------------
banyand/kv/kv.go | 6 +++---
banyand/measure/measure.go | 3 ++-
banyand/measure/metadata.go | 5 +++--
banyand/tsdb/block.go | 3 ++-
banyand/tsdb/tsdb.go | 5 +++--
dist/LICENSE | 2 +-
go.mod | 2 +-
go.sum | 4 ++--
pkg/index/lsm/iterator.go | 3 ---
pkg/index/lsm/lsm.go | 2 +-
pkg/query/logical/common.go | 3 ---
pkg/query/logical/measure/measure_analyzer.go | 4 +++-
pkg/query/logical/stream/stream_analyzer.go | 4 +++-
14 files changed, 27 insertions(+), 44 deletions(-)
diff --git a/banyand/kv/badger.go b/banyand/kv/badger.go
index f5a4489..7f616d2 100644
--- a/banyand/kv/badger.go
+++ b/banyand/kv/badger.go
@@ -44,28 +44,9 @@ var (
)
type badgerTSS struct {
- dbOpts badger.Options
badger.TSet
- db *badger.DB
-}
-
-func (b *badgerTSS) Context(key []byte, ts uint64, n int) (pre Iterator, next Iterator) {
- preOpts := badger.DefaultIteratorOptions
- preOpts.PrefetchSize = n
- preOpts.PrefetchValues = false
- preOpts.Prefix = key
- preOpts.Reverse = false
- nextOpts := badger.DefaultIteratorOptions
- nextOpts.PrefetchSize = n
- nextOpts.PrefetchValues = false
- nextOpts.Prefix = key
- nextOpts.Reverse = true
- seekKey := y.KeyWithTs(key, ts)
- preIter := b.db.NewIterator(preOpts)
- preIter.Seek(seekKey)
- nextIter := b.db.NewIterator(nextOpts)
- nextIter.Seek(seekKey)
- return &iterator{delegated: preIter}, &iterator{delegated: nextIter, reverse: true}
+ db *badger.DB
+ dbOpts badger.Options
}
func (b *badgerTSS) Stats() (s observability.Statistics) {
@@ -138,8 +119,8 @@ func (i mergedIter) Value() y.ValueStruct {
}
type badgerDB struct {
- dbOpts badger.Options
db *badger.DB
+ dbOpts badger.Options
}
func (b *badgerDB) Stats() observability.Statistics {
diff --git a/banyand/kv/kv.go b/banyand/kv/kv.go
index e3a3f35..8f85432 100644
--- a/banyand/kv/kv.go
+++ b/banyand/kv/kv.go
@@ -89,7 +89,6 @@ type TimeSeriesWriter interface {
type TimeSeriesReader interface {
// Get a value by its key and timestamp/version
Get(key []byte, ts uint64) ([]byte, error)
- Context(key []byte, ts uint64, n int) (pre, next Iterator)
}
// TimeSeriesStore is time series storage.
@@ -115,7 +114,7 @@ func TSSWithLogger(l *logger.Logger) TimeSeriesOptions {
}
// TSSWithEncoding sets encoding and decoding pools for building chunks.
-func TSSWithEncoding(encoderPool encoding.SeriesEncoderPool, decoderPool encoding.SeriesDecoderPool) TimeSeriesOptions {
+func TSSWithEncoding(encoderPool encoding.SeriesEncoderPool, decoderPool encoding.SeriesDecoderPool, chunkSize int) TimeSeriesOptions {
return func(store TimeSeriesStore) {
if btss, ok := store.(*badgerTSS); ok {
btss.dbOpts = btss.dbOpts.WithKeyBasedEncoder(
@@ -123,7 +122,7 @@ func TSSWithEncoding(encoderPool encoding.SeriesEncoderPool, decoderPool encodin
encoderPool,
}, &decoderPoolDelegate{
decoderPool,
- })
+ }, chunkSize)
}
}
}
@@ -187,6 +186,7 @@ func OpenTimeSeriesStore(path string, options ...TimeSeriesOptions) (TimeSeriesS
opt(btss)
}
// Put all values into LSM
+ btss.dbOpts = btss.dbOpts.WithNumVersionsToKeep(math.MaxUint32)
btss.dbOpts = btss.dbOpts.WithVLogPercentile(1.0)
if btss.dbOpts.MemTableSize < 8<<20 {
btss.dbOpts = btss.dbOpts.WithValueThreshold(1 << 10)
diff --git a/banyand/measure/measure.go b/banyand/measure/measure.go
index 2099859..4648525 100644
--- a/banyand/measure/measure.go
+++ b/banyand/measure/measure.go
@@ -36,7 +36,8 @@ import (
const (
plainChunkSize = 1 << 20
- intChunkSize = 120
+ intChunkNum = 120
+ intChunkSize = 4 * 1024
)
type measure struct {
diff --git a/banyand/measure/metadata.go b/banyand/measure/metadata.go
index 5361bee..aa16588 100644
--- a/banyand/measure/metadata.go
+++ b/banyand/measure/metadata.go
@@ -221,8 +221,9 @@ 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, intChunkSize, s.l),
- DecoderPool: newDecoderPool(name, intChunkSize, s.l),
+ EncoderPool: newEncoderPool(name, intChunkNum, s.l),
+ DecoderPool: newDecoderPool(name, intChunkNum, s.l),
+ ChunkSizeInBytes: intChunkSize,
}
opts.CompressionMethod = tsdb.CompressionMethod{
Type: tsdb.CompressionTypeZSTD,
diff --git a/banyand/tsdb/block.go b/banyand/tsdb/block.go
index 0f1cd0a..70e94db 100644
--- a/banyand/tsdb/block.go
+++ b/banyand/tsdb/block.go
@@ -134,7 +134,8 @@ func (b *block) options(ctx context.Context) {
options = o.(DatabaseOpts)
}
if options.EncodingMethod.EncoderPool != nil && options.EncodingMethod.DecoderPool != nil {
- b.openOpts.store = append(b.openOpts.store, kv.TSSWithEncoding(options.EncodingMethod.EncoderPool, options.EncodingMethod.DecoderPool))
+ b.openOpts.store = append(b.openOpts.store, kv.TSSWithEncoding(options.EncodingMethod.EncoderPool,
+ options.EncodingMethod.DecoderPool, options.EncodingMethod.ChunkSizeInBytes))
}
if options.CompressionMethod.Type == CompressionTypeZSTD {
b.openOpts.store = append(b.openOpts.store, kv.TSSWithZSTDCompression(options.CompressionMethod.ChunkSizeInBytes))
diff --git a/banyand/tsdb/tsdb.go b/banyand/tsdb/tsdb.go
index fc45804..9e65920 100644
--- a/banyand/tsdb/tsdb.go
+++ b/banyand/tsdb/tsdb.go
@@ -117,8 +117,9 @@ type InvertedIndexOpts struct {
// EncodingMethod wraps encoder/decoder pools to flush/compact data on disk.
type EncodingMethod struct {
- EncoderPool encoding.SeriesEncoderPool
- DecoderPool encoding.SeriesDecoderPool
+ EncoderPool encoding.SeriesEncoderPool
+ DecoderPool encoding.SeriesDecoderPool
+ ChunkSizeInBytes int
}
// CompressionMethod denotes how to compress a single chunk.
diff --git a/dist/LICENSE b/dist/LICENSE
index 5e6915c..a60b3a9 100644
--- a/dist/LICENSE
+++ b/dist/LICENSE
@@ -178,7 +178,7 @@
Apache-2.0 licenses
========================================================================
- github.com/SkyAPM/badger/v3 v3.0.0-20221224092806-866b662cda58 Apache-2.0
+ github.com/SkyAPM/badger/v3 v3.0.0-20221227124922-b88a2f7d336f 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
diff --git a/go.mod b/go.mod
index 1ab479b..249590a 100644
--- a/go.mod
+++ b/go.mod
@@ -136,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-20221224092806-866b662cda58
+ github.com/dgraph-io/badger/v3 v3.2011.1 => github.com/SkyAPM/badger/v3 v3.0.0-20221227124922-b88a2f7d336f
)
diff --git a/go.sum b/go.sum
index cdc2cc1..11c7fbc 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-20221224092806-866b662cda58 h1:Q/XKyXMro7kvpFE0nxkd+iryR9AqnqbA6SXU2Z+RDfw=
-github.com/SkyAPM/badger/v3 v3.0.0-20221224092806-866b662cda58/go.mod h1:4WETftF8A4mEeFgIsYB/VvGo5kfTVl/neYgEqlVW9Ag=
+github.com/SkyAPM/badger/v3 v3.0.0-20221227124922-b88a2f7d336f h1:fcfR1l/cO03QQWnwhlYdgXkjZBV58ZmxwgRJj7HiNQw=
+github.com/SkyAPM/badger/v3 v3.0.0-20221227124922-b88a2f7d336f/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=
diff --git a/pkg/index/lsm/iterator.go b/pkg/index/lsm/iterator.go
index dc82435..642c761 100644
--- a/pkg/index/lsm/iterator.go
+++ b/pkg/index/lsm/iterator.go
@@ -106,9 +106,6 @@ func newFieldIteratorTemplate(l *logger.Logger, fieldKey index.FieldKey, termRan
term = termRange.Upper
reverse = true
}
- if order == modelv1.Sort_SORT_DESC {
- reverse = true
- }
iter := iterable.NewIterator(kv.ScanOpts{
Prefix: fieldKey.Marshal(),
Reverse: reverse,
diff --git a/pkg/index/lsm/lsm.go b/pkg/index/lsm/lsm.go
index 73adf6b..2737d97 100644
--- a/pkg/index/lsm/lsm.go
+++ b/pkg/index/lsm/lsm.go
@@ -64,7 +64,7 @@ func NewStore(opts StoreOpts) (index.Store, error) {
var err error
var lsm kv.Store
if lsm, err = kv.OpenStore(
- opts.Path+"/lsm",
+ opts.Path,
kv.StoreWithLogger(opts.Logger),
kv.StoreWithMemTableSize(opts.MemTableSize)); err != nil {
return nil, err
diff --git a/pkg/query/logical/common.go b/pkg/query/logical/common.go
index 8ddb295..c5c754c 100644
--- a/pkg/query/logical/common.go
+++ b/pkg/query/logical/common.go
@@ -144,9 +144,6 @@ func ExecuteForShard(l *logger.Logger, series tsdb.SeriesList, timeRange timesta
return itersInShard, closers, nil
}
-// DefaultLimit is the default limit size of a querying.
-var DefaultLimit uint32 = 20
-
// Tag represents the combination of tag family and tag name.
// It's a tag's identity.
type Tag struct {
diff --git a/pkg/query/logical/measure/measure_analyzer.go b/pkg/query/logical/measure/measure_analyzer.go
index e516c23..9433065 100644
--- a/pkg/query/logical/measure/measure_analyzer.go
+++ b/pkg/query/logical/measure/measure_analyzer.go
@@ -27,6 +27,8 @@ import (
"github.com/apache/skywalking-banyandb/pkg/query/logical"
)
+const defaultLimit uint32 = 100
+
// BuildSchema returns Schema loaded from the metadata repository.
func BuildSchema(measureSchema measure.Measure) (logical.Schema, error) {
md := measureSchema.GetSchema()
@@ -97,7 +99,7 @@ func Analyze(_ context.Context, criteria *measurev1.QueryRequest, metadata *comm
// parse limit and offset
limitParameter := criteria.GetLimit()
if limitParameter == 0 {
- limitParameter = logical.DefaultLimit
+ limitParameter = defaultLimit
}
plan = limit(plan, criteria.GetOffset(), limitParameter)
diff --git a/pkg/query/logical/stream/stream_analyzer.go b/pkg/query/logical/stream/stream_analyzer.go
index 5a9ee5e..466819e 100644
--- a/pkg/query/logical/stream/stream_analyzer.go
+++ b/pkg/query/logical/stream/stream_analyzer.go
@@ -26,6 +26,8 @@ import (
"github.com/apache/skywalking-banyandb/pkg/query/logical"
)
+const defaultLimit uint32 = 20
+
// BuildSchema returns Schema loaded from the metadata repository.
func BuildSchema(streamSchema stream.Stream) (logical.Schema, error) {
sm := streamSchema.GetSchema()
@@ -68,7 +70,7 @@ func Analyze(_ context.Context, criteria *streamv1.QueryRequest, metadata *commo
// parse limit
limitParameter := criteria.GetLimit()
if limitParameter == 0 {
- limitParameter = logical.DefaultLimit
+ limitParameter = defaultLimit
}
plan = logical.NewLimit(plan, limitParameter)