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 2023/12/18 23:49:40 UTC
(skywalking-banyandb) 03/04: Fix series write
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch storage-column
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit 5ea42eb3f200147e40a3e52e084a5bb63975b89a
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Tue Dec 19 06:19:24 2023 +0800
Fix series write
Signed-off-by: Gao Hongtao <ha...@gmail.com>
---
banyand/measure/query.go | 18 ++++++++++++++----
banyand/measure/write.go | 20 ++++++++++----------
pkg/index/index.go | 5 +++--
pkg/index/inverted/inverted.go | 16 ++++++++++++----
4 files changed, 39 insertions(+), 20 deletions(-)
diff --git a/banyand/measure/query.go b/banyand/measure/query.go
index bf60dd31..f60a07b3 100644
--- a/banyand/measure/query.go
+++ b/banyand/measure/query.go
@@ -364,10 +364,16 @@ func (qr QueryResult) Len() int {
}
func (qr QueryResult) Less(i, j int) bool {
+ leftTS := qr.data[i].timestamps[qr.data[i].idx]
+ rightTS := qr.data[j].timestamps[qr.data[j].idx]
+ leftVersion := qr.data[i].p.partMetadata.Version
+ rightVersion := qr.data[j].p.partMetadata.Version
if qr.orderByTS {
- leftTS := qr.data[i].timestamps[qr.data[i].idx]
- rightTS := qr.data[j].timestamps[qr.data[j].idx]
if leftTS == rightTS {
+ if qr.data[i].bm.seriesID == qr.data[j].bm.seriesID {
+ // sort version in descending order if timestamps and seriesID are equal
+ return leftVersion > rightVersion
+ }
// sort seriesID in ascending order if timestamps are equal
return qr.data[i].bm.seriesID < qr.data[j].bm.seriesID
}
@@ -379,8 +385,12 @@ func (qr QueryResult) Less(i, j int) bool {
leftSIDIndex := qr.sidToIndex[qr.data[i].bm.seriesID]
rightSIDIndex := qr.sidToIndex[qr.data[j].bm.seriesID]
if leftSIDIndex == rightSIDIndex {
+ if leftTS == rightTS {
+ // sort version in descending order if timestamps and seriesID are equal
+ return leftVersion > rightVersion
+ }
// sort timestamps in ascending order if seriesID are equal
- return qr.data[i].timestamps[qr.data[i].idx] < qr.data[j].timestamps[qr.data[j].idx]
+ return leftTS < rightTS
}
return leftSIDIndex < rightSIDIndex
}
@@ -424,7 +434,7 @@ func (qr *QueryResult) merge() *pbv1.Result {
if len(result.Timestamps) > 0 &&
topBC.timestamps[topBC.idx] == result.Timestamps[len(result.Timestamps)-1] {
if topBC.p.partMetadata.Version > lastPartVersion {
- lastPartVersion = topBC.p.partMetadata.Version
+ logger.Panicf("following parts version should be less or equal to the previous one")
}
} else {
topBC.copyTo(result)
diff --git a/banyand/measure/write.go b/banyand/measure/write.go
index b0d6b241..4eadbc43 100644
--- a/banyand/measure/write.go
+++ b/banyand/measure/write.go
@@ -101,12 +101,12 @@ func (w *writeCallback) handle(dst map[string]*dataPointsInGroup, writeEvent *me
if fLen > len(stm.schema.GetTagFamilies()) {
return nil, fmt.Errorf("%s has more tag families than expected", req.Metadata)
}
- series, err := tsdb.Register(shardID, &pbv1.Series{
+ series := &pbv1.Series{
Subject: req.Metadata.Name,
EntityValues: writeEvent.EntityValues,
- })
- if err != nil {
- return nil, fmt.Errorf("cannot register series: %w", err)
+ }
+ if err := series.Marshal(); err != nil {
+ return nil, fmt.Errorf("cannot marshal series: %w", err)
}
dpt.dataPoints.seriesIDs = append(dpt.dataPoints.seriesIDs, series.ID)
field := nameValues{}
@@ -187,12 +187,12 @@ func (w *writeCallback) handle(dst map[string]*dataPointsInGroup, writeEvent *me
})
}
}
- if len(fields) > 0 {
- dpg.docs = append(dpg.docs, index.Document{
- DocID: uint64(series.ID),
- Fields: fields,
- })
- }
+
+ dpg.docs = append(dpg.docs, index.Document{
+ DocID: uint64(series.ID),
+ EntityValues: series.Buffer,
+ Fields: fields,
+ })
return dst, nil
}
diff --git a/pkg/index/index.go b/pkg/index/index.go
index 27eacb88..d0776c1b 100644
--- a/pkg/index/index.go
+++ b/pkg/index/index.go
@@ -192,8 +192,9 @@ type PostingValue struct {
}
type Document struct {
- Fields []Field
- DocID uint64
+ Fields []Field
+ EntityValues []byte
+ DocID uint64
}
type Documents []Document
diff --git a/pkg/index/inverted/inverted.go b/pkg/index/inverted/inverted.go
index 554090ba..bf1f3038 100644
--- a/pkg/index/inverted/inverted.go
+++ b/pkg/index/inverted/inverted.go
@@ -363,9 +363,12 @@ func (s *store) run() {
for _, d := range docs {
// TODO: generate a segment directly.
- fk := d.Fields[0].Key
+ var fk *index.FieldKey
+ if len(d.Fields) > 0 {
+ fk = &d.Fields[0].Key
+ }
docIDBuffer.Reset()
- if fk.HasSeriesID() {
+ if fk != nil && fk.HasSeriesID() {
docIDBuffer.Write(fk.SeriesID.Marshal())
}
docIDBuffer.Write(convert.Uint64ToBytes(d.DocID))
@@ -380,11 +383,16 @@ func (s *store) run() {
WithAnalyzer(analyzers[f.Key.Analyzer]))
}
}
- if toAddSeriesIDField && fk.HasSeriesID() {
+ if fk != nil && toAddSeriesIDField && fk.HasSeriesID() {
doc.AddField(bluge.NewKeywordFieldBytes(seriesIDField, fk.SeriesID.Marshal()))
}
+
+ if d.EntityValues != nil {
+ doc.AddField(bluge.NewKeywordFieldBytes(entityField, d.EntityValues).StoreValue())
+ }
+
size++
- batch.Update(doc.ID(), doc)
+ batch.Insert(doc)
}
if isBatch || size >= batchSize {
flush()