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/12/24 13:52:09 UTC
[skywalking-banyandb] branch main updated: Support IN operation for the global index scanning. (#232)
This is an automated email from the ASF dual-hosted git repository.
lujiajing pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
The following commit(s) were added to refs/heads/main by this push:
new 6edcc05 Support IN operation for the global index scanning. (#232)
6edcc05 is described below
commit 6edcc05ab48e7ceca9a852793f0467fe5943bb5a
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Sat Dec 24 21:52:03 2022 +0800
Support IN operation for the global index scanning. (#232)
Signed-off-by: Gao Hongtao <ha...@gmail.com>
---
pkg/query/logical/index_filter.go | 11 ++-
.../logical/stream/stream_plan_indexscan_global.go | 89 +++++++++++-----------
test/cases/stream/data/input/global_indices.yaml | 36 +++++++++
test/cases/stream/data/want/global_indices.yaml | 56 ++++++++++++++
test/cases/stream/stream.go | 1 +
5 files changed, 147 insertions(+), 46 deletions(-)
diff --git a/pkg/query/logical/index_filter.go b/pkg/query/logical/index_filter.go
index f084b04..f7f2882 100644
--- a/pkg/query/logical/index_filter.go
+++ b/pkg/query/logical/index_filter.go
@@ -63,9 +63,14 @@ func BuildLocalFilter(criteria *modelv1.Criteria, schema Schema, entityDict map[
}
if ok, indexRule := schema.IndexDefined(cond.Name); ok {
if indexRule.Location == databasev1.IndexRule_LOCATION_GLOBAL {
- return nil, nil, GlobalIndexError{
- IndexRule: indexRule,
- Expr: expr,
+ switch cond.Op {
+ case modelv1.Condition_BINARY_OP_EQ, modelv1.Condition_BINARY_OP_IN:
+ return nil, nil, GlobalIndexError{
+ IndexRule: indexRule,
+ Expr: expr,
+ }
+ default:
+ return nil, nil, errors.Wrapf(errUnsupportedConditionOp, "gobal index conf:%s", cond)
}
}
return parseCondition(cond, indexRule, expr, entity)
diff --git a/pkg/query/logical/stream/stream_plan_indexscan_global.go b/pkg/query/logical/stream/stream_plan_indexscan_global.go
index d2e1842..4ec4fc3 100644
--- a/pkg/query/logical/stream/stream_plan_indexscan_global.go
+++ b/pkg/query/logical/stream/stream_plan_indexscan_global.go
@@ -77,55 +77,58 @@ func (t *globalIndexScan) Execute(ec executor.StreamExecutionContext) ([]*stream
func (t *globalIndexScan) executeForShard(ec executor.StreamExecutionContext, shard tsdb.Shard) ([]*streamv1.Element, error) {
var elementsInShard []*streamv1.Element
- itemIDs, err := shard.Index().Seek(index.Field{
- Key: index.FieldKey{
- SeriesID: tsdb.GlobalSeriesID(t.schema.Scope()),
- IndexRuleID: t.globalIndexRule.GetMetadata().GetId(),
- },
- Term: t.expr.Bytes()[0],
- })
- if err != nil || len(itemIDs) < 1 {
- return nil, err
- }
- for _, itemID := range itemIDs {
- segShard, err := ec.Shard(itemID.ShardID)
- if err != nil {
- return elementsInShard, errors.WithStack(err)
- }
- series, err := segShard.Series().GetByID(itemID.SeriesID)
+ for _, term := range t.expr.Bytes() {
+ itemIDs, err := shard.Index().Seek(index.Field{
+ Key: index.FieldKey{
+ SeriesID: tsdb.GlobalSeriesID(t.schema.Scope()),
+ IndexRuleID: t.globalIndexRule.GetMetadata().GetId(),
+ },
+ Term: term,
+ })
if err != nil {
- return elementsInShard, errors.WithStack(err)
+ return nil, err
}
- err = func() error {
- ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
- defer cancel()
- item, closer, errInner := series.Get(ctx, itemID)
- defer func(closer io.Closer) {
- if closer != nil {
- _ = closer.Close()
- }
- }(closer)
- if errInner != nil {
- return errors.WithStack(errInner)
+ for _, itemID := range itemIDs {
+ segShard, err := ec.Shard(itemID.ShardID)
+ if err != nil {
+ return elementsInShard, errors.WithStack(err)
}
- tagFamilies, errInner := logical.ProjectItem(ec, item, t.projectionTagRefs)
- if errInner != nil {
- return errors.WithStack(errInner)
+ series, err := segShard.Series().GetByID(itemID.SeriesID)
+ if err != nil {
+ return elementsInShard, errors.WithStack(err)
}
- elementID, errInner := ec.ParseElementID(item)
- if errInner != nil {
- return errors.WithStack(errInner)
+ err = func() error {
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+ defer cancel()
+ item, closer, errInner := series.Get(ctx, itemID)
+ defer func(closer io.Closer) {
+ if closer != nil {
+ _ = closer.Close()
+ }
+ }(closer)
+ if errInner != nil {
+ return errors.WithStack(errInner)
+ }
+ tagFamilies, errInner := logical.ProjectItem(ec, item, t.projectionTagRefs)
+ if errInner != nil {
+ return errors.WithStack(errInner)
+ }
+ elementID, errInner := ec.ParseElementID(item)
+ if errInner != nil {
+ return errors.WithStack(errInner)
+ }
+ elementsInShard = append(elementsInShard, &streamv1.Element{
+ ElementId: elementID,
+ Timestamp: timestamppb.New(time.Unix(0, int64(item.Time()))),
+ TagFamilies: tagFamilies,
+ })
+ return nil
+ }()
+ if err != nil {
+ return nil, err
}
- elementsInShard = append(elementsInShard, &streamv1.Element{
- ElementId: elementID,
- Timestamp: timestamppb.New(time.Unix(0, int64(item.Time()))),
- TagFamilies: tagFamilies,
- })
- return nil
- }()
- if err != nil {
- return nil, err
}
}
+
return elementsInShard, nil
}
diff --git a/test/cases/stream/data/input/global_indices.yaml b/test/cases/stream/data/input/global_indices.yaml
new file mode 100644
index 0000000..4a5c0fa
--- /dev/null
+++ b/test/cases/stream/data/input/global_indices.yaml
@@ -0,0 +1,36 @@
+# Licensed to Apache Software Foundation (ASF) under one or more contributor
+# license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright
+# ownership. Apache Software Foundation (ASF) licenses this file to you under
+# the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+metadata:
+ group: "default"
+ name: "sw"
+projection:
+ tagFamilies:
+ - name: "searchable"
+ tags: ["trace_id"]
+ - name: "data"
+ tags: ["data_binary"]
+criteria:
+ le:
+ op: "LOGICAL_OP_AND"
+ left:
+ condition:
+ name: "trace_id"
+ op: "BINARY_OP_IN"
+ value:
+ str_array:
+ value: ["1", "3", "5", "unknown"]
diff --git a/test/cases/stream/data/want/global_indices.yaml b/test/cases/stream/data/want/global_indices.yaml
new file mode 100644
index 0000000..ba5e651
--- /dev/null
+++ b/test/cases/stream/data/want/global_indices.yaml
@@ -0,0 +1,56 @@
+# Licensed to Apache Software Foundation (ASF) under one or more contributor
+# license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright
+# ownership. Apache Software Foundation (ASF) licenses this file to you under
+# the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+elements:
+ - elementId: "0"
+ tagFamilies:
+ - name: searchable
+ tags:
+ - key: trace_id
+ value:
+ str:
+ value: "1"
+ - name: data
+ tags:
+ - key: data_binary
+ value:
+ binaryData: YWJjMTIzIT8kKiYoKSctPUB+
+ - elementId: "2"
+ tagFamilies:
+ - name: searchable
+ tags:
+ - key: trace_id
+ value:
+ str:
+ value: "3"
+ - name: data
+ tags:
+ - key: data_binary
+ value:
+ binaryData: YWJjMTIzIT8kKiYoKSctPUB+
+ - elementId: "4"
+ tagFamilies:
+ - name: searchable
+ tags:
+ - key: trace_id
+ value:
+ str:
+ value: "5"
+ - name: data
+ tags:
+ - key: data_binary
+ value:
+ binaryData: YWJjMTIzIT8kKiYoKSctPUB+
\ No newline at end of file
diff --git a/test/cases/stream/stream.go b/test/cases/stream/stream.go
index 058cd55..5f9cbd2 100644
--- a/test/cases/stream/stream.go
+++ b/test/cases/stream/stream.go
@@ -57,6 +57,7 @@ var _ = g.DescribeTable("Scanning Streams", func(args helpers.Args) {
}),
g.Entry("sort desc", helpers.Args{Input: "sort_desc", Duration: 1 * time.Hour}),
g.Entry("global index", helpers.Args{Input: "global_index", Duration: 1 * time.Hour}),
+ g.Entry("multi-global index", helpers.Args{Input: "global_indices", Duration: 1 * time.Hour}),
g.Entry("filter by non-indexed tag", helpers.Args{Input: "filter_tag", Duration: 1 * time.Hour}),
g.Entry("get empty result by non-indexed tag", helpers.Args{Input: "filter_tag_empty", Duration: 1 * time.Hour, WantEmpty: true}),
g.Entry("numeric local index: less", helpers.Args{Input: "less", Duration: 1 * time.Hour}),