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 2024/02/22 00:07:56 UTC
(skywalking-banyandb) branch main updated: get entity value from series index when query stream data (#386)
This is an automated email from the ASF dual-hosted git repository.
hanahmily 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 26df3134 get entity value from series index when query stream data (#386)
26df3134 is described below
commit 26df313468d2b545fdd3704c244e0a2a2011cce6
Author: A3bz <59...@users.noreply.github.com>
AuthorDate: Thu Feb 22 08:07:52 2024 +0800
get entity value from series index when query stream data (#386)
* feat: when query stream data, get entity value from series index
* fix: correct stream query
---------
Signed-off-by: A3bz <ze...@163.com>
Co-authored-by: Gao Hongtao <ha...@gmail.com>
---
banyand/measure/query.go | 3 +
banyand/measure/write.go | 25 +++-----
banyand/stream/block.go | 2 +-
banyand/stream/iter.go | 5 +-
banyand/stream/part.go | 10 +--
banyand/stream/query.go | 160 ++++++++++++++++++++++++++++++++++++++++++----
banyand/stream/tstable.go | 8 +--
banyand/stream/write.go | 21 +++---
8 files changed, 185 insertions(+), 49 deletions(-)
diff --git a/banyand/measure/query.go b/banyand/measure/query.go
index f140af13..ece4e1b1 100644
--- a/banyand/measure/query.go
+++ b/banyand/measure/query.go
@@ -393,6 +393,9 @@ func (qr *queryResult) Pull() *pbv1.MeasureResult {
offset := qr.tagNameIndex[tagProj]
tagFamilySpec := qr.schema.GetTagFamilies()[offset.FamilyOffset]
tagSpec := tagFamilySpec.GetTags()[offset.TagOffset]
+ if tagSpec.IndexedOnly {
+ continue
+ }
valueType := pbv1.MustTagValueToValueType(series.EntityValues[entityPos-1])
qr.data[i].tagFamilies[tagFamilyPos-1].columns = append(qr.data[i].tagFamilies[tagFamilyPos-1].columns[:j],
append([]column{{
diff --git a/banyand/measure/write.go b/banyand/measure/write.go
index e7f16f19..1b9b2095 100644
--- a/banyand/measure/write.go
+++ b/banyand/measure/write.go
@@ -126,6 +126,7 @@ func (w *writeCallback) handle(dst map[string]*dataPointsInGroup, writeEvent *me
}
dpt.dataPoints.fields = append(dpt.dataPoints.fields, field)
tagFamilies := make([]nameValues, len(stm.schema.TagFamilies))
+ tagFamiliesForIndexWrite := make([]nameValues, len(stm.schema.TagFamilies))
dpt.dataPoints.tagFamilies = append(dpt.dataPoints.tagFamilies, tagFamilies)
entityMap := make(map[string]bool)
@@ -148,11 +149,16 @@ func (w *writeCallback) handle(dst map[string]*dataPointsInGroup, writeEvent *me
} else {
tagValue = tagFamily.Tags[j]
}
- tagFamilies[i].values = append(tagFamilies[i].values, encodeTagValue(
+
+ nameValue := encodeTagValue(
tagFamilySpec.Tags[j].Name,
tagFamilySpec.Tags[j].Type,
- tagValue,
- ))
+ tagValue)
+ tagFamiliesForIndexWrite[i].values = append(tagFamiliesForIndexWrite[i].values, nameValue)
+ if tagFamilySpec.Tags[j].IndexedOnly || entityMap[tagFamilySpec.Tags[j].Name] {
+ continue
+ }
+ tagFamilies[i].values = append(tagFamilies[i].values, nameValue)
}
}
@@ -168,7 +174,7 @@ func (w *writeCallback) handle(dst map[string]*dataPointsInGroup, writeEvent *me
}
var fields []index.Field
for _, ruleIndex := range stm.indexRuleLocators {
- nv := getIndexValue(ruleIndex, tagFamilies)
+ nv := getIndexValue(ruleIndex, tagFamiliesForIndexWrite)
if nv == nil {
continue
}
@@ -193,17 +199,6 @@ func (w *writeCallback) handle(dst map[string]*dataPointsInGroup, writeEvent *me
})
}
}
-
- for i := range tagFamilies {
- values := make([]*nameValue, len(tagFamilies[i].values))
- copy(values, tagFamilies[i].values)
- tagFamilies[i].values = tagFamilies[i].values[:0]
- for j, tagValue := range values {
- if !entityMap[tagValue.name] {
- tagFamilies[i].values = append(tagFamilies[i].values, values[j])
- }
- }
- }
dpg.docs = append(dpg.docs, index.Document{
DocID: uint64(series.ID),
EntityValues: series.Buffer,
diff --git a/banyand/stream/block.go b/banyand/stream/block.go
index 75ef388c..ee254902 100644
--- a/banyand/stream/block.go
+++ b/banyand/stream/block.go
@@ -566,8 +566,8 @@ func (bc *blockCursor) loadData(tmpBlock *block) bool {
tmpBlock.tagFamilies[i].tags[blockIndex].name, len(tmpBlock.tagFamilies[i].tags[blockIndex].values), len(tmpBlock.timestamps))
}
t.values = append(t.values, tmpBlock.tagFamilies[i].tags[blockIndex].values[start:end+1]...)
- blockIndex++
}
+ blockIndex++
tf.tags = append(tf.tags, t)
}
bc.tagFamilies = append(bc.tagFamilies, tf)
diff --git a/banyand/stream/iter.go b/banyand/stream/iter.go
index 0c9d2662..ed36be2c 100644
--- a/banyand/stream/iter.go
+++ b/banyand/stream/iter.go
@@ -99,8 +99,9 @@ type item struct {
seriesID common.SeriesID
}
-func (i *item) Element() (*element, error) {
- return i.table.getElement(i.seriesID, i.itemID, i.tagProjection)
+func (i *item) Element() (*element, int, error) {
+ e, count, err := i.table.getElement(i.seriesID, i.itemID, i.tagProjection)
+ return e, count, err
}
func (i *item) Time() uint64 {
diff --git a/banyand/stream/part.go b/banyand/stream/part.go
index 2484817c..bdca61a1 100644
--- a/banyand/stream/part.go
+++ b/banyand/stream/part.go
@@ -139,7 +139,7 @@ func (p *part) String() string {
return fmt.Sprintf("part %d", p.partMetadata.ID)
}
-func (p *part) getElement(seriesID common.SeriesID, timestamp common.ItemID, tagProjection []pbv1.TagProjection) (*element, error) {
+func (p *part) getElement(seriesID common.SeriesID, timestamp common.ItemID, tagProjection []pbv1.TagProjection) (*element, int, error) {
// TODO: refactor to column-based query
// TODO: cache blocks
for i, primaryMeta := range p.primaryBlockMetadata {
@@ -158,12 +158,12 @@ func (p *part) getElement(seriesID common.SeriesID, timestamp common.ItemID, tag
primaryBuf := make([]byte, 0)
primaryBuf, err = zstd.Decompress(primaryBuf[:0], compressedPrimaryBuf)
if err != nil {
- return nil, fmt.Errorf("cannot decompress index block: %w", err)
+ return nil, 0, fmt.Errorf("cannot decompress index block: %w", err)
}
bm := make([]blockMetadata, 0)
bm, err = unmarshalBlockMetadata(bm, primaryBuf)
if err != nil {
- return nil, fmt.Errorf("cannot unmarshal index block: %w", err)
+ return nil, 0, fmt.Errorf("cannot unmarshal index block: %w", err)
}
var targetBlockMetadata blockMetadata
for _, blockMetadata := range bm {
@@ -196,14 +196,14 @@ func (p *part) getElement(seriesID common.SeriesID, timestamp common.ItemID, tag
elementID: elementIDs[i],
tagFamilies: tfs,
index: i,
- }, nil
+ }, len(timestamps), nil
}
if common.ItemID(ts) > timestamp {
break
}
}
}
- return nil, errors.New("element not found")
+ return nil, 0, errors.New("element not found")
}
func unmarshalTagFamily(decoder *encoding.BytesBlockDecoder, name string,
diff --git a/banyand/stream/query.go b/banyand/stream/query.go
index de0d6595..f8ae2167 100644
--- a/banyand/stream/query.go
+++ b/banyand/stream/query.go
@@ -27,11 +27,13 @@ import (
"go.uber.org/multierr"
"github.com/apache/skywalking-banyandb/api/common"
+ 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/banyand/internal/storage"
"github.com/apache/skywalking-banyandb/pkg/convert"
itersort "github.com/apache/skywalking-banyandb/pkg/iter/sort"
"github.com/apache/skywalking-banyandb/pkg/logger"
+ "github.com/apache/skywalking-banyandb/pkg/partition"
pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
)
@@ -41,6 +43,16 @@ type queryOptions struct {
maxTimestamp int64
}
+func mustEncodeTagValue(name string, tagType databasev1.TagType, tagValue *modelv1.TagValue, num int) [][]byte {
+ values := make([][]byte, num)
+ nv := encodeTagValue(name, tagType, tagValue)
+ value := nv.marshal()
+ for i := 0; i < num; i++ {
+ values[i] = value
+ }
+ return values
+}
+
func mustDecodeTagValue(valueType pbv1.ValueType, value []byte) *modelv1.TagValue {
if value == nil {
switch valueType {
@@ -140,12 +152,16 @@ func strArrTagValue(values []string) *modelv1.TagValue {
}
type queryResult struct {
- sidToIndex map[common.SeriesID]int
- data []*blockCursor
- snapshots []*snapshot
- loaded bool
- orderByTS bool
- ascTS bool
+ entityMap map[string]int
+ sidToIndex map[common.SeriesID]int
+ tagNameIndex map[string]partition.TagLocator
+ schema *databasev1.Stream
+ data []*blockCursor
+ snapshots []*snapshot
+ seriesList pbv1.SeriesList
+ loaded bool
+ orderByTS bool
+ ascTS bool
}
func (qr *queryResult) Pull() *pbv1.StreamResult {
@@ -161,6 +177,46 @@ func (qr *queryResult) Pull() *pbv1.StreamResult {
qr.data = append(qr.data[:i], qr.data[i+1:]...)
i--
}
+ if qr.schema.GetEntity() == nil || len(qr.schema.GetEntity().GetTagNames()) == 0 {
+ continue
+ }
+ sidIndex := qr.sidToIndex[qr.data[i].bm.seriesID]
+ series := qr.seriesList[sidIndex]
+ entityMap := make(map[string]int)
+ tagFamilyMap := make(map[string]int)
+ for idx, entity := range qr.schema.GetEntity().GetTagNames() {
+ entityMap[entity] = idx + 1
+ }
+ for idx, tagFamily := range qr.data[i].tagFamilies {
+ tagFamilyMap[tagFamily.name] = idx + 1
+ }
+ for _, tagFamilyProj := range qr.data[i].tagProjection {
+ for j, tagProj := range tagFamilyProj.Names {
+ offset := qr.tagNameIndex[tagProj]
+ tagFamilySpec := qr.schema.GetTagFamilies()[offset.FamilyOffset]
+ tagSpec := tagFamilySpec.GetTags()[offset.TagOffset]
+ if tagSpec.IndexedOnly {
+ continue
+ }
+ entityPos := entityMap[tagProj]
+ tagFamilyPos := tagFamilyMap[tagFamilyProj.Family]
+ if entityPos == 0 {
+ continue
+ }
+ if tagFamilyPos == 0 {
+ qr.data[i].tagFamilies[tagFamilyPos-1] = tagFamily{
+ name: tagFamilyProj.Family,
+ tags: make([]tag, 0),
+ }
+ }
+ valueType := pbv1.MustTagValueToValueType(series.EntityValues[entityPos-1])
+ qr.data[i].tagFamilies[tagFamilyPos-1].tags[j] = tag{
+ name: tagProj,
+ values: mustEncodeTagValue(tagProj, tagSpec.GetType(), series.EntityValues[entityPos-1], len(qr.data[i].timestamps)),
+ valueType: valueType,
+ }
+ }
+ }
if qr.orderByTimestampDesc() {
qr.data[i].idx = len(qr.data[i].timestamps) - 1
}
@@ -267,6 +323,39 @@ func (qr *queryResult) merge() *pbv1.StreamResult {
return result
}
+func (s *stream) genIndex(tagProj []pbv1.TagProjection, seriesList pbv1.SeriesList) (map[string]int, map[string]*databasev1.TagSpec,
+ map[string]partition.TagLocator, map[common.SeriesID]int,
+) {
+ entityMap := make(map[string]int)
+ for idx, entity := range s.schema.GetEntity().GetTagNames() {
+ entityMap[entity] = idx + 1
+ }
+ tagSpecIndex := make(map[string]*databasev1.TagSpec)
+ for _, tagFamilySpec := range s.schema.GetTagFamilies() {
+ for _, tagSpec := range tagFamilySpec.Tags {
+ tagSpecIndex[tagSpec.GetName()] = tagSpec
+ }
+ }
+ tagProjIndex := make(map[string]partition.TagLocator)
+ for i, tagFamilyProj := range tagProj {
+ for j, tagProj := range tagFamilyProj.Names {
+ if entityMap[tagProj] == 0 {
+ continue
+ }
+ tagProjIndex[tagProj] = partition.TagLocator{
+ FamilyOffset: i,
+ TagOffset: j,
+ }
+ }
+ }
+ sidToIndex := make(map[common.SeriesID]int)
+ for idx, series := range seriesList {
+ sidToIndex[series.ID] = idx
+ }
+
+ return entityMap, tagSpecIndex, tagProjIndex, sidToIndex
+}
+
func (s *stream) Filter(ctx context.Context, sfo pbv1.StreamFilterOptions) (sfr pbv1.StreamFilterResult, err error) {
if sfo.TimeRange == nil || sfo.Entities == nil {
return nil, errors.New("invalid query options: timeRange and series are required")
@@ -294,6 +383,7 @@ func (s *stream) Filter(ctx context.Context, sfo pbv1.StreamFilterOptions) (sfr
seriesList = seriesList.Merge(sl)
}
+ entityMap, tagSpecIndex, tagProjIndex, sidToIndex := s.genIndex(sfo.TagProjection, seriesList)
ces := newColumnElements()
for _, tw := range tabWrappers {
if len(ces.timestamp) >= sfo.MaxElementSize {
@@ -308,10 +398,25 @@ func (s *stream) Filter(ctx context.Context, sfo pbv1.StreamFilterOptions) (sfr
erl = erl[:sfo.MaxElementSize-len(ces.timestamp)]
}
for _, er := range erl {
- e, err := tw.Table().getElement(er.seriesID, common.ItemID(er.timestamp), sfo.TagProjection)
+ e, count, err := tw.Table().getElement(er.seriesID, common.ItemID(er.timestamp), sfo.TagProjection)
if err != nil {
return nil, err
}
+ if len(tagProjIndex) != 0 {
+ for entity, offset := range tagProjIndex {
+ tagSpec := tagSpecIndex[entity]
+ if tagSpec.IndexedOnly {
+ continue
+ }
+ series := seriesList[sidToIndex[er.seriesID]]
+ entityPos := entityMap[entity] - 1
+ e.tagFamilies[offset.FamilyOffset].tags[offset.TagOffset] = tag{
+ name: entity,
+ values: mustEncodeTagValue(entity, tagSpec.GetType(), series.EntityValues[entityPos], count),
+ valueType: pbv1.MustTagValueToValueType(series.EntityValues[entityPos]),
+ }
+ }
+ }
ces.BuildFromElement(e, sfo.TagProjection)
}
}
@@ -342,6 +447,8 @@ func (s *stream) Sort(ctx context.Context, sso pbv1.StreamSortOptions) (ssr pbv1
seriesList = seriesList.Merge(sl)
}
+ entityMap, tagSpecIndex, tagProjIndex, sidToIndex := s.genIndex(sso.TagProjection, seriesList)
+
var iters []*searcherIterator
for _, series := range seriesList {
seekerBuilder := newIterBuilder(tabWrappers, series.ID, sso)
@@ -366,10 +473,25 @@ func (s *stream) Sort(ctx context.Context, sso pbv1.StreamSortOptions) (ssr pbv1
ces := newColumnElements()
for it.Next() {
nextItem := it.Val()
- e, err := nextItem.Element()
+ e, count, err := nextItem.Element()
if err != nil {
return nil, err
}
+ if len(tagProjIndex) != 0 {
+ for entity, offset := range tagProjIndex {
+ tagSpec := tagSpecIndex[entity]
+ if tagSpec.IndexedOnly {
+ continue
+ }
+ series := seriesList[sidToIndex[nextItem.seriesID]]
+ entityPos := entityMap[entity] - 1
+ e.tagFamilies[offset.FamilyOffset].tags[offset.TagOffset] = tag{
+ name: entity,
+ values: mustEncodeTagValue(entity, tagSpec.GetType(), series.EntityValues[entityPos], count),
+ valueType: pbv1.MustTagValueToValueType(series.EntityValues[entityPos]),
+ }
+ }
+ }
ces.BuildFromElement(e, sso.TagProjection)
if len(ces.timestamp) >= sso.MaxElementSize {
break
@@ -442,6 +564,24 @@ func (s *stream) Query(ctx context.Context, sqo pbv1.StreamQueryOptions) (pbv1.S
if tstIter.Error() != nil {
return nil, fmt.Errorf("cannot iterate tstIter: %w", tstIter.Error())
}
+
+ entityMap, _, _, sidToIndex := s.genIndex(sqo.TagProjection, sl)
+ result.entityMap = entityMap
+ result.sidToIndex = sidToIndex
+ result.tagNameIndex = make(map[string]partition.TagLocator)
+ result.schema = s.schema
+ result.seriesList = sl
+ for i, si := range originalSids {
+ result.sidToIndex[si] = i
+ }
+ for i, tagFamilySpec := range s.schema.GetTagFamilies() {
+ for j, tagSpec := range tagFamilySpec.GetTags() {
+ result.tagNameIndex[tagSpec.GetName()] = partition.TagLocator{
+ FamilyOffset: i,
+ TagOffset: j,
+ }
+ }
+ }
if sqo.Order == nil {
result.orderByTS = true
result.ascTS = true
@@ -455,10 +595,6 @@ func (s *stream) Query(ctx context.Context, sqo pbv1.StreamQueryOptions) (pbv1.S
return &result, nil
}
- result.sidToIndex = make(map[common.SeriesID]int)
- for i, si := range originalSids {
- result.sidToIndex[si] = i
- }
return &result, nil
}
diff --git a/banyand/stream/tstable.go b/banyand/stream/tstable.go
index aefd3076..3dc584eb 100644
--- a/banyand/stream/tstable.go
+++ b/banyand/stream/tstable.go
@@ -267,19 +267,19 @@ func (tst *tsTable) mustAddElements(es *elements) {
}
}
-func (tst *tsTable) getElement(seriesID common.SeriesID, timestamp common.ItemID, tagProjection []pbv1.TagProjection) (*element, error) {
+func (tst *tsTable) getElement(seriesID common.SeriesID, timestamp common.ItemID, tagProjection []pbv1.TagProjection) (*element, int, error) {
tst.RLock()
defer tst.RUnlock()
for _, p := range tst.currentSnapshot().parts {
if !p.p.containTimestamp(timestamp) {
continue
}
- elem, err := p.p.getElement(seriesID, timestamp, tagProjection)
+ elem, count, err := p.p.getElement(seriesID, timestamp, tagProjection)
if err == nil {
- return elem, nil
+ return elem, count, nil
}
}
- return nil, fmt.Errorf("cannot find element with seriesID %d and timestamp %d", seriesID, timestamp)
+ return nil, 0, fmt.Errorf("cannot find element with seriesID %d and timestamp %d", seriesID, timestamp)
}
type tstIter struct {
diff --git a/banyand/stream/write.go b/banyand/stream/write.go
index 6672bee9..4cc06aa4 100644
--- a/banyand/stream/write.go
+++ b/banyand/stream/write.go
@@ -113,6 +113,11 @@ func (w *writeCallback) handle(dst map[string]*elementsInGroup, writeEvent *stre
tagFamilies := make([]tagValues, len(stm.schema.TagFamilies))
tagFamiliesForIndexWrite := make([]tagValues, len(stm.schema.TagFamilies))
+ entityMap := make(map[string]bool)
+ et.elements.tagFamilies = append(et.elements.tagFamilies, tagFamilies)
+ for _, entity := range stm.GetSchema().GetEntity().GetTagNames() {
+ entityMap[entity] = true
+ }
for i := range stm.GetSchema().GetTagFamilies() {
var tagFamily *modelv1.TagFamilyForWrite
if len(req.Element.TagFamilies) <= i {
@@ -129,22 +134,18 @@ func (w *writeCallback) handle(dst map[string]*elementsInGroup, writeEvent *stre
} else {
tagValue = tagFamily.Tags[j]
}
- tagFamiliesForIndexWrite[i].values = append(tagFamiliesForIndexWrite[i].values, encodeTagValue(
+
+ encodeTagValue := encodeTagValue(
tagFamilySpec.Tags[j].Name,
tagFamilySpec.Tags[j].Type,
- tagValue,
- ))
- if tagFamilySpec.Tags[j].IndexedOnly {
+ tagValue)
+ tagFamiliesForIndexWrite[i].values = append(tagFamiliesForIndexWrite[i].values, encodeTagValue)
+ if tagFamilySpec.Tags[j].IndexedOnly || entityMap[tagFamilySpec.Tags[j].Name] {
continue
}
- tagFamilies[i].values = append(tagFamilies[i].values, encodeTagValue(
- tagFamilySpec.Tags[j].Name,
- tagFamilySpec.Tags[j].Type,
- tagValue,
- ))
+ tagFamilies[i].values = append(tagFamilies[i].values, encodeTagValue)
}
}
- et.elements.tagFamilies = append(et.elements.tagFamilies, tagFamilies)
var fields []index.Field
for _, indexRule := range stm.indexRuleLocators {