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 2023/12/19 05:30:40 UTC
(skywalking-banyandb) 01/02: Remove single point index writer
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch storage-column
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit cdeebb687f1a14a9f1635f412de94e9884126c98
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Tue Dec 19 05:05:47 2023 +0000
Remove single point index writer
Signed-off-by: Gao Hongtao <ha...@gmail.com>
---
banyand/internal/storage/index.go | 24 -------------------
banyand/internal/storage/index_test.go | 37 ++++++++++++++++++------------
banyand/internal/storage/storage.go | 1 -
banyand/internal/storage/tsdb.go | 9 --------
pkg/index/index.go | 1 -
pkg/index/inverted/inverted.go | 13 +----------
pkg/index/inverted/inverted_series.go | 13 -----------
pkg/index/inverted/inverted_series_test.go | 17 ++++++--------
pkg/index/inverted/inverted_test.go | 2 +-
pkg/pb/v1/value_test.go | 10 +-------
10 files changed, 32 insertions(+), 95 deletions(-)
diff --git a/banyand/internal/storage/index.go b/banyand/internal/storage/index.go
index ecca0736..d44b172c 100644
--- a/banyand/internal/storage/index.go
+++ b/banyand/internal/storage/index.go
@@ -61,30 +61,6 @@ func newSeriesIndex(ctx context.Context, root string) (*seriesIndex, error) {
return si, nil
}
-var entityKey = index.FieldKey{}
-
-func (s *seriesIndex) createPrimary(series *pbv1.Series) (*pbv1.Series, error) {
- if err := series.Marshal(); err != nil {
- return nil, err
- }
- id, err := s.store.Search(series.Buffer)
- if err != nil {
- return nil, err
- }
- if id > 0 {
- return series, nil
- }
- evv := make([]byte, len(series.Buffer))
- copy(evv, series.Buffer)
- if err := s.store.Create(index.Series{
- ID: series.ID,
- EntityValues: evv,
- }); err != nil {
- return nil, errors.WithMessagef(err, "create entity values -> seriesID: %s", string(evv))
- }
- return series, nil
-}
-
func (s *seriesIndex) Write(docs index.Documents) error {
return s.store.Batch(docs)
}
diff --git a/banyand/internal/storage/index_test.go b/banyand/internal/storage/index_test.go
index 97280850..ad435f07 100644
--- a/banyand/internal/storage/index_test.go
+++ b/banyand/internal/storage/index_test.go
@@ -26,6 +26,7 @@ import (
"github.com/stretchr/testify/require"
modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
+ "github.com/apache/skywalking-banyandb/pkg/index"
"github.com/apache/skywalking-banyandb/pkg/logger"
pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
"github.com/apache/skywalking-banyandb/pkg/test"
@@ -36,14 +37,14 @@ var testSeriesPool pbv1.SeriesPool
func TestSeriesIndex_Primary(t *testing.T) {
ctx := context.Background()
- tester := assert.New(t)
path, fn := setUp(require.New(t))
si, err := newSeriesIndex(ctx, path)
- tester.NoError(err)
+ require.NoError(t, err)
defer func() {
- tester.NoError(si.Close())
+ require.NoError(t, si.Close())
fn()
}()
+ var docs index.Documents
for i := 0; i < 100; i++ {
series := testSeriesPool.Get()
series.Subject = "service_instance_latency"
@@ -53,17 +54,23 @@ func TestSeriesIndex_Primary(t *testing.T) {
}
// Initialize test data
- series, err = si.createPrimary(series)
- if err != nil {
- t.Fatalf("Failed to create primary series: %v", err)
+ if err = series.Marshal(); err != nil {
+ t.Fatalf("Failed to marshal series: %v", err)
}
- tester.True(series.ID > 0)
+ require.True(t, series.ID > 0)
+ doc := index.Document{
+ DocID: uint64(series.ID),
+ EntityValues: make([]byte, len(series.Buffer)),
+ }
+ copy(doc.EntityValues, series.Buffer)
+ docs = append(docs, doc)
testSeriesPool.Put(series)
}
+ require.NoError(t, si.Write(docs))
// Restart the index
- tester.NoError(si.Close())
+ require.NoError(t, si.Close())
si, err = newSeriesIndex(ctx, path)
- tester.NoError(err)
+ require.NoError(t, err)
tests := []struct {
name string
subject string
@@ -115,12 +122,12 @@ func TestSeriesIndex_Primary(t *testing.T) {
seriesQuery.Subject = tt.subject
seriesQuery.EntityValues = tt.entityValues
sl, err := si.searchPrimary(ctx, seriesQuery)
- tester.NoError(err)
- tester.Equal(1, len(sl))
- tester.Equal(tt.subject, sl[0].Subject)
- tester.Equal(tt.expected[0].GetStr().GetValue(), sl[0].EntityValues[0].GetStr().GetValue())
- tester.Equal(tt.expected[1].GetStr().GetValue(), sl[0].EntityValues[1].GetStr().GetValue())
- tester.True(sl[0].ID > 0)
+ require.NoError(t, err)
+ require.Equal(t, 1, len(sl))
+ assert.Equal(t, tt.subject, sl[0].Subject)
+ assert.Equal(t, tt.expected[0].GetStr().GetValue(), sl[0].EntityValues[0].GetStr().GetValue())
+ assert.Equal(t, tt.expected[1].GetStr().GetValue(), sl[0].EntityValues[1].GetStr().GetValue())
+ assert.True(t, sl[0].ID > 0)
})
}
}
diff --git a/banyand/internal/storage/storage.go b/banyand/internal/storage/storage.go
index d2dd5d72..e2aca767 100644
--- a/banyand/internal/storage/storage.go
+++ b/banyand/internal/storage/storage.go
@@ -71,7 +71,6 @@ type IndexDB interface {
// TSDB allows listing and getting shard details.
type TSDB[T TSTable] interface {
io.Closer
- Register(shardID common.ShardID, series *pbv1.Series) (*pbv1.Series, error)
Lookup(ctx context.Context, series *pbv1.Series) (pbv1.SeriesList, error)
CreateTSTableIfNotExist(shardID common.ShardID, ts time.Time) (TSTableWrapper[T], error)
SelectTSTables(timeRange timestamp.TimeRange) []TSTableWrapper[T]
diff --git a/banyand/internal/storage/tsdb.go b/banyand/internal/storage/tsdb.go
index bb3e1169..8f8dd9b7 100644
--- a/banyand/internal/storage/tsdb.go
+++ b/banyand/internal/storage/tsdb.go
@@ -36,7 +36,6 @@ import (
"github.com/apache/skywalking-banyandb/api/common"
"github.com/apache/skywalking-banyandb/pkg/logger"
- pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
"github.com/apache/skywalking-banyandb/pkg/timestamp"
)
@@ -115,14 +114,6 @@ func OpenTSDB[T TSTable](ctx context.Context, opts TSDBOpts[T]) (TSDB[T], error)
return db, nil
}
-func (d *database[T]) Register(shardID common.ShardID, series *pbv1.Series) (*pbv1.Series, error) {
- var err error
- if series, err = d.index.createPrimary(series); err != nil {
- return nil, err
- }
- return series, nil
-}
-
func (d *database[T]) CreateTSTableIfNotExist(shardID common.ShardID, ts time.Time) (TSTableWrapper[T], error) {
id := uint32(shardID)
if id >= atomic.LoadUint32(&d.sLen) {
diff --git a/pkg/index/index.go b/pkg/index/index.go
index d0776c1b..2ee7782a 100644
--- a/pkg/index/index.go
+++ b/pkg/index/index.go
@@ -234,7 +234,6 @@ type Series struct {
type SeriesStore interface {
Store
- Create(Series) error
Search([]byte) (common.SeriesID, error)
SearchPrefix([]byte) ([]Series, error)
SearchWildcard([]byte) ([]Series, error)
diff --git a/pkg/index/inverted/inverted.go b/pkg/index/inverted/inverted.go
index bf1f3038..1a2224d9 100644
--- a/pkg/index/inverted/inverted.go
+++ b/pkg/index/inverted/inverted.go
@@ -39,7 +39,6 @@ import (
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/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/index/posting/roaring"
@@ -133,6 +132,7 @@ func NewStore(opts StoreOpts) (index.SeriesStore, error) {
}
func (s *store) Close() error {
+ s.flush()
s.closer.CloseThenWait()
if s.errClosing.Load() != nil {
return *s.errClosing.Load()
@@ -324,7 +324,6 @@ func (s *store) run() {
}
defer flush()
var docIDBuffer bytes.Buffer
- seriesIDBuffer := make([]byte, 8)
for {
timer := time.NewTimer(s.batchInterval)
select {
@@ -340,16 +339,6 @@ func (s *store) run() {
case flushEvent:
flush()
close(d.onComplete)
- case index.Series:
- seriesIDBuffer = seriesIDBuffer[:0]
- seriesIDBuffer = encoding.Uint64ToBytes(seriesIDBuffer, uint64(d.ID))
- doc := bluge.NewDocument(string(seriesIDBuffer))
- doc.AddField(bluge.NewKeywordFieldBytes(entityField, d.EntityValues).StoreValue())
- size++
- batch.Update(doc.ID(), doc)
- if size >= batchSize {
- flush()
- }
case index.Document, index.Documents:
var docs []index.Document
var isBatch bool
diff --git a/pkg/index/inverted/inverted_series.go b/pkg/index/inverted/inverted_series.go
index 31d56406..b6296896 100644
--- a/pkg/index/inverted/inverted_series.go
+++ b/pkg/index/inverted/inverted_series.go
@@ -30,19 +30,6 @@ import (
"github.com/apache/skywalking-banyandb/pkg/index"
)
-// Create implements index.SeriesStore.
-func (s *store) Create(series index.Series) error {
- if !s.closer.AddRunning() {
- return nil
- }
- defer s.closer.Done()
- select {
- case <-s.closer.CloseNotify():
- case s.ch <- series:
- }
- return nil
-}
-
// Search implements index.SeriesStore.
func (s *store) Search(term []byte) (common.SeriesID, error) {
reader, err := s.writer.Reader()
diff --git a/pkg/index/inverted/inverted_series_test.go b/pkg/index/inverted/inverted_series_test.go
index 609f70bd..03860f83 100644
--- a/pkg/index/inverted/inverted_series_test.go
+++ b/pkg/index/inverted/inverted_series_test.go
@@ -193,23 +193,20 @@ func TestStore_SearchPrefix(t *testing.T) {
}
func setupData(tester *assert.Assertions, s index.SeriesStore) {
- series1 := index.Series{
- ID: common.SeriesID(1),
+ series1 := index.Document{
+ DocID: 1,
EntityValues: []byte("test1"),
}
- tester.NoError(s.Create(series1))
- series2 := index.Series{
- ID: common.SeriesID(2),
+ series2 := index.Document{
+ DocID: 2,
EntityValues: []byte("test2"),
}
- tester.NoError(s.Create(series2))
- series3 := index.Series{
- ID: common.SeriesID(3),
+ series3 := index.Document{
+ DocID: 3,
EntityValues: []byte("test3"),
}
- tester.NoError(s.Create(series3))
- tester.NoError(s.Create(series3)) // duplicated data
+ tester.NoError(s.Batch([]index.Document{series1, series2, series3, series3}))
s.(*store).flush()
}
diff --git a/pkg/index/inverted/inverted_test.go b/pkg/index/inverted/inverted_test.go
index 4d818b99..d352302c 100644
--- a/pkg/index/inverted/inverted_test.go
+++ b/pkg/index/inverted/inverted_test.go
@@ -171,7 +171,7 @@ func TestStore_SeriesMatch(t *testing.T) {
},
{
matches: []string{"root"},
- want: roaring.NewPostingListWithInitialData(),
+ want: roaring.NewPostingListWithInitialData(2),
},
}
for _, tt := range tests {
diff --git a/pkg/pb/v1/value_test.go b/pkg/pb/v1/value_test.go
index 17f77fed..710fb2cf 100644
--- a/pkg/pb/v1/value_test.go
+++ b/pkg/pb/v1/value_test.go
@@ -53,13 +53,6 @@ func TestMarshalAndUnmarshalTagValue(t *testing.T) {
dest := []byte{}
// Test marshalTagValue
dest, err := marshalTagValue(dest, tt.src)
- if tt.name == "unsupported type" {
- assert.Error(t, err)
- dest = dest[:0]
- _, _, _, err = unmarshalTagValue(dest, []byte("unsupported type"))
- assert.Error(t, err)
- return
- }
marshaled := make([]byte, len(dest))
copy(marshaled, dest)
@@ -69,11 +62,10 @@ func TestMarshalAndUnmarshalTagValue(t *testing.T) {
// Test unmarshalTagValue
dest = dest[:0]
- dest, marshaled, unmarshaled, err := unmarshalTagValue(dest, marshaled)
+ _, marshaled, unmarshaled, err := unmarshalTagValue(dest, marshaled)
// Add assertions
assert.NoError(t, err)
- assert.True(t, len(dest) > 0)
assert.True(t, len(marshaled) == 0)
assert.NotNil(t, unmarshaled)