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:37 UTC
(skywalking-banyandb) branch storage-column updated (00d456e9 -> 1943ef01)
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a change to branch storage-column
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
from 00d456e9 Pass integration test
new c57690bf Fix topn test cases
new 2d1d0098 Fix stream test
new 5ea42eb3 Fix series write
new 1943ef01 fix distributed tests
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
banyand/liaison/grpc/measure.go | 2 +-
banyand/liaison/grpc/stream.go | 2 +-
banyand/measure/part_iter.go | 2 +-
banyand/measure/query.go | 44 +++++++++++++++++++---
banyand/measure/topn.go | 2 +-
banyand/measure/write.go | 38 +++++++++++++------
banyand/queue/sub/sub.go | 2 +-
banyand/stream/metadata.go | 2 +-
banyand/stream/stream_write.go | 39 +++++++++++++++----
pkg/bus/bus.go | 7 +++-
pkg/index/index.go | 5 ++-
pkg/index/inverted/inverted.go | 16 ++++++--
pkg/pb/v1/metadata.go | 13 +++++--
pkg/pb/v1/value.go | 6 ++-
.../logical/measure/measure_plan_distributed.go | 3 --
pkg/query/logical/measure/topn_plan_localscan.go | 4 +-
test/cases/measure/measure.go | 2 +-
test/cases/topn/data/input/null_group.yaml | 3 +-
.../query_ondisk/query_ondisk_suite_test.go | 1 +
19 files changed, 142 insertions(+), 51 deletions(-)
(skywalking-banyandb) 01/04: Fix topn test cases
Posted by ha...@apache.org.
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 c57690bf9621ed85916c55fb185f3d860d709a0e
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Mon Dec 18 22:30:03 2023 +0800
Fix topn test cases
Signed-off-by: Gao Hongtao <ha...@gmail.com>
---
banyand/measure/part_iter.go | 2 +-
banyand/measure/query.go | 26 ++++++++++++++++++++++--
pkg/pb/v1/metadata.go | 13 +++++++++---
pkg/pb/v1/value.go | 6 ++++--
pkg/query/logical/measure/topn_plan_localscan.go | 4 ++--
test/cases/measure/measure.go | 2 +-
test/cases/topn/data/input/null_group.yaml | 3 +--
7 files changed, 43 insertions(+), 13 deletions(-)
diff --git a/banyand/measure/part_iter.go b/banyand/measure/part_iter.go
index 87cc007e..6c08de04 100644
--- a/banyand/measure/part_iter.go
+++ b/banyand/measure/part_iter.go
@@ -121,7 +121,7 @@ func (pi *partIter) searchTargetSeriesID(sid common.SeriesID) bool {
}
sids := pi.sids[pi.sidIdx:]
pi.sidIdx += sort.Search(len(sids), func(i int) bool {
- return sid < sids[i]
+ return sid <= sids[i]
})
if pi.sidIdx >= len(pi.sids) {
pi.sidIdx = len(pi.sids)
diff --git a/banyand/measure/query.go b/banyand/measure/query.go
index 6a41ca45..bf60dd31 100644
--- a/banyand/measure/query.go
+++ b/banyand/measure/query.go
@@ -138,7 +138,20 @@ func (s *measure) Query(ctx context.Context, mqo pbv1.MeasureQueryOptions) (pbv1
func mustDecodeTagValue(valueType pbv1.ValueType, value []byte) *modelv1.TagValue {
if value == nil {
- return pbv1.NullTagValue
+ switch valueType {
+ case pbv1.ValueTypeInt64:
+ logger.Panicf("int64 can be nil")
+ case pbv1.ValueTypeStr:
+ return pbv1.EmptyStrTagValue
+ case pbv1.ValueTypeStrArr:
+ return pbv1.EmptyStrArrTagValue
+ case pbv1.ValueTypeInt64Arr:
+ return pbv1.EmptyIntArrTagValue
+ case pbv1.ValueTypeBinaryData:
+ return pbv1.EmptyBinaryTagValue
+ default:
+ return pbv1.NullTagValue
+ }
}
switch valueType {
case pbv1.ValueTypeInt64:
@@ -223,7 +236,16 @@ func strArrTagValue(values []string) *modelv1.TagValue {
func mustDecodeFieldValue(valueType pbv1.ValueType, value []byte) *modelv1.FieldValue {
if value == nil {
- return pbv1.NullFieldValue
+ switch valueType {
+ case pbv1.ValueTypeInt64, pbv1.ValueTypeFloat64:
+ logger.Panicf("int64 and float64 can't be nil")
+ case pbv1.ValueTypeStr:
+ return pbv1.EmptyStrFieldValue
+ case pbv1.ValueTypeBinaryData:
+ return pbv1.EmptyBinaryFieldValue
+ default:
+ return pbv1.NullFieldValue
+ }
}
switch valueType {
case pbv1.ValueTypeInt64:
diff --git a/pkg/pb/v1/metadata.go b/pkg/pb/v1/metadata.go
index 451304e6..a54edcea 100644
--- a/pkg/pb/v1/metadata.go
+++ b/pkg/pb/v1/metadata.go
@@ -125,7 +125,14 @@ type MeasureQueryResult interface {
}
var (
- NullFieldValue = &modelv1.FieldValue{Value: &modelv1.FieldValue_Null{}}
- NullTagFamily = &modelv1.TagFamilyForWrite{}
- NullTagValue = &modelv1.TagValue{Value: &modelv1.TagValue_Null{}}
+ NullFieldValue = &modelv1.FieldValue{Value: &modelv1.FieldValue_Null{}}
+ EmptyStrFieldValue = &modelv1.FieldValue{Value: &modelv1.FieldValue_Str{Str: &modelv1.Str{Value: ""}}}
+ EmptyBinaryFieldValue = &modelv1.FieldValue{Value: &modelv1.FieldValue_BinaryData{BinaryData: []byte{}}}
+
+ NullTagFamily = &modelv1.TagFamilyForWrite{}
+ NullTagValue = &modelv1.TagValue{Value: &modelv1.TagValue_Null{}}
+ EmptyStrTagValue = &modelv1.TagValue{Value: &modelv1.TagValue_Str{Str: &modelv1.Str{Value: ""}}}
+ EmptyStrArrTagValue = &modelv1.TagValue{Value: &modelv1.TagValue_StrArray{StrArray: &modelv1.StrArray{Value: []string{}}}}
+ EmptyIntArrTagValue = &modelv1.TagValue{Value: &modelv1.TagValue_IntArray{IntArray: &modelv1.IntArray{Value: []int64{}}}}
+ EmptyBinaryTagValue = &modelv1.TagValue{Value: &modelv1.TagValue_BinaryData{BinaryData: []byte{}}}
)
diff --git a/pkg/pb/v1/value.go b/pkg/pb/v1/value.go
index 0046f617..7f55d2a6 100644
--- a/pkg/pb/v1/value.go
+++ b/pkg/pb/v1/value.go
@@ -67,6 +67,7 @@ func marshalTagValue(dest []byte, tv *modelv1.TagValue) ([]byte, error) {
dest = append(dest, byte(MustTagValueToValueType(tv)))
switch tv.Value.(type) {
case *modelv1.TagValue_Null:
+ dest = marshalEntityValue(dest, nil)
case *modelv1.TagValue_Str:
dest = marshalEntityValue(dest, convert.StringToBytes(tv.GetStr().Value))
case *modelv1.TagValue_Int:
@@ -88,7 +89,8 @@ func unmarshalTagValue(dest []byte, src []byte) ([]byte, []byte, *modelv1.TagVal
vt := ValueType(src[0])
switch vt {
case ValueTypeUnknown:
- return dest, src[1:], NullTagValue, nil
+ // skip ValueType and entityDelimiter
+ return dest, src[2:], NullTagValue, nil
case ValueTypeStr:
if dest, src, err = unmarshalEntityValue(dest, src[1:]); err != nil {
return nil, nil, nil, errors.WithMessage(err, "unmarshal string tag value")
@@ -107,7 +109,7 @@ func unmarshalTagValue(dest []byte, src []byte) ([]byte, []byte, *modelv1.TagVal
return dest, src, &modelv1.TagValue{
Value: &modelv1.TagValue_Int{
Int: &modelv1.Int{
- Value: convert.BytesToInt64(dest),
+ Value: encoding.BytesToInt64(dest),
},
},
}, nil
diff --git a/pkg/query/logical/measure/topn_plan_localscan.go b/pkg/query/logical/measure/topn_plan_localscan.go
index 41d901d3..f1c48d1d 100644
--- a/pkg/query/logical/measure/topn_plan_localscan.go
+++ b/pkg/query/logical/measure/topn_plan_localscan.go
@@ -101,7 +101,7 @@ func (uls *unresolvedLocalScan) locateEntity(entityList []string) ([]*modelv1.Ta
entity[0] = &modelv1.TagValue{
Value: &modelv1.TagValue_Int{
Int: &modelv1.Int{
- Value: int64(uls.sort),
+ Value: int64(uls.sort.Number()),
},
},
}
@@ -119,7 +119,7 @@ func (uls *unresolvedLocalScan) locateEntity(entityList []string) ([]*modelv1.Ta
if entityIdx, ok := entityMap[pairQuery.GetName()]; ok {
entity[entityIdx] = pairQuery.Value
switch pairQuery.GetValue().GetValue().(type) {
- case *modelv1.TagValue_Str, *modelv1.TagValue_Int:
+ case *modelv1.TagValue_Str, *modelv1.TagValue_Int, *modelv1.TagValue_Null:
entity[entityIdx] = pairQuery.Value
default:
return nil, errors.New("unsupported condition tag type for entity")
diff --git a/test/cases/measure/measure.go b/test/cases/measure/measure.go
index f011fe29..3cc814ac 100644
--- a/test/cases/measure/measure.go
+++ b/test/cases/measure/measure.go
@@ -39,7 +39,7 @@ var (
}
)
-var _ = g.FDescribeTable("Scanning Measures", verify,
+var _ = g.DescribeTable("Scanning Measures", verify,
g.Entry("all", helpers.Args{Input: "all", Duration: 25 * time.Minute, Offset: -20 * time.Minute}),
g.Entry("all only fields", helpers.Args{Input: "all_only_fields", Duration: 25 * time.Minute, Offset: -20 * time.Minute}),
g.Entry("filter by tag", helpers.Args{Input: "tag_filter", Duration: 25 * time.Minute, Offset: -20 * time.Minute}),
diff --git a/test/cases/topn/data/input/null_group.yaml b/test/cases/topn/data/input/null_group.yaml
index 5020488f..7cb92e1b 100644
--- a/test/cases/topn/data/input/null_group.yaml
+++ b/test/cases/topn/data/input/null_group.yaml
@@ -25,5 +25,4 @@ conditions:
- name: http.uri
op: 1
value:
- str:
- value: ""
+ "null": null
(skywalking-banyandb) 03/04: Fix series write
Posted by ha...@apache.org.
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()
(skywalking-banyandb) 02/04: Fix stream test
Posted by ha...@apache.org.
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 2d1d009857d314d568983fa8bf04d9fc904fb432
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Mon Dec 18 22:45:09 2023 +0800
Fix stream test
Signed-off-by: Gao Hongtao <ha...@gmail.com>
---
banyand/stream/metadata.go | 2 +-
banyand/stream/stream_write.go | 28 ++++++++++++++++++++--------
2 files changed, 21 insertions(+), 9 deletions(-)
diff --git a/banyand/stream/metadata.go b/banyand/stream/metadata.go
index 58710cdd..e1273539 100644
--- a/banyand/stream/metadata.go
+++ b/banyand/stream/metadata.go
@@ -225,7 +225,7 @@ func newSupplier(path string, metadata metadata.Repo, bufferSize int64, dbOpts t
func (s *supplier) OpenResource(shardNum uint32, db resourceSchema.Supplier, resource resourceSchema.Resource) (io.Closer, error) {
streamSchema := resource.Schema().(*databasev1.Stream)
- return openStream(shardNum, &tsdbSupplier{db: db.(tsdb.Database)}, streamSpec{
+ return openStream(shardNum, &tsdbSupplier{db: db.SupplyTSDB().(tsdb.Database)}, streamSpec{
schema: streamSchema,
indexRules: resource.IndexRules(),
}, s.l), nil
diff --git a/banyand/stream/stream_write.go b/banyand/stream/stream_write.go
index 8c47dc7e..1d4e4e9b 100644
--- a/banyand/stream/stream_write.go
+++ b/banyand/stream/stream_write.go
@@ -147,20 +147,32 @@ func setUpWriteCallback(l *logger.Logger, schemaRepo *schemaRepo) *writeCallback
}
func (w *writeCallback) Rev(message bus.Message) (resp bus.Message) {
- writeEvent, ok := message.Data().(*streamv1.InternalWriteRequest)
+ events, ok := message.Data().([]any)
if !ok {
w.l.Warn().Msg("invalid event data type")
return
}
- stm, ok := w.schemaRepo.loadStream(writeEvent.GetRequest().GetMetadata())
- if !ok {
- w.l.Warn().Msg("cannot find stream definition")
+ if len(events) < 1 {
+ w.l.Warn().Msg("empty event")
return
}
- err := stm.write(common.ShardID(writeEvent.GetShardId()), writeEvent.SeriesHash,
- tsdb.DecodeEntityValues(writeEvent.GetEntityValues()), writeEvent.GetRequest().GetElement())
- if err != nil {
- w.l.Error().Err(err).RawJSON("written", logger.Proto(writeEvent)).Msg("fail to write entity")
+ for _, event := range events {
+ writeEvent, ok := event.(*streamv1.InternalWriteRequest)
+ if !ok {
+ w.l.Warn().Msg("invalid event data type")
+ return
+ }
+ stm, ok := w.schemaRepo.loadStream(writeEvent.GetRequest().GetMetadata())
+ if !ok {
+ w.l.Warn().Msg("cannot find stream definition")
+ return
+ }
+ err := stm.write(common.ShardID(writeEvent.GetShardId()), writeEvent.SeriesHash,
+ tsdb.DecodeEntityValues(writeEvent.GetEntityValues()), writeEvent.GetRequest().GetElement())
+ if err != nil {
+ w.l.Error().Err(err).RawJSON("written", logger.Proto(writeEvent)).Msg("fail to write entity")
+ }
}
+
return
}
(skywalking-banyandb) 04/04: fix distributed tests
Posted by ha...@apache.org.
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 1943ef01c4bd33eb01e699c845d03ba2e18153b6
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Tue Dec 19 07:49:05 2023 +0800
fix distributed tests
Signed-off-by: Gao Hongtao <ha...@gmail.com>
---
banyand/liaison/grpc/measure.go | 2 +-
banyand/liaison/grpc/stream.go | 2 +-
banyand/measure/topn.go | 2 +-
banyand/measure/write.go | 18 +++++++++++++++++-
banyand/queue/sub/sub.go | 2 +-
banyand/stream/stream_write.go | 19 +++++++++++++++----
pkg/bus/bus.go | 7 ++++++-
pkg/query/logical/measure/measure_plan_distributed.go | 3 ---
.../query_ondisk/query_ondisk_suite_test.go | 1 +
9 files changed, 43 insertions(+), 13 deletions(-)
diff --git a/banyand/liaison/grpc/measure.go b/banyand/liaison/grpc/measure.go
index afc07fab..9517b6cb 100644
--- a/banyand/liaison/grpc/measure.go
+++ b/banyand/liaison/grpc/measure.go
@@ -125,7 +125,7 @@ func (ms *measureService) Write(measure measurev1.MeasureService_WriteServer) er
reply(writeRequest.GetMetadata(), modelv1.Status_STATUS_INTERNAL_ERROR, writeRequest.GetMessageId(), measure, ms.sampled)
continue
}
- message := bus.NewMessageWithNode(bus.MessageID(time.Now().UnixNano()), nodeID, iwr)
+ message := bus.NewBatchMessageWithNode(bus.MessageID(time.Now().UnixNano()), nodeID, iwr)
_, errWritePub := publisher.Publish(data.TopicMeasureWrite, message)
if errWritePub != nil {
ms.sampled.Error().Err(errWritePub).RawJSON("written", logger.Proto(writeRequest)).Str("nodeID", nodeID).Msg("failed to send a message")
diff --git a/banyand/liaison/grpc/stream.go b/banyand/liaison/grpc/stream.go
index d2aca976..37dc44ea 100644
--- a/banyand/liaison/grpc/stream.go
+++ b/banyand/liaison/grpc/stream.go
@@ -126,7 +126,7 @@ func (s *streamService) Write(stream streamv1.StreamService_WriteServer) error {
reply(writeEntity.GetMetadata(), modelv1.Status_STATUS_INTERNAL_ERROR, writeEntity.GetMessageId(), stream, s.sampled)
continue
}
- message := bus.NewMessageWithNode(bus.MessageID(time.Now().UnixNano()), nodeID, iwr)
+ message := bus.NewBatchMessageWithNode(bus.MessageID(time.Now().UnixNano()), nodeID, iwr)
_, errWritePub := publisher.Publish(data.TopicStreamWrite, message)
if errWritePub != nil {
s.sampled.Error().Err(errWritePub).RawJSON("written", logger.Proto(writeEntity)).Str("nodeID", nodeID).Msg("failed to send a message")
diff --git a/banyand/measure/topn.go b/banyand/measure/topn.go
index 3f41c4e2..9763c3f4 100644
--- a/banyand/measure/topn.go
+++ b/banyand/measure/topn.go
@@ -220,7 +220,7 @@ func (t *topNStreamingProcessor) writeData(publisher queue.BatchPublisher, event
EntityValues: series.EntityValues,
ShardId: uint32(shardID),
}
- message := bus.NewMessageWithNode(bus.MessageID(time.Now().UnixNano()), "local", iwr)
+ message := bus.NewBatchMessageWithNode(bus.MessageID(time.Now().UnixNano()), "local", iwr)
_, errWritePub := publisher.Publish(apiData.TopicMeasureWrite, message)
return errWritePub
}
diff --git a/banyand/measure/write.go b/banyand/measure/write.go
index 4eadbc43..0c4993e7 100644
--- a/banyand/measure/write.go
+++ b/banyand/measure/write.go
@@ -22,6 +22,8 @@ import (
"fmt"
"time"
+ "google.golang.org/protobuf/types/known/anypb"
+
"github.com/apache/skywalking-banyandb/api/common"
databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
measurev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/measure/v1"
@@ -208,8 +210,22 @@ func (w *writeCallback) Rev(message bus.Message) (resp bus.Message) {
}
groups := make(map[string]*dataPointsInGroup)
for i := range events {
+ var writeEvent *measurev1.InternalWriteRequest
+ switch e := events[i].(type) {
+ case *measurev1.InternalWriteRequest:
+ writeEvent = e
+ case *anypb.Any:
+ writeEvent = &measurev1.InternalWriteRequest{}
+ if err := e.UnmarshalTo(writeEvent); err != nil {
+ w.l.Error().Err(err).RawJSON("written", logger.Proto(e)).Msg("fail to unmarshal event")
+ continue
+ }
+ default:
+ w.l.Warn().Msg("invalid event data type")
+ continue
+ }
var err error
- if groups, err = w.handle(groups, events[i].(*measurev1.InternalWriteRequest)); err != nil {
+ if groups, err = w.handle(groups, writeEvent); err != nil {
w.l.Error().Err(err).Msg("cannot handle write event")
continue
}
diff --git a/banyand/queue/sub/sub.go b/banyand/queue/sub/sub.go
index 4d8a8d31..9846c4db 100644
--- a/banyand/queue/sub/sub.go
+++ b/banyand/queue/sub/sub.go
@@ -62,7 +62,7 @@ func (s *server) Send(stream clusterv1.Service_SendServer) error {
reply(writeEntity, err, "no listener found")
return nil
}
- _ = listener.Rev(bus.NewMessage(bus.MessageID(writeEntity.MessageId), dataCollection))
+ _ = listener.Rev(bus.NewMessage(bus.MessageID(0), dataCollection))
return nil
}
if err != nil {
diff --git a/banyand/stream/stream_write.go b/banyand/stream/stream_write.go
index 1d4e4e9b..a2781d72 100644
--- a/banyand/stream/stream_write.go
+++ b/banyand/stream/stream_write.go
@@ -24,6 +24,7 @@ import (
"github.com/pkg/errors"
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/client_golang/prometheus/promauto"
+ "google.golang.org/protobuf/types/known/anypb"
"github.com/apache/skywalking-banyandb/api/common"
streamv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/stream/v1"
@@ -157,15 +158,25 @@ func (w *writeCallback) Rev(message bus.Message) (resp bus.Message) {
return
}
for _, event := range events {
- writeEvent, ok := event.(*streamv1.InternalWriteRequest)
- if !ok {
+ var writeEvent *streamv1.InternalWriteRequest
+ switch e := event.(type) {
+ case *streamv1.InternalWriteRequest:
+ writeEvent = e
+ case *anypb.Any:
+ writeEvent = &streamv1.InternalWriteRequest{}
+ if err := e.UnmarshalTo(writeEvent); err != nil {
+ w.l.Error().Err(err).RawJSON("written", logger.Proto(e)).Msg("fail to unmarshal event")
+ continue
+ }
+ default:
w.l.Warn().Msg("invalid event data type")
- return
+ continue
+
}
stm, ok := w.schemaRepo.loadStream(writeEvent.GetRequest().GetMetadata())
if !ok {
w.l.Warn().Msg("cannot find stream definition")
- return
+ continue
}
err := stm.write(common.ShardID(writeEvent.GetShardId()), writeEvent.SeriesHash,
tsdb.DecodeEntityValues(writeEvent.GetEntityValues()), writeEvent.GetRequest().GetElement())
diff --git a/pkg/bus/bus.go b/pkg/bus/bus.go
index 763de2d2..26c5be17 100644
--- a/pkg/bus/bus.go
+++ b/pkg/bus/bus.go
@@ -75,9 +75,14 @@ func NewMessage(id MessageID, data interface{}) Message {
return Message{id: id, node: "local", payload: data}
}
+// NewBatchMessageWithNode returns a new Message with a MessageID and NodeID and embed data.
+func NewBatchMessageWithNode(id MessageID, node string, data interface{}) Message {
+ return Message{id: id, node: node, payload: data, batchMode: true}
+}
+
// NewMessageWithNode returns a new Message with a MessageID and NodeID and embed data.
func NewMessageWithNode(id MessageID, node string, data interface{}) Message {
- return Message{id: id, node: node, payload: data, batchMode: true}
+ return Message{id: id, node: node, payload: data}
}
// MessageListener is the signature of functions that can handle an EventMessage.
diff --git a/pkg/query/logical/measure/measure_plan_distributed.go b/pkg/query/logical/measure/measure_plan_distributed.go
index bbe4d0d4..8f35f4b4 100644
--- a/pkg/query/logical/measure/measure_plan_distributed.go
+++ b/pkg/query/logical/measure/measure_plan_distributed.go
@@ -48,9 +48,6 @@ func newUnresolvedDistributed(query *measurev1.QueryRequest) logical.UnresolvedP
}
func (ud *unresolvedDistributed) Analyze(s logical.Schema) (logical.Plan, error) {
- if ud.originalQuery.TagProjection == nil {
- return nil, fmt.Errorf("tag projection is required")
- }
projectionTags := logical.ToTags(ud.originalQuery.GetTagProjection())
if len(projectionTags) > 0 {
var err error
diff --git a/test/integration/standalone/query_ondisk/query_ondisk_suite_test.go b/test/integration/standalone/query_ondisk/query_ondisk_suite_test.go
index ac89d108..d6846ee3 100644
--- a/test/integration/standalone/query_ondisk/query_ondisk_suite_test.go
+++ b/test/integration/standalone/query_ondisk/query_ondisk_suite_test.go
@@ -42,6 +42,7 @@ import (
)
func TestIntegrationQueryOnDisk(t *testing.T) {
+ t.Skip("skip on-disk integration test until measure parts flushing is supported")
RegisterFailHandler(Fail)
RunSpecs(t, "Integration Query OnDisk Suite", Label(integration_standalone.Labels...))
}