hanahmily commented on code in PR #665:
URL:
https://github.com/apache/skywalking-banyandb/pull/665#discussion_r2114034672
##########
pkg/query/logical/stream/index_filter.go:
##########
@@ -113,7 +117,10 @@ func parseConditionToFilter(cond *modelv1.Condition,
indexRule *databasev1.Index
case modelv1.Condition_BINARY_OP_EQ:
return newEq(indexRule, expr), [][]*modelv1.TagValue{entity},
nil
case modelv1.Condition_BINARY_OP_MATCH:
- return newMatch(indexRule, expr, cond.MatchOption),
[][]*modelv1.TagValue{entity}, nil
+ if indexRule.Type == databasev1.IndexRule_TYPE_INVERTED {
+ return newMatch(indexRule, expr, cond.MatchOption),
[][]*modelv1.TagValue{entity}, nil
+ }
+ return ENode, [][]*modelv1.TagValue{entity}, nil
Review Comment:
You should return error if it's not an inverted index. Tag filter can not
handle the matching operations.
##########
banyand/stream/block.go:
##########
@@ -103,6 +105,21 @@ func (b *block) processTags(tf tagValues, tagFamilyIdx, i
int, elementsLen int)
tags[j].resizeValues(elementsLen)
tags[j].valueType = t.valueType
tags[j].values[i] = t.marshal()
+ if !t.indexed {
+ continue
+ }
+ if tags[j].filter == nil {
+ tags[j].filter = filter.NewBloomFilter(elementsLen)
Review Comment:
You should use the pool to cache the filter.
##########
pkg/query/logical/interface.go:
##########
@@ -58,4 +58,5 @@ type ComparableExpr interface {
Compare(LiteralExpr) (int, bool)
BelongTo(LiteralExpr) bool
Contains(LiteralExpr) bool
+ Match(LiteralExpr) bool
Review Comment:
Expr can not support Match because the written data is not tokenized, which
should be supported through the analyzer.
##########
banyand/stream/tag.go:
##########
@@ -39,6 +40,10 @@ func (t *tag) reset() {
values[i] = nil
}
t.values = values[:0]
+
+ t.filter = nil
+ t.min = t.min[:0]
+ t.max = t.max[:0]
Review Comment:
Create tagFilter.reset function
##########
banyand/stream/tag_metadata.go:
##########
@@ -28,26 +28,38 @@ import (
type tagMetadata struct {
name string
+ min []byte
+ max []byte
dataBlock
- valueType pbv1.ValueType
+ valueType pbv1.ValueType
+ filterBlock dataBlock
}
func (tm *tagMetadata) reset() {
tm.name = ""
tm.valueType = 0
tm.dataBlock.reset()
+ tm.min = tm.min[:0]
+ tm.max = tm.max[:0]
Review Comment:
```suggestion
tm.min = nil
tm.max = nil
```
##########
banyand/stream/tag_filter.go:
##########
@@ -0,0 +1,200 @@
+// 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 stream
+
+import (
+ "bytes"
+
+ pkgbytes "github.com/apache/skywalking-banyandb/pkg/bytes"
+ "github.com/apache/skywalking-banyandb/pkg/encoding"
+ "github.com/apache/skywalking-banyandb/pkg/filter"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
+ "github.com/apache/skywalking-banyandb/pkg/index"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
+ "github.com/apache/skywalking-banyandb/pkg/pool"
+)
+
+func encodeBloomFilter(dst []byte, bf *filter.BloomFilter) []byte {
+ dst = encoding.Int64ToBytes(dst, int64(bf.N()))
+ dst = encoding.EncodeUint64Block(dst, bf.Bits())
+ return dst
+}
+
+func decodeBloomFilter(src []byte, bf *filter.BloomFilter) *filter.BloomFilter
{
+ n := encoding.BytesToInt64(src)
+ bf.SetN(int(n))
+
+ m := n * filter.B
+ bits := make([]uint64, 0)
+ bits, _, err := encoding.DecodeUint64Block(bits[:0], src[8:],
uint64((m+63)/64))
+ if err != nil {
+ logger.Panicf("failed to decode Bloom filter: %v", err)
+ }
+ bf.SetBits(bits)
+
+ return bf
+}
+
+func generateBloomFilter() *filter.BloomFilter {
+ v := bloomFilterPool.Get()
+ if v == nil {
+ return filter.NewBloomFilter(0)
+ }
+ return v
+}
+
+func releaseBloomFilter(bf *filter.BloomFilter) {
+ bf.Reset()
+ bloomFilterPool.Put(bf)
+}
+
+var bloomFilterPool = pool.Register[*filter.BloomFilter]("stream-bloomFilter")
+
+type tagFilter struct {
+ filter *filter.BloomFilter
+ min []byte
+ max []byte
+}
+
+type tagFamilyFilter map[string]*tagFilter
+
+func (tff *tagFamilyFilter) reset() {
+ clear(*tff)
+}
+
+func (tff tagFamilyFilter) unmarshal(tagFamilyMetadataBlock *dataBlock,
metaReader, filterReader fs.Reader) {
+ bb := bigValuePool.Generate()
+ bb.Buf = pkgbytes.ResizeExact(bb.Buf, int(tagFamilyMetadataBlock.size))
+ fs.MustReadData(metaReader, int64(tagFamilyMetadataBlock.offset),
bb.Buf)
+ tfm := generateTagFamilyMetadata()
+ defer releaseTagFamilyMetadata(tfm)
+ err := tfm.unmarshal(bb.Buf)
+ if err != nil {
+ logger.Panicf("%s: cannot unmarshal tagFamilyMetadata: %v",
metaReader.Path(), err)
+ }
+ bigValuePool.Release(bb)
+ for _, tm := range tfm.tagMetadata {
+ if tm.filterBlock.size == 0 {
+ continue
+ }
+ bb.Buf = pkgbytes.ResizeExact(bb.Buf, int(tm.filterBlock.size))
+ fs.MustReadData(filterReader, int64(tm.filterBlock.offset),
bb.Buf)
+ bf := generateBloomFilter()
+ bf = decodeBloomFilter(bb.Buf, bf)
+ tf := &tagFilter{
+ filter: bf,
+ }
+ if tm.valueType == pbv1.ValueTypeInt64 {
+ tf.min = tm.min
+ tf.max = tm.max
+ }
+ tff[tm.name] = tf
+ }
+}
+
+func generateTagFamilyFilter() *tagFamilyFilter {
+ v := tagFamilyFilterPool.Get()
+ if v == nil {
+ return &tagFamilyFilter{}
+ }
+ return v
+}
+
+func releaseTagFamilyFilter(tff *tagFamilyFilter) {
+ for _, tf := range *tff {
+ releaseBloomFilter(tf.filter)
+ }
+ tff.reset()
+ tagFamilyFilterPool.Put(tff)
+}
+
+var tagFamilyFilterPool =
pool.Register[*tagFamilyFilter]("stream-tagFamilyFilter")
+
+type tagFamilyFilters struct {
+ tagFamilyFilters []*tagFamilyFilter
+}
+
+func (tfs *tagFamilyFilters) reset() {
+ tfs.tagFamilyFilters = tfs.tagFamilyFilters[:0]
+}
+
+func (tfs *tagFamilyFilters) unmarshal(tagFamilies map[string]*dataBlock,
metaReader, filterReader map[string]fs.Reader) {
+ for tf := range tagFamilies {
+ tff := generateTagFamilyFilter()
+ tff.unmarshal(tagFamilies[tf], metaReader[tf], filterReader[tf])
+ tfs.tagFamilyFilters = append(tfs.tagFamilyFilters, tff)
+ }
+}
+
+func (tfs *tagFamilyFilters) Eq(tagName string, tagValue string) bool {
+ for _, tff := range tfs.tagFamilyFilters {
+ if tf, ok := (*tff)[tagName]; ok {
+ return tf.filter.MightContain([]byte(tagValue))
+ }
+ }
+ return true
+}
+
+func (tfs *tagFamilyFilters) Range(tagName string, rangeOpts index.RangeOpts)
bool {
+ for _, tff := range tfs.tagFamilyFilters {
+ if tf, ok := (*tff)[tagName]; ok {
+ if rangeOpts.Lower != nil {
+ lower, ok :=
rangeOpts.Lower.(*index.FloatTermValue)
+ if !ok {
+ logger.Panicf("lower is not a float
value: %v", rangeOpts.Lower)
Review Comment:
You should return false instead of a panic.
##########
banyand/stream/tag_filter.go:
##########
@@ -0,0 +1,200 @@
+// 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 stream
+
+import (
+ "bytes"
+
+ pkgbytes "github.com/apache/skywalking-banyandb/pkg/bytes"
+ "github.com/apache/skywalking-banyandb/pkg/encoding"
+ "github.com/apache/skywalking-banyandb/pkg/filter"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
+ "github.com/apache/skywalking-banyandb/pkg/index"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
+ "github.com/apache/skywalking-banyandb/pkg/pool"
+)
+
+func encodeBloomFilter(dst []byte, bf *filter.BloomFilter) []byte {
+ dst = encoding.Int64ToBytes(dst, int64(bf.N()))
+ dst = encoding.EncodeUint64Block(dst, bf.Bits())
+ return dst
+}
+
+func decodeBloomFilter(src []byte, bf *filter.BloomFilter) *filter.BloomFilter
{
+ n := encoding.BytesToInt64(src)
+ bf.SetN(int(n))
+
+ m := n * filter.B
+ bits := make([]uint64, 0)
+ bits, _, err := encoding.DecodeUint64Block(bits[:0], src[8:],
uint64((m+63)/64))
+ if err != nil {
+ logger.Panicf("failed to decode Bloom filter: %v", err)
+ }
+ bf.SetBits(bits)
+
+ return bf
+}
+
+func generateBloomFilter() *filter.BloomFilter {
+ v := bloomFilterPool.Get()
+ if v == nil {
+ return filter.NewBloomFilter(0)
+ }
+ return v
+}
+
+func releaseBloomFilter(bf *filter.BloomFilter) {
+ bf.Reset()
+ bloomFilterPool.Put(bf)
+}
+
+var bloomFilterPool = pool.Register[*filter.BloomFilter]("stream-bloomFilter")
+
+type tagFilter struct {
+ filter *filter.BloomFilter
+ min []byte
+ max []byte
+}
+
+type tagFamilyFilter map[string]*tagFilter
+
+func (tff *tagFamilyFilter) reset() {
+ clear(*tff)
+}
+
+func (tff tagFamilyFilter) unmarshal(tagFamilyMetadataBlock *dataBlock,
metaReader, filterReader fs.Reader) {
+ bb := bigValuePool.Generate()
+ bb.Buf = pkgbytes.ResizeExact(bb.Buf, int(tagFamilyMetadataBlock.size))
+ fs.MustReadData(metaReader, int64(tagFamilyMetadataBlock.offset),
bb.Buf)
+ tfm := generateTagFamilyMetadata()
+ defer releaseTagFamilyMetadata(tfm)
+ err := tfm.unmarshal(bb.Buf)
+ if err != nil {
+ logger.Panicf("%s: cannot unmarshal tagFamilyMetadata: %v",
metaReader.Path(), err)
+ }
+ bigValuePool.Release(bb)
+ for _, tm := range tfm.tagMetadata {
+ if tm.filterBlock.size == 0 {
+ continue
+ }
+ bb.Buf = pkgbytes.ResizeExact(bb.Buf, int(tm.filterBlock.size))
+ fs.MustReadData(filterReader, int64(tm.filterBlock.offset),
bb.Buf)
+ bf := generateBloomFilter()
+ bf = decodeBloomFilter(bb.Buf, bf)
+ tf := &tagFilter{
Review Comment:
You should add a pool for tagFilter.
##########
banyand/stream/part_iter.go:
##########
@@ -221,13 +226,26 @@ func (pi *partIter) findBlock() bool {
bhs = bhs[1:]
continue
}
+
if bm.timestamps.min > pi.maxTimestamp {
if !pi.nextSeriesID() {
return false
}
continue
}
+ if pi.blockFilter != nil && pi.blockFilter !=
logicalstream.ENode {
+ tfs := generateTagFamilyFilters()
+ defer releaseTagFamilyFilters(tfs)
+ tfs.unmarshal(bm.tagFamilies, pi.p.tagFamilyMetadata,
pi.p.tagFamilyFilter)
+ if !pi.blockFilter.ShouldNotSkip(tfs) {
Review Comment:
Could you change the function name to 'ShouldSkip'? It makes more sense.
##########
banyand/stream/tag_filter.go:
##########
@@ -0,0 +1,200 @@
+// 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 stream
+
+import (
+ "bytes"
+
+ pkgbytes "github.com/apache/skywalking-banyandb/pkg/bytes"
+ "github.com/apache/skywalking-banyandb/pkg/encoding"
+ "github.com/apache/skywalking-banyandb/pkg/filter"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
+ "github.com/apache/skywalking-banyandb/pkg/index"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
+ "github.com/apache/skywalking-banyandb/pkg/pool"
+)
+
+func encodeBloomFilter(dst []byte, bf *filter.BloomFilter) []byte {
+ dst = encoding.Int64ToBytes(dst, int64(bf.N()))
+ dst = encoding.EncodeUint64Block(dst, bf.Bits())
+ return dst
+}
+
+func decodeBloomFilter(src []byte, bf *filter.BloomFilter) *filter.BloomFilter
{
+ n := encoding.BytesToInt64(src)
+ bf.SetN(int(n))
+
+ m := n * filter.B
+ bits := make([]uint64, 0)
+ bits, _, err := encoding.DecodeUint64Block(bits[:0], src[8:],
uint64((m+63)/64))
+ if err != nil {
+ logger.Panicf("failed to decode Bloom filter: %v", err)
+ }
+ bf.SetBits(bits)
+
+ return bf
+}
+
+func generateBloomFilter() *filter.BloomFilter {
+ v := bloomFilterPool.Get()
+ if v == nil {
+ return filter.NewBloomFilter(0)
+ }
+ return v
+}
+
+func releaseBloomFilter(bf *filter.BloomFilter) {
+ bf.Reset()
+ bloomFilterPool.Put(bf)
+}
+
+var bloomFilterPool = pool.Register[*filter.BloomFilter]("stream-bloomFilter")
+
+type tagFilter struct {
+ filter *filter.BloomFilter
+ min []byte
+ max []byte
+}
+
+type tagFamilyFilter map[string]*tagFilter
+
+func (tff *tagFamilyFilter) reset() {
+ clear(*tff)
+}
+
+func (tff tagFamilyFilter) unmarshal(tagFamilyMetadataBlock *dataBlock,
metaReader, filterReader fs.Reader) {
+ bb := bigValuePool.Generate()
+ bb.Buf = pkgbytes.ResizeExact(bb.Buf, int(tagFamilyMetadataBlock.size))
+ fs.MustReadData(metaReader, int64(tagFamilyMetadataBlock.offset),
bb.Buf)
+ tfm := generateTagFamilyMetadata()
+ defer releaseTagFamilyMetadata(tfm)
+ err := tfm.unmarshal(bb.Buf)
+ if err != nil {
+ logger.Panicf("%s: cannot unmarshal tagFamilyMetadata: %v",
metaReader.Path(), err)
+ }
+ bigValuePool.Release(bb)
+ for _, tm := range tfm.tagMetadata {
+ if tm.filterBlock.size == 0 {
+ continue
+ }
+ bb.Buf = pkgbytes.ResizeExact(bb.Buf, int(tm.filterBlock.size))
+ fs.MustReadData(filterReader, int64(tm.filterBlock.offset),
bb.Buf)
+ bf := generateBloomFilter()
+ bf = decodeBloomFilter(bb.Buf, bf)
+ tf := &tagFilter{
+ filter: bf,
+ }
+ if tm.valueType == pbv1.ValueTypeInt64 {
+ tf.min = tm.min
+ tf.max = tm.max
+ }
+ tff[tm.name] = tf
+ }
+}
+
+func generateTagFamilyFilter() *tagFamilyFilter {
+ v := tagFamilyFilterPool.Get()
+ if v == nil {
+ return &tagFamilyFilter{}
+ }
+ return v
+}
+
+func releaseTagFamilyFilter(tff *tagFamilyFilter) {
+ for _, tf := range *tff {
+ releaseBloomFilter(tf.filter)
+ }
+ tff.reset()
+ tagFamilyFilterPool.Put(tff)
+}
+
+var tagFamilyFilterPool =
pool.Register[*tagFamilyFilter]("stream-tagFamilyFilter")
+
+type tagFamilyFilters struct {
+ tagFamilyFilters []*tagFamilyFilter
+}
+
+func (tfs *tagFamilyFilters) reset() {
+ tfs.tagFamilyFilters = tfs.tagFamilyFilters[:0]
+}
+
+func (tfs *tagFamilyFilters) unmarshal(tagFamilies map[string]*dataBlock,
metaReader, filterReader map[string]fs.Reader) {
Review Comment:
Could you add a benchmark to verify the allocation status of unmarshal?
##########
banyand/stream/part_iter.go:
##########
@@ -221,13 +226,26 @@ func (pi *partIter) findBlock() bool {
bhs = bhs[1:]
continue
}
+
if bm.timestamps.min > pi.maxTimestamp {
if !pi.nextSeriesID() {
return false
}
continue
}
+ if pi.blockFilter != nil && pi.blockFilter !=
logicalstream.ENode {
+ tfs := generateTagFamilyFilters()
+ defer releaseTagFamilyFilters(tfs)
Review Comment:
Do not use a defer in a loop.
##########
banyand/internal/storage/versions.yml:
##########
@@ -14,4 +14,5 @@
# limitations under the License.
versions:
+- 1.3.0
- 1.2.0
Review Comment:
You can not support 1.2.0 anymore. The relevant unmarshal operations does
not support 1.2.0
##########
pkg/query/logical/expr_literal.go:
##########
@@ -266,6 +276,14 @@ func (s *strLiteral) BelongTo(other LiteralExpr) bool {
return false
}
+func (s *strLiteral) Match(other LiteralExpr) bool {
+ if o, ok := other.(*strLiteral); ok {
Review Comment:
strLiteral can not support matching since the written str is not tokenized.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscr...@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
