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 <[email protected]> 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 <[email protected]> --- 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)
