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/09/10 00:57:53 UTC
[skywalking-banyandb] 01/03: Support remote dequeue
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch distributed-query
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit 04dbe50dcb4118c6100d9a24fca80329b2d26743
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Mon Sep 4 02:43:06 2023 +0000
Support remote dequeue
Signed-off-by: Gao Hongtao <ha...@gmail.com>
---
api/proto/banyandb/cluster/v1/rpc.proto | 18 +-
banyand/queue/pub/pub.go | 129 +++++--
banyand/queue/sub/server.go | 2 +-
banyand/queue/sub/sub.go | 56 ++-
docs/api-reference.md | 628 ++++++++++++++++----------------
5 files changed, 471 insertions(+), 362 deletions(-)
diff --git a/api/proto/banyandb/cluster/v1/rpc.proto b/api/proto/banyandb/cluster/v1/rpc.proto
index 076c21e6..732bbe1f 100644
--- a/api/proto/banyandb/cluster/v1/rpc.proto
+++ b/api/proto/banyandb/cluster/v1/rpc.proto
@@ -19,22 +19,22 @@ syntax = "proto3";
package banyandb.cluster.v1;
-import "banyandb/measure/v1/write.proto";
-import "banyandb/stream/v1/write.proto";
+import "google/protobuf/any.proto";
option go_package = "github.com/apache/skywalking-banyandb/api/proto/banyandb/cluster/v1";
-message WriteRequest {
+message SendRequest {
string topic = 1;
uint64 message_id = 2;
- oneof request {
- stream.v1.InternalWriteRequest stream = 3;
- measure.v1.InternalWriteRequest measure = 4;
- }
+ google.protobuf.Any body = 3;
}
-message WriteResponse {}
+message SendResponse {
+ uint64 message_id = 1;
+ string error = 2;
+ google.protobuf.Any body = 3;
+}
service Service {
- rpc Write(stream WriteRequest) returns (stream WriteResponse);
+ rpc Send(stream SendRequest) returns (stream SendResponse);
}
diff --git a/banyand/queue/pub/pub.go b/banyand/queue/pub/pub.go
index 1f3e3cf4..1a953857 100644
--- a/banyand/queue/pub/pub.go
+++ b/banyand/queue/pub/pub.go
@@ -20,15 +20,17 @@ package pub
import (
"context"
+ "errors"
"fmt"
+ "io"
"sync"
"time"
"go.uber.org/multierr"
+ "google.golang.org/protobuf/proto"
+ "google.golang.org/protobuf/types/known/anypb"
clusterv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/cluster/v1"
- measurev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/measure/v1"
- streamv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/stream/v1"
"github.com/apache/skywalking-banyandb/banyand/metadata"
"github.com/apache/skywalking-banyandb/banyand/metadata/schema"
"github.com/apache/skywalking-banyandb/banyand/queue"
@@ -66,8 +68,38 @@ func (p *pub) Serve() run.StopNotify {
return p.closer.CloseNotify()
}
-func (*pub) Publish(_ bus.Topic, _ ...bus.Message) (bus.Future, error) {
- panic("unimplemented")
+func (p *pub) Publish(topic bus.Topic, messages ...bus.Message) (bus.Future, error) {
+ var err error
+ f := &future{}
+ handleMessage := func(m bus.Message, err error) error {
+ r, errSend := messageToRequest(topic, m)
+ if errSend != nil {
+ return multierr.Append(err, fmt.Errorf("failed to marshal message %T: %w", m, errSend))
+ }
+ node := m.Node()
+ p.mu.Lock()
+ client, ok := p.clients[node]
+ p.mu.Unlock()
+ if !ok {
+ return multierr.Append(err, fmt.Errorf("failed to get client for node %s", node))
+ }
+ ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
+ stream, errCreateStream := client.client.Send(ctx)
+ if err != nil {
+ return multierr.Append(err, fmt.Errorf("failed to get stream for node %s: %w", node, errCreateStream))
+ }
+ errSend = stream.Send(r)
+ if errSend != nil {
+ return multierr.Append(err, fmt.Errorf("failed to send message to node %s: %w", node, errSend))
+ }
+ f.clients = append(f.clients, stream)
+ return err
+ }
+ for _, m := range messages {
+ err = handleMessage(m, err)
+ }
+ return f, err
}
// NewBatchPublisher returns a new batch publisher.
@@ -96,7 +128,7 @@ func (p *pub) PreRun(context.Context) error {
}
type writeStream struct {
- client clusterv1.Service_WriteClient
+ client clusterv1.Service_SendClient
cancel func()
}
@@ -113,27 +145,14 @@ func (bp *batchPublisher) Close() (err error) {
return err
}
-func (bp *batchPublisher) Publish(topic bus.Topic, message ...bus.Message) (bus.Future, error) {
- var err error
- for _, m := range message {
- if !m.IsRemote() {
- continue
- }
- r := &clusterv1.WriteRequest{
- Topic: topic.String(),
- MessageId: uint64(m.ID()),
- }
- switch d := m.Data().(type) {
- case *streamv1.InternalWriteRequest:
- r.Request = &clusterv1.WriteRequest_Stream{Stream: d}
- case *measurev1.InternalWriteRequest:
- r.Request = &clusterv1.WriteRequest_Measure{Measure: d}
- default:
- err = multierr.Append(err, fmt.Errorf("invalid message type %T", d))
+func (bp *batchPublisher) Publish(topic bus.Topic, messages ...bus.Message) (bus.Future, error) {
+ for _, m := range messages {
+ r, err := messageToRequest(topic, m)
+ if err != nil {
+ err = multierr.Append(err, fmt.Errorf("failed to marshal message %T: %w", m, err))
continue
}
node := m.Node()
-
sendData := func() (success bool) {
if stream, ok := bp.streams[node]; !ok {
defer func() {
@@ -149,7 +168,7 @@ func (bp *batchPublisher) Publish(topic bus.Topic, message ...bus.Message) (bus.
}
errSend := stream.client.Send(r)
if errSend != nil {
- err = multierr.Append(errSend, fmt.Errorf("failed to send message to node %s: %w", node, err))
+ err = multierr.Append(err, fmt.Errorf("failed to send message to node %s: %w", node, errSend))
return false
}
_, errRecv := stream.client.Recv()
@@ -170,7 +189,7 @@ func (bp *batchPublisher) Publish(topic bus.Topic, message ...bus.Message) (bus.
}
//nolint: govet
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
- stream, errCreateStream := client.client.Write(ctx)
+ stream, errCreateStream := client.client.Send(ctx)
if err != nil {
err = multierr.Append(err, fmt.Errorf("failed to get stream for node %s: %w", node, errCreateStream))
continue
@@ -182,5 +201,63 @@ func (bp *batchPublisher) Publish(topic bus.Topic, message ...bus.Message) (bus.
_ = sendData()
}
//nolint: govet
- return nil, err
+ return nil, nil
+}
+
+func messageToRequest(topic bus.Topic, m bus.Message) (*clusterv1.SendRequest, error) {
+ if !m.IsRemote() {
+ return nil, fmt.Errorf("message %d is not remote", m.ID())
+ }
+ r := &clusterv1.SendRequest{
+ Topic: topic.String(),
+ MessageId: uint64(m.ID()),
+ }
+ message, ok := m.Data().(proto.Message)
+ if !ok {
+ return nil, fmt.Errorf("invalid message type %T", m)
+ }
+ anyMessage, err := anypb.New(message)
+ if err != nil {
+ return nil, fmt.Errorf("failed to marshal message %T: %w", m, err)
+ }
+ r.Body = anyMessage
+ return r, nil
+}
+
+type future struct {
+ clients []clusterv1.Service_SendClient
+}
+
+func (l *future) Get() (bus.Message, error) {
+ if len(l.clients) < 1 {
+ return bus.Message{}, io.EOF
+ }
+ c := l.clients[0]
+ defer func() {
+ l.clients = l.clients[1:]
+ }()
+ resp, err := c.Recv()
+ if err != nil {
+ return bus.Message{}, err
+ }
+ return bus.NewMessage(
+ bus.MessageID(resp.MessageId),
+ resp.Body,
+ ), nil
+}
+
+func (l *future) GetAll() ([]bus.Message, error) {
+ var globalErr error
+ ret := make([]bus.Message, 0, len(l.clients))
+ for {
+ m, err := l.Get()
+ if errors.Is(err, io.EOF) {
+ return ret, globalErr
+ }
+ if err != nil {
+ globalErr = multierr.Append(globalErr, err)
+ continue
+ }
+ ret = append(ret, m)
+ }
}
diff --git a/banyand/queue/sub/server.go b/banyand/queue/sub/server.go
index d7e415da..78833713 100644
--- a/banyand/queue/sub/server.go
+++ b/banyand/queue/sub/server.go
@@ -59,7 +59,7 @@ type server struct {
creds credentials.TransportCredentials
log *logger.Logger
ser *grpclib.Server
- listeners map[bus.Topic][]bus.MessageListener
+ listeners map[bus.Topic]bus.MessageListener
*clusterv1.UnimplementedServiceServer
addr string
certFile string
diff --git a/banyand/queue/sub/sub.go b/banyand/queue/sub/sub.go
index d53d4fe1..c689eef0 100644
--- a/banyand/queue/sub/sub.go
+++ b/banyand/queue/sub/sub.go
@@ -22,15 +22,20 @@ import (
"io"
"github.com/pkg/errors"
+ "google.golang.org/protobuf/proto"
+ "google.golang.org/protobuf/types/known/anypb"
clusterv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/cluster/v1"
"github.com/apache/skywalking-banyandb/pkg/bus"
)
-// Write implements v1.ServiceServer.
-func (s *server) Write(stream clusterv1.Service_WriteServer) error {
- reply := func(stream clusterv1.Service_WriteServer) {
- if errResp := stream.Send(&clusterv1.WriteResponse{}); errResp != nil {
+func (s *server) Send(stream clusterv1.Service_SendServer) error {
+ reply := func(writeEntity *clusterv1.SendRequest, err error, message string) {
+ s.log.Error().Stringer("written", writeEntity).Err(err).Msg(message)
+ if errResp := stream.Send(&clusterv1.SendResponse{
+ MessageId: writeEntity.MessageId,
+ Error: message,
+ }); errResp != nil {
s.log.Err(errResp).Msg("failed to send response")
}
}
@@ -47,8 +52,7 @@ func (s *server) Write(stream clusterv1.Service_WriteServer) error {
return nil
}
if err != nil {
- s.log.Error().Stringer("written", writeEntity).Err(err).Msg("failed to receive message")
- reply(stream)
+ reply(writeEntity, err, "failed to receive message")
continue
}
if writeEntity.Topic != "" && topic == nil {
@@ -56,14 +60,36 @@ func (s *server) Write(stream clusterv1.Service_WriteServer) error {
topic = &t
}
if topic == nil {
- s.log.Error().Stringer("written", writeEntity).Msg("topic is empty")
- reply(stream)
+ reply(writeEntity, err, "topic is empty")
continue
}
- for _, listener := range s.getListeners(*topic) {
- _ = listener.Rev(bus.NewMessage(bus.MessageID(writeEntity.MessageId), writeEntity.Request))
+ listener := s.getListeners(*topic)
+ if listener == nil {
+ reply(writeEntity, err, "no listener found")
+ continue
+ }
+ m := listener.Rev(bus.NewMessage(bus.MessageID(writeEntity.MessageId), writeEntity.Body))
+ if m.Data() == nil {
+ reply(writeEntity, err, "no response")
+ continue
+ }
+ message, ok := m.Data().(proto.Message)
+ if !ok {
+ reply(writeEntity, err, "invalid response")
+ continue
+ }
+ anyMessage, err := anypb.New(message)
+ if err != nil {
+ reply(writeEntity, err, "failed to marshal message")
+ continue
+ }
+ if err := stream.Send(&clusterv1.SendResponse{
+ MessageId: writeEntity.MessageId,
+ Body: anyMessage,
+ }); err != nil {
+ reply(writeEntity, err, "failed to send response")
+ continue
}
- reply(stream)
}
}
@@ -71,13 +97,13 @@ func (s *server) Subscribe(topic bus.Topic, listener bus.MessageListener) error
s.listenersLock.Lock()
defer s.listenersLock.Unlock()
if _, ok := s.listeners[topic]; !ok {
- s.listeners[topic] = make([]bus.MessageListener, 0)
+ s.listeners[topic] = listener
+ return nil
}
- s.listeners[topic] = append(s.listeners[topic], listener)
- return nil
+ return errors.New("topic already exists")
}
-func (s *server) getListeners(topic bus.Topic) []bus.MessageListener {
+func (s *server) getListeners(topic bus.Topic) bus.MessageListener {
s.listenersLock.RLock()
defer s.listenersLock.RUnlock()
return s.listeners[topic]
diff --git a/docs/api-reference.md b/docs/api-reference.md
index 537ac018..27a53eec 100644
--- a/docs/api-reference.md
+++ b/docs/api-reference.md
@@ -3,6 +3,12 @@
## Table of Contents
+- [banyandb/cluster/v1/rpc.proto](#banyandb_cluster_v1_rpc-proto)
+ - [SendRequest](#banyandb-cluster-v1-SendRequest)
+ - [SendResponse](#banyandb-cluster-v1-SendResponse)
+
+ - [Service](#banyandb-cluster-v1-Service)
+
- [banyandb/common/v1/common.proto](#banyandb_common_v1_common-proto)
- [Group](#banyandb-common-v1-Group)
- [IntervalRule](#banyandb-common-v1-IntervalRule)
@@ -12,6 +18,12 @@
- [Catalog](#banyandb-common-v1-Catalog)
- [IntervalRule.Unit](#banyandb-common-v1-IntervalRule-Unit)
+- [banyandb/database/v1/database.proto](#banyandb_database_v1_database-proto)
+ - [Node](#banyandb-database-v1-Node)
+ - [Shard](#banyandb-database-v1-Shard)
+
+ - [Role](#banyandb-database-v1-Role)
+
- [banyandb/model/v1/common.proto](#banyandb_model_v1_common-proto)
- [FieldValue](#banyandb-model-v1-FieldValue)
- [Float](#banyandb-model-v1-Float)
@@ -24,30 +36,6 @@
- [AggregationFunction](#banyandb-model-v1-AggregationFunction)
-- [banyandb/measure/v1/write.proto](#banyandb_measure_v1_write-proto)
- - [DataPointValue](#banyandb-measure-v1-DataPointValue)
- - [InternalWriteRequest](#banyandb-measure-v1-InternalWriteRequest)
- - [WriteRequest](#banyandb-measure-v1-WriteRequest)
- - [WriteResponse](#banyandb-measure-v1-WriteResponse)
-
-- [banyandb/stream/v1/write.proto](#banyandb_stream_v1_write-proto)
- - [ElementValue](#banyandb-stream-v1-ElementValue)
- - [InternalWriteRequest](#banyandb-stream-v1-InternalWriteRequest)
- - [WriteRequest](#banyandb-stream-v1-WriteRequest)
- - [WriteResponse](#banyandb-stream-v1-WriteResponse)
-
-- [banyandb/cluster/v1/rpc.proto](#banyandb_cluster_v1_rpc-proto)
- - [WriteRequest](#banyandb-cluster-v1-WriteRequest)
- - [WriteResponse](#banyandb-cluster-v1-WriteResponse)
-
- - [Service](#banyandb-cluster-v1-Service)
-
-- [banyandb/database/v1/database.proto](#banyandb_database_v1_database-proto)
- - [Node](#banyandb-database-v1-Node)
- - [Shard](#banyandb-database-v1-Shard)
-
- - [Role](#banyandb-database-v1-Role)
-
- [banyandb/model/v1/query.proto](#banyandb_model_v1_query-proto)
- [Condition](#banyandb-model-v1-Condition)
- [Criteria](#banyandb-model-v1-Criteria)
@@ -180,6 +168,12 @@
- [TopNRequest](#banyandb-measure-v1-TopNRequest)
- [TopNResponse](#banyandb-measure-v1-TopNResponse)
+- [banyandb/measure/v1/write.proto](#banyandb_measure_v1_write-proto)
+ - [DataPointValue](#banyandb-measure-v1-DataPointValue)
+ - [InternalWriteRequest](#banyandb-measure-v1-InternalWriteRequest)
+ - [WriteRequest](#banyandb-measure-v1-WriteRequest)
+ - [WriteResponse](#banyandb-measure-v1-WriteResponse)
+
- [banyandb/measure/v1/rpc.proto](#banyandb_measure_v1_rpc-proto)
- [MeasureService](#banyandb-measure-v1-MeasureService)
@@ -206,6 +200,12 @@
- [QueryRequest](#banyandb-stream-v1-QueryRequest)
- [QueryResponse](#banyandb-stream-v1-QueryResponse)
+- [banyandb/stream/v1/write.proto](#banyandb_stream_v1_write-proto)
+ - [ElementValue](#banyandb-stream-v1-ElementValue)
+ - [InternalWriteRequest](#banyandb-stream-v1-InternalWriteRequest)
+ - [WriteRequest](#banyandb-stream-v1-WriteRequest)
+ - [WriteResponse](#banyandb-stream-v1-WriteResponse)
+
- [banyandb/stream/v1/rpc.proto](#banyandb_stream_v1_rpc-proto)
- [StreamService](#banyandb-stream-v1-StreamService)
@@ -213,6 +213,66 @@
+<a name="banyandb_cluster_v1_rpc-proto"></a>
+<p align="right"><a href="#top">Top</a></p>
+
+## banyandb/cluster/v1/rpc.proto
+
+
+
+<a name="banyandb-cluster-v1-SendRequest"></a>
+
+### SendRequest
+
+
+
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| topic | [string](#string) | | |
+| message_id | [uint64](#uint64) | | |
+| body | [google.protobuf.Any](#google-protobuf-Any) | | |
+
+
+
+
+
+
+<a name="banyandb-cluster-v1-SendResponse"></a>
+
+### SendResponse
+
+
+
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| message_id | [uint64](#uint64) | | |
+| error | [string](#string) | | |
+| body | [google.protobuf.Any](#google-protobuf-Any) | | |
+
+
+
+
+
+
+
+
+
+
+
+
+<a name="banyandb-cluster-v1-Service"></a>
+
+### Service
+
+
+| Method Name | Request Type | Response Type | Description |
+| ----------- | ------------ | ------------- | ------------|
+| Send | [SendRequest](#banyandb-cluster-v1-SendRequest) stream | [SendResponse](#banyandb-cluster-v1-SendResponse) stream | |
+
+
+
+
+
<a name="banyandb_common_v1_common-proto"></a>
<p align="right"><a href="#top">Top</a></p>
@@ -326,136 +386,47 @@ Metadata is for multi-tenant, multi-model use
-<a name="banyandb_model_v1_common-proto"></a>
+<a name="banyandb_database_v1_database-proto"></a>
<p align="right"><a href="#top">Top</a></p>
-## banyandb/model/v1/common.proto
-
-
-
-<a name="banyandb-model-v1-FieldValue"></a>
-
-### FieldValue
-
-
-
-| Field | Type | Label | Description |
-| ----- | ---- | ----- | ----------- |
-| null | [google.protobuf.NullValue](#google-protobuf-NullValue) | | |
-| str | [Str](#banyandb-model-v1-Str) | | |
-| int | [Int](#banyandb-model-v1-Int) | | |
-| binary_data | [bytes](#bytes) | | |
-| float | [Float](#banyandb-model-v1-Float) | | |
-
-
-
-
-
-
-<a name="banyandb-model-v1-Float"></a>
-
-### Float
-
-
-
-| Field | Type | Label | Description |
-| ----- | ---- | ----- | ----------- |
-| value | [double](#double) | | |
-
-
-
-
-
-
-<a name="banyandb-model-v1-Int"></a>
-
-### Int
-
-
-
-| Field | Type | Label | Description |
-| ----- | ---- | ----- | ----------- |
-| value | [int64](#int64) | | |
-
-
-
-
-
-
-<a name="banyandb-model-v1-IntArray"></a>
-
-### IntArray
-
-
-
-| Field | Type | Label | Description |
-| ----- | ---- | ----- | ----------- |
-| value | [int64](#int64) | repeated | |
-
-
-
-
-
-
-<a name="banyandb-model-v1-Str"></a>
-
-### Str
-
-
-
-| Field | Type | Label | Description |
-| ----- | ---- | ----- | ----------- |
-| value | [string](#string) | | |
-
-
-
-
-
-
-<a name="banyandb-model-v1-StrArray"></a>
-
-### StrArray
-
-
-
-| Field | Type | Label | Description |
-| ----- | ---- | ----- | ----------- |
-| value | [string](#string) | repeated | |
-
-
-
+## banyandb/database/v1/database.proto
-<a name="banyandb-model-v1-TagFamilyForWrite"></a>
+<a name="banyandb-database-v1-Node"></a>
-### TagFamilyForWrite
+### Node
| Field | Type | Label | Description |
| ----- | ---- | ----- | ----------- |
-| tags | [TagValue](#banyandb-model-v1-TagValue) | repeated | |
+| metadata | [banyandb.common.v1.Metadata](#banyandb-common-v1-Metadata) | | |
+| roles | [Role](#banyandb-database-v1-Role) | repeated | |
+| grpc_address | [string](#string) | | |
+| http_address | [string](#string) | | |
+| created_at | [google.protobuf.Timestamp](#google-protobuf-Timestamp) | | |
-<a name="banyandb-model-v1-TagValue"></a>
+<a name="banyandb-database-v1-Shard"></a>
-### TagValue
+### Shard
| Field | Type | Label | Description |
| ----- | ---- | ----- | ----------- |
-| null | [google.protobuf.NullValue](#google-protobuf-NullValue) | | |
-| str | [Str](#banyandb-model-v1-Str) | | |
-| str_array | [StrArray](#banyandb-model-v1-StrArray) | | |
-| int | [Int](#banyandb-model-v1-Int) | | |
-| int_array | [IntArray](#banyandb-model-v1-IntArray) | | |
-| binary_data | [bytes](#bytes) | | |
+| id | [uint64](#uint64) | | |
+| metadata | [banyandb.common.v1.Metadata](#banyandb-common-v1-Metadata) | | |
+| catalog | [banyandb.common.v1.Catalog](#banyandb-common-v1-Catalog) | | |
+| node | [string](#string) | | |
+| total | [uint32](#uint32) | | |
+| updated_at | [google.protobuf.Timestamp](#google-protobuf-Timestamp) | | |
+| created_at | [google.protobuf.Timestamp](#google-protobuf-Timestamp) | | |
@@ -464,97 +435,19 @@ Metadata is for multi-tenant, multi-model use
-<a name="banyandb-model-v1-AggregationFunction"></a>
+<a name="banyandb-database-v1-Role"></a>
-### AggregationFunction
+### Role
| Name | Number | Description |
| ---- | ------ | ----------- |
-| AGGREGATION_FUNCTION_UNSPECIFIED | 0 | |
-| AGGREGATION_FUNCTION_MEAN | 1 | |
-| AGGREGATION_FUNCTION_MAX | 2 | |
-| AGGREGATION_FUNCTION_MIN | 3 | |
-| AGGREGATION_FUNCTION_COUNT | 4 | |
-| AGGREGATION_FUNCTION_SUM | 5 | |
-
-
-
-
-
-
-
-
-
-
-<a name="banyandb_measure_v1_write-proto"></a>
-<p align="right"><a href="#top">Top</a></p>
-
-## banyandb/measure/v1/write.proto
-
-
-
-<a name="banyandb-measure-v1-DataPointValue"></a>
-
-### DataPointValue
-DataPointValue is the data point for writing. It only contains values.
-
-
-| Field | Type | Label | Description |
-| ----- | ---- | ----- | ----------- |
-| timestamp | [google.protobuf.Timestamp](#google-protobuf-Timestamp) | | timestamp is in the timeunit of milliseconds. |
-| tag_families | [banyandb.model.v1.TagFamilyForWrite](#banyandb-model-v1-TagFamilyForWrite) | repeated | the order of tag_families' items match the measure schema |
-| fields | [banyandb.model.v1.FieldValue](#banyandb-model-v1-FieldValue) | repeated | the order of fields match the measure schema |
-
-
-
-
-
-
-<a name="banyandb-measure-v1-InternalWriteRequest"></a>
-
-### InternalWriteRequest
-
-
-
-| Field | Type | Label | Description |
-| ----- | ---- | ----- | ----------- |
-| shard_id | [uint32](#uint32) | | |
-| series_hash | [bytes](#bytes) | | |
-| entity_values | [banyandb.model.v1.TagValue](#banyandb-model-v1-TagValue) | repeated | |
-| request | [WriteRequest](#banyandb-measure-v1-WriteRequest) | | |
-
-
-
-
-
-
-<a name="banyandb-measure-v1-WriteRequest"></a>
-
-### WriteRequest
-WriteRequest is the request contract for write
-
-
-| Field | Type | Label | Description |
-| ----- | ---- | ----- | ----------- |
-| metadata | [banyandb.common.v1.Metadata](#banyandb-common-v1-Metadata) | | the metadata is required. |
-| data_point | [DataPointValue](#banyandb-measure-v1-DataPointValue) | | the data_point is required. |
-
-
-
-
-
-
-<a name="banyandb-measure-v1-WriteResponse"></a>
-
-### WriteResponse
-WriteResponse is the response contract for write
-
-
-
-
+| ROLE_UNSPECIFIED | 0 | |
+| ROLE_META | 1 | |
+| ROLE_DATA | 2 | |
+| ROLE_QUERY | 3 | |
+| ROLE_LIAISON | 4 | |
-
@@ -564,178 +457,136 @@ WriteResponse is the response contract for write
-<a name="banyandb_stream_v1_write-proto"></a>
+<a name="banyandb_model_v1_common-proto"></a>
<p align="right"><a href="#top">Top</a></p>
-## banyandb/stream/v1/write.proto
-
-
-
-<a name="banyandb-stream-v1-ElementValue"></a>
-
-### ElementValue
-
-
-
-| Field | Type | Label | Description |
-| ----- | ---- | ----- | ----------- |
-| element_id | [string](#string) | | element_id could be span_id of a Span or segment_id of a Segment in the context of stream |
-| timestamp | [google.protobuf.Timestamp](#google-protobuf-Timestamp) | | timestamp is in the timeunit of milliseconds. It represents 1) either the start time of a Span/Segment, 2) or the timestamp of a log |
-| tag_families | [banyandb.model.v1.TagFamilyForWrite](#banyandb-model-v1-TagFamilyForWrite) | repeated | the order of tag_families' items match the stream schema |
-
-
-
-
-
-
-<a name="banyandb-stream-v1-InternalWriteRequest"></a>
-
-### InternalWriteRequest
-
-
-
-| Field | Type | Label | Description |
-| ----- | ---- | ----- | ----------- |
-| shard_id | [uint32](#uint32) | | |
-| series_hash | [bytes](#bytes) | | |
-| entity_values | [banyandb.model.v1.TagValue](#banyandb-model-v1-TagValue) | repeated | |
-| request | [WriteRequest](#banyandb-stream-v1-WriteRequest) | | |
-
-
-
+## banyandb/model/v1/common.proto
-<a name="banyandb-stream-v1-WriteRequest"></a>
+<a name="banyandb-model-v1-FieldValue"></a>
-### WriteRequest
+### FieldValue
| Field | Type | Label | Description |
| ----- | ---- | ----- | ----------- |
-| metadata | [banyandb.common.v1.Metadata](#banyandb-common-v1-Metadata) | | the metadata is only required in the first write. |
-| element | [ElementValue](#banyandb-stream-v1-ElementValue) | | the element is required. |
+| null | [google.protobuf.NullValue](#google-protobuf-NullValue) | | |
+| str | [Str](#banyandb-model-v1-Str) | | |
+| int | [Int](#banyandb-model-v1-Int) | | |
+| binary_data | [bytes](#bytes) | | |
+| float | [Float](#banyandb-model-v1-Float) | | |
-<a name="banyandb-stream-v1-WriteResponse"></a>
+<a name="banyandb-model-v1-Float"></a>
-### WriteResponse
+### Float
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| value | [double](#double) | | |
-
-
-
-
+<a name="banyandb-model-v1-Int"></a>
+### Int
-<a name="banyandb_cluster_v1_rpc-proto"></a>
-<p align="right"><a href="#top">Top</a></p>
-## banyandb/cluster/v1/rpc.proto
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| value | [int64](#int64) | | |
-<a name="banyandb-cluster-v1-WriteRequest"></a>
-### WriteRequest
+<a name="banyandb-model-v1-IntArray"></a>
-| Field | Type | Label | Description |
-| ----- | ---- | ----- | ----------- |
-| topic | [string](#string) | | |
-| message_id | [uint64](#uint64) | | |
-| stream | [banyandb.stream.v1.InternalWriteRequest](#banyandb-stream-v1-InternalWriteRequest) | | |
-| measure | [banyandb.measure.v1.InternalWriteRequest](#banyandb-measure-v1-InternalWriteRequest) | | |
+### IntArray
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| value | [int64](#int64) | repeated | |
-<a name="banyandb-cluster-v1-WriteResponse"></a>
-### WriteResponse
+<a name="banyandb-model-v1-Str"></a>
+### Str
-
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| value | [string](#string) | | |
-
-
-<a name="banyandb-cluster-v1-Service"></a>
-### Service
+<a name="banyandb-model-v1-StrArray"></a>
+
+### StrArray
-| Method Name | Request Type | Response Type | Description |
-| ----------- | ------------ | ------------- | ------------|
-| Write | [WriteRequest](#banyandb-cluster-v1-WriteRequest) stream | [WriteResponse](#banyandb-cluster-v1-WriteResponse) stream | |
-
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| value | [string](#string) | repeated | |
-<a name="banyandb_database_v1_database-proto"></a>
-<p align="right"><a href="#top">Top</a></p>
-## banyandb/database/v1/database.proto
-<a name="banyandb-database-v1-Node"></a>
+<a name="banyandb-model-v1-TagFamilyForWrite"></a>
-### Node
+### TagFamilyForWrite
| Field | Type | Label | Description |
| ----- | ---- | ----- | ----------- |
-| metadata | [banyandb.common.v1.Metadata](#banyandb-common-v1-Metadata) | | |
-| roles | [Role](#banyandb-database-v1-Role) | repeated | |
-| grpc_address | [string](#string) | | |
-| http_address | [string](#string) | | |
-| created_at | [google.protobuf.Timestamp](#google-protobuf-Timestamp) | | |
+| tags | [TagValue](#banyandb-model-v1-TagValue) | repeated | |
-<a name="banyandb-database-v1-Shard"></a>
+<a name="banyandb-model-v1-TagValue"></a>
-### Shard
+### TagValue
| Field | Type | Label | Description |
| ----- | ---- | ----- | ----------- |
-| id | [uint64](#uint64) | | |
-| metadata | [banyandb.common.v1.Metadata](#banyandb-common-v1-Metadata) | | |
-| catalog | [banyandb.common.v1.Catalog](#banyandb-common-v1-Catalog) | | |
-| node | [string](#string) | | |
-| total | [uint32](#uint32) | | |
-| updated_at | [google.protobuf.Timestamp](#google-protobuf-Timestamp) | | |
-| created_at | [google.protobuf.Timestamp](#google-protobuf-Timestamp) | | |
+| null | [google.protobuf.NullValue](#google-protobuf-NullValue) | | |
+| str | [Str](#banyandb-model-v1-Str) | | |
+| str_array | [StrArray](#banyandb-model-v1-StrArray) | | |
+| int | [Int](#banyandb-model-v1-Int) | | |
+| int_array | [IntArray](#banyandb-model-v1-IntArray) | | |
+| binary_data | [bytes](#bytes) | | |
@@ -744,18 +595,19 @@ WriteResponse is the response contract for write
-<a name="banyandb-database-v1-Role"></a>
+<a name="banyandb-model-v1-AggregationFunction"></a>
-### Role
+### AggregationFunction
| Name | Number | Description |
| ---- | ------ | ----------- |
-| ROLE_UNSPECIFIED | 0 | |
-| ROLE_META | 1 | |
-| ROLE_DATA | 2 | |
-| ROLE_QUERY | 3 | |
-| ROLE_LIAISON | 4 | |
+| AGGREGATION_FUNCTION_UNSPECIFIED | 0 | |
+| AGGREGATION_FUNCTION_MEAN | 1 | |
+| AGGREGATION_FUNCTION_MAX | 2 | |
+| AGGREGATION_FUNCTION_MIN | 3 | |
+| AGGREGATION_FUNCTION_COUNT | 4 | |
+| AGGREGATION_FUNCTION_SUM | 5 | |
@@ -2647,6 +2499,83 @@ TopNResponse is the response for a query to the Query module.
+<a name="banyandb_measure_v1_write-proto"></a>
+<p align="right"><a href="#top">Top</a></p>
+
+## banyandb/measure/v1/write.proto
+
+
+
+<a name="banyandb-measure-v1-DataPointValue"></a>
+
+### DataPointValue
+DataPointValue is the data point for writing. It only contains values.
+
+
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| timestamp | [google.protobuf.Timestamp](#google-protobuf-Timestamp) | | timestamp is in the timeunit of milliseconds. |
+| tag_families | [banyandb.model.v1.TagFamilyForWrite](#banyandb-model-v1-TagFamilyForWrite) | repeated | the order of tag_families' items match the measure schema |
+| fields | [banyandb.model.v1.FieldValue](#banyandb-model-v1-FieldValue) | repeated | the order of fields match the measure schema |
+
+
+
+
+
+
+<a name="banyandb-measure-v1-InternalWriteRequest"></a>
+
+### InternalWriteRequest
+
+
+
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| shard_id | [uint32](#uint32) | | |
+| series_hash | [bytes](#bytes) | | |
+| entity_values | [banyandb.model.v1.TagValue](#banyandb-model-v1-TagValue) | repeated | |
+| request | [WriteRequest](#banyandb-measure-v1-WriteRequest) | | |
+
+
+
+
+
+
+<a name="banyandb-measure-v1-WriteRequest"></a>
+
+### WriteRequest
+WriteRequest is the request contract for write
+
+
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| metadata | [banyandb.common.v1.Metadata](#banyandb-common-v1-Metadata) | | the metadata is required. |
+| data_point | [DataPointValue](#banyandb-measure-v1-DataPointValue) | | the data_point is required. |
+
+
+
+
+
+
+<a name="banyandb-measure-v1-WriteResponse"></a>
+
+### WriteResponse
+WriteResponse is the response contract for write
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
<a name="banyandb_measure_v1_rpc-proto"></a>
<p align="right"><a href="#top">Top</a></p>
@@ -2964,6 +2893,83 @@ QueryResponse is the response for a query to the Query module.
+<a name="banyandb_stream_v1_write-proto"></a>
+<p align="right"><a href="#top">Top</a></p>
+
+## banyandb/stream/v1/write.proto
+
+
+
+<a name="banyandb-stream-v1-ElementValue"></a>
+
+### ElementValue
+
+
+
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| element_id | [string](#string) | | element_id could be span_id of a Span or segment_id of a Segment in the context of stream |
+| timestamp | [google.protobuf.Timestamp](#google-protobuf-Timestamp) | | timestamp is in the timeunit of milliseconds. It represents 1) either the start time of a Span/Segment, 2) or the timestamp of a log |
+| tag_families | [banyandb.model.v1.TagFamilyForWrite](#banyandb-model-v1-TagFamilyForWrite) | repeated | the order of tag_families' items match the stream schema |
+
+
+
+
+
+
+<a name="banyandb-stream-v1-InternalWriteRequest"></a>
+
+### InternalWriteRequest
+
+
+
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| shard_id | [uint32](#uint32) | | |
+| series_hash | [bytes](#bytes) | | |
+| entity_values | [banyandb.model.v1.TagValue](#banyandb-model-v1-TagValue) | repeated | |
+| request | [WriteRequest](#banyandb-stream-v1-WriteRequest) | | |
+
+
+
+
+
+
+<a name="banyandb-stream-v1-WriteRequest"></a>
+
+### WriteRequest
+
+
+
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| metadata | [banyandb.common.v1.Metadata](#banyandb-common-v1-Metadata) | | the metadata is only required in the first write. |
+| element | [ElementValue](#banyandb-stream-v1-ElementValue) | | the element is required. |
+
+
+
+
+
+
+<a name="banyandb-stream-v1-WriteResponse"></a>
+
+### WriteResponse
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
<a name="banyandb_stream_v1_rpc-proto"></a>
<p align="right"><a href="#top">Top</a></p>