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:58:59 UTC
[skywalking-banyandb] branch feature/optimize-query created (now 9fb7b35)
This is an automated email from the ASF dual-hosted git repository.
lujiajing pushed a change to branch feature/optimize-query
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git.
at 9fb7b35 check iters
This branch includes the following new commits:
new 9fb7b35 check iters
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: check iters
Posted by lu...@apache.org.
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 {