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/17 22:04:31 UTC
(skywalking-banyandb) 01/01: Pass integration test
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 00d456e9595ce41d9edc36385126933cf05d66de
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Mon Dec 18 05:45:33 2023 +0800
Pass integration test
Signed-off-by: Gao Hongtao <ha...@gmail.com>
---
banyand/measure/block.go | 19 +++++++++++---
banyand/measure/query.go | 4 +++
banyand/measure/tstable_test.go | 30 +++++++++++-----------
pkg/index/inverted/inverted_series.go | 11 ++++++--
pkg/index/inverted/inverted_series_test.go | 1 +
pkg/query/logical/index_filter.go | 2 +-
.../measure/measure_plan_indexscan_local.go | 20 ++++++++-------
test/cases/measure/measure.go | 4 +--
8 files changed, 59 insertions(+), 32 deletions(-)
diff --git a/banyand/measure/block.go b/banyand/measure/block.go
index fd0661e1..1730d022 100644
--- a/banyand/measure/block.go
+++ b/banyand/measure/block.go
@@ -213,9 +213,10 @@ func (b *block) unmarshalTagFamily(decoder *encoding.BytesBlockDecoder, tfIndex
cc := b.tagFamilies[tfIndex].resizeColumns(len(tagProjection))
for j := range tagProjection {
- for i := range cc {
+ for i := range cfm.columnMetadata {
if tagProjection[j] == cfm.columnMetadata[i].name {
- cc[i].mustReadValues(decoder, valueReader, cfm.columnMetadata[i], uint64(b.Len()))
+ cc[j].mustReadValues(decoder, valueReader, cfm.columnMetadata[i], uint64(b.Len()))
+ break
}
}
}
@@ -372,7 +373,7 @@ func (bc *blockCursor) init(p *part, bm blockMetadata, queryOpts QueryOptions) {
func (bc *blockCursor) copyAllTo(r *pbv1.Result) {
r.SID = bc.bm.seriesID
- r.Timestamps = bc.timestamps
+ r.Timestamps = append(r.Timestamps, bc.timestamps...)
for _, cf := range bc.tagFamilies {
tf := pbv1.TagFamily{
Name: cf.Name,
@@ -479,6 +480,12 @@ func (bc *blockCursor) loadData(tmpBlock *block) bool {
Name: cf.Columns[i].Name,
ValueType: cf.Columns[i].ValueType,
}
+ if len(cf.Columns[i].Values) == 0 {
+ continue
+ }
+ if len(cf.Columns[i].Values) != len(tmpBlock.timestamps) {
+ logger.Panicf("unexpected number of values for tags %q: got %d; want %d", cf.Columns[i].Name, len(cf.Columns[i].Values), len(tmpBlock.timestamps))
+ }
column.Values = append(column.Values, cf.Columns[i].Values[start:end]...)
tf.Columns = append(tf.Columns, column)
}
@@ -486,6 +493,12 @@ func (bc *blockCursor) loadData(tmpBlock *block) bool {
}
bc.fields.Name = tmpBlock.field.Name
for i := range tmpBlock.field.Columns {
+ if len(tmpBlock.field.Columns[i].Values) == 0 {
+ continue
+ }
+ if len(tmpBlock.field.Columns[i].Values) != len(tmpBlock.timestamps) {
+ logger.Panicf("unexpected number of values for fields %q: got %d; want %d", tmpBlock.field.Columns[i].Name, len(tmpBlock.field.Columns[i].Values), len(tmpBlock.timestamps))
+ }
column := Column{
Name: tmpBlock.field.Columns[i].Name,
ValueType: tmpBlock.field.Columns[i].ValueType,
diff --git a/banyand/measure/query.go b/banyand/measure/query.go
index 15a9193f..6a41ca45 100644
--- a/banyand/measure/query.go
+++ b/banyand/measure/query.go
@@ -70,6 +70,9 @@ func (s *measure) Query(ctx context.Context, mqo pbv1.MeasureQueryOptions) (pbv1
if mqo.TimeRange == nil || mqo.Entity == nil {
return nil, errors.New("invalid query options: timeRange and series are required")
}
+ if len(mqo.TagProjection) == 0 && len(mqo.FieldProjection) == 0 {
+ return nil, errors.New("invalid query options: tagProjection or fieldProjection is required")
+ }
tsdb := s.databaseSupplier.SupplyTSDB().(storage.TSDB[*tsTable])
tabWrappers := tsdb.SelectTSTables(*mqo.TimeRange)
sl, err := tsdb.IndexDB().Search(ctx, &pbv1.Series{Subject: mqo.Name, EntityValues: mqo.Entity}, mqo.Filter, mqo.Order)
@@ -123,6 +126,7 @@ func (s *measure) Query(ctx context.Context, mqo pbv1.MeasureQueryOptions) (pbv1
if mqo.Order.Sort == modelv1.Sort_SORT_ASC || mqo.Order.Sort == modelv1.Sort_SORT_UNSPECIFIED {
result.ascTS = true
}
+ return &result, nil
}
result.sidToIndex = make(map[common.SeriesID]int)
diff --git a/banyand/measure/tstable_test.go b/banyand/measure/tstable_test.go
index a56846ed..20b750a0 100644
--- a/banyand/measure/tstable_test.go
+++ b/banyand/measure/tstable_test.go
@@ -124,9 +124,9 @@ func Test_tstIter(t *testing.T) {
minTimestamp: 1,
maxTimestamp: 1,
want: []blockMetadata{
- {seriesID: 1, count: 2, uncompressedSizeBytes: 3426},
- {seriesID: 2, count: 2, uncompressedSizeBytes: 106},
- {seriesID: 3, count: 2, uncompressedSizeBytes: 48},
+ {seriesID: 1, count: 1, uncompressedSizeBytes: 1676},
+ {seriesID: 2, count: 1, uncompressedSizeBytes: 55},
+ {seriesID: 3, count: 1, uncompressedSizeBytes: 24},
},
},
{
@@ -136,12 +136,12 @@ func Test_tstIter(t *testing.T) {
minTimestamp: 1,
maxTimestamp: 2,
want: []blockMetadata{
- {seriesID: 1, count: 2, uncompressedSizeBytes: 3426},
- {seriesID: 1, count: 2, uncompressedSizeBytes: 3426},
- {seriesID: 2, count: 2, uncompressedSizeBytes: 106},
- {seriesID: 2, count: 2, uncompressedSizeBytes: 106},
- {seriesID: 3, count: 2, uncompressedSizeBytes: 48},
- {seriesID: 3, count: 2, uncompressedSizeBytes: 48},
+ {seriesID: 1, count: 1, uncompressedSizeBytes: 1676},
+ {seriesID: 1, count: 1, uncompressedSizeBytes: 1676},
+ {seriesID: 2, count: 1, uncompressedSizeBytes: 55},
+ {seriesID: 2, count: 1, uncompressedSizeBytes: 55},
+ {seriesID: 3, count: 1, uncompressedSizeBytes: 24},
+ {seriesID: 3, count: 1, uncompressedSizeBytes: 24},
},
},
{
@@ -151,12 +151,12 @@ func Test_tstIter(t *testing.T) {
minTimestamp: 1,
maxTimestamp: 2,
want: []blockMetadata{
- {seriesID: 1, count: 2, uncompressedSizeBytes: 3426},
- {seriesID: 1, count: 2, uncompressedSizeBytes: 3426},
- {seriesID: 2, count: 2, uncompressedSizeBytes: 106},
- {seriesID: 2, count: 2, uncompressedSizeBytes: 106},
- {seriesID: 3, count: 2, uncompressedSizeBytes: 48},
- {seriesID: 3, count: 2, uncompressedSizeBytes: 48},
+ {seriesID: 1, count: 1, uncompressedSizeBytes: 1676},
+ {seriesID: 1, count: 1, uncompressedSizeBytes: 1676},
+ {seriesID: 2, count: 1, uncompressedSizeBytes: 55},
+ {seriesID: 2, count: 1, uncompressedSizeBytes: 55},
+ {seriesID: 3, count: 1, uncompressedSizeBytes: 24},
+ {seriesID: 3, count: 1, uncompressedSizeBytes: 24},
},
},
}
diff --git a/pkg/index/inverted/inverted_series.go b/pkg/index/inverted/inverted_series.go
index f3a477ed..31d56406 100644
--- a/pkg/index/inverted/inverted_series.go
+++ b/pkg/index/inverted/inverted_series.go
@@ -105,11 +105,16 @@ func (s *store) SearchWildcard(wildcard []byte) ([]index.Series, error) {
func parseResult(dmi search.DocumentMatchIterator) ([]index.Series, error) {
result := make([]index.Series, 0, 10)
next, err := dmi.Next()
+ docIDMap := make(map[uint64]struct{})
for err == nil && next != nil {
var series index.Series
err = next.VisitStoredFields(func(field string, value []byte) bool {
if field == docIDField {
- series.ID = common.SeriesID(convert.BytesToUint64(value))
+ id := convert.BytesToUint64(value)
+ if _, ok := docIDMap[id]; !ok {
+ series.ID = common.SeriesID(convert.BytesToUint64(value))
+ docIDMap[id] = struct{}{}
+ }
}
if field == entityField {
series.EntityValues = value
@@ -119,7 +124,9 @@ func parseResult(dmi search.DocumentMatchIterator) ([]index.Series, error) {
if err != nil {
return nil, errors.WithMessage(err, "visit stored fields")
}
- result = append(result, series)
+ if series.ID > 0 {
+ result = append(result, series)
+ }
next, err = dmi.Next()
}
if err != nil {
diff --git a/pkg/index/inverted/inverted_series_test.go b/pkg/index/inverted/inverted_series_test.go
index 64daaaa2..609f70bd 100644
--- a/pkg/index/inverted/inverted_series_test.go
+++ b/pkg/index/inverted/inverted_series_test.go
@@ -210,5 +210,6 @@ func setupData(tester *assert.Assertions, s index.SeriesStore) {
EntityValues: []byte("test3"),
}
tester.NoError(s.Create(series3))
+ tester.NoError(s.Create(series3)) // duplicated data
s.(*store).flush()
}
diff --git a/pkg/query/logical/index_filter.go b/pkg/query/logical/index_filter.go
index 78a210c5..ac2df54e 100644
--- a/pkg/query/logical/index_filter.go
+++ b/pkg/query/logical/index_filter.go
@@ -512,7 +512,7 @@ func parseEntities(op modelv1.LogicalExpression_LogicalOp, input []*modelv1.TagV
anyEntity := func(entities [][]*modelv1.TagValue) bool {
for _, entity := range entities {
for _, entry := range entity {
- if entry == pbv1.AnyTagValue {
+ if entry != pbv1.AnyTagValue {
return false
}
}
diff --git a/pkg/query/logical/measure/measure_plan_indexscan_local.go b/pkg/query/logical/measure/measure_plan_indexscan_local.go
index 620a17bc..e4b1ccc3 100644
--- a/pkg/query/logical/measure/measure_plan_indexscan_local.go
+++ b/pkg/query/logical/measure/measure_plan_indexscan_local.go
@@ -137,15 +137,11 @@ func (i *localIndexScan) Sort(order *logical.OrderBy) {
func (i *localIndexScan) Execute(ctx context.Context) (mit executor.MIterator, err error) {
var orderBy *pbv1.OrderBy
- if i.order.Index != nil {
+ if i.order != nil {
orderBy = &pbv1.OrderBy{
Index: i.order.Index,
Sort: i.order.Sort,
}
- } else if i.groupByEntity {
- orderBy = &pbv1.OrderBy{
- Sort: i.order.Sort,
- }
}
ec := executor.FromMeasureExecutionContext(ctx)
var results []pbv1.MeasureQueryResult
@@ -205,15 +201,20 @@ func indexScan(startTime, endTime time.Time, metadata *commonv1.Metadata, projec
}
type resultMIterator struct {
- results []pbv1.MeasureQueryResult
- current []*measurev1.DataPoint
- index int
+ results []pbv1.MeasureQueryResult
+ current []*measurev1.DataPoint
+ index int
+ currentIndex int
}
func (ei *resultMIterator) Next() bool {
if ei.index >= len(ei.results) {
return false
}
+ ei.currentIndex++
+ if ei.currentIndex < len(ei.current) {
+ return true
+ }
r := ei.results[ei.index].Pull()
if r == nil {
@@ -221,6 +222,7 @@ func (ei *resultMIterator) Next() bool {
return ei.Next()
}
ei.current = ei.current[:0]
+ ei.currentIndex = 0
for i := range r.Timestamps {
dp := &measurev1.DataPoint{
Timestamp: timestamppb.New(time.Unix(0, int64(r.Timestamps[i]))),
@@ -251,7 +253,7 @@ func (ei *resultMIterator) Next() bool {
}
func (ei *resultMIterator) Current() []*measurev1.DataPoint {
- return ei.current
+ return []*measurev1.DataPoint{ei.current[ei.currentIndex]}
}
func (ei *resultMIterator) Close() error {
diff --git a/test/cases/measure/measure.go b/test/cases/measure/measure.go
index bfcb6642..f011fe29 100644
--- a/test/cases/measure/measure.go
+++ b/test/cases/measure/measure.go
@@ -39,9 +39,9 @@ var (
}
)
-var _ = g.DescribeTable("Scanning Measures", verify,
+var _ = g.FDescribeTable("Scanning Measures", verify,
g.Entry("all", helpers.Args{Input: "all", Duration: 25 * time.Minute, Offset: -20 * time.Minute}),
- g.FEntry("all only fields", helpers.Args{Input: "all_only_fields", Duration: 25 * time.Minute, Offset: -20 * time.Minute}),
+ g.Entry("all only fields", helpers.Args{Input: "all_only_fields", Duration: 25 * time.Minute, Offset: -20 * time.Minute}),
g.Entry("filter by tag", helpers.Args{Input: "tag_filter", Duration: 25 * time.Minute, Offset: -20 * time.Minute}),
g.Entry("filter by a integer tag", helpers.Args{Input: "tag_filter_int", Duration: 25 * time.Minute, Offset: -20 * time.Minute}),
g.Entry("filter by an unknown tag", helpers.Args{Input: "tag_filter_unknown", Duration: 25 * time.Minute, Offset: -20 * time.Minute, WantEmpty: true}),