You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by lu...@apache.org on 2022/02/18 07:59:00 UTC
[skywalking-banyandb] 01/01: check iters
This is an automated email from the ASF dual-hosted git repository.
lujiajing pushed a commit to branch feature/optimize-query
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit 9fb7b3503fc5f13db00fe6bdc4ee975de751071e
Author: Megrez Lu <lu...@gmail.com>
AuthorDate: Fri Feb 18 15:58:46 2022 +0800
check iters
---
banyand/query/processor_test.go | 18 ++++++++++++++++++
banyand/stream/stream_write.go | 3 ++-
pkg/query/logical/common.go | 3 +++
pkg/query/logical/plan_indexscan_global.go | 11 +++++++----
pkg/query/logical/plan_indexscan_local.go | 21 ++++++++++++++++-----
5 files changed, 46 insertions(+), 10 deletions(-)
diff --git a/banyand/query/processor_test.go b/banyand/query/processor_test.go
index d3a659a..0f5e0d6 100644
--- a/banyand/query/processor_test.go
+++ b/banyand/query/processor_test.go
@@ -22,6 +22,7 @@ import (
"embed"
"encoding/base64"
"encoding/json"
+ "math"
"os"
"strconv"
"testing"
@@ -243,6 +244,23 @@ func TestQueryProcessor(t *testing.T) {
checker: withDataBinaryChecker,
},
{
+ name: "query max valid time-range which covers all the segments with data binary projection",
+ queryGenerator: func(baseTs time.Time) *streamv1.QueryRequest {
+ return pb.NewQueryRequestBuilder().
+ Limit(10).
+ Offset(0).
+ Metadata("default", "sw").
+ // min: 1677-09-21T00:12:43.145224194Z
+ // max: 2262-04-11T23:47:16.854775806Z
+ TimeRange(time.Unix(0, math.MinInt64), time.Unix(0, math.MaxInt64)).
+ Projection("searchable", "trace_id").
+ Projection("data", "data_binary").
+ Build()
+ },
+ wantLen: 5,
+ checker: withDataBinaryChecker,
+ },
+ {
name: "query given timeRange which covers all the segments and sort by duration DESC",
queryGenerator: func(baseTs time.Time) *streamv1.QueryRequest {
return pb.NewQueryRequestBuilder().
diff --git a/banyand/stream/stream_write.go b/banyand/stream/stream_write.go
index 38843cf..83c4094 100644
--- a/banyand/stream/stream_write.go
+++ b/banyand/stream/stream_write.go
@@ -112,7 +112,8 @@ func (s *stream) write(shardID common.ShardID, seriesHashKey []byte, value *stre
Interface("data", value).
Uint64("series_id", uint64(series.ID())).
Int("shard_id", int(shardID)).
- Msg("write measure")
+ Str("stream", sm.Metadata.GetName()).
+ Msg("write stream")
return writer, errWrite
}
writer, err := writeFn()
diff --git a/pkg/query/logical/common.go b/pkg/query/logical/common.go
index 9a6b1fd..78a2132 100644
--- a/pkg/query/logical/common.go
+++ b/pkg/query/logical/common.go
@@ -48,6 +48,9 @@ func createComparator(sortDirection modelv1.Sort) comparator {
func projectItem(ec executor.ExecutionContext, item tsdb.Item, projectionFieldRefs [][]*FieldRef) ([]*modelv1.TagFamily, error) {
tagFamily := make([]*modelv1.TagFamily, len(projectionFieldRefs))
for i, refs := range projectionFieldRefs {
+ if refs == nil || len(refs) == 0 {
+ continue
+ }
tags := make([]*modelv1.Tag, len(refs))
familyName := refs[0].tag.GetFamilyName()
parsedTagFamily, err := ec.ParseTagFamily(familyName, item)
diff --git a/pkg/query/logical/plan_indexscan_global.go b/pkg/query/logical/plan_indexscan_global.go
index bb1bbe0..6f5bbb6 100644
--- a/pkg/query/logical/plan_indexscan_global.go
+++ b/pkg/query/logical/plan_indexscan_global.go
@@ -86,9 +86,10 @@ func (t *globalIndexScan) Execute(ec executor.ExecutionContext) ([]*streamv1.Ele
}
var elements []*streamv1.Element
for _, shard := range shards {
- elementsInShard, err := t.executeForShard(ec, shard)
- if err != nil {
- return elements, err
+ elementsInShard, shardErr := t.executeForShard(ec, shard)
+ if shardErr != nil {
+ // TODO: shall we proceed?
+ return elements, shardErr
}
elements = append(elements, elementsInShard...)
}
@@ -119,7 +120,9 @@ func (t *globalIndexScan) executeForShard(ec executor.ExecutionContext, shard ts
err = func() error {
item, closer, errInner := series.Get(itemID)
defer func(closer io.Closer) {
- _ = closer.Close()
+ if closer != nil {
+ _ = closer.Close()
+ }
}(closer)
if errInner != nil {
return errors.WithStack(errInner)
diff --git a/pkg/query/logical/plan_indexscan_local.go b/pkg/query/logical/plan_indexscan_local.go
index 4233b27..1ab0e4f 100644
--- a/pkg/query/logical/plan_indexscan_local.go
+++ b/pkg/query/logical/plan_indexscan_local.go
@@ -137,16 +137,23 @@ func (i *localIndexScan) Execute(ec executor.ExecutionContext) ([]*streamv1.Elem
}
var iters []tsdb.Iterator
for _, shard := range shards {
- itersInShard, err := i.executeInShard(shard)
- if err != nil {
- return nil, err
+ itersInShard, innerErr := i.executeInShard(shard)
+ if innerErr != nil {
+ return nil, innerErr
+ }
+ if itersInShard == nil {
+ continue
}
iters = append(iters, itersInShard...)
}
- c := createComparator(i.sort)
-
var elems []*streamv1.Element
+
+ if iters == nil || len(iters) == 0 {
+ return elems, nil
+ }
+
+ c := createComparator(i.sort)
it := NewItemIter(iters, c)
for it.HasNext() {
nextItem := it.Next()
@@ -173,6 +180,10 @@ func (i *localIndexScan) executeInShard(shard tsdb.Shard) ([]tsdb.Iterator, erro
return nil, err
}
+ if seriesList == nil || len(seriesList) == 0 {
+ return nil, nil
+ }
+
var builders []seekerBuilder
if i.index != nil {