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 2022/05/22 10:47:11 UTC
[skywalking-banyandb] 01/01: Support filter tags based on non-indexed tags for meausre. Fix "having" operation
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch query-index
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit 49ffa4b0d89e735dcf56744d3080c80042ade4c1
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Sun May 22 10:36:13 2022 +0000
Support filter tags based on non-indexed tags for meausre.
Fix "having" operation
Signed-off-by: Gao Hongtao <ha...@gmail.com>
---
banyand/metadata/metadata_test.go | 1 +
banyand/stream/stream_query_test.go | 146 +++++++++++
banyand/stream/stream_write_test.go | 5 +-
banyand/stream/testdata/multiple_shards.json | 27 ++-
banyand/tsdb/index/writer.go | 166 +++++++------
pkg/bus/bus_test.go | 38 +--
pkg/index/tree.go | 30 +--
pkg/pb/v1/write.go | 99 +++++++-
pkg/query/logical/expr.go | 12 +-
pkg/query/logical/expr_literal.go | 138 +++++++++++
pkg/query/logical/interface.go | 9 +
pkg/query/logical/stream_analyzer.go | 4 +-
pkg/query/logical/stream_analyzer_test.go | 29 ++-
pkg/query/logical/stream_plan_execution_test.go | 59 +++--
pkg/query/logical/stream_plan_indexscan_global.go | 18 +-
pkg/query/logical/stream_plan_indexscan_local.go | 129 +---------
pkg/query/logical/stream_plan_tag_filter.go | 269 +++++++++++++++++++++
pkg/query/logical/testdata/global_index.json | 19 +-
pkg/test/stream/testdata/index_rule_binding.json | 3 +-
.../stream/testdata/index_rules/extended_tags.json | 13 +
pkg/test/stream/testdata/stream.json | 8 +
21 files changed, 922 insertions(+), 300 deletions(-)
diff --git a/banyand/metadata/metadata_test.go b/banyand/metadata/metadata_test.go
index d866b46..fc3c7e5 100644
--- a/banyand/metadata/metadata_test.go
+++ b/banyand/metadata/metadata_test.go
@@ -77,6 +77,7 @@ func Test_service_RulesBySubject(t *testing.T) {
"mq.broker",
"mq.queue",
"mq.topic",
+ "extended_tags",
},
},
{
diff --git a/banyand/stream/stream_query_test.go b/banyand/stream/stream_query_test.go
index 35501f7..ba748cf 100644
--- a/banyand/stream/stream_query_test.go
+++ b/banyand/stream/stream_query_test.go
@@ -641,6 +641,152 @@ var _ = Describe("Write", func() {
},
false,
),
+ Entry(
+
+ "filter by extended tags c",
+ queryOpts{
+ entity: tsdb.Entity{tsdb.AnyEntry, tsdb.AnyEntry, tsdb.AnyEntry},
+ duration: 1 * time.Hour,
+ buildFn: func(builder tsdb.SeekerBuilder) {
+ rule := &databasev1.IndexRule{
+ Metadata: &commonv1.Metadata{
+ Name: "extended_tags",
+ Group: "default",
+ Id: 11,
+ },
+ Tags: []string{"extended_tags"},
+ Type: databasev1.IndexRule_TYPE_INVERTED,
+ Location: databasev1.IndexRule_LOCATION_SERIES,
+ }
+ builder.Filter(rule, tsdb.Condition{
+ "extended_tags": []index.ConditionValue{
+ {
+ Op: modelv1.Condition_BINARY_OP_HAVING,
+ Values: [][]byte{[]byte("c")},
+ },
+ },
+ })
+ },
+ },
+ shardsForTest{
+ {
+ id: 0,
+ location: l1,
+ elements: []string{"4"},
+ },
+ {
+ id: 1,
+ location: l2,
+ },
+ {
+ id: 1,
+ location: l3,
+ },
+ {
+ id: 1,
+ location: l4,
+ elements: []string{"3", "5"},
+ },
+ },
+ false,
+ ),
+ Entry(
+
+ "filter by extended tags abc",
+ queryOpts{
+ entity: tsdb.Entity{tsdb.AnyEntry, tsdb.AnyEntry, tsdb.AnyEntry},
+ duration: 1 * time.Hour,
+ buildFn: func(builder tsdb.SeekerBuilder) {
+ rule := &databasev1.IndexRule{
+ Metadata: &commonv1.Metadata{
+ Name: "extended_tags",
+ Group: "default",
+ Id: 11,
+ },
+ Tags: []string{"extended_tags"},
+ Type: databasev1.IndexRule_TYPE_INVERTED,
+ Location: databasev1.IndexRule_LOCATION_SERIES,
+ }
+ builder.Filter(rule, tsdb.Condition{
+ "extended_tags": []index.ConditionValue{
+ {
+ Op: modelv1.Condition_BINARY_OP_HAVING,
+ Values: [][]byte{[]byte("c"), []byte("a"), []byte("b")},
+ },
+ },
+ })
+ },
+ },
+ shardsForTest{
+ {
+ id: 0,
+ location: l1,
+ },
+ {
+ id: 1,
+ location: l2,
+ },
+ {
+ id: 1,
+ location: l3,
+ },
+ {
+ id: 1,
+ location: l4,
+ elements: []string{"5"},
+ },
+ },
+ false,
+ ),
+ Entry(
+
+ "filter by extended tags bc",
+ queryOpts{
+ entity: tsdb.Entity{tsdb.AnyEntry, tsdb.AnyEntry, tsdb.AnyEntry},
+ duration: 1 * time.Hour,
+ buildFn: func(builder tsdb.SeekerBuilder) {
+ rule := &databasev1.IndexRule{
+ Metadata: &commonv1.Metadata{
+ Name: "extended_tags",
+ Group: "default",
+ Id: 11,
+ },
+ Tags: []string{"extended_tags"},
+ Type: databasev1.IndexRule_TYPE_INVERTED,
+ Location: databasev1.IndexRule_LOCATION_SERIES,
+ }
+ builder.Filter(rule, tsdb.Condition{
+ "extended_tags": []index.ConditionValue{
+ {
+ Op: modelv1.Condition_BINARY_OP_HAVING,
+ Values: [][]byte{[]byte("c"), []byte("b")},
+ },
+ },
+ })
+ },
+ },
+ shardsForTest{
+ {
+ id: 0,
+ location: l1,
+ elements: []string{"4"},
+ },
+ {
+ id: 1,
+ location: l2,
+ },
+ {
+ id: 1,
+ location: l3,
+ },
+ {
+ id: 1,
+ location: l4,
+ elements: []string{"5"},
+ },
+ },
+ false,
+ ),
)
})
})
diff --git a/banyand/stream/stream_write_test.go b/banyand/stream/stream_write_test.go
index ab99f2f..c4abea4 100644
--- a/banyand/stream/stream_write_test.go
+++ b/banyand/stream/stream_write_test.go
@@ -99,7 +99,7 @@ var _ = Describe("Write", func() {
300,
1622933202000000000,
"GET",
- "200",
+ 200,
),
},
},
@@ -116,6 +116,7 @@ var _ = Describe("Write", func() {
1622933202000000000,
nil,
nil,
+ nil,
"MySQL",
"10.1.1.2",
),
@@ -136,6 +137,7 @@ var _ = Describe("Write", func() {
nil,
nil,
nil,
+ nil,
"test_topic",
"10.0.0.1",
"broker",
@@ -174,6 +176,7 @@ var _ = Describe("Write", func() {
nil,
nil,
nil,
+ nil,
"test_topic",
"10.0.0.1",
"broker",
diff --git a/banyand/stream/testdata/multiple_shards.json b/banyand/stream/testdata/multiple_shards.json
index 402964d..edd9017 100644
--- a/banyand/stream/testdata/multiple_shards.json
+++ b/banyand/stream/testdata/multiple_shards.json
@@ -31,7 +31,14 @@
{"int":{"value": 30}},
{"int":{"value": 1622933202000000000}},
{"str":{"value": "GET"}},
- {"int":{"value": 500}}
+ {"int":{"value": 500}},
+ {"null":0},
+ {"null":0},
+ {"null":0},
+ {"null":0},
+ {"null":0},
+ {"null":0},
+ {"str_array":{"value": ["c"]}}
]
},
{
@@ -44,7 +51,14 @@
{"int":{"value": 60}},
{"int":{"value": 1622933202000000000}},
{"str":{"value": "GET"}},
- {"int":{"value": 400}}
+ {"int":{"value": 400}},
+ {"null":0},
+ {"null":0},
+ {"null":0},
+ {"null":0},
+ {"null":0},
+ {"null":0},
+ {"str_array":{"value": ["b", "c"]}}
]
},
{
@@ -57,7 +71,14 @@
{"int":{"value": 300}},
{"int":{"value": 1622933202000000000}},
{"str":{"value": "GET"}},
- {"int":{"value": 500}}
+ {"int":{"value": 500}},
+ {"null":0},
+ {"null":0},
+ {"null":0},
+ {"null":0},
+ {"null":0},
+ {"null":0},
+ {"str_array":{"value": ["a", "b", "c"]}}
]
}
diff --git a/banyand/tsdb/index/writer.go b/banyand/tsdb/index/writer.go
index bb459a0..c99c67c 100644
--- a/banyand/tsdb/index/writer.go
+++ b/banyand/tsdb/index/writer.go
@@ -127,94 +127,112 @@ func (s *Writer) bootIndexGenerator() {
//TODO: should listen to pipeline in a distributed cluster
func (s *Writer) writeGlobalIndex(scope tsdb.Entry, ruleIndex *partition.IndexRuleLocator, ref tsdb.GlobalItemID, value Value) error {
- val, _, err := getIndexValue(ruleIndex, value)
+ values, _, err := getIndexValue(ruleIndex, value)
if err != nil {
return err
}
- indexShardID, err := partition.ShardID(val, s.shardNum)
- if err != nil {
- return err
- }
- shard, err := s.db.SupplyTSDB().Shard(common.ShardID(indexShardID))
- if err != nil {
- return err
- }
- builder := shard.Index().WriterBuilder()
- indexWriter, err := builder.
- Scope(scope).
- GlobalItemID(ref).
- Time(value.Timestamp).
- Build()
- if err != nil {
- return err
- }
- rule := ruleIndex.Rule
- switch rule.GetType() {
- case databasev1.IndexRule_TYPE_INVERTED:
- return indexWriter.WriteInvertedIndex(index.Field{
- Key: index.FieldKey{
- IndexRuleID: rule.GetMetadata().GetId(),
- },
- Term: val,
- })
- case databasev1.IndexRule_TYPE_TREE:
- return indexWriter.WriteLSMIndex(index.Field{
- Key: index.FieldKey{
- IndexRuleID: rule.GetMetadata().GetId(),
- },
- Term: val,
- })
+ var errWriting error
+ for _, val := range values {
+ indexShardID, err := partition.ShardID(val, s.shardNum)
+ if err != nil {
+ return err
+ }
+ shard, err := s.db.SupplyTSDB().Shard(common.ShardID(indexShardID))
+ if err != nil {
+ return err
+ }
+ builder := shard.Index().WriterBuilder()
+ indexWriter, err := builder.
+ Scope(scope).
+ GlobalItemID(ref).
+ Time(value.Timestamp).
+ Build()
+ if err != nil {
+ return err
+ }
+ rule := ruleIndex.Rule
+ switch rule.GetType() {
+ case databasev1.IndexRule_TYPE_INVERTED:
+ errWriting = multierr.Append(errWriting, indexWriter.WriteInvertedIndex(index.Field{
+ Key: index.FieldKey{
+ IndexRuleID: rule.GetMetadata().GetId(),
+ },
+ Term: val,
+ }))
+ case databasev1.IndexRule_TYPE_TREE:
+ errWriting = multierr.Append(errWriting, indexWriter.WriteLSMIndex(index.Field{
+ Key: index.FieldKey{
+ IndexRuleID: rule.GetMetadata().GetId(),
+ },
+ Term: val,
+ }))
+ }
}
- return err
+ return errWriting
}
func writeLocalIndex(writer tsdb.Writer, ruleIndex *partition.IndexRuleLocator, value Value) (err error) {
- val, _, err := getIndexValue(ruleIndex, value)
+ values, _, err := getIndexValue(ruleIndex, value)
if err != nil {
return err
}
- rule := ruleIndex.Rule
- switch rule.GetType() {
- case databasev1.IndexRule_TYPE_INVERTED:
- return writer.WriteInvertedIndex(index.Field{
- Key: index.FieldKey{
- IndexRuleID: rule.GetMetadata().GetId(),
- },
- Term: val,
- })
- case databasev1.IndexRule_TYPE_TREE:
- return writer.WriteLSMIndex(index.Field{
- Key: index.FieldKey{
- IndexRuleID: rule.GetMetadata().GetId(),
- },
- Term: val,
- })
+ var errWriting error
+ for _, val := range values {
+ rule := ruleIndex.Rule
+ switch rule.GetType() {
+ case databasev1.IndexRule_TYPE_INVERTED:
+ errWriting = multierr.Append(errWriting, writer.WriteInvertedIndex(index.Field{
+ Key: index.FieldKey{
+ IndexRuleID: rule.GetMetadata().GetId(),
+ },
+ Term: val,
+ }))
+ case databasev1.IndexRule_TYPE_TREE:
+ errWriting = multierr.Append(errWriting, writer.WriteLSMIndex(index.Field{
+ Key: index.FieldKey{
+ IndexRuleID: rule.GetMetadata().GetId(),
+ },
+ Term: val,
+ }))
+ }
}
- return err
+
+ return errWriting
}
-func getIndexValue(ruleIndex *partition.IndexRuleLocator, value Value) (val []byte, isInt bool, err error) {
- val = make([]byte, 0, len(ruleIndex.TagIndices))
+var ErrUnsupportedIndexType = errors.New("unsupported index type")
+
+func getIndexValue(ruleIndex *partition.IndexRuleLocator, value Value) (val [][]byte, isInt bool, err error) {
+ val = make([][]byte, 0)
var existInt bool
- for _, tIndex := range ruleIndex.TagIndices {
- tag, err := partition.GetTagByOffset(value.TagFamilies, tIndex.FamilyOffset, tIndex.TagOffset)
- if errors.Is(err, partition.ErrMalformedElement) {
- continue
- }
- if err != nil {
- return nil, false, errors.WithMessagef(err, "index rule:%v", ruleIndex.Rule.Metadata)
- }
- if tag.GetInt() != nil {
- existInt = true
- }
- v, err := pbv1.MarshalIndexFieldValue(tag)
- if err != nil {
- return nil, false, err
- }
- val = append(val, v...)
+ if len(ruleIndex.TagIndices) != 1 {
+ return nil, false, errors.Wrap(ErrUnsupportedIndexType, "the index rule didn't support composited tags")
}
- if len(ruleIndex.TagIndices) == 1 && existInt {
- return val, true, nil
+ tIndex := ruleIndex.TagIndices[0]
+ tag, err := partition.GetTagByOffset(value.TagFamilies, tIndex.FamilyOffset, tIndex.TagOffset)
+
+ if errors.Is(err, partition.ErrMalformedElement) {
+ return val, false, nil
+ }
+ if err != nil {
+ return nil, false, errors.WithMessagef(err, "index rule:%v", ruleIndex.Rule.Metadata)
+ }
+ if tag.GetInt() != nil {
+ existInt = true
+ }
+ fv, err := pbv1.ParseIndexFieldValue(tag)
+ if err != nil {
+ return nil, false, err
+ }
+ v := fv.GetValue()
+ if v != nil {
+ val = append(val, v)
+ return val, existInt, nil
+ }
+ arr := fv.GetArr()
+ if arr != nil {
+ val = append(val, arr...)
+
}
- return val, false, nil
+ return val, existInt, nil
}
diff --git a/pkg/bus/bus_test.go b/pkg/bus/bus_test.go
index 5e4c295..c3c4357 100644
--- a/pkg/bus/bus_test.go
+++ b/pkg/bus/bus_test.go
@@ -169,23 +169,23 @@ func TestBus_PubAndSub(t *testing.T) {
t.Errorf("Subscribe() error = %v, wantErr %v", err, l.wantErr)
}
}
- go func() {
- for _, m := range tt.messages {
- mm := make([]Message, 0)
- for _, id := range m.messageIDS {
- mm = append(mm, NewMessage(id, nil))
- }
- f, err := e.Publish(m.topic, mm...)
- if err != nil && !m.wantErr {
- t.Errorf("Publish() error = %v, wantErr %v", err, m.wantErr)
- continue
- }
- if f == nil {
- continue
- }
+ for _, m := range tt.messages {
+ mm := make([]Message, 0)
+ for _, id := range m.messageIDS {
+ mm = append(mm, NewMessage(id, nil))
+ }
+ f, err := e.Publish(m.topic, mm...)
+ if err != nil && !m.wantErr {
+ t.Errorf("Publish() error = %v, wantErr %v", err, m.wantErr)
+ continue
+ }
+ if f == nil {
+ continue
+ }
+ go func(want []MessageID) {
ret, errRet := f.GetAll()
if errRet == ErrEmptyFuture {
- continue
+ return
} else if errRet != nil {
t.Errorf("Publish()'s return message error = %v", err)
}
@@ -194,14 +194,14 @@ func TestBus_PubAndSub(t *testing.T) {
ids = append(ids, ret[i].ID())
}
ids = sortMessage(ids)
- if !reflect.DeepEqual(ids, m.wantRet) {
- t.Errorf("Publish()'s return = %v, want %v", ret, m.wantRet)
+ if !reflect.DeepEqual(ids, want) {
+ t.Errorf("Publish()'s return = %v, want %v", ret, want)
}
for i := 0; i < len(ret); i++ {
wg.Done()
}
- }
- }()
+ }(m.wantRet)
+ }
if waitTimeout(&wg, 10*time.Second) {
t.Error("message receiving is time out")
}
diff --git a/pkg/index/tree.go b/pkg/index/tree.go
index 9b51a2c..dd39f64 100644
--- a/pkg/index/tree.go
+++ b/pkg/index/tree.go
@@ -88,12 +88,12 @@ func BuildTree(searcher Searcher, condMap Condition) (Tree, error) {
case modelv1.Condition_BINARY_OP_NE:
root.addNot(key, root.newEq(key, cond.Values))
case modelv1.Condition_BINARY_OP_HAVING:
- n := root.addOrNode(len(cond.Values))
+ n := root.addAndNode(len(cond.Values))
for _, v := range cond.Values {
n.addEq(key, [][]byte{v})
}
case modelv1.Condition_BINARY_OP_NOT_HAVING:
- n := root.newOrNode(len(cond.Values))
+ n := root.newAndNode(len(cond.Values))
for _, v := range cond.Values {
n.addEq(key, [][]byte{v})
}
@@ -160,8 +160,8 @@ func (n *node) addRangeLeaf(key FieldKey) *rangeOp {
return r
}
-func (n *node) newOrNode(size int) *orNode {
- return &orNode{
+func (n *node) newAndNode(size int) *andNode {
+ return &andNode{
node: &node{
searcher: n.searcher,
SubNodes: make([]Executor, 0, size),
@@ -169,8 +169,8 @@ func (n *node) newOrNode(size int) *orNode {
}
}
-func (n *node) addOrNode(size int) *orNode {
- on := n.newOrNode(size)
+func (n *node) addAndNode(size int) *andNode {
+ on := n.newAndNode(size)
n.SubNodes = append(n.SubNodes, on)
return on
}
@@ -245,24 +245,6 @@ func (an *andNode) MarshalJSON() ([]byte, error) {
return json.Marshal(data)
}
-type orNode struct {
- *node
-}
-
-func (on *orNode) merge(list posting.List) error {
- return on.value.Union(list)
-}
-
-func (on *orNode) Execute() (posting.List, error) {
- return execute(on.node, on)
-}
-
-func (on *orNode) MarshalJSON() ([]byte, error) {
- data := make(map[string]interface{}, 1)
- data["or"] = on.node.SubNodes
- return json.Marshal(data)
-}
-
type leaf struct {
Executor
Key FieldKey
diff --git a/pkg/pb/v1/write.go b/pkg/pb/v1/write.go
index 55a72ec..0169535 100644
--- a/pkg/pb/v1/write.go
+++ b/pkg/pb/v1/write.go
@@ -19,7 +19,6 @@ package v1
import (
"bytes"
- "strings"
"time"
"github.com/pkg/errors"
@@ -34,30 +33,108 @@ import (
type ID string
-const strDelimiter = "\n"
+var strDelimiter = []byte("\n")
var ErrUnsupportedTagForIndexField = errors.New("the tag type(for example, null) can not be as the index field value")
func MarshalIndexFieldValue(tagValue *modelv1.TagValue) ([]byte, error) {
+ fv, err := ParseIndexFieldValue(tagValue)
+ if err != nil {
+ return nil, err
+ }
+ val := fv.GetValue()
+ if val != nil {
+ return val, nil
+ }
+ return fv.marshalArr(), nil
+}
+
+type FieldValue struct {
+ value []byte
+ arr [][]byte
+ splitter []byte
+}
+
+func newValue(value []byte) FieldValue {
+ return FieldValue{
+ value: value,
+ }
+}
+
+func newValueWithSplitter(splitter []byte) *FieldValue {
+ return &FieldValue{
+ splitter: splitter,
+ }
+}
+
+func appendValue(fv *FieldValue, value []byte) *FieldValue {
+ if fv == nil {
+ fv = &FieldValue{}
+ }
+ fv.arr = append(fv.arr, value)
+ return fv
+}
+
+func (fv FieldValue) GetValue() []byte {
+ if len(fv.value) < 1 {
+ return nil
+ }
+ return fv.value
+}
+func (fv FieldValue) GetArr() [][]byte {
+ if len(fv.arr) < 1 {
+ return nil
+ }
+ return fv.arr
+}
+
+func (fv *FieldValue) marshalArr() []byte {
+ switch len(fv.arr) {
+ case 0:
+ return nil
+ case 1:
+ return fv.arr[0]
+ }
+ n := len(fv.splitter) * (len(fv.arr) - 1)
+ for i := 0; i < len(fv.arr); i++ {
+ n += len(fv.arr[i])
+ }
+ buf := bytes.NewBuffer(nil)
+ buf.Grow(n)
+ buf.Write(fv.arr[0])
+ for _, v := range fv.arr[1:] {
+ if fv.splitter != nil {
+ buf.Write(fv.splitter)
+ }
+ buf.Write(v)
+ }
+ return buf.Bytes()
+}
+
+func ParseIndexFieldValue(tagValue *modelv1.TagValue) (FieldValue, error) {
switch x := tagValue.GetValue().(type) {
case *modelv1.TagValue_Str:
- return []byte(x.Str.GetValue()), nil
+ return newValue([]byte(x.Str.GetValue())), nil
case *modelv1.TagValue_Int:
- return convert.Int64ToBytes(x.Int.GetValue()), nil
+ return newValue(convert.Int64ToBytes(x.Int.GetValue())), nil
case *modelv1.TagValue_StrArray:
- return []byte(strings.Join(x.StrArray.GetValue(), strDelimiter)), nil
+ fv := newValueWithSplitter(strDelimiter)
+ for _, v := range x.StrArray.GetValue() {
+ fv = appendValue(fv, []byte(v))
+ }
+ return *fv, nil
case *modelv1.TagValue_IntArray:
- buf := bytes.NewBuffer(nil)
+ var fv *FieldValue
for _, i := range x.IntArray.GetValue() {
- buf.Write(convert.Int64ToBytes(i))
+ fv = appendValue(fv, convert.Int64ToBytes(i))
}
- return buf.Bytes(), nil
+ return *fv, nil
case *modelv1.TagValue_BinaryData:
- return x.BinaryData, nil
+ return newValue(x.BinaryData), nil
case *modelv1.TagValue_Id:
- return []byte(x.Id.GetValue()), nil
+ return newValue([]byte(x.Id.GetValue())), nil
}
- return nil, ErrUnsupportedTagForIndexField
+ return FieldValue{}, ErrUnsupportedTagForIndexField
}
type StreamWriteRequestBuilder struct {
diff --git a/pkg/query/logical/expr.go b/pkg/query/logical/expr.go
index e3681e9..200a533 100644
--- a/pkg/query/logical/expr.go
+++ b/pkg/query/logical/expr.go
@@ -80,8 +80,8 @@ func NewTagRef(familyName, tagName string) *TagRef {
}
}
-// NewSearchableFieldRef is a short-handed method for creating a TagRef to the tag in the searchable family
-func NewSearchableFieldRef(tagName string) *TagRef {
+// NewSearchableTagRef is a short-handed method for creating a TagRef to the tag in the searchable family
+func NewSearchableTagRef(tagName string) *TagRef {
return &TagRef{
tag: NewTag("searchable", tagName),
}
@@ -168,6 +168,14 @@ func (b *binaryExpr) Resolve(s Schema) error {
return nil
}
+func (b *binaryExpr) Execute(tag *modelv1.Tag) bool {
+ switch b.op {
+ case modelv1.Condition_BINARY_OP_EQ:
+
+ }
+ return false
+}
+
func (b *binaryExpr) String() string {
return fmt.Sprintf("%s %s %s", b.l.String(), b.op.String(), b.r.String())
}
diff --git a/pkg/query/logical/expr_literal.go b/pkg/query/logical/expr_literal.go
index 0e47643..ec5bbe7 100644
--- a/pkg/query/logical/expr_literal.go
+++ b/pkg/query/logical/expr_literal.go
@@ -24,15 +24,38 @@ import (
"github.com/google/go-cmp/cmp"
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"
)
var _ LiteralExpr = (*int64Literal)(nil)
+var _ ComparableExpr = (*int64Literal)(nil)
type int64Literal struct {
int64
}
+func (i *int64Literal) Compare(tagValue *modelv1.TagValue) (int, bool) {
+ intValue := tagValue.GetInt()
+ if intValue == nil {
+ return 0, false
+ }
+ return int(i.int64 - intValue.Value), true
+}
+
+func (i *int64Literal) BelongTo(tagValue *modelv1.TagValue) bool {
+ intValue := tagValue.GetInt()
+ if intValue != nil {
+ return i.int64 == intValue.Value
+
+ }
+ intArray := tagValue.GetIntArray()
+ if intArray == nil {
+ return false
+ }
+ return int64SliceContains(i.int64, intArray.Value)
+}
+
func (i *int64Literal) Bytes() [][]byte {
return [][]byte{convert.Int64ToBytes(i.int64)}
}
@@ -58,11 +81,41 @@ func (i *int64Literal) String() string {
}
var _ LiteralExpr = (*int64ArrLiteral)(nil)
+var _ ComparableExpr = (*int64ArrLiteral)(nil)
type int64ArrLiteral struct {
arr []int64
}
+func (i *int64ArrLiteral) Compare(tagValue *modelv1.TagValue) (int, bool) {
+ intArray := tagValue.GetIntArray()
+ if intArray == nil {
+ return 0, false
+ }
+ if int64SlicesEqual(i.arr, intArray.Value) {
+ return 0, true
+ }
+ return 0, false
+}
+
+func (i *int64ArrLiteral) BelongTo(tagValue *modelv1.TagValue) bool {
+ intValue := tagValue.GetInt()
+ if intValue != nil {
+ return int64SliceContains(intValue.Value, i.arr)
+
+ }
+ intArray := tagValue.GetIntArray()
+ if intArray == nil {
+ return false
+ }
+ for _, v := range intArray.Value {
+ if !int64SliceContains(v, i.arr) {
+ return false
+ }
+ }
+ return true
+}
+
func (i *int64ArrLiteral) Bytes() [][]byte {
b := make([][]byte, 0, len(i.arr))
for _, i := range i.arr {
@@ -94,11 +147,36 @@ func (i *int64ArrLiteral) String() string {
}
var _ LiteralExpr = (*strLiteral)(nil)
+var _ ComparableExpr = (*strLiteral)(nil)
type strLiteral struct {
string
}
+func (s *strLiteral) Compare(tagValue *modelv1.TagValue) (int, bool) {
+ strValue := tagValue.GetStr()
+ if strValue == nil {
+ return 0, false
+ }
+ if strValue.Value == s.string {
+ return 0, true
+ }
+ return 0, false
+}
+
+func (s *strLiteral) BelongTo(tagValue *modelv1.TagValue) bool {
+ strValue := tagValue.GetStr()
+ if strValue != nil {
+ return s.string == strValue.Value
+
+ }
+ strArray := tagValue.GetStrArray()
+ if strArray == nil {
+ return false
+ }
+ return stringSliceContains(s.string, strArray.Value)
+}
+
func (s *strLiteral) Bytes() [][]byte {
return [][]byte{[]byte(s.string)}
}
@@ -124,11 +202,41 @@ func (s *strLiteral) String() string {
}
var _ LiteralExpr = (*strArrLiteral)(nil)
+var _ ComparableExpr = (*strArrLiteral)(nil)
type strArrLiteral struct {
arr []string
}
+func (s *strArrLiteral) Compare(tagValue *modelv1.TagValue) (int, bool) {
+ strArray := tagValue.GetStrArray()
+ if strArray == nil {
+ return 0, false
+ }
+ if stringSlicesEqual(s.arr, strArray.Value) {
+ return 0, true
+ }
+ return 0, false
+}
+
+func (s *strArrLiteral) BelongTo(tagValue *modelv1.TagValue) bool {
+ strValue := tagValue.GetStr()
+ if strValue != nil {
+ return stringSliceContains(strValue.Value, s.arr)
+
+ }
+ strArray := tagValue.GetStrArray()
+ if strArray == nil {
+ return false
+ }
+ for _, v := range strArray.Value {
+ if !stringSliceContains(v, s.arr) {
+ return false
+ }
+ }
+ return true
+}
+
func (s *strArrLiteral) Bytes() [][]byte {
b := make([][]byte, 0, len(s.arr))
for _, str := range s.arr {
@@ -188,3 +296,33 @@ func (s *idLiteral) DataType() int32 {
func (s *idLiteral) String() string {
return s.string
}
+
+func int64SlicesEqual(a, b []int64) bool {
+ if len(a) != len(b) {
+ return false
+ }
+ for i, v := range a {
+ if v != b[i] {
+ return false
+ }
+ }
+ return true
+}
+
+func int64SliceContains(a int64, slice []int64) bool {
+ for _, v := range slice {
+ if v == a {
+ return true
+ }
+ }
+ return false
+}
+
+func stringSliceContains(a string, slice []string) bool {
+ for _, v := range slice {
+ if v == a {
+ return true
+ }
+ }
+ return false
+}
diff --git a/pkg/query/logical/interface.go b/pkg/query/logical/interface.go
index e37c97c..f572de8 100644
--- a/pkg/query/logical/interface.go
+++ b/pkg/query/logical/interface.go
@@ -19,6 +19,8 @@ package logical
import (
"fmt"
+
+ modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
)
type PlanType uint8
@@ -31,6 +33,7 @@ const (
PlanGroupBy
PlanAggregation
PlanTop
+ PlanTagFilter
)
type UnresolvedPlan interface {
@@ -60,3 +63,9 @@ type ResolvableExpr interface {
Expr
Resolve(Schema) error
}
+
+type ComparableExpr interface {
+ Expr
+ Compare(tagValue *modelv1.TagValue) (int, bool)
+ BelongTo(tagValue *modelv1.TagValue) bool
+}
diff --git a/pkg/query/logical/stream_analyzer.go b/pkg/query/logical/stream_analyzer.go
index 5cfdd70..b376de1 100644
--- a/pkg/query/logical/stream_analyzer.go
+++ b/pkg/query/logical/stream_analyzer.go
@@ -122,7 +122,7 @@ func (a *StreamAnalyzer) Analyze(_ context.Context, criteria *streamv1.QueryRequ
// parse orderBy
queryOrder := criteria.GetOrderBy()
if queryOrder != nil {
- if v, ok := plan.(*unresolvedStreamIndexScan); ok {
+ if v, ok := plan.(*unresolvedTagFilter); ok {
v.unresolvedOrderBy = OrderBy(queryOrder.GetIndexRuleName(), queryOrder.GetSort())
}
}
@@ -209,6 +209,6 @@ func parseStreamFields(criteria *streamv1.QueryRequest, metadata *commonv1.Metad
}
}
- return IndexScan(timeRange.GetBegin().AsTime(), timeRange.GetEnd().AsTime(), metadata,
+ return TagFilter(timeRange.GetBegin().AsTime(), timeRange.GetEnd().AsTime(), metadata,
tagExprs, entity, nil, projTags...), nil
}
diff --git a/pkg/query/logical/stream_analyzer_test.go b/pkg/query/logical/stream_analyzer_test.go
index 45f6929..6348642 100644
--- a/pkg/query/logical/stream_analyzer_test.go
+++ b/pkg/query/logical/stream_analyzer_test.go
@@ -29,6 +29,7 @@ import (
modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
"github.com/apache/skywalking-banyandb/banyand/metadata"
"github.com/apache/skywalking-banyandb/banyand/tsdb"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
pb "github.com/apache/skywalking-banyandb/pkg/pb/v1"
"github.com/apache/skywalking-banyandb/pkg/query/logical"
teststream "github.com/apache/skywalking-banyandb/pkg/test/stream"
@@ -37,6 +38,14 @@ import (
// setUpStreamAnalyzer creates a default analyzer for testing.
// You have to close the underlying metadata after teststream
func setUpStreamAnalyzer() (*logical.StreamAnalyzer, func(), error) {
+ err := logger.Init(logger.Logging{
+ Env: "dev",
+ Level: "warn",
+ })
+ if err != nil {
+ return nil, func() {
+ }, err
+ }
metadataService, err := metadata.NewService(context.TODO())
if err != nil {
return nil, func() {
@@ -101,7 +110,7 @@ func TestStreamAnalyzer_SimpleTimeScan(t *testing.T) {
assert.NotNil(plan)
correctPlan, err := logical.Limit(
logical.Offset(
- logical.IndexScan(sT, eT, metadata, nil, tsdb.Entity{tsdb.AnyEntry, tsdb.AnyEntry, tsdb.AnyEntry}, nil),
+ logical.TagFilter(sT, eT, metadata, nil, tsdb.Entity{tsdb.AnyEntry, tsdb.AnyEntry, tsdb.AnyEntry}, nil),
0),
20).
Analyze(schema)
@@ -141,10 +150,10 @@ func TestStreamAnalyzer_ComplexQuery(t *testing.T) {
correctPlan, err := logical.Limit(
logical.Offset(
- logical.IndexScan(sT, eT, metadata,
+ logical.TagFilter(sT, eT, metadata,
[]logical.Expr{
- logical.Eq(logical.NewSearchableFieldRef("mq.topic"), logical.Str("event_topic")),
- logical.Eq(logical.NewSearchableFieldRef("http.method"), logical.Str("GET")),
+ logical.Eq(logical.NewSearchableTagRef("mq.topic"), logical.Str("event_topic")),
+ logical.Eq(logical.NewSearchableTagRef("http.method"), logical.Str("GET")),
}, tsdb.Entity{tsdb.Entry("my_app"), tsdb.AnyEntry, tsdb.AnyEntry},
logical.OrderBy("duration", modelv1.Sort_SORT_DESC),
logical.NewTags("searchable", "http.method", "service_id", "duration")),
@@ -180,8 +189,8 @@ func TestStreamAnalyzer_TraceIDQuery(t *testing.T) {
assert.NoError(err)
assert.NotNil(plan)
correctPlan, err := logical.Limit(
- logical.Offset(logical.IndexScan(time.Now(), time.Now(), metadata, []logical.Expr{
- logical.Eq(logical.NewSearchableFieldRef("trace_id"), logical.Str("123")),
+ logical.Offset(logical.TagFilter(time.Now(), time.Now(), metadata, []logical.Expr{
+ logical.Eq(logical.NewSearchableTagRef("trace_id"), logical.Str("123")),
}, nil, nil),
0),
100).Analyze(schema)
@@ -217,7 +226,7 @@ func TestStreamAnalyzer_OrderBy_IndexNotDefined(t *testing.T) {
assert.ErrorIs(err, logical.ErrIndexNotDefined)
}
-func TestStreamAnalyzer_OrderBy_FieldNotDefined(t *testing.T) {
+func TestStreamAnalyzer_OrderBy_TagNotDefined(t *testing.T) {
assert := require.New(t)
ana, stopFunc, err := setUpStreamAnalyzer()
@@ -243,7 +252,7 @@ func TestStreamAnalyzer_OrderBy_FieldNotDefined(t *testing.T) {
assert.ErrorIs(err, logical.ErrIndexNotDefined)
}
-func TestStreamAnalyzer_Projection_FieldNotDefined(t *testing.T) {
+func TestStreamAnalyzer_Projection_TagNotDefined(t *testing.T) {
assert := require.New(t)
ana, stopFunc, err := setUpStreamAnalyzer()
@@ -269,7 +278,7 @@ func TestStreamAnalyzer_Projection_FieldNotDefined(t *testing.T) {
assert.ErrorIs(err, logical.ErrTagNotDefined)
}
-func TestStreamAnalyzer_Fields_IndexNotDefined(t *testing.T) {
+func TestStreamAnalyzer_Tags_IndexNotDefined(t *testing.T) {
assert := require.New(t)
ana, stopFunc, err := setUpStreamAnalyzer()
@@ -292,5 +301,5 @@ func TestStreamAnalyzer_Fields_IndexNotDefined(t *testing.T) {
assert.NoError(err)
_, err = ana.Analyze(context.TODO(), criteria, metadata, schema)
- assert.ErrorIs(err, logical.ErrIndexNotDefined)
+ assert.NoError(err)
}
diff --git a/pkg/query/logical/stream_plan_execution_test.go b/pkg/query/logical/stream_plan_execution_test.go
index e4bee62..d336e02 100644
--- a/pkg/query/logical/stream_plan_execution_test.go
+++ b/pkg/query/logical/stream_plan_execution_test.go
@@ -55,17 +55,17 @@ func TestPlanExecution_TableScan_Limit(t *testing.T) {
}{
{
name: "Limit 1",
- unresolvedPlan: logical.Limit(logical.IndexScan(sT, eT, metadata, nil, tsdb.Entity{tsdb.AnyEntry, tsdb.AnyEntry, tsdb.AnyEntry}, nil), 1),
+ unresolvedPlan: logical.Limit(logical.TagFilter(sT, eT, metadata, nil, tsdb.Entity{tsdb.AnyEntry, tsdb.AnyEntry, tsdb.AnyEntry}, nil), 1),
wantLength: 1,
},
{
name: "Limit 5",
- unresolvedPlan: logical.Limit(logical.IndexScan(sT, eT, metadata, nil, tsdb.Entity{tsdb.AnyEntry, tsdb.AnyEntry, tsdb.AnyEntry}, nil), 5),
+ unresolvedPlan: logical.Limit(logical.TagFilter(sT, eT, metadata, nil, tsdb.Entity{tsdb.AnyEntry, tsdb.AnyEntry, tsdb.AnyEntry}, nil), 5),
wantLength: 5,
},
{
name: "Limit 10",
- unresolvedPlan: logical.Limit(logical.IndexScan(sT, eT, metadata, nil, tsdb.Entity{tsdb.AnyEntry, tsdb.AnyEntry, tsdb.AnyEntry}, nil), 10),
+ unresolvedPlan: logical.Limit(logical.TagFilter(sT, eT, metadata, nil, tsdb.Entity{tsdb.AnyEntry, tsdb.AnyEntry, tsdb.AnyEntry}, nil), 10),
wantLength: 5,
},
}
@@ -112,17 +112,17 @@ func TestPlanExecution_Offset(t *testing.T) {
}{
{
name: "Offset 0",
- unresolvedPlan: logical.Offset(logical.IndexScan(sT, eT, metadata, nil, tsdb.Entity{tsdb.AnyEntry, tsdb.AnyEntry, tsdb.AnyEntry}, nil), 0),
+ unresolvedPlan: logical.Offset(logical.TagFilter(sT, eT, metadata, nil, tsdb.Entity{tsdb.AnyEntry, tsdb.AnyEntry, tsdb.AnyEntry}, nil), 0),
wantLength: 5,
},
{
name: "Offset 3",
- unresolvedPlan: logical.Offset(logical.IndexScan(sT, eT, metadata, nil, tsdb.Entity{tsdb.AnyEntry, tsdb.AnyEntry, tsdb.AnyEntry}, nil), 3),
+ unresolvedPlan: logical.Offset(logical.TagFilter(sT, eT, metadata, nil, tsdb.Entity{tsdb.AnyEntry, tsdb.AnyEntry, tsdb.AnyEntry}, nil), 3),
wantLength: 2,
},
{
name: "Limit 5",
- unresolvedPlan: logical.Offset(logical.IndexScan(sT, eT, metadata, nil, tsdb.Entity{tsdb.AnyEntry, tsdb.AnyEntry, tsdb.AnyEntry}, nil), 5),
+ unresolvedPlan: logical.Offset(logical.TagFilter(sT, eT, metadata, nil, tsdb.Entity{tsdb.AnyEntry, tsdb.AnyEntry, tsdb.AnyEntry}, nil), 5),
wantLength: 0,
},
}
@@ -148,7 +148,7 @@ func TestPlanExecution_TraceIDFetch(t *testing.T) {
tester := require.New(t)
streamSvc, metaService, deferFunc := setup(tester)
defer deferFunc()
- _ = setupQueryData(t, "multiple_shards.json", streamSvc)
+ _ = setupQueryData(t, "global_index.json", streamSvc)
m := &commonv1.Metadata{
Name: "sw",
@@ -162,21 +162,29 @@ func TestPlanExecution_TraceIDFetch(t *testing.T) {
tests := []struct {
name string
traceID string
+ expr logical.Expr
wantLength int
}{
{
name: "traceID = 1",
traceID: "1",
- wantLength: 1,
+ wantLength: 2,
},
{
name: "traceID = 2",
traceID: "2",
+ wantLength: 3,
+ },
+ {
+ name: "traceID = 1 and spanID = 2",
+ traceID: "1",
+ expr: logical.Eq(logical.NewTagRef("searchable", "span_id"), logical.Str("2")),
wantLength: 1,
},
{
- name: "traceID = 3",
- traceID: "3",
+ name: "traceID = 2 and spanID = 3",
+ traceID: "1",
+ expr: logical.Eq(logical.NewTagRef("searchable", "span_id"), logical.Str("2")),
wantLength: 1,
},
}
@@ -187,16 +195,23 @@ func TestPlanExecution_TraceIDFetch(t *testing.T) {
s, err := analyzer.BuildStreamSchema(context.TODO(), m)
tester.NoError(err)
- p, err := logical.GlobalIndexScan(m, []logical.Expr{
+ exprs := []logical.Expr{
logical.Eq(logical.NewTagRef("searchable", "trace_id"), logical.Str(tt.traceID)),
- }, logical.NewTags("searchable", "trace_id")).Analyze(s)
+ }
+ if tt.expr != nil {
+ exprs = append(exprs, tt.expr)
+ }
+ p, err := logical.GlobalIndexScan(m, exprs,
+ logical.NewTags("searchable", "trace_id"), logical.NewTags("searchable", "span_id")).
+ Analyze(s)
tester.NoError(err)
tester.NotNil(p)
entities, err := p.(executor.StreamExecutable).Execute(streamSvc)
tester.NoError(err)
for _, entity := range entities {
- tester.Len(entity.GetTagFamilies(), 1)
+ tester.Len(entity.GetTagFamilies(), 2)
tester.Len(entity.GetTagFamilies()[0].GetTags(), 1)
+ tester.Len(entity.GetTagFamilies()[1].GetTags(), 1)
tester.Equal(entity.GetTagFamilies()[0].GetTags()[0].GetValue().GetStr().GetValue(), tt.traceID)
}
tester.Len(entities, tt.wantLength)
@@ -228,35 +243,35 @@ func TestPlanExecution_IndexScan(t *testing.T) {
}{
{
name: "Single Index Search using POST without entity returns nothing",
- unresolvedPlan: logical.IndexScan(sT, eT, metadata, []logical.Expr{
+ unresolvedPlan: logical.TagFilter(sT, eT, metadata, []logical.Expr{
logical.Eq(logical.NewTagRef("searchable", "http.method"), logical.Str("POST")),
}, tsdb.Entity{tsdb.AnyEntry, tsdb.AnyEntry, tsdb.AnyEntry}, nil),
wantLength: 0,
},
{
name: "Single Index Search using inverted index",
- unresolvedPlan: logical.IndexScan(sT, eT, metadata, []logical.Expr{
+ unresolvedPlan: logical.TagFilter(sT, eT, metadata, []logical.Expr{
logical.Eq(logical.NewTagRef("searchable", "http.method"), logical.Str("GET")),
}, tsdb.Entity{tsdb.AnyEntry, tsdb.AnyEntry, tsdb.AnyEntry}, nil),
wantLength: 3,
},
{
name: "Single Index Search using LSM tree index",
- unresolvedPlan: logical.IndexScan(sT, eT, metadata, []logical.Expr{
+ unresolvedPlan: logical.TagFilter(sT, eT, metadata, []logical.Expr{
logical.Lt(logical.NewTagRef("searchable", "duration"), logical.Int(100)),
}, tsdb.Entity{tsdb.AnyEntry, tsdb.AnyEntry, tsdb.AnyEntry}, nil),
wantLength: 2,
},
{
name: "Single Index Search without entity returns results",
- unresolvedPlan: logical.IndexScan(sT, eT, metadata, []logical.Expr{
+ unresolvedPlan: logical.TagFilter(sT, eT, metadata, []logical.Expr{
logical.Eq(logical.NewTagRef("searchable", "endpoint_id"), logical.Str("/home_id")),
}, tsdb.Entity{tsdb.AnyEntry, tsdb.AnyEntry, tsdb.AnyEntry}, nil),
wantLength: 2,
},
{
name: "Multiple Index Search",
- unresolvedPlan: logical.IndexScan(sT, eT, metadata, []logical.Expr{
+ unresolvedPlan: logical.TagFilter(sT, eT, metadata, []logical.Expr{
logical.Eq(logical.NewTagRef("searchable", "http.method"), logical.Str("GET")),
logical.Eq(logical.NewTagRef("searchable", "endpoint_id"), logical.Str("/home_id")),
}, tsdb.Entity{tsdb.AnyEntry, tsdb.AnyEntry, tsdb.AnyEntry}, nil),
@@ -264,7 +279,7 @@ func TestPlanExecution_IndexScan(t *testing.T) {
},
{
name: "Multiple Index Search with a combination of numerical index and textual index",
- unresolvedPlan: logical.IndexScan(sT, eT, metadata, []logical.Expr{
+ unresolvedPlan: logical.TagFilter(sT, eT, metadata, []logical.Expr{
logical.Eq(logical.NewTagRef("searchable", "http.method"), logical.Str("GET")),
logical.Lt(logical.NewTagRef("searchable", "duration"), logical.Int(100)),
}, tsdb.Entity{tsdb.AnyEntry, tsdb.AnyEntry, tsdb.AnyEntry}, nil),
@@ -272,7 +287,7 @@ func TestPlanExecution_IndexScan(t *testing.T) {
},
{
name: "Multiple Index With One Empty Result(ChunkID)",
- unresolvedPlan: logical.IndexScan(sT, eT, metadata, []logical.Expr{
+ unresolvedPlan: logical.TagFilter(sT, eT, metadata, []logical.Expr{
logical.Eq(logical.NewTagRef("searchable", "http.method"), logical.Str("GET")),
logical.Eq(logical.NewTagRef("searchable", "endpoint_id"), logical.Str("/unknown")),
}, tsdb.Entity{tsdb.AnyEntry, tsdb.AnyEntry, tsdb.AnyEntry}, nil),
@@ -356,7 +371,7 @@ func TestPlanExecution_OrderBy(t *testing.T) {
tester.NotNil(schema)
if tt.targetIndexRule == "" {
- p, err := logical.IndexScan(sT, eT, metadata, nil, tsdb.Entity{tsdb.AnyEntry, tsdb.AnyEntry, tsdb.AnyEntry},
+ p, err := logical.TagFilter(sT, eT, metadata, nil, tsdb.Entity{tsdb.AnyEntry, tsdb.AnyEntry, tsdb.AnyEntry},
logical.OrderBy("", tt.sortDirection), logical.NewTags("searchable", "start_time")).
Analyze(schema)
tester.NoError(err)
@@ -368,7 +383,7 @@ func TestPlanExecution_OrderBy(t *testing.T) {
tester.True(logical.SortedByTimestamp(entities, tt.sortDirection))
} else {
- p, err := logical.IndexScan(sT, eT, metadata, nil, tsdb.Entity{tsdb.AnyEntry, tsdb.AnyEntry, tsdb.AnyEntry},
+ p, err := logical.TagFilter(sT, eT, metadata, nil, tsdb.Entity{tsdb.AnyEntry, tsdb.AnyEntry, tsdb.AnyEntry},
logical.OrderBy(tt.targetIndexRule, tt.sortDirection), logical.NewTags("searchable", tt.targetIndexRule)).
Analyze(schema)
tester.NoError(err)
diff --git a/pkg/query/logical/stream_plan_indexscan_global.go b/pkg/query/logical/stream_plan_indexscan_global.go
index 48ca477..cfe5504 100644
--- a/pkg/query/logical/stream_plan_indexscan_global.go
+++ b/pkg/query/logical/stream_plan_indexscan_global.go
@@ -37,21 +37,21 @@ import (
var _ Plan = (*globalIndexScan)(nil)
type globalIndexScan struct {
- schema Schema
- metadata *commonv1.Metadata
- globalIndexRule *databasev1.IndexRule
- expr Expr
- projectionFieldRefs [][]*TagRef
+ schema Schema
+ metadata *commonv1.Metadata
+ globalIndexRule *databasev1.IndexRule
+ expr Expr
+ projectionTagRefs [][]*TagRef
}
func (t *globalIndexScan) String() string {
- if len(t.projectionFieldRefs) == 0 {
+ if len(t.projectionTagRefs) == 0 {
return fmt.Sprintf("GlobalIndexScan: Metadata{group=%s,name=%s},condition=%s; projection=None",
t.metadata.GetGroup(), t.metadata.GetName(), t.expr.String())
}
return fmt.Sprintf("GlobalIndexScan: Metadata{group=%s,name=%s},conditions=%s; projection=%s",
t.metadata.GetGroup(), t.metadata.GetName(),
- t.expr.String(), formatTagRefs(", ", t.projectionFieldRefs...))
+ t.expr.String(), formatTagRefs(", ", t.projectionTagRefs...))
}
func (t *globalIndexScan) Children() []Plan {
@@ -73,7 +73,7 @@ func (t *globalIndexScan) Equal(plan Plan) bool {
other := plan.(*globalIndexScan)
return t.metadata.GetGroup() == other.metadata.GetGroup() &&
t.metadata.GetName() == other.metadata.GetName() &&
- cmp.Equal(t.projectionFieldRefs, other.projectionFieldRefs) &&
+ cmp.Equal(t.projectionTagRefs, other.projectionTagRefs) &&
cmp.Equal(t.schema, other.schema) &&
cmp.Equal(t.globalIndexRule.GetMetadata().GetId(), other.globalIndexRule.GetMetadata().GetId()) &&
cmp.Equal(t.expr, other.expr)
@@ -126,7 +126,7 @@ func (t *globalIndexScan) executeForShard(ec executor.StreamExecutionContext, sh
if errInner != nil {
return errors.WithStack(errInner)
}
- tagFamilies, errInner := projectItem(ec, item, t.projectionFieldRefs)
+ tagFamilies, errInner := projectItem(ec, item, t.projectionTagRefs)
if errInner != nil {
return errors.WithStack(errInner)
}
diff --git a/pkg/query/logical/stream_plan_indexscan_local.go b/pkg/query/logical/stream_plan_indexscan_local.go
index ed15e62..bec23df 100644
--- a/pkg/query/logical/stream_plan_indexscan_local.go
+++ b/pkg/query/logical/stream_plan_indexscan_local.go
@@ -24,7 +24,6 @@ import (
"time"
"github.com/google/go-cmp/cmp"
- "github.com/pkg/errors"
"google.golang.org/protobuf/types/known/timestamppb"
commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
@@ -36,98 +35,16 @@ import (
"github.com/apache/skywalking-banyandb/pkg/timestamp"
)
-var _ UnresolvedPlan = (*unresolvedStreamIndexScan)(nil)
-
-type unresolvedStreamIndexScan struct {
- unresolvedOrderBy *UnresolvedOrderBy
- startTime time.Time
- endTime time.Time
- metadata *commonv1.Metadata
- conditions []Expr
- projectionFields [][]*Tag
- entity tsdb.Entity
-}
-
-func (uis *unresolvedStreamIndexScan) Analyze(s Schema) (Plan, error) {
- localConditionMap := make(map[*databasev1.IndexRule][]Expr)
- globalConditions := make([]interface{}, 0)
- for _, cond := range uis.conditions {
- if resolvable, ok := cond.(ResolvableExpr); ok {
- err := resolvable.Resolve(s)
- if err != nil {
- return nil, err
- }
-
- if bCond, ok := cond.(*binaryExpr); ok {
- tag := bCond.l.(*TagRef).tag
- if defined, indexObj := s.IndexDefined(tag); defined {
- if indexObj.GetLocation() == databasev1.IndexRule_LOCATION_SERIES {
- if v, exist := localConditionMap[indexObj]; exist {
- v = append(v, cond)
- localConditionMap[indexObj] = v
- } else {
- localConditionMap[indexObj] = []Expr{cond}
- }
- } else if indexObj.GetLocation() == databasev1.IndexRule_LOCATION_GLOBAL {
- globalConditions = append(globalConditions, indexObj, cond)
- }
- } else {
- return nil, errors.Wrap(ErrIndexNotDefined, tag.GetCompoundName())
- }
- }
- }
- }
-
- var projFieldsRefs [][]*TagRef
- if len(uis.projectionFields) > 0 {
- var err error
- projFieldsRefs, err = s.CreateTagRef(uis.projectionFields...)
- if err != nil {
- return nil, err
- }
- }
-
- if len(globalConditions) > 0 {
- if len(globalConditions)/2 > 1 {
- return nil, ErrMultipleGlobalIndexes
- }
- return &globalIndexScan{
- schema: s,
- projectionFieldRefs: projFieldsRefs,
- metadata: uis.metadata,
- globalIndexRule: globalConditions[0].(*databasev1.IndexRule),
- expr: globalConditions[1].(Expr),
- }, nil
- }
-
- // resolve sub-plan with the projected view of streamSchema
- orderBySubPlan, err := uis.unresolvedOrderBy.analyze(s.ProjTags(projFieldsRefs...))
-
- if err != nil {
- return nil, err
- }
-
- return &localIndexScan{
- orderBy: orderBySubPlan,
- timeRange: timestamp.NewInclusiveTimeRange(uis.startTime, uis.endTime),
- schema: s,
- projectionFieldRefs: projFieldsRefs,
- metadata: uis.metadata,
- conditionMap: localConditionMap,
- entity: uis.entity,
- }, nil
-}
-
var _ Plan = (*localIndexScan)(nil)
type localIndexScan struct {
*orderBy
- timeRange timestamp.TimeRange
- schema Schema
- metadata *commonv1.Metadata
- conditionMap map[*databasev1.IndexRule][]Expr
- projectionFieldRefs [][]*TagRef
- entity tsdb.Entity
+ timeRange timestamp.TimeRange
+ schema Schema
+ metadata *commonv1.Metadata
+ conditionMap map[*databasev1.IndexRule][]Expr
+ projectionTagRefs [][]*TagRef
+ entity tsdb.Entity
}
func (i *localIndexScan) Execute(ec executor.StreamExecutionContext) ([]*streamv1.Element, error) {
@@ -157,7 +74,7 @@ func (i *localIndexScan) Execute(ec executor.StreamExecutionContext) ([]*streamv
it := NewItemIter(iters, c)
for it.HasNext() {
nextItem := it.Next()
- tagFamilies, innerErr := projectItem(ec, nextItem, i.projectionFieldRefs)
+ tagFamilies, innerErr := projectItem(ec, nextItem, i.projectionTagRefs)
if innerErr != nil {
return nil, innerErr
}
@@ -216,13 +133,13 @@ func (i *localIndexScan) String() string {
}
exprStr = append(exprStr, fmt.Sprintf("(%s)", strings.Join(conditionStr, " AND ")))
}
- if len(i.projectionFieldRefs) == 0 {
+ if len(i.projectionTagRefs) == 0 {
return fmt.Sprintf("IndexScan: startTime=%d,endTime=%d,Metadata{group=%s,name=%s},conditions=%s; projection=None",
i.timeRange.Start.Unix(), i.timeRange.End.Unix(), i.metadata.GetGroup(), i.metadata.GetName(), strings.Join(exprStr, " AND "))
}
return fmt.Sprintf("IndexScan: startTime=%d,endTime=%d,Metadata{group=%s,name=%s},conditions=%s; projection=%s",
i.timeRange.Start.Unix(), i.timeRange.End.Unix(), i.metadata.GetGroup(), i.metadata.GetName(),
- strings.Join(exprStr, " AND "), formatTagRefs(", ", i.projectionFieldRefs...))
+ strings.Join(exprStr, " AND "), formatTagRefs(", ", i.projectionTagRefs...))
}
func (i *localIndexScan) Type() PlanType {
@@ -234,10 +151,10 @@ func (i *localIndexScan) Children() []Plan {
}
func (i *localIndexScan) Schema() Schema {
- if i.projectionFieldRefs == nil || len(i.projectionFieldRefs) == 0 {
+ if i.projectionTagRefs == nil || len(i.projectionTagRefs) == 0 {
return i.schema
}
- return i.schema.ProjTags(i.projectionFieldRefs...)
+ return i.schema.ProjTags(i.projectionTagRefs...)
}
func (i *localIndexScan) Equal(plan Plan) bool {
@@ -251,34 +168,12 @@ func (i *localIndexScan) Equal(plan Plan) bool {
i.timeRange.End.UnixNano() == other.timeRange.End.UnixNano() &&
len(i.entity) == len(other.entity) &&
bytes.Equal(i.entity.Marshal(), other.entity.Marshal()) &&
- cmp.Equal(i.projectionFieldRefs, other.projectionFieldRefs) &&
+ cmp.Equal(i.projectionTagRefs, other.projectionTagRefs) &&
cmp.Equal(i.schema, other.schema) &&
cmp.Equal(i.conditionMap, other.conditionMap) &&
cmp.Equal(i.orderBy, other.orderBy)
}
-func IndexScan(startTime, endTime time.Time, metadata *commonv1.Metadata, conditions []Expr, entity tsdb.Entity,
- orderBy *UnresolvedOrderBy, projection ...[]*Tag) UnresolvedPlan {
- return &unresolvedStreamIndexScan{
- unresolvedOrderBy: orderBy,
- startTime: startTime,
- endTime: endTime,
- metadata: metadata,
- conditions: conditions,
- projectionFields: projection,
- entity: entity,
- }
-}
-
-// GlobalIndexScan is a short-handed method for composing a globalIndexScan plan
-func GlobalIndexScan(metadata *commonv1.Metadata, conditions []Expr, projection ...[]*Tag) UnresolvedPlan {
- return &unresolvedStreamIndexScan{
- metadata: metadata,
- conditions: conditions,
- projectionFields: projection,
- }
-}
-
func exprToCondition(exprs []Expr) tsdb.Condition {
cond := make(map[string][]index.ConditionValue)
for _, expr := range exprs {
diff --git a/pkg/query/logical/stream_plan_tag_filter.go b/pkg/query/logical/stream_plan_tag_filter.go
new file mode 100644
index 0000000..d1479f0
--- /dev/null
+++ b/pkg/query/logical/stream_plan_tag_filter.go
@@ -0,0 +1,269 @@
+// 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.
+
+package logical
+
+import (
+ "time"
+
+ commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
+ databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+ modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
+ streamv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/stream/v1"
+ "github.com/apache/skywalking-banyandb/banyand/tsdb"
+ "github.com/apache/skywalking-banyandb/pkg/query/executor"
+ "github.com/apache/skywalking-banyandb/pkg/timestamp"
+)
+
+var _ UnresolvedPlan = (*unresolvedTagFilter)(nil)
+
+type unresolvedTagFilter struct {
+ unresolvedOrderBy *UnresolvedOrderBy
+ startTime time.Time
+ endTime time.Time
+ metadata *commonv1.Metadata
+ conditions []Expr
+ projectionTags [][]*Tag
+ entity tsdb.Entity
+}
+
+func (uis *unresolvedTagFilter) Analyze(s Schema) (Plan, error) {
+ ctx := newStreamAnalyzerContext(s)
+ for _, cond := range uis.conditions {
+ resolvable, ok := cond.(ResolvableExpr)
+ if !ok {
+ continue
+ }
+ err := resolvable.Resolve(s)
+ if err != nil {
+ return nil, err
+ }
+ bCond, ok := cond.(*binaryExpr)
+ if !ok {
+ continue
+ }
+ tag := bCond.l.(*TagRef).tag
+ defined, indexObj := s.IndexDefined(tag)
+ if !defined {
+ ctx.tagFilters = append(ctx.tagFilters, bCond)
+ continue
+ }
+ switch indexObj.GetLocation() {
+ case databasev1.IndexRule_LOCATION_SERIES:
+ if v, exist := ctx.localConditionMap[indexObj]; exist {
+ v = append(v, cond)
+ ctx.localConditionMap[indexObj] = v
+ } else {
+ ctx.localConditionMap[indexObj] = []Expr{cond}
+ }
+ case databasev1.IndexRule_LOCATION_GLOBAL:
+ ctx.globalConditions = append(ctx.globalConditions, indexObj, cond)
+ }
+ }
+
+ if len(uis.projectionTags) > 0 {
+ var err error
+ ctx.projTagsRefs, err = s.CreateTagRef(uis.projectionTags...)
+ if err != nil {
+ return nil, err
+ }
+ }
+ plan, err := uis.selectIndexScanner(ctx)
+ if err != nil {
+ return nil, err
+ }
+ if len(ctx.tagFilters) > 0 {
+ plan = NewTagFilter(s, plan, ctx.tagFilters)
+ }
+ return plan, err
+}
+
+func (uis *unresolvedTagFilter) selectIndexScanner(ctx *streamAnalyzeContext) (Plan, error) {
+ if len(ctx.globalConditions) > 0 {
+ if len(ctx.globalConditions)/2 > 1 {
+ return nil, ErrMultipleGlobalIndexes
+ }
+ return &globalIndexScan{
+ schema: ctx.s,
+ projectionTagRefs: ctx.projTagsRefs,
+ metadata: uis.metadata,
+ globalIndexRule: ctx.globalConditions[0].(*databasev1.IndexRule),
+ expr: ctx.globalConditions[1].(Expr),
+ }, nil
+ }
+
+ // resolve sub-plan with the projected view of streamSchema
+ orderBySubPlan, err := uis.unresolvedOrderBy.analyze(ctx.s.ProjTags(ctx.projTagsRefs...))
+
+ if err != nil {
+ return nil, err
+ }
+
+ return &localIndexScan{
+ orderBy: orderBySubPlan,
+ timeRange: timestamp.NewInclusiveTimeRange(uis.startTime, uis.endTime),
+ schema: ctx.s,
+ projectionTagRefs: ctx.projTagsRefs,
+ metadata: uis.metadata,
+ conditionMap: ctx.localConditionMap,
+ entity: uis.entity,
+ }, nil
+}
+
+func TagFilter(startTime, endTime time.Time, metadata *commonv1.Metadata, conditions []Expr, entity tsdb.Entity,
+ orderBy *UnresolvedOrderBy, projection ...[]*Tag) UnresolvedPlan {
+ return &unresolvedTagFilter{
+ unresolvedOrderBy: orderBy,
+ startTime: startTime,
+ endTime: endTime,
+ metadata: metadata,
+ conditions: conditions,
+ projectionTags: projection,
+ entity: entity,
+ }
+}
+
+// GlobalIndexScan is a short-handed method for composing a globalIndexScan plan
+func GlobalIndexScan(metadata *commonv1.Metadata, conditions []Expr, projection ...[]*Tag) UnresolvedPlan {
+ return &unresolvedTagFilter{
+ metadata: metadata,
+ conditions: conditions,
+ projectionTags: projection,
+ }
+}
+
+type streamAnalyzeContext struct {
+ s Schema
+ localConditionMap map[*databasev1.IndexRule][]Expr
+ globalConditions []interface{}
+ projTagsRefs [][]*TagRef
+ tagFilters []*binaryExpr
+}
+
+func newStreamAnalyzerContext(s Schema) *streamAnalyzeContext {
+ return &streamAnalyzeContext{
+ localConditionMap: make(map[*databasev1.IndexRule][]Expr),
+ globalConditions: make([]interface{}, 0),
+ s: s,
+ }
+}
+
+var (
+ _ Plan = (*tagFilter)(nil)
+ _ executor.StreamExecutable = (*tagFilter)(nil)
+)
+
+type tagFilter struct {
+ s Schema
+ parent Plan
+ tagFilters []*binaryExpr
+}
+
+func NewTagFilter(s Schema, parent Plan, tagFilters []*binaryExpr) Plan {
+ return &tagFilter{
+ s: s,
+ parent: parent,
+ tagFilters: tagFilters,
+ }
+}
+
+func (t *tagFilter) Execute(ec executor.StreamExecutionContext) ([]*streamv1.Element, error) {
+ entities, err := t.parent.(executor.StreamExecutable).Execute(ec)
+ if err != nil {
+ return nil, err
+ }
+ filteredElements := make([]*streamv1.Element, 0)
+ for _, e := range entities {
+ if t.check(e) {
+ filteredElements = append(filteredElements, e)
+ }
+ }
+ return filteredElements, nil
+}
+
+func (*tagFilter) String() string {
+ panic("unimplemented")
+}
+
+func (t *tagFilter) Children() []Plan {
+ return []Plan{t.parent}
+}
+
+func (*tagFilter) Equal(another Plan) bool {
+ panic("unimplemented")
+}
+
+func (t *tagFilter) Schema() Schema {
+ return t.s
+}
+
+func (*tagFilter) Type() PlanType {
+ return PlanTagFilter
+}
+
+func (t *tagFilter) check(element *streamv1.Element) bool {
+ compare := func(ce ComparableExpr, tagValue *modelv1.TagValue, exp func(result int) bool) bool {
+ c, b := ce.Compare(tagValue)
+ if b {
+ return exp(c)
+ }
+ return false
+ }
+ for _, filter := range t.tagFilters {
+ l := filter.l.(*TagRef)
+ tagValue, exist := tagValue(element, l)
+ if !exist {
+ return false
+ }
+ r, ok := filter.r.(ComparableExpr)
+ if !ok {
+ continue
+ }
+ switch filter.op {
+ case modelv1.Condition_BINARY_OP_EQ:
+ return compare(r, tagValue, func(c int) bool { return c == 0 })
+ case modelv1.Condition_BINARY_OP_GE:
+ return compare(r, tagValue, func(c int) bool { return c >= 0 })
+ case modelv1.Condition_BINARY_OP_GT:
+ return compare(r, tagValue, func(c int) bool { return c > 0 })
+ case modelv1.Condition_BINARY_OP_LE:
+ return compare(r, tagValue, func(c int) bool { return c <= 0 })
+ case modelv1.Condition_BINARY_OP_LT:
+ return compare(r, tagValue, func(c int) bool { return c < 0 })
+ case modelv1.Condition_BINARY_OP_HAVING:
+ return r.BelongTo(tagValue)
+ case modelv1.Condition_BINARY_OP_NOT_HAVING:
+ return !r.BelongTo(tagValue)
+ }
+ }
+
+ return true
+}
+
+func tagValue(element *streamv1.Element, tagRef *TagRef) (*modelv1.TagValue, bool) {
+ for _, tf := range element.TagFamilies {
+ if tf.Name != tagRef.tag.familyName {
+ continue
+ }
+ for _, t := range tf.Tags {
+ if t.Key == tagRef.tag.name {
+ return t.Value, true
+ }
+ }
+ }
+ return nil, false
+}
diff --git a/pkg/query/logical/testdata/global_index.json b/pkg/query/logical/testdata/global_index.json
index 73c0414..2020ff9 100644
--- a/pkg/query/logical/testdata/global_index.json
+++ b/pkg/query/logical/testdata/global_index.json
@@ -7,7 +7,10 @@
{"str":{"value": "10.0.0.1_id"}},
{"str":{"value": "/home_id"}},
{"int":{"value": 1000}},
- {"int":{"value": 1622933202000000000}}
+ {"int":{"value": 1622933202000000000}},
+ {"null":0},
+ {"null":0},
+ {"str":{"value": "1"}}
]
},
{
@@ -18,7 +21,10 @@
{"str":{"value": "10.0.0.3_id"}},
{"str":{"value": "/product_id"}},
{"int":{"value": 500}},
- {"int":{"value": 1622933202000000000}}
+ {"int":{"value": 1622933202000000000}},
+ {"null":0},
+ {"null":0},
+ {"str":{"value": "1"}}
]
},
{
@@ -31,7 +37,8 @@
{"int":{"value": 30}},
{"int":{"value": 1622933202000000000}},
{"str":{"value": "GET"}},
- {"int":{"value": 500}}
+ {"int":{"value": 500}},
+ {"str":{"value": "2"}}
]
},
{
@@ -44,7 +51,8 @@
{"int":{"value": 60}},
{"int":{"value": 1622933202000000000}},
{"str":{"value": "GET"}},
- {"int":{"value": 400}}
+ {"int":{"value": 400}},
+ {"str":{"value": "2"}}
]
},
{
@@ -57,7 +65,8 @@
{"int":{"value": 300}},
{"int":{"value": 1622933202000000000}},
{"str":{"value": "GET"}},
- {"int":{"value": 500}}
+ {"int":{"value": 500}},
+ {"str":{"value": "3"}}
]
}
diff --git a/pkg/test/stream/testdata/index_rule_binding.json b/pkg/test/stream/testdata/index_rule_binding.json
index a1cc114..e0d21b5 100644
--- a/pkg/test/stream/testdata/index_rule_binding.json
+++ b/pkg/test/stream/testdata/index_rule_binding.json
@@ -13,7 +13,8 @@
"db.type",
"mq.broker",
"mq.queue",
- "mq.topic"
+ "mq.topic",
+ "extended_tags"
],
"subject":{
"catalog": "CATALOG_STREAM",
diff --git a/pkg/test/stream/testdata/index_rules/extended_tags.json b/pkg/test/stream/testdata/index_rules/extended_tags.json
new file mode 100644
index 0000000..38269c5
--- /dev/null
+++ b/pkg/test/stream/testdata/index_rules/extended_tags.json
@@ -0,0 +1,13 @@
+{
+ "metadata": {
+ "id": 11,
+ "name": "extended_tags",
+ "group": "default"
+ },
+ "tags": [
+ "extended_tags"
+ ],
+ "type": "TYPE_INVERTED",
+ "location": "LOCATION_SERIES",
+ "updated_at": "2021-04-15T01:30:15.01Z"
+}
diff --git a/pkg/test/stream/testdata/stream.json b/pkg/test/stream/testdata/stream.json
index 313dad1..6a3f303 100644
--- a/pkg/test/stream/testdata/stream.json
+++ b/pkg/test/stream/testdata/stream.json
@@ -52,6 +52,10 @@
"name": "status_code",
"type": "TAG_TYPE_INT"
},
+ {
+ "name": "span_id",
+ "type": "TAG_TYPE_STRING"
+ },
{
"name": "db.type",
"type": "TAG_TYPE_STRING"
@@ -71,6 +75,10 @@
{
"name": "mq.broker",
"type": "TAG_TYPE_STRING"
+ },
+ {
+ "name": "extended_tags",
+ "type": "TAG_TYPE_STRING_ARRAY"
}
]
}