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/02/23 02:19:51 UTC
[skywalking-banyandb] branch index-iterator created (now fb7c8f7)
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a change to branch index-iterator
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git.
at fb7c8f7 Fix the index iterator overflow
This branch includes the following new commits:
new fb7c8f7 Fix the index iterator overflow
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: Fix the index iterator overflow
Posted by ha...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch index-iterator
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit fb7c8f7c320104443bcffeb3fd5e29847d86d8e9
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Wed Feb 23 02:00:11 2022 +0000
Fix the index iterator overflow
Signed-off-by: Gao Hongtao <ha...@gmail.com>
---
banyand/stream/stream_write.go | 1 +
banyand/tsdb/block.go | 13 ++--
banyand/tsdb/series_write.go | 13 ++++
banyand/tsdb/seriesdb.go | 2 +-
pkg/index/inverted/inverted.go | 5 +-
pkg/index/iterator.go | 138 ++++++++++++++++++++++++++++++-----------
pkg/index/lsm/lsm.go | 2 +
pkg/index/lsm/search.go | 37 ++++++-----
pkg/query/logical/common.go | 9 +++
pkg/run/run.go | 3 +-
10 files changed, 163 insertions(+), 60 deletions(-)
diff --git a/banyand/stream/stream_write.go b/banyand/stream/stream_write.go
index 83c4094..10fcc74 100644
--- a/banyand/stream/stream_write.go
+++ b/banyand/stream/stream_write.go
@@ -111,6 +111,7 @@ func (s *stream) write(shardID common.ShardID, seriesHashKey []byte, value *stre
Int("ts_nano", t.Nanosecond()).
Interface("data", value).
Uint64("series_id", uint64(series.ID())).
+ Uint64("item_id", uint64(writer.ItemID().ID)).
Int("shard_id", int(shardID)).
Str("stream", sm.Metadata.GetName()).
Msg("write stream")
diff --git a/banyand/tsdb/block.go b/banyand/tsdb/block.go
index 77a7ad3..42a1a03 100644
--- a/banyand/tsdb/block.go
+++ b/banyand/tsdb/block.go
@@ -103,26 +103,26 @@ func (b *block) open() (err error) {
0,
b.path+"/store",
kv.TSSWithEncoding(b.encodingMethod.EncoderPool, b.encodingMethod.DecoderPool),
- kv.TSSWithLogger(b.l),
+ kv.TSSWithLogger(b.l.Named("main-store")),
); err != nil {
return err
}
if b.primaryIndex, err = lsm.NewStore(lsm.StoreOpts{
Path: b.path + "/primary",
- Logger: b.l,
+ Logger: b.l.Named("primary-lsm"),
}); err != nil {
return err
}
b.closableLst = append(b.closableLst, b.store, b.primaryIndex)
if b.invertedIndex, err = inverted.NewStore(inverted.StoreOpts{
Path: b.path + "/inverted",
- Logger: b.l,
+ Logger: b.l.Named("secondary-inverted"),
}); err != nil {
return err
}
if b.lsmIndex, err = lsm.NewStore(lsm.StoreOpts{
Path: b.path + "/lsm",
- Logger: b.l,
+ Logger: b.l.Named("secondary-lsm"),
}); err != nil {
return err
}
@@ -186,6 +186,7 @@ type blockDelegate interface {
primaryIndexReader() index.Searcher
identity() (segID uint16, blockID uint16)
startTime() time.Time
+ String() string
}
var _ blockDelegate = (*bDelegate)(nil)
@@ -244,6 +245,10 @@ func (d *bDelegate) contains(ts time.Time) bool {
return d.delegate.Contains(uint64(ts.UnixNano()))
}
+func (d *bDelegate) String() string {
+ return d.delegate.String()
+}
+
func (d *bDelegate) Close() error {
d.delegate.dscRef()
return nil
diff --git a/banyand/tsdb/series_write.go b/banyand/tsdb/series_write.go
index ec14e33..9d8d97d 100644
--- a/banyand/tsdb/series_write.go
+++ b/banyand/tsdb/series_write.go
@@ -19,6 +19,7 @@ package tsdb
import (
"bytes"
+ "fmt"
"time"
"github.com/pkg/errors"
@@ -39,6 +40,7 @@ type Writer interface {
IndexWriter
Write() (GlobalItemID, error)
ItemID() GlobalItemID
+ String() string
}
var _ WriterBuilder = (*writerBuilder)(nil)
@@ -151,6 +153,17 @@ func (w *writer) WriteInvertedIndex(field index.Field) error {
return w.block.writeInvertedIndex(field, w.itemID.ID)
}
+func (w *writer) String() string {
+ var buf []byte
+ buf = append(buf, "block:"...)
+ buf = append(buf, w.block.String()...)
+ buf = append(buf, ",column:"...)
+ buf = append(buf, fmt.Sprintf("%+v", w.columns)...)
+ buf = append(buf, ",ts:"...)
+ buf = append(buf, w.ts.String()...)
+ return string(buf)
+}
+
type dataBucket struct {
seriesID common.SeriesID
family []byte
diff --git a/banyand/tsdb/seriesdb.go b/banyand/tsdb/seriesdb.go
index 4c36389..4a8f745 100644
--- a/banyand/tsdb/seriesdb.go
+++ b/banyand/tsdb/seriesdb.go
@@ -191,7 +191,7 @@ func (s *seriesDB) List(path Path) (SeriesList, error) {
s.l.Debug().
Hex("path", path.prefix).
Uint64("series_id", uint64(seriesID)).
- Msg("got a series")
+ Msg("got a series with a full path")
return []Series{newSeries(s.context(), seriesID, s)}, nil
}
s.l.Debug().Hex("path", path.prefix).Msg("doesn't get any series")
diff --git a/pkg/index/inverted/inverted.go b/pkg/index/inverted/inverted.go
index 7c19920..7387b63 100644
--- a/pkg/index/inverted/inverted.go
+++ b/pkg/index/inverted/inverted.go
@@ -42,6 +42,8 @@ type store struct {
memTable *memTable
immutableMemTable *memTable
rwMutex sync.RWMutex
+
+ l *logger.Logger
}
type StoreOpts struct {
@@ -65,6 +67,7 @@ func NewStore(opts StoreOpts) (index.Store, error) {
memTable: newMemTable(),
diskTable: diskTable,
termMetadata: md,
+ l: opts.Logger,
}, nil
}
@@ -166,7 +169,7 @@ func (s *store) Iterator(fieldKey index.FieldKey, termRange index.RangeOpts,
}
iters = append(iters, it)
}
- it, err := index.NewFieldIteratorTemplate(fieldKey, termRange, order, s.diskTable, s.termMetadata,
+ it, err := index.NewFieldIteratorTemplate(s.l, fieldKey, termRange, order, s.diskTable, s.termMetadata,
func(term, val []byte, delegated kv.Iterator) (*index.PostingValue, error) {
list := roaring.NewPostingList()
err := list.Unmarshall(val)
diff --git a/pkg/index/iterator.go b/pkg/index/iterator.go
index a933c0d..beaf15a 100644
--- a/pkg/index/iterator.go
+++ b/pkg/index/iterator.go
@@ -27,6 +27,7 @@ import (
"github.com/apache/skywalking-banyandb/banyand/kv"
"github.com/apache/skywalking-banyandb/pkg/convert"
"github.com/apache/skywalking-banyandb/pkg/index/metadata"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
)
type CompositePostingValueFn = func(term, value []byte, delegated kv.Iterator) (*PostingValue, error)
@@ -36,17 +37,15 @@ var DefaultUpper = convert.Uint64ToBytes(math.MaxUint64)
var DefaultLower = convert.Uint64ToBytes(0)
type FieldIteratorTemplate struct {
- delegated kv.Iterator
-
- init bool
- curr *PostingValue
- err error
- termRange RangeOpts
- fn CompositePostingValueFn
- reverse bool
- seekKey []byte
- fieldKeyBytes []byte
- parseKeyFn func(key []byte) (Field, error)
+ delegated *delegateIterator
+
+ init bool
+ cur *PostingValue
+ err error
+ termRange RangeOpts
+ fn CompositePostingValueFn
+ reverse bool
+ seekKey []byte
}
func (f *FieldIteratorTemplate) Next() bool {
@@ -57,15 +56,7 @@ func (f *FieldIteratorTemplate) Next() bool {
if !f.delegated.Valid() {
return false
}
- field, err := f.parseKeyFn(f.delegated.Key())
- if err != nil {
- f.err = err
- return false
- }
- if !bytes.Equal(field.Key.Marshal(), f.fieldKeyBytes) {
- return false
- }
- pv, err := f.fn(field.Term, f.delegated.Val(), f.delegated)
+ pv, err := f.fn(f.delegated.Field().Term, f.delegated.Val(), f.delegated)
if err != nil {
f.err = err
return false
@@ -83,19 +74,19 @@ func (f *FieldIteratorTemplate) Next() bool {
}
return f.Next()
}
- f.curr = pv
+ f.cur = pv
return true
}
func (f *FieldIteratorTemplate) Val() *PostingValue {
- return f.curr
+ return f.cur
}
func (f *FieldIteratorTemplate) Close() error {
return multierr.Append(f.err, f.delegated.Close())
}
-func NewFieldIteratorTemplate(fieldKey FieldKey, termRange RangeOpts, order modelv1.Sort, iterable kv.Iterable,
+func NewFieldIteratorTemplate(l *logger.Logger, fieldKey FieldKey, termRange RangeOpts, order modelv1.Sort, iterable kv.Iterable,
metadata metadata.Term, fn CompositePostingValueFn) (*FieldIteratorTemplate, error) {
if termRange.Upper == nil {
termRange.Upper = DefaultUpper
@@ -129,25 +120,25 @@ func NewFieldIteratorTemplate(fieldKey FieldKey, termRange RangeOpts, order mode
return nil, err
}
return &FieldIteratorTemplate{
- delegated: iter,
+ delegated: newDelegateIterator(iter, fieldKey, metadata, l),
termRange: termRange,
fn: fn,
reverse: reverse,
- parseKeyFn: func(key []byte) (Field, error) {
- f := &Field{
- Key: fieldKey,
- }
- errParseKey := f.Unmarshal(metadata, key)
- if errParseKey != nil {
- return *f, err
- }
- return *f, nil
- },
- seekKey: seekKey,
- fieldKeyBytes: fieldKey.Marshal(),
+ seekKey: seekKey,
}, nil
}
+func parseKey(fieldKey FieldKey, metadata metadata.Term, key []byte) (Field, error) {
+ f := &Field{
+ Key: fieldKey,
+ }
+ err := f.Unmarshal(metadata, key)
+ if err != nil {
+ return *f, err
+ }
+ return *f, nil
+}
+
type SwitchFn = func(a, b []byte) bool
var _ FieldIterator = (*mergedIterator)(nil)
@@ -234,3 +225,78 @@ func (m *mergedIterator) drain(index int) {
func (m *mergedIterator) allDrained() bool {
return m.drainedCount == len(m.inner)
}
+
+var _ kv.Iterator = (*delegateIterator)(nil)
+
+type delegateIterator struct {
+ delegated kv.Iterator
+ fieldKey FieldKey
+ fieldKeyBytes []byte
+ metadata metadata.Term
+ l *logger.Logger
+
+ curField Field
+ closed bool
+}
+
+func newDelegateIterator(delegated kv.Iterator, fieldKey FieldKey, metadata metadata.Term, l *logger.Logger) *delegateIterator {
+ fieldKeyBytes := fieldKey.Marshal()
+ return &delegateIterator{
+ delegated: delegated,
+ fieldKey: fieldKey,
+ fieldKeyBytes: fieldKeyBytes,
+ metadata: metadata,
+ l: l,
+ }
+}
+
+func (di *delegateIterator) Next() {
+ di.delegated.Next()
+}
+
+func (di *delegateIterator) Rewind() {
+ di.delegated.Rewind()
+}
+
+func (di *delegateIterator) Seek(key []byte) {
+ di.delegated.Seek(key)
+}
+
+func (di *delegateIterator) Key() []byte {
+ return di.delegated.Key()
+}
+
+func (di *delegateIterator) Field() Field {
+ return di.curField
+}
+
+func (di *delegateIterator) Val() []byte {
+ return di.delegated.Val()
+}
+
+func (di *delegateIterator) Valid() bool {
+ if di.closed || !di.delegated.Valid() {
+ return false
+ }
+ var err error
+ di.curField, err = parseKey(di.fieldKey, di.metadata, di.Key())
+ if err != nil {
+ di.l.Error().Err(err).Msg("fail to parse field from key")
+ di.Close()
+ return false
+ }
+ if !bytes.Equal(di.curField.Key.Marshal(), di.fieldKeyBytes) {
+ di.l.Debug().
+ Uint64("series_id", uint64(di.fieldKey.SeriesID)).
+ Uint32("index_rule_id", di.fieldKey.IndexRuleID).
+ Msg("reached the limitation of the field(series_id+index_rule_id)")
+ di.Close()
+ return false
+ }
+ return true
+}
+
+func (di *delegateIterator) Close() error {
+ di.closed = true
+ return nil
+}
diff --git a/pkg/index/lsm/lsm.go b/pkg/index/lsm/lsm.go
index 3a7cddd..58f8f93 100644
--- a/pkg/index/lsm/lsm.go
+++ b/pkg/index/lsm/lsm.go
@@ -33,6 +33,7 @@ var _ index.Store = (*store)(nil)
type store struct {
lsm kv.Store
termMetadata metadata.Term
+ l *logger.Logger
}
func (s *store) Close() error {
@@ -69,5 +70,6 @@ func NewStore(opts StoreOpts) (index.Store, error) {
return &store{
lsm: lsm,
termMetadata: md,
+ l: opts.Logger,
}, nil
}
diff --git a/pkg/index/lsm/search.go b/pkg/index/lsm/search.go
index 69a84d9..43f1948 100644
--- a/pkg/index/lsm/search.go
+++ b/pkg/index/lsm/search.go
@@ -66,23 +66,26 @@ func (s *store) Range(fieldKey index.FieldKey, opts index.RangeOpts) (list posti
}
func (s *store) Iterator(fieldKey index.FieldKey, termRange index.RangeOpts, order modelv1.Sort) (index.FieldIterator, error) {
- return index.NewFieldIteratorTemplate(fieldKey, termRange, order, s.lsm, s.termMetadata, func(term, value []byte, delegated kv.Iterator) (*index.PostingValue, error) {
- pv := &index.PostingValue{
- Term: term,
- Value: roaring.NewPostingListWithInitialData(convert.BytesToUint64(value)),
- }
-
- for ; delegated.Valid(); delegated.Next() {
- f := index.Field{}
- err := f.Unmarshal(s.termMetadata, delegated.Key())
- if err != nil {
- return nil, err
+ return index.NewFieldIteratorTemplate(s.l, fieldKey, termRange, order, s.lsm, s.termMetadata,
+ func(term, value []byte, delegated kv.Iterator) (*index.PostingValue, error) {
+ pv := &index.PostingValue{
+ Term: term,
+ Value: roaring.NewPostingListWithInitialData(convert.BytesToUint64(value)),
}
- if !bytes.Equal(f.Term, term) {
- break
+
+ for ; delegated.Valid(); delegated.Next() {
+ f := index.Field{}
+ err := f.Unmarshal(s.termMetadata, delegated.Key())
+ if err != nil {
+ return nil, err
+ }
+ if !bytes.Equal(f.Term, term) {
+ break
+ }
+ itemID := convert.BytesToUint64(delegated.Val())
+ s.l.Debug().Uint64("item_id", itemID).Msg("add item id")
+ pv.Value.Insert(common.ItemID(itemID))
}
- pv.Value.Insert(common.ItemID(convert.BytesToUint64(delegated.Val())))
- }
- return pv, nil
- })
+ return pv, nil
+ })
}
diff --git a/pkg/query/logical/common.go b/pkg/query/logical/common.go
index 33b12f9..2960c7b 100644
--- a/pkg/query/logical/common.go
+++ b/pkg/query/logical/common.go
@@ -20,12 +20,16 @@ package logical
import (
"bytes"
+ "github.com/pkg/errors"
+
modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
"github.com/apache/skywalking-banyandb/banyand/tsdb"
"github.com/apache/skywalking-banyandb/pkg/query/executor"
"github.com/apache/skywalking-banyandb/pkg/timestamp"
)
+var ErrInvalidData = errors.New("data is invalid")
+
type (
seekerBuilder func(builder tsdb.SeekerBuilder)
comparator func(a, b tsdb.Item) bool
@@ -57,6 +61,11 @@ func projectItem(ec executor.ExecutionContext, item tsdb.Item, projectionFieldRe
if err != nil {
return nil, err
}
+ if len(refs) > len(parsedTagFamily.Tags) {
+ return nil, errors.Wrapf(ErrInvalidData,
+ "the number of tags %d in %s is less then expected %d",
+ len(parsedTagFamily.Tags), familyName, len(refs))
+ }
for j, ref := range refs {
tags[j] = parsedTagFamily.GetTags()[ref.Spec.TagIdx]
}
diff --git a/pkg/run/run.go b/pkg/run/run.go
index f630b65..f5b4156 100644
--- a/pkg/run/run.go
+++ b/pkg/run/run.go
@@ -25,6 +25,7 @@ import (
"time"
"github.com/oklog/run"
+ "github.com/pkg/errors"
"github.com/spf13/pflag"
"go.uber.org/multierr"
@@ -231,7 +232,7 @@ func (g *Group) RunConfig() (interrupted bool, err error) {
// Load config from env and file
if err = config.Load(g.f.Name, g.f.FlagSet); err != nil {
- return false, err
+ return false, errors.Wrapf(err, "%s fails to load config", g.f.Name)
}
// bail early on help or version requests