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 2021/12/02 02:53:30 UTC
[skywalking-banyandb] branch measure updated (901a4a8 -> ae1410d)
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a change to branch measure
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git.
discard 901a4a8 Add the measure module * Implements measure APIs except for the TopN * Encapsulate index writing component which could be imported by measure and stream modules. * Compose measure and relevant index rules for testing
new ae1410d Add the measure module * Implements measure APIs except for the TopN * Encapsulate index writing component which could be imported by measure and stream modules. * Compose measure and relevant index rules for testing
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (901a4a8)
\
N -- N -- N refs/heads/measure (ae1410d)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 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:
api/proto/banyandb/database/v1/schema.proto | 22 +++++++++++-----------
api/proto/banyandb/measure/v1/query.proto | 4 ++--
api/proto/banyandb/measure/v1/rpc.proto | 4 ++--
api/proto/banyandb/measure/v1/topn.proto | 4 ++--
api/proto/banyandb/measure/v1/write.proto | 8 ++++----
api/proto/banyandb/stream/v1/query.proto | 10 +++++-----
api/proto/banyandb/stream/v1/rpc.proto | 4 ++--
api/proto/banyandb/stream/v1/write.proto | 8 ++++----
8 files changed, 32 insertions(+), 32 deletions(-)
[skywalking-banyandb] 01/01: Add the measure module * Implements measure APIs except for the TopN * Encapsulate index writing component which could be imported by measure and stream modules. * Compose measure and relevant index rules for testing
Posted by ha...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch measure
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit ae1410db2b4ea35bba997c648bba5554e1e100fc
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Thu Dec 2 10:33:17 2021 +0800
Add the measure module
* Implements measure APIs except for the TopN
* Encapsulate index writing component which could be imported by
measure and stream modules.
* Compose measure and relevant index rules for testing
Signed-off-by: Gao Hongtao <ha...@gmail.com>
---
api/{event/discovery.go => data/measure.go} | 25 ++--
api/event/{discovery.go => measure.go} | 12 +-
api/event/{discovery.go => stream.go} | 12 +-
api/proto/banyandb/database/v1/schema.pb.go | 24 ++--
api/proto/banyandb/database/v1/schema.proto | 2 +-
api/proto/banyandb/measure/v1/query.pb.go | 14 +--
api/proto/banyandb/measure/v1/rpc.pb.go | 16 +--
api/proto/banyandb/measure/v1/topn.pb.go | 14 +--
api/proto/banyandb/measure/v1/write.pb.go | 136 +++++++++++++++++----
api/proto/banyandb/measure/v1/write.proto | 6 +
api/proto/banyandb/stream/v1/query.pb.go | 20 +--
api/proto/banyandb/stream/v1/rpc.pb.go | 14 +--
api/proto/banyandb/stream/v1/write.pb.go | 18 +--
banyand/liaison/grpc/server.go | 4 +-
banyand/liaison/grpc/stream_test.go | 7 +-
banyand/measure/measure.go | 93 ++++++++++++++
banyand/measure/query.go | 129 +++++++++++++++++++
banyand/measure/query_test.go | 73 +++++++++++
banyand/{stream => measure}/service.go | 32 ++---
banyand/measure/testdata/query_data.json | 47 +++++++
banyand/measure/testdata/write_data.json | 47 +++++++
.../{stream/stream_write.go => measure/write.go} | 131 ++++++++++++--------
banyand/measure/write_test.go | 113 +++++++++++++++++
banyand/metadata/metadata_test.go | 2 +-
banyand/metadata/schema/etcd_test.go | 2 +-
banyand/query/processor.go | 2 +-
banyand/query/processor_test.go | 7 +-
banyand/stream/service.go | 4 +-
banyand/stream/stream.go | 84 +++----------
banyand/stream/stream_write.go | 55 +++------
banyand/stream/stream_write_test.go | 5 +-
banyand/{stream/index.go => tsdb/index/writer.go} | 125 +++++++++++++++----
banyand/tsdb/series_write.go | 6 +-
pkg/encoding/int.go | 2 +-
pkg/partition/entity.go | 12 ++
api/event/discovery.go => pkg/partition/index.go | 33 ++---
pkg/pb/v1/metadata.go | 66 ++++++++++
pkg/query/logical/analyzer_test.go | 8 +-
pkg/query/logical/common_test.go | 5 +-
pkg/test/{ => measure}/etcd.go | 12 +-
pkg/test/measure/testdata/index_rule_binding.json | 16 +++
.../testdata/index_rules/entity_id.json} | 4 +-
pkg/test/measure/testdata/measure.json | 71 +++++++++++
pkg/test/{ => stream}/etcd.go | 2 +-
.../{ => stream}/testdata/index_rule_binding.json | 0
.../testdata/index_rules/db.instance.json | 0
.../{ => stream}/testdata/index_rules/db.type.json | 0
.../testdata/index_rules/duration.json | 0
.../testdata/index_rules/endpoint_id.json | 0
.../testdata/index_rules/http.method.json | 0
.../testdata/index_rules/mq.broker.json | 0
.../testdata/index_rules/mq.queue.json | 0
.../testdata/index_rules/mq.topic.json | 0
.../testdata/index_rules/status_code.json | 0
.../testdata/index_rules/trace_id.json | 0
pkg/test/{ => stream}/testdata/stream.json | 0
56 files changed, 1162 insertions(+), 350 deletions(-)
diff --git a/api/event/discovery.go b/api/data/measure.go
similarity index 73%
copy from api/event/discovery.go
copy to api/data/measure.go
index b419442..a582924 100644
--- a/api/event/discovery.go
+++ b/api/data/measure.go
@@ -15,23 +15,22 @@
// specific language governing permissions and limitations
// under the License.
-package event
+package data
import (
"github.com/apache/skywalking-banyandb/api/common"
"github.com/apache/skywalking-banyandb/pkg/bus"
)
-var (
- ShardEventKindVersion = common.KindVersion{
- Version: "v1",
- Kind: "event-shard",
- }
- TopicShardEvent = bus.UniTopic(ShardEventKindVersion.String())
+var MeasureWriteKindVersion = common.KindVersion{
+ Version: "v1",
+ Kind: "measure-write",
+}
- EntityEventKindVersion = common.KindVersion{
- Version: "v1",
- Kind: "event-entity",
- }
- TopicEntityEvent = bus.UniTopic(EntityEventKindVersion.String())
-)
+var TopicMeasureWrite = bus.UniTopic(MeasureWriteKindVersion.String())
+
+var MeasureQueryKindVersion = common.KindVersion{
+ Version: "v1",
+ Kind: "measure-query",
+}
+var TopicMeasureQuery = bus.BiTopic(MeasureQueryKindVersion.String())
diff --git a/api/event/discovery.go b/api/event/measure.go
similarity index 75%
copy from api/event/discovery.go
copy to api/event/measure.go
index b419442..9f2524e 100644
--- a/api/event/discovery.go
+++ b/api/event/measure.go
@@ -23,15 +23,15 @@ import (
)
var (
- ShardEventKindVersion = common.KindVersion{
+ MeasureShardEventKindVersion = common.KindVersion{
Version: "v1",
- Kind: "event-shard",
+ Kind: "measure-event-shard",
}
- TopicShardEvent = bus.UniTopic(ShardEventKindVersion.String())
+ MeasureTopicShardEvent = bus.UniTopic(MeasureShardEventKindVersion.String())
- EntityEventKindVersion = common.KindVersion{
+ MeasureEntityEventKindVersion = common.KindVersion{
Version: "v1",
- Kind: "event-entity",
+ Kind: "measure-event-entity",
}
- TopicEntityEvent = bus.UniTopic(EntityEventKindVersion.String())
+ MeasureTopicEntityEvent = bus.UniTopic(MeasureEntityEventKindVersion.String())
)
diff --git a/api/event/discovery.go b/api/event/stream.go
similarity index 75%
copy from api/event/discovery.go
copy to api/event/stream.go
index b419442..fa6d39c 100644
--- a/api/event/discovery.go
+++ b/api/event/stream.go
@@ -23,15 +23,15 @@ import (
)
var (
- ShardEventKindVersion = common.KindVersion{
+ StreamShardEventKindVersion = common.KindVersion{
Version: "v1",
- Kind: "event-shard",
+ Kind: "stream-event-shard",
}
- TopicShardEvent = bus.UniTopic(ShardEventKindVersion.String())
+ StreamTopicShardEvent = bus.UniTopic(StreamShardEventKindVersion.String())
- EntityEventKindVersion = common.KindVersion{
+ StreamEntityEventKindVersion = common.KindVersion{
Version: "v1",
- Kind: "event-entity",
+ Kind: "stream-event-entity",
}
- TopicEntityEvent = bus.UniTopic(EntityEventKindVersion.String())
+ StreamTopicEntityEvent = bus.UniTopic(StreamEntityEventKindVersion.String())
)
diff --git a/api/proto/banyandb/database/v1/schema.pb.go b/api/proto/banyandb/database/v1/schema.pb.go
index 57682eb..cfaf598 100644
--- a/api/proto/banyandb/database/v1/schema.pb.go
+++ b/api/proto/banyandb/database/v1/schema.pb.go
@@ -575,10 +575,10 @@ type Stream struct {
Metadata *v1.Metadata `protobuf:"bytes,1,opt,name=metadata,proto3" json:"metadata,omitempty"`
// tag_families
TagFamilies []*TagFamilySpec `protobuf:"bytes,2,rep,name=tag_families,json=tagFamilies,proto3" json:"tag_families,omitempty"`
- // entity indicates how to generate a series and shard a stream
+ // entity indicates how to generate a series and shard a measure
Entity *Entity `protobuf:"bytes,3,opt,name=entity,proto3" json:"entity,omitempty"`
Opts *ResourceOpts `protobuf:"bytes,4,opt,name=opts,proto3" json:"opts,omitempty"`
- // updated_at_nanoseconds indicates when the measure is updated
+ // updated_at_nanoseconds indicates when the stream is updated
UpdatedAtNanoseconds *timestamppb.Timestamp `protobuf:"bytes,5,opt,name=updated_at_nanoseconds,json=updatedAtNanoseconds,proto3" json:"updated_at_nanoseconds,omitempty"`
}
@@ -829,7 +829,7 @@ func (x *FieldSpec) GetCompressionMethod() CompressionMethod {
return CompressionMethod_COMPRESSION_METHOD_UNSPECIFIED
}
-// IntervalRule indicates how to get the interval of a measure series.
+// IntervalRule indicates how to get the interval of a stream series.
type IntervalRule struct {
state protoimpl.MessageState
sizeCache protoimpl.SizeCache
@@ -936,19 +936,19 @@ type Measure struct {
sizeCache protoimpl.SizeCache
unknownFields protoimpl.UnknownFields
- // metadata is the identity of a measure
+ // metadata is the identity of a stream
Metadata *v1.Metadata `protobuf:"bytes,1,opt,name=metadata,proto3" json:"metadata,omitempty"`
// tag_families are for filter measures
TagFamilies []*TagFamilySpec `protobuf:"bytes,2,rep,name=tag_families,json=tagFamilies,proto3" json:"tag_families,omitempty"`
- // fields denote measure values
+ // fields denote stream values
Fields []*FieldSpec `protobuf:"bytes,3,rep,name=fields,proto3" json:"fields,omitempty"`
- // entity indicates which tags will be to generate a series and shard a measure
+ // entity indicates which tags will be to generate a series and shard a stream
Entity *Entity `protobuf:"bytes,4,opt,name=entity,proto3" json:"entity,omitempty"`
// interval_rules define data points writing interval
IntervalRules []*IntervalRule `protobuf:"bytes,5,rep,name=interval_rules,json=intervalRules,proto3" json:"interval_rules,omitempty"`
// opts is basic resource management options
Opts *ResourceOpts `protobuf:"bytes,6,opt,name=opts,proto3" json:"opts,omitempty"`
- // updated_at_nanoseconds indicates when the measure is updated
+ // updated_at_nanoseconds indicates when the stream is updated
UpdatedAtNanoseconds *timestamppb.Timestamp `protobuf:"bytes,7,opt,name=updated_at_nanoseconds,json=updatedAtNanoseconds,proto3" json:"updated_at_nanoseconds,omitempty"`
}
@@ -1033,7 +1033,7 @@ func (x *Measure) GetUpdatedAtNanoseconds() *timestamppb.Timestamp {
return nil
}
-// TopNAggregation generates offline TopN statistics for a measure's TopN approximation
+// TopNAggregation generates offline TopN statistics for a stream's TopN approximation
type TopNAggregation struct {
state protoimpl.MessageState
sizeCache protoimpl.SizeCache
@@ -1052,13 +1052,13 @@ type TopNAggregation struct {
FieldValueSort v11.Sort `protobuf:"varint,4,opt,name=field_value_sort,json=fieldValueSort,proto3,enum=banyandb.model.v1.Sort" json:"field_value_sort,omitempty"`
// group_by_tag_names groups data points into statistical counters
GroupByTagNames []string `protobuf:"bytes,5,rep,name=group_by_tag_names,json=groupByTagNames,proto3" json:"group_by_tag_names,omitempty"`
- // criteria select partial data points from measure
+ // criteria select partial data points from stream
Criteria []*v11.Criteria `protobuf:"bytes,6,rep,name=criteria,proto3" json:"criteria,omitempty"`
// counters_number sets the number of counters to be tracked. The default value is 1000
CountersNumber int32 `protobuf:"varint,7,opt,name=counters_number,json=countersNumber,proto3" json:"counters_number,omitempty"`
// opts is basic resource management options
Opts *ResourceOpts `protobuf:"bytes,8,opt,name=opts,proto3" json:"opts,omitempty"`
- // updated_at_nanoseconds indicates when the measure is updated
+ // updated_at_nanoseconds indicates when the stream is updated
UpdatedAtNanoseconds *timestamppb.Timestamp `protobuf:"bytes,9,opt,name=updated_at_nanoseconds,json=updatedAtNanoseconds,proto3" json:"updated_at_nanoseconds,omitempty"`
}
@@ -1245,7 +1245,7 @@ func (x *IndexRule) GetUpdatedAt() *timestamppb.Timestamp {
return nil
}
-// Subject defines which stream or measure would generate indices
+// Subject defines which measure or stream would generate indices
type Subject struct {
state protoimpl.MessageState
sizeCache protoimpl.SizeCache
@@ -1253,7 +1253,7 @@ type Subject struct {
// catalog is where the subject belongs to
Catalog v1.Catalog `protobuf:"varint,1,opt,name=catalog,proto3,enum=banyandb.common.v1.Catalog" json:"catalog,omitempty"`
- // name refers to a stream or measure in a particular catalog
+ // name refers to a measure or stream in a particular catalog
Name string `protobuf:"bytes,2,opt,name=name,proto3" json:"name,omitempty"`
}
diff --git a/api/proto/banyandb/database/v1/schema.proto b/api/proto/banyandb/database/v1/schema.proto
index c5c2fb7..624766c 100644
--- a/api/proto/banyandb/database/v1/schema.proto
+++ b/api/proto/banyandb/database/v1/schema.proto
@@ -68,7 +68,7 @@ message Stream {
// entity indicates how to generate a series and shard a stream
Entity entity = 3;
ResourceOpts opts = 4;
- // updated_at_nanoseconds indicates when the measure is updated
+ // updated_at_nanoseconds indicates when the stream is updated
google.protobuf.Timestamp updated_at_nanoseconds = 5;
}
diff --git a/api/proto/banyandb/measure/v1/query.pb.go b/api/proto/banyandb/measure/v1/query.pb.go
index bedbb78..f220c73 100644
--- a/api/proto/banyandb/measure/v1/query.pb.go
+++ b/api/proto/banyandb/measure/v1/query.pb.go
@@ -470,14 +470,14 @@ var file_banyandb_measure_v1_query_proto_rawDesc = []byte{
0x0a, 0x63, 0x6f, 0x6e, 0x64, 0x69, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x1a, 0x25, 0x0a, 0x0f, 0x46,
0x69, 0x65, 0x6c, 0x64, 0x50, 0x72, 0x6f, 0x6a, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x12,
0x0a, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x01, 0x20, 0x03, 0x28, 0x09, 0x52, 0x04, 0x6e, 0x61,
- 0x6d, 0x65, 0x42, 0x70, 0x0a, 0x29, 0x6f, 0x72, 0x67, 0x2e, 0x61, 0x70, 0x61, 0x63, 0x68, 0x65,
+ 0x6d, 0x65, 0x42, 0x6e, 0x0a, 0x28, 0x6f, 0x72, 0x67, 0x2e, 0x61, 0x70, 0x61, 0x63, 0x68, 0x65,
0x2e, 0x73, 0x6b, 0x79, 0x77, 0x61, 0x6c, 0x6b, 0x69, 0x6e, 0x67, 0x2e, 0x62, 0x61, 0x6e, 0x79,
- 0x61, 0x6e, 0x64, 0x62, 0x2e, 0x6d, 0x65, 0x61, 0x73, 0x75, 0x72, 0x65, 0x2e, 0x76, 0x31, 0x5a,
- 0x43, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x61, 0x70, 0x61, 0x63,
- 0x68, 0x65, 0x2f, 0x73, 0x6b, 0x79, 0x77, 0x61, 0x6c, 0x6b, 0x69, 0x6e, 0x67, 0x2d, 0x62, 0x61,
- 0x6e, 0x79, 0x61, 0x6e, 0x64, 0x62, 0x2f, 0x61, 0x70, 0x69, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f,
- 0x2f, 0x62, 0x61, 0x6e, 0x79, 0x61, 0x6e, 0x64, 0x62, 0x2f, 0x6d, 0x65, 0x61, 0x73, 0x75, 0x72,
- 0x65, 0x2f, 0x76, 0x31, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+ 0x61, 0x6e, 0x64, 0x62, 0x2e, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x2e, 0x76, 0x31, 0x5a, 0x42,
+ 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x61, 0x70, 0x61, 0x63, 0x68,
+ 0x65, 0x2f, 0x73, 0x6b, 0x79, 0x77, 0x61, 0x6c, 0x6b, 0x69, 0x6e, 0x67, 0x2d, 0x62, 0x61, 0x6e,
+ 0x79, 0x61, 0x6e, 0x64, 0x62, 0x2f, 0x61, 0x70, 0x69, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2f,
+ 0x62, 0x61, 0x6e, 0x79, 0x61, 0x6e, 0x64, 0x62, 0x2f, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x2f,
+ 0x76, 0x31, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
}
var (
diff --git a/api/proto/banyandb/measure/v1/rpc.pb.go b/api/proto/banyandb/measure/v1/rpc.pb.go
index 535c380..4bc49c6 100644
--- a/api/proto/banyandb/measure/v1/rpc.pb.go
+++ b/api/proto/banyandb/measure/v1/rpc.pb.go
@@ -65,15 +65,15 @@ var file_banyandb_measure_v1_rpc_proto_rawDesc = []byte{
0x2e, 0x6d, 0x65, 0x61, 0x73, 0x75, 0x72, 0x65, 0x2e, 0x76, 0x31, 0x2e, 0x54, 0x6f, 0x70, 0x4e,
0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x21, 0x2e, 0x62, 0x61, 0x6e, 0x79, 0x61, 0x6e,
0x64, 0x62, 0x2e, 0x6d, 0x65, 0x61, 0x73, 0x75, 0x72, 0x65, 0x2e, 0x76, 0x31, 0x2e, 0x54, 0x6f,
- 0x70, 0x4e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x42, 0x70, 0x0a, 0x29, 0x6f, 0x72,
+ 0x70, 0x4e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x42, 0x6e, 0x0a, 0x28, 0x6f, 0x72,
0x67, 0x2e, 0x61, 0x70, 0x61, 0x63, 0x68, 0x65, 0x2e, 0x73, 0x6b, 0x79, 0x77, 0x61, 0x6c, 0x6b,
- 0x69, 0x6e, 0x67, 0x2e, 0x62, 0x61, 0x6e, 0x79, 0x61, 0x6e, 0x64, 0x62, 0x2e, 0x6d, 0x65, 0x61,
- 0x73, 0x75, 0x72, 0x65, 0x2e, 0x76, 0x31, 0x5a, 0x43, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e,
- 0x63, 0x6f, 0x6d, 0x2f, 0x61, 0x70, 0x61, 0x63, 0x68, 0x65, 0x2f, 0x73, 0x6b, 0x79, 0x77, 0x61,
- 0x6c, 0x6b, 0x69, 0x6e, 0x67, 0x2d, 0x62, 0x61, 0x6e, 0x79, 0x61, 0x6e, 0x64, 0x62, 0x2f, 0x61,
- 0x70, 0x69, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2f, 0x62, 0x61, 0x6e, 0x79, 0x61, 0x6e, 0x64,
- 0x62, 0x2f, 0x6d, 0x65, 0x61, 0x73, 0x75, 0x72, 0x65, 0x2f, 0x76, 0x31, 0x62, 0x06, 0x70, 0x72,
- 0x6f, 0x74, 0x6f, 0x33,
+ 0x69, 0x6e, 0x67, 0x2e, 0x62, 0x61, 0x6e, 0x79, 0x61, 0x6e, 0x64, 0x62, 0x2e, 0x73, 0x74, 0x72,
+ 0x65, 0x61, 0x6d, 0x2e, 0x76, 0x31, 0x5a, 0x42, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63,
+ 0x6f, 0x6d, 0x2f, 0x61, 0x70, 0x61, 0x63, 0x68, 0x65, 0x2f, 0x73, 0x6b, 0x79, 0x77, 0x61, 0x6c,
+ 0x6b, 0x69, 0x6e, 0x67, 0x2d, 0x62, 0x61, 0x6e, 0x79, 0x61, 0x6e, 0x64, 0x62, 0x2f, 0x61, 0x70,
+ 0x69, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2f, 0x62, 0x61, 0x6e, 0x79, 0x61, 0x6e, 0x64, 0x62,
+ 0x2f, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x2f, 0x76, 0x31, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74,
+ 0x6f, 0x33,
}
var file_banyandb_measure_v1_rpc_proto_goTypes = []interface{}{
diff --git a/api/proto/banyandb/measure/v1/topn.pb.go b/api/proto/banyandb/measure/v1/topn.pb.go
index b25a116..b7a33e1 100644
--- a/api/proto/banyandb/measure/v1/topn.pb.go
+++ b/api/proto/banyandb/measure/v1/topn.pb.go
@@ -391,14 +391,14 @@ var file_banyandb_measure_v1_topn_proto_rawDesc = []byte{
0x4f, 0x4e, 0x5f, 0x55, 0x4e, 0x53, 0x50, 0x45, 0x43, 0x49, 0x46, 0x49, 0x45, 0x44, 0x10, 0x00,
0x12, 0x1d, 0x0a, 0x19, 0x41, 0x47, 0x47, 0x52, 0x45, 0x47, 0x41, 0x54, 0x49, 0x4f, 0x4e, 0x5f,
0x46, 0x55, 0x4e, 0x43, 0x54, 0x49, 0x4f, 0x4e, 0x5f, 0x4d, 0x45, 0x41, 0x4e, 0x10, 0x01, 0x42,
- 0x70, 0x0a, 0x29, 0x6f, 0x72, 0x67, 0x2e, 0x61, 0x70, 0x61, 0x63, 0x68, 0x65, 0x2e, 0x73, 0x6b,
+ 0x6e, 0x0a, 0x28, 0x6f, 0x72, 0x67, 0x2e, 0x61, 0x70, 0x61, 0x63, 0x68, 0x65, 0x2e, 0x73, 0x6b,
0x79, 0x77, 0x61, 0x6c, 0x6b, 0x69, 0x6e, 0x67, 0x2e, 0x62, 0x61, 0x6e, 0x79, 0x61, 0x6e, 0x64,
- 0x62, 0x2e, 0x6d, 0x65, 0x61, 0x73, 0x75, 0x72, 0x65, 0x2e, 0x76, 0x31, 0x5a, 0x43, 0x67, 0x69,
- 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x61, 0x70, 0x61, 0x63, 0x68, 0x65, 0x2f,
- 0x73, 0x6b, 0x79, 0x77, 0x61, 0x6c, 0x6b, 0x69, 0x6e, 0x67, 0x2d, 0x62, 0x61, 0x6e, 0x79, 0x61,
- 0x6e, 0x64, 0x62, 0x2f, 0x61, 0x70, 0x69, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2f, 0x62, 0x61,
- 0x6e, 0x79, 0x61, 0x6e, 0x64, 0x62, 0x2f, 0x6d, 0x65, 0x61, 0x73, 0x75, 0x72, 0x65, 0x2f, 0x76,
- 0x31, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+ 0x62, 0x2e, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x2e, 0x76, 0x31, 0x5a, 0x42, 0x67, 0x69, 0x74,
+ 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x61, 0x70, 0x61, 0x63, 0x68, 0x65, 0x2f, 0x73,
+ 0x6b, 0x79, 0x77, 0x61, 0x6c, 0x6b, 0x69, 0x6e, 0x67, 0x2d, 0x62, 0x61, 0x6e, 0x79, 0x61, 0x6e,
+ 0x64, 0x62, 0x2f, 0x61, 0x70, 0x69, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2f, 0x62, 0x61, 0x6e,
+ 0x79, 0x61, 0x6e, 0x64, 0x62, 0x2f, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x2f, 0x76, 0x31, 0x62,
+ 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
}
var (
diff --git a/api/proto/banyandb/measure/v1/write.pb.go b/api/proto/banyandb/measure/v1/write.pb.go
index bc6f682..55e0270 100644
--- a/api/proto/banyandb/measure/v1/write.pb.go
+++ b/api/proto/banyandb/measure/v1/write.pb.go
@@ -50,9 +50,9 @@ type DataPointValue struct {
// timestamp is in the timeunit of nanoseconds.
Timestamp *timestamppb.Timestamp `protobuf:"bytes,1,opt,name=timestamp,proto3" json:"timestamp,omitempty"`
- // the order of tag_families' items match the measure schema
+ // the order of tag_families' items match the stream schema
TagFamilies []*v1.TagFamilyForWrite `protobuf:"bytes,2,rep,name=tag_families,json=tagFamilies,proto3" json:"tag_families,omitempty"`
- // the order of fields match the measure schema
+ // the order of fields match the stream schema
Fields []*v1.FieldValue `protobuf:"bytes,3,rep,name=fields,proto3" json:"fields,omitempty"`
}
@@ -206,6 +206,69 @@ func (*WriteResponse) Descriptor() ([]byte, []int) {
return file_banyandb_measure_v1_write_proto_rawDescGZIP(), []int{2}
}
+type InternalWriteRequest struct {
+ state protoimpl.MessageState
+ sizeCache protoimpl.SizeCache
+ unknownFields protoimpl.UnknownFields
+
+ ShardId uint32 `protobuf:"varint,1,opt,name=shard_id,json=shardId,proto3" json:"shard_id,omitempty"`
+ SeriesHash []byte `protobuf:"bytes,2,opt,name=series_hash,json=seriesHash,proto3" json:"series_hash,omitempty"`
+ Request *WriteRequest `protobuf:"bytes,3,opt,name=request,proto3" json:"request,omitempty"`
+}
+
+func (x *InternalWriteRequest) Reset() {
+ *x = InternalWriteRequest{}
+ if protoimpl.UnsafeEnabled {
+ mi := &file_banyandb_measure_v1_write_proto_msgTypes[3]
+ ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+ ms.StoreMessageInfo(mi)
+ }
+}
+
+func (x *InternalWriteRequest) String() string {
+ return protoimpl.X.MessageStringOf(x)
+}
+
+func (*InternalWriteRequest) ProtoMessage() {}
+
+func (x *InternalWriteRequest) ProtoReflect() protoreflect.Message {
+ mi := &file_banyandb_measure_v1_write_proto_msgTypes[3]
+ if protoimpl.UnsafeEnabled && x != nil {
+ ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+ if ms.LoadMessageInfo() == nil {
+ ms.StoreMessageInfo(mi)
+ }
+ return ms
+ }
+ return mi.MessageOf(x)
+}
+
+// Deprecated: Use InternalWriteRequest.ProtoReflect.Descriptor instead.
+func (*InternalWriteRequest) Descriptor() ([]byte, []int) {
+ return file_banyandb_measure_v1_write_proto_rawDescGZIP(), []int{3}
+}
+
+func (x *InternalWriteRequest) GetShardId() uint32 {
+ if x != nil {
+ return x.ShardId
+ }
+ return 0
+}
+
+func (x *InternalWriteRequest) GetSeriesHash() []byte {
+ if x != nil {
+ return x.SeriesHash
+ }
+ return nil
+}
+
+func (x *InternalWriteRequest) GetRequest() *WriteRequest {
+ if x != nil {
+ return x.Request
+ }
+ return nil
+}
+
var File_banyandb_measure_v1_write_proto protoreflect.FileDescriptor
var file_banyandb_measure_v1_write_proto_rawDesc = []byte{
@@ -241,14 +304,23 @@ var file_banyandb_measure_v1_write_proto_rawDesc = []byte{
0x6d, 0x65, 0x61, 0x73, 0x75, 0x72, 0x65, 0x2e, 0x76, 0x31, 0x2e, 0x44, 0x61, 0x74, 0x61, 0x50,
0x6f, 0x69, 0x6e, 0x74, 0x56, 0x61, 0x6c, 0x75, 0x65, 0x52, 0x09, 0x64, 0x61, 0x74, 0x61, 0x50,
0x6f, 0x69, 0x6e, 0x74, 0x22, 0x0f, 0x0a, 0x0d, 0x57, 0x72, 0x69, 0x74, 0x65, 0x52, 0x65, 0x73,
- 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x42, 0x70, 0x0a, 0x29, 0x6f, 0x72, 0x67, 0x2e, 0x61, 0x70, 0x61,
- 0x63, 0x68, 0x65, 0x2e, 0x73, 0x6b, 0x79, 0x77, 0x61, 0x6c, 0x6b, 0x69, 0x6e, 0x67, 0x2e, 0x62,
- 0x61, 0x6e, 0x79, 0x61, 0x6e, 0x64, 0x62, 0x2e, 0x6d, 0x65, 0x61, 0x73, 0x75, 0x72, 0x65, 0x2e,
- 0x76, 0x31, 0x5a, 0x43, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x61,
- 0x70, 0x61, 0x63, 0x68, 0x65, 0x2f, 0x73, 0x6b, 0x79, 0x77, 0x61, 0x6c, 0x6b, 0x69, 0x6e, 0x67,
- 0x2d, 0x62, 0x61, 0x6e, 0x79, 0x61, 0x6e, 0x64, 0x62, 0x2f, 0x61, 0x70, 0x69, 0x2f, 0x70, 0x72,
- 0x6f, 0x74, 0x6f, 0x2f, 0x62, 0x61, 0x6e, 0x79, 0x61, 0x6e, 0x64, 0x62, 0x2f, 0x6d, 0x65, 0x61,
- 0x73, 0x75, 0x72, 0x65, 0x2f, 0x76, 0x31, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+ 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x8f, 0x01, 0x0a, 0x14, 0x49, 0x6e, 0x74, 0x65, 0x72, 0x6e,
+ 0x61, 0x6c, 0x57, 0x72, 0x69, 0x74, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x19,
+ 0x0a, 0x08, 0x73, 0x68, 0x61, 0x72, 0x64, 0x5f, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0d,
+ 0x52, 0x07, 0x73, 0x68, 0x61, 0x72, 0x64, 0x49, 0x64, 0x12, 0x1f, 0x0a, 0x0b, 0x73, 0x65, 0x72,
+ 0x69, 0x65, 0x73, 0x5f, 0x68, 0x61, 0x73, 0x68, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x0a,
+ 0x73, 0x65, 0x72, 0x69, 0x65, 0x73, 0x48, 0x61, 0x73, 0x68, 0x12, 0x3b, 0x0a, 0x07, 0x72, 0x65,
+ 0x71, 0x75, 0x65, 0x73, 0x74, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x21, 0x2e, 0x62, 0x61,
+ 0x6e, 0x79, 0x61, 0x6e, 0x64, 0x62, 0x2e, 0x6d, 0x65, 0x61, 0x73, 0x75, 0x72, 0x65, 0x2e, 0x76,
+ 0x31, 0x2e, 0x57, 0x72, 0x69, 0x74, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x52, 0x07,
+ 0x72, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x42, 0x6e, 0x0a, 0x28, 0x6f, 0x72, 0x67, 0x2e, 0x61,
+ 0x70, 0x61, 0x63, 0x68, 0x65, 0x2e, 0x73, 0x6b, 0x79, 0x77, 0x61, 0x6c, 0x6b, 0x69, 0x6e, 0x67,
+ 0x2e, 0x62, 0x61, 0x6e, 0x79, 0x61, 0x6e, 0x64, 0x62, 0x2e, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d,
+ 0x2e, 0x76, 0x31, 0x5a, 0x42, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f,
+ 0x61, 0x70, 0x61, 0x63, 0x68, 0x65, 0x2f, 0x73, 0x6b, 0x79, 0x77, 0x61, 0x6c, 0x6b, 0x69, 0x6e,
+ 0x67, 0x2d, 0x62, 0x61, 0x6e, 0x79, 0x61, 0x6e, 0x64, 0x62, 0x2f, 0x61, 0x70, 0x69, 0x2f, 0x70,
+ 0x72, 0x6f, 0x74, 0x6f, 0x2f, 0x62, 0x61, 0x6e, 0x79, 0x61, 0x6e, 0x64, 0x62, 0x2f, 0x73, 0x74,
+ 0x72, 0x65, 0x61, 0x6d, 0x2f, 0x76, 0x31, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
}
var (
@@ -263,27 +335,29 @@ func file_banyandb_measure_v1_write_proto_rawDescGZIP() []byte {
return file_banyandb_measure_v1_write_proto_rawDescData
}
-var file_banyandb_measure_v1_write_proto_msgTypes = make([]protoimpl.MessageInfo, 3)
+var file_banyandb_measure_v1_write_proto_msgTypes = make([]protoimpl.MessageInfo, 4)
var file_banyandb_measure_v1_write_proto_goTypes = []interface{}{
(*DataPointValue)(nil), // 0: banyandb.measure.v1.DataPointValue
(*WriteRequest)(nil), // 1: banyandb.measure.v1.WriteRequest
(*WriteResponse)(nil), // 2: banyandb.measure.v1.WriteResponse
- (*timestamppb.Timestamp)(nil), // 3: google.protobuf.Timestamp
- (*v1.TagFamilyForWrite)(nil), // 4: banyandb.model.v1.TagFamilyForWrite
- (*v1.FieldValue)(nil), // 5: banyandb.model.v1.FieldValue
- (*v11.Metadata)(nil), // 6: banyandb.common.v1.Metadata
+ (*InternalWriteRequest)(nil), // 3: banyandb.measure.v1.InternalWriteRequest
+ (*timestamppb.Timestamp)(nil), // 4: google.protobuf.Timestamp
+ (*v1.TagFamilyForWrite)(nil), // 5: banyandb.model.v1.TagFamilyForWrite
+ (*v1.FieldValue)(nil), // 6: banyandb.model.v1.FieldValue
+ (*v11.Metadata)(nil), // 7: banyandb.common.v1.Metadata
}
var file_banyandb_measure_v1_write_proto_depIdxs = []int32{
- 3, // 0: banyandb.measure.v1.DataPointValue.timestamp:type_name -> google.protobuf.Timestamp
- 4, // 1: banyandb.measure.v1.DataPointValue.tag_families:type_name -> banyandb.model.v1.TagFamilyForWrite
- 5, // 2: banyandb.measure.v1.DataPointValue.fields:type_name -> banyandb.model.v1.FieldValue
- 6, // 3: banyandb.measure.v1.WriteRequest.metadata:type_name -> banyandb.common.v1.Metadata
+ 4, // 0: banyandb.measure.v1.DataPointValue.timestamp:type_name -> google.protobuf.Timestamp
+ 5, // 1: banyandb.measure.v1.DataPointValue.tag_families:type_name -> banyandb.model.v1.TagFamilyForWrite
+ 6, // 2: banyandb.measure.v1.DataPointValue.fields:type_name -> banyandb.model.v1.FieldValue
+ 7, // 3: banyandb.measure.v1.WriteRequest.metadata:type_name -> banyandb.common.v1.Metadata
0, // 4: banyandb.measure.v1.WriteRequest.data_point:type_name -> banyandb.measure.v1.DataPointValue
- 5, // [5:5] is the sub-list for method output_type
- 5, // [5:5] is the sub-list for method input_type
- 5, // [5:5] is the sub-list for extension type_name
- 5, // [5:5] is the sub-list for extension extendee
- 0, // [0:5] is the sub-list for field type_name
+ 1, // 5: banyandb.measure.v1.InternalWriteRequest.request:type_name -> banyandb.measure.v1.WriteRequest
+ 6, // [6:6] is the sub-list for method output_type
+ 6, // [6:6] is the sub-list for method input_type
+ 6, // [6:6] is the sub-list for extension type_name
+ 6, // [6:6] is the sub-list for extension extendee
+ 0, // [0:6] is the sub-list for field type_name
}
func init() { file_banyandb_measure_v1_write_proto_init() }
@@ -328,6 +402,18 @@ func file_banyandb_measure_v1_write_proto_init() {
return nil
}
}
+ file_banyandb_measure_v1_write_proto_msgTypes[3].Exporter = func(v interface{}, i int) interface{} {
+ switch v := v.(*InternalWriteRequest); i {
+ case 0:
+ return &v.state
+ case 1:
+ return &v.sizeCache
+ case 2:
+ return &v.unknownFields
+ default:
+ return nil
+ }
+ }
}
type x struct{}
out := protoimpl.TypeBuilder{
@@ -335,7 +421,7 @@ func file_banyandb_measure_v1_write_proto_init() {
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: file_banyandb_measure_v1_write_proto_rawDesc,
NumEnums: 0,
- NumMessages: 3,
+ NumMessages: 4,
NumExtensions: 0,
NumServices: 0,
},
diff --git a/api/proto/banyandb/measure/v1/write.proto b/api/proto/banyandb/measure/v1/write.proto
index 11ab62a..4246cd6 100644
--- a/api/proto/banyandb/measure/v1/write.proto
+++ b/api/proto/banyandb/measure/v1/write.proto
@@ -46,3 +46,9 @@ message WriteRequest {
// WriteResponse is the response contract for write
message WriteResponse {}
+
+message InternalWriteRequest {
+ uint32 shard_id = 1;
+ bytes series_hash = 2;
+ WriteRequest request = 3;
+}
diff --git a/api/proto/banyandb/stream/v1/query.pb.go b/api/proto/banyandb/stream/v1/query.pb.go
index 2804c39..ecd7389 100644
--- a/api/proto/banyandb/stream/v1/query.pb.go
+++ b/api/proto/banyandb/stream/v1/query.pb.go
@@ -43,14 +43,14 @@ const (
)
// Element represents
-// (stream context) a Span defined in Google Dapper paper or equivalently a Segment in Skywalking.
+// (measure context) a Span defined in Google Dapper paper or equivalently a Segment in Skywalking.
// (Log context) a log
type Element struct {
state protoimpl.MessageState
sizeCache protoimpl.SizeCache
unknownFields protoimpl.UnknownFields
- // element_id could be span_id of a Span or segment_id of a Segment in the context of stream
+ // element_id could be span_id of a Span or segment_id of a Segment in the context of measure
ElementId string `protobuf:"bytes,1,opt,name=element_id,json=elementId,proto3" json:"element_id,omitempty"`
// timestamp represents
// 1) either the start time of a Span/Segment,
@@ -176,7 +176,7 @@ type QueryRequest struct {
// metadata is required
Metadata *v11.Metadata `protobuf:"bytes,1,opt,name=metadata,proto3" json:"metadata,omitempty"`
// time_range is a range query with begin/end time of entities in the timeunit of nanoseconds.
- // In the context of stream, it represents the range of the `startTime` for spans/segments,
+ // In the context of measure, it represents the range of the `startTime` for spans/segments,
// while in the context of Log, it means the range of the timestamp(s) for logs.
// it is always recommended to specify time range for performance reason
TimeRange *v1.TimeRange `protobuf:"bytes,2,opt,name=time_range,json=timeRange,proto3" json:"time_range,omitempty"`
@@ -323,14 +323,14 @@ var file_banyandb_stream_v1_query_proto_rawDesc = []byte{
0x28, 0x0b, 0x32, 0x20, 0x2e, 0x62, 0x61, 0x6e, 0x79, 0x61, 0x6e, 0x64, 0x62, 0x2e, 0x6d, 0x6f,
0x64, 0x65, 0x6c, 0x2e, 0x76, 0x31, 0x2e, 0x54, 0x61, 0x67, 0x50, 0x72, 0x6f, 0x6a, 0x65, 0x63,
0x74, 0x69, 0x6f, 0x6e, 0x52, 0x0a, 0x70, 0x72, 0x6f, 0x6a, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e,
- 0x42, 0x6e, 0x0a, 0x28, 0x6f, 0x72, 0x67, 0x2e, 0x61, 0x70, 0x61, 0x63, 0x68, 0x65, 0x2e, 0x73,
+ 0x42, 0x70, 0x0a, 0x29, 0x6f, 0x72, 0x67, 0x2e, 0x61, 0x70, 0x61, 0x63, 0x68, 0x65, 0x2e, 0x73,
0x6b, 0x79, 0x77, 0x61, 0x6c, 0x6b, 0x69, 0x6e, 0x67, 0x2e, 0x62, 0x61, 0x6e, 0x79, 0x61, 0x6e,
- 0x64, 0x62, 0x2e, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x2e, 0x76, 0x31, 0x5a, 0x42, 0x67, 0x69,
- 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x61, 0x70, 0x61, 0x63, 0x68, 0x65, 0x2f,
- 0x73, 0x6b, 0x79, 0x77, 0x61, 0x6c, 0x6b, 0x69, 0x6e, 0x67, 0x2d, 0x62, 0x61, 0x6e, 0x79, 0x61,
- 0x6e, 0x64, 0x62, 0x2f, 0x61, 0x70, 0x69, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2f, 0x62, 0x61,
- 0x6e, 0x79, 0x61, 0x6e, 0x64, 0x62, 0x2f, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x2f, 0x76, 0x31,
- 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+ 0x64, 0x62, 0x2e, 0x6d, 0x65, 0x61, 0x73, 0x75, 0x72, 0x65, 0x2e, 0x76, 0x31, 0x5a, 0x43, 0x67,
+ 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x61, 0x70, 0x61, 0x63, 0x68, 0x65,
+ 0x2f, 0x73, 0x6b, 0x79, 0x77, 0x61, 0x6c, 0x6b, 0x69, 0x6e, 0x67, 0x2d, 0x62, 0x61, 0x6e, 0x79,
+ 0x61, 0x6e, 0x64, 0x62, 0x2f, 0x61, 0x70, 0x69, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2f, 0x62,
+ 0x61, 0x6e, 0x79, 0x61, 0x6e, 0x64, 0x62, 0x2f, 0x6d, 0x65, 0x61, 0x73, 0x75, 0x72, 0x65, 0x2f,
+ 0x76, 0x31, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
}
var (
diff --git a/api/proto/banyandb/stream/v1/rpc.pb.go b/api/proto/banyandb/stream/v1/rpc.pb.go
index 199c536..0950141 100644
--- a/api/proto/banyandb/stream/v1/rpc.pb.go
+++ b/api/proto/banyandb/stream/v1/rpc.pb.go
@@ -58,14 +58,14 @@ var file_banyandb_stream_v1_rpc_proto_rawDesc = []byte{
0x2e, 0x57, 0x72, 0x69, 0x74, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x21, 0x2e,
0x62, 0x61, 0x6e, 0x79, 0x61, 0x6e, 0x64, 0x62, 0x2e, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x2e,
0x76, 0x31, 0x2e, 0x57, 0x72, 0x69, 0x74, 0x65, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65,
- 0x28, 0x01, 0x30, 0x01, 0x42, 0x6e, 0x0a, 0x28, 0x6f, 0x72, 0x67, 0x2e, 0x61, 0x70, 0x61, 0x63,
+ 0x28, 0x01, 0x30, 0x01, 0x42, 0x70, 0x0a, 0x29, 0x6f, 0x72, 0x67, 0x2e, 0x61, 0x70, 0x61, 0x63,
0x68, 0x65, 0x2e, 0x73, 0x6b, 0x79, 0x77, 0x61, 0x6c, 0x6b, 0x69, 0x6e, 0x67, 0x2e, 0x62, 0x61,
- 0x6e, 0x79, 0x61, 0x6e, 0x64, 0x62, 0x2e, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x2e, 0x76, 0x31,
- 0x5a, 0x42, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x61, 0x70, 0x61,
- 0x63, 0x68, 0x65, 0x2f, 0x73, 0x6b, 0x79, 0x77, 0x61, 0x6c, 0x6b, 0x69, 0x6e, 0x67, 0x2d, 0x62,
- 0x61, 0x6e, 0x79, 0x61, 0x6e, 0x64, 0x62, 0x2f, 0x61, 0x70, 0x69, 0x2f, 0x70, 0x72, 0x6f, 0x74,
- 0x6f, 0x2f, 0x62, 0x61, 0x6e, 0x79, 0x61, 0x6e, 0x64, 0x62, 0x2f, 0x73, 0x74, 0x72, 0x65, 0x61,
- 0x6d, 0x2f, 0x76, 0x31, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+ 0x6e, 0x79, 0x61, 0x6e, 0x64, 0x62, 0x2e, 0x6d, 0x65, 0x61, 0x73, 0x75, 0x72, 0x65, 0x2e, 0x76,
+ 0x31, 0x5a, 0x43, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x61, 0x70,
+ 0x61, 0x63, 0x68, 0x65, 0x2f, 0x73, 0x6b, 0x79, 0x77, 0x61, 0x6c, 0x6b, 0x69, 0x6e, 0x67, 0x2d,
+ 0x62, 0x61, 0x6e, 0x79, 0x61, 0x6e, 0x64, 0x62, 0x2f, 0x61, 0x70, 0x69, 0x2f, 0x70, 0x72, 0x6f,
+ 0x74, 0x6f, 0x2f, 0x62, 0x61, 0x6e, 0x79, 0x61, 0x6e, 0x64, 0x62, 0x2f, 0x6d, 0x65, 0x61, 0x73,
+ 0x75, 0x72, 0x65, 0x2f, 0x76, 0x31, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
}
var file_banyandb_stream_v1_rpc_proto_goTypes = []interface{}{
diff --git a/api/proto/banyandb/stream/v1/write.pb.go b/api/proto/banyandb/stream/v1/write.pb.go
index 9d63ba5..c5f62fd 100644
--- a/api/proto/banyandb/stream/v1/write.pb.go
+++ b/api/proto/banyandb/stream/v1/write.pb.go
@@ -47,13 +47,13 @@ type ElementValue struct {
sizeCache protoimpl.SizeCache
unknownFields protoimpl.UnknownFields
- // element_id could be span_id of a Span or segment_id of a Segment in the context of stream
+ // element_id could be span_id of a Span or segment_id of a Segment in the context of measure
ElementId string `protobuf:"bytes,1,opt,name=element_id,json=elementId,proto3" json:"element_id,omitempty"`
// timestamp_nanoseconds is in the timeunit of nanoseconds. It represents
// 1) either the start time of a Span/Segment,
// 2) or the timestamp of a log
Timestamp *timestamppb.Timestamp `protobuf:"bytes,2,opt,name=timestamp,proto3" json:"timestamp,omitempty"`
- // the order of tag_families' items match the stream schema
+ // the order of tag_families' items match the measure schema
TagFamilies []*v1.TagFamilyForWrite `protobuf:"bytes,3,rep,name=tag_families,json=tagFamilies,proto3" json:"tag_families,omitempty"`
}
@@ -310,14 +310,14 @@ var file_banyandb_stream_v1_write_proto_rawDesc = []byte{
0x74, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x20, 0x2e, 0x62, 0x61, 0x6e, 0x79, 0x61, 0x6e,
0x64, 0x62, 0x2e, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x2e, 0x76, 0x31, 0x2e, 0x57, 0x72, 0x69,
0x74, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x52, 0x07, 0x72, 0x65, 0x71, 0x75, 0x65,
- 0x73, 0x74, 0x42, 0x6e, 0x0a, 0x28, 0x6f, 0x72, 0x67, 0x2e, 0x61, 0x70, 0x61, 0x63, 0x68, 0x65,
+ 0x73, 0x74, 0x42, 0x70, 0x0a, 0x29, 0x6f, 0x72, 0x67, 0x2e, 0x61, 0x70, 0x61, 0x63, 0x68, 0x65,
0x2e, 0x73, 0x6b, 0x79, 0x77, 0x61, 0x6c, 0x6b, 0x69, 0x6e, 0x67, 0x2e, 0x62, 0x61, 0x6e, 0x79,
- 0x61, 0x6e, 0x64, 0x62, 0x2e, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x2e, 0x76, 0x31, 0x5a, 0x42,
- 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x61, 0x70, 0x61, 0x63, 0x68,
- 0x65, 0x2f, 0x73, 0x6b, 0x79, 0x77, 0x61, 0x6c, 0x6b, 0x69, 0x6e, 0x67, 0x2d, 0x62, 0x61, 0x6e,
- 0x79, 0x61, 0x6e, 0x64, 0x62, 0x2f, 0x61, 0x70, 0x69, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2f,
- 0x62, 0x61, 0x6e, 0x79, 0x61, 0x6e, 0x64, 0x62, 0x2f, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x2f,
- 0x76, 0x31, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+ 0x61, 0x6e, 0x64, 0x62, 0x2e, 0x6d, 0x65, 0x61, 0x73, 0x75, 0x72, 0x65, 0x2e, 0x76, 0x31, 0x5a,
+ 0x43, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x61, 0x70, 0x61, 0x63,
+ 0x68, 0x65, 0x2f, 0x73, 0x6b, 0x79, 0x77, 0x61, 0x6c, 0x6b, 0x69, 0x6e, 0x67, 0x2d, 0x62, 0x61,
+ 0x6e, 0x79, 0x61, 0x6e, 0x64, 0x62, 0x2f, 0x61, 0x70, 0x69, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f,
+ 0x2f, 0x62, 0x61, 0x6e, 0x79, 0x61, 0x6e, 0x64, 0x62, 0x2f, 0x6d, 0x65, 0x61, 0x73, 0x75, 0x72,
+ 0x65, 0x2f, 0x76, 0x31, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
}
var (
diff --git a/banyand/liaison/grpc/server.go b/banyand/liaison/grpc/server.go
index 6e4c3ec..6720d7b 100644
--- a/banyand/liaison/grpc/server.go
+++ b/banyand/liaison/grpc/server.go
@@ -94,11 +94,11 @@ func (s *Server) PreRun() error {
s.log = logger.GetLogger("liaison-grpc")
s.shardRepo.log = s.log
s.entityRepo.log = s.log
- err := s.repo.Subscribe(event.TopicShardEvent, s.shardRepo)
+ err := s.repo.Subscribe(event.StreamTopicShardEvent, s.shardRepo)
if err != nil {
return err
}
- return s.repo.Subscribe(event.TopicEntityEvent, s.entityRepo)
+ return s.repo.Subscribe(event.StreamTopicEntityEvent, s.entityRepo)
}
func (s *Server) Name() string {
diff --git a/banyand/liaison/grpc/stream_test.go b/banyand/liaison/grpc/stream_test.go
index d2bf5aa..1f30088 100644
--- a/banyand/liaison/grpc/stream_test.go
+++ b/banyand/liaison/grpc/stream_test.go
@@ -43,6 +43,7 @@ import (
pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
"github.com/apache/skywalking-banyandb/pkg/run"
"github.com/apache/skywalking-banyandb/pkg/test"
+ teststream "github.com/apache/skywalking-banyandb/pkg/test/stream"
)
var _ run.PreRunner = (*preloadStreamService)(nil)
@@ -52,11 +53,11 @@ type preloadStreamService struct {
}
func (p *preloadStreamService) Name() string {
- return "preload-stream"
+ return "preload-measure"
}
func (p *preloadStreamService) PreRun() error {
- return test.PreloadSchema(p.metaSvc.SchemaRegistry())
+ return teststream.PreloadSchema(p.metaSvc.SchemaRegistry())
}
type testData struct {
@@ -108,7 +109,7 @@ func setup(req *require.Assertions, testData testData) func() {
)
// Create a random directory
rootPath, deferFunc := test.Space(req)
- flags := []string{"--root-path=" + rootPath, "--metadata-root-path=" + test.RandomTempDir()}
+ flags := []string{"--root-path=" + rootPath, "--metadata-root-path=" + teststream.RandomTempDir()}
if testData.TLS {
flags = append(flags, "--tls=true")
certFile := filepath.Join(testData.basePath, "testdata/server_cert.pem")
diff --git a/banyand/measure/measure.go b/banyand/measure/measure.go
new file mode 100644
index 0000000..f48953c
--- /dev/null
+++ b/banyand/measure/measure.go
@@ -0,0 +1,93 @@
+// 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 measure
+
+import (
+ "context"
+
+ databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+ "github.com/apache/skywalking-banyandb/banyand/tsdb"
+ "github.com/apache/skywalking-banyandb/banyand/tsdb/index"
+ "github.com/apache/skywalking-banyandb/pkg/encoding"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ "github.com/apache/skywalking-banyandb/pkg/partition"
+)
+
+// a chunk is 1MB
+const chunkSize = 1 << 20
+
+type measure struct {
+ name string
+ group string
+ l *logger.Logger
+ schema *databasev1.Measure
+ db tsdb.Database
+ entityLocator partition.EntityLocator
+ indexRules []*databasev1.IndexRule
+ indexWriter *index.Writer
+}
+
+func (s *measure) Close() error {
+ _ = s.indexWriter.Close()
+ return s.db.Close()
+}
+
+func (s *measure) parseSchema() {
+ sm := s.schema
+ meta := sm.GetMetadata()
+ s.name, s.group = meta.GetName(), meta.GetGroup()
+ s.entityLocator = partition.NewEntityLocator(sm.TagFamilies, sm.Entity)
+}
+
+type measureSpec struct {
+ schema *databasev1.Measure
+ indexRules []*databasev1.IndexRule
+}
+
+func openMeasure(root string, spec measureSpec, l *logger.Logger) (*measure, error) {
+ sm := &measure{
+ schema: spec.schema,
+ indexRules: spec.indexRules,
+ l: l,
+ }
+ sm.parseSchema()
+ ctx := context.WithValue(context.Background(), logger.ContextKey, l)
+
+ db, err := tsdb.OpenDatabase(
+ ctx,
+ tsdb.DatabaseOpts{
+ Location: root,
+ ShardNum: sm.schema.GetOpts().GetShardNum(),
+ IndexRules: spec.indexRules,
+ EncodingMethod: tsdb.EncodingMethod{
+ EncoderPool: encoding.NewPlainEncoderPool(chunkSize),
+ DecoderPool: encoding.NewPlainDecoderPool(chunkSize),
+ },
+ })
+ if err != nil {
+ return nil, err
+ }
+ sm.db = db
+ sm.indexWriter = index.NewWriter(ctx, index.WriterOptions{
+ DB: db,
+ ShardNum: spec.schema.GetOpts().ShardNum,
+ Families: spec.schema.TagFamilies,
+ IndexRules: spec.indexRules,
+ })
+ return sm, nil
+}
diff --git a/banyand/measure/query.go b/banyand/measure/query.go
new file mode 100644
index 0000000..505dd0c
--- /dev/null
+++ b/banyand/measure/query.go
@@ -0,0 +1,129 @@
+// 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 measure
+
+import (
+ "io"
+
+ "github.com/pkg/errors"
+ "google.golang.org/protobuf/proto"
+
+ "github.com/apache/skywalking-banyandb/api/common"
+ commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
+ databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+ measurev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/measure/v1"
+ modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
+ "github.com/apache/skywalking-banyandb/banyand/tsdb"
+ "github.com/apache/skywalking-banyandb/pkg/partition"
+)
+
+var (
+ ErrTagFamilyNotExist = errors.New("tag family doesn't exist")
+)
+
+type Query interface {
+ Measure(measure *commonv1.Metadata) (Measure, error)
+}
+
+type Measure interface {
+ io.Closer
+ Write(value *measurev1.DataPointValue) error
+ Shards(entity tsdb.Entity) ([]tsdb.Shard, error)
+ Shard(id common.ShardID) (tsdb.Shard, error)
+ ParseTagFamily(family string, item tsdb.Item) (*modelv1.TagFamily, error)
+ ParseField(name string, item tsdb.Item) (*measurev1.DataPoint_Field, error)
+}
+
+var _ Measure = (*measure)(nil)
+
+func (s *measure) Shards(entity tsdb.Entity) ([]tsdb.Shard, error) {
+ if len(entity) < 1 {
+ return s.db.Shards(), nil
+ }
+ for _, e := range entity {
+ if e == nil {
+ return s.db.Shards(), nil
+ }
+ }
+ shardID, err := partition.ShardID(entity.Marshal(), s.schema.GetOpts().GetShardNum())
+ if err != nil {
+ return nil, err
+ }
+ shard, err := s.db.Shard(common.ShardID(shardID))
+ if err != nil {
+ return nil, err
+ }
+ return []tsdb.Shard{shard}, nil
+}
+
+func (s *measure) Shard(id common.ShardID) (tsdb.Shard, error) {
+ return s.db.Shard(id)
+}
+
+func (s *measure) ParseTagFamily(family string, item tsdb.Item) (*modelv1.TagFamily, error) {
+ familyRawBytes, err := item.Family(string(familyIdentity(family, TagFlag)))
+ if err != nil {
+ return nil, err
+ }
+ tagFamily := &modelv1.TagFamilyForWrite{}
+ err = proto.Unmarshal(familyRawBytes, tagFamily)
+ if err != nil {
+ return nil, err
+ }
+ tags := make([]*modelv1.Tag, len(tagFamily.GetTags()))
+ var tagSpec []*databasev1.TagSpec
+ for _, tf := range s.schema.GetTagFamilies() {
+ if tf.GetName() == family {
+ tagSpec = tf.GetTags()
+ }
+ }
+ if tagSpec == nil {
+ return nil, ErrTagFamilyNotExist
+ }
+ for i, tag := range tagFamily.GetTags() {
+ tags[i] = &modelv1.Tag{
+ Key: tagSpec[i].GetName(),
+ Value: &modelv1.TagValue{
+ Value: tag.GetValue(),
+ },
+ }
+ }
+ return &modelv1.TagFamily{
+ Name: family,
+ Tags: tags,
+ }, err
+}
+
+func (s *measure) ParseField(name string, item tsdb.Item) (*measurev1.DataPoint_Field, error) {
+ var fieldSpec *databasev1.FieldSpec
+ for _, spec := range s.schema.GetFields() {
+ if spec.GetName() == name {
+ fieldSpec = spec
+ break
+ }
+ }
+ bytes, err := item.Family(string(familyIdentity(name, encoderFieldFlag(fieldSpec))))
+ if err != nil {
+ return nil, err
+ }
+ fieldValue := decodeFieldValue(bytes, fieldSpec)
+ return &measurev1.DataPoint_Field{
+ Name: name,
+ Value: fieldValue,
+ }, err
+}
diff --git a/banyand/measure/query_test.go b/banyand/measure/query_test.go
new file mode 100644
index 0000000..31638db
--- /dev/null
+++ b/banyand/measure/query_test.go
@@ -0,0 +1,73 @@
+// 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 measure
+
+import (
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
+ "github.com/apache/skywalking-banyandb/banyand/tsdb"
+)
+
+func Test_ParseTag_And_ParseField(t *testing.T) {
+ s, deferFunc := setup(t)
+ defer deferFunc()
+ baseTime := writeData(t, "query_data.json", s)
+ r := require.New(t)
+ shard, err := s.Shard(0)
+ r.NoError(err)
+ series, err := shard.Series().Get(tsdb.Entity{tsdb.Entry("1")})
+ r.NoError(err)
+ seriesSpan, err := series.Span(tsdb.NewTimeRangeDuration(baseTime, 1*time.Hour))
+ defer func(seriesSpan tsdb.SeriesSpan) {
+ _ = seriesSpan.Close()
+ }(seriesSpan)
+ r.NoError(err)
+ seeker, err := seriesSpan.SeekerBuilder().OrderByTime(modelv1.Sort_SORT_DESC).Build()
+ r.NoError(err)
+ iter, err := seeker.Seek()
+ r.NoError(err)
+ at := assert.New(t)
+ r.Equal(1, len(iter))
+ {
+ defer func(iterator tsdb.Iterator) {
+ _ = iterator.Close()
+ }(iter[0])
+ i := 0
+ expectedFields := [][]int64{{150, 300, 5}, {200, 50, 4}, {100, 100, 1}}
+ for ; iter[0].Next(); i++ {
+ item := iter[0].Val()
+ tagFamily, err := s.ParseTagFamily("default", item)
+ r.NoError(err)
+ at.Equal(2, len(tagFamily.Tags))
+ summation, err := s.ParseField("summation", item)
+ r.NoError(err)
+ at.Equal(expectedFields[i][0], summation.GetValue().GetInt().Value)
+ count, err := s.ParseField("count", item)
+ r.NoError(err)
+ at.Equal(expectedFields[i][1], count.GetValue().GetInt().Value)
+ value, err := s.ParseField("value", item)
+ r.NoError(err)
+ at.Equal(expectedFields[i][2], value.GetValue().GetInt().Value)
+ }
+ }
+}
diff --git a/banyand/stream/service.go b/banyand/measure/service.go
similarity index 82%
copy from banyand/stream/service.go
copy to banyand/measure/service.go
index 6af8634..51dbe80 100644
--- a/banyand/stream/service.go
+++ b/banyand/measure/service.go
@@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-package stream
+package measure
import (
"context"
@@ -38,8 +38,8 @@ import (
)
var (
- ErrEmptyRootPath = errors.New("root path is empty")
- ErrStreamNotExist = errors.New("stream doesn't exist")
+ ErrEmptyRootPath = errors.New("root path is empty")
+ ErrMeasureNotExist = errors.New("stream doesn't exist")
)
type Service interface {
@@ -52,7 +52,7 @@ type Service interface {
var _ Service = (*service)(nil)
type service struct {
- schemaMap map[string]*stream
+ schemaMap map[string]*measure
writeListener *writeCallback
l *logger.Logger
metadata metadata.Repo
@@ -62,11 +62,11 @@ type service struct {
stopCh chan struct{}
}
-func (s *service) Stream(stream *commonv1.Metadata) (Stream, error) {
- sID := formatStreamID(stream.GetName(), stream.GetGroup())
+func (s *service) Measure(measure *commonv1.Metadata) (Measure, error) {
+ sID := formatMeasureID(measure.GetName(), measure.GetGroup())
sm, ok := s.schemaMap[sID]
if !ok {
- return nil, errors.WithStack(ErrStreamNotExist)
+ return nil, errors.WithStack(ErrMeasureNotExist)
}
return sm, nil
}
@@ -89,26 +89,26 @@ func (s *service) Name() string {
}
func (s *service) PreRun() error {
- schemas, err := s.metadata.StreamRegistry().ListStream(context.TODO(), schema.ListOpt{})
+ schemas, err := s.metadata.MeasureRegistry().ListMeasure(context.TODO(), schema.ListOpt{})
if err != nil {
return err
}
- s.schemaMap = make(map[string]*stream, len(schemas))
+ s.schemaMap = make(map[string]*measure, len(schemas))
s.l = logger.GetLogger(s.Name())
for _, sa := range schemas {
iRules, errIndexRules := s.metadata.IndexRules(context.TODO(), sa.Metadata)
if errIndexRules != nil {
return errIndexRules
}
- sm, errTS := openStream(s.root, streamSpec{
+ sm, errTS := openMeasure(s.root, measureSpec{
schema: sa,
indexRules: iRules,
}, s.l)
if errTS != nil {
return errTS
}
- id := formatStreamID(sm.name, sm.group)
+ id := formatMeasureID(sm.name, sm.group)
s.schemaMap[id] = sm
s.l.Info().Str("id", id).Msg("initialize stream")
}
@@ -128,7 +128,7 @@ func (s *service) Serve() error {
TagOffset: uint32(tagLocator.TagOffset),
})
}
- _, err := s.repo.Publish(event.TopicEntityEvent, bus.NewMessage(bus.MessageID(now), &databasev1.EntityEvent{
+ _, err := s.repo.Publish(event.MeasureTopicEntityEvent, bus.NewMessage(bus.MessageID(now), &databasev1.EntityEvent{
Subject: &commonv1.Metadata{
Name: sMeta.name,
Group: sMeta.group,
@@ -141,7 +141,7 @@ func (s *service) Serve() error {
return err
}
for i := 0; i < int(sMeta.schema.GetOpts().GetShardNum()); i++ {
- _, errShard := s.repo.Publish(event.TopicShardEvent, bus.NewMessage(bus.MessageID(now), &databasev1.ShardEvent{
+ _, errShard := s.repo.Publish(event.MeasureTopicShardEvent, bus.NewMessage(bus.MessageID(now), &databasev1.ShardEvent{
Shard: &databasev1.Shard{
Id: uint64(i),
Total: sMeta.schema.GetOpts().GetShardNum(),
@@ -166,7 +166,7 @@ func (s *service) Serve() error {
}
}
}
- errWrite := s.pipeline.Subscribe(data.TopicStreamWrite, s.writeListener)
+ errWrite := s.pipeline.Subscribe(data.TopicMeasureWrite, s.writeListener)
if errWrite != nil {
return errWrite
}
@@ -192,3 +192,7 @@ func NewService(_ context.Context, metadata metadata.Repo, repo discovery.Servic
pipeline: pipeline,
}, nil
}
+
+func formatMeasureID(name, group string) string {
+ return name + ":" + group
+}
diff --git a/banyand/measure/testdata/query_data.json b/banyand/measure/testdata/query_data.json
new file mode 100644
index 0000000..11a4dea
--- /dev/null
+++ b/banyand/measure/testdata/query_data.json
@@ -0,0 +1,47 @@
+[
+ {
+ "tag_families": [
+ {
+ "tags": [
+ {"str":{"value": "1"}},
+ {"str":{"value": "minute"}}
+ ]
+ }
+ ],
+ "fields": [
+ {"int":{"value": 100}},
+ {"int":{"value": 100}},
+ {"int":{"value": 1}}
+ ]
+ },
+ {
+ "tag_families": [
+ {
+ "tags": [
+ {"str":{"value": "1"}},
+ {"str":{"value": "minute"}}
+ ]
+ }
+ ],
+ "fields": [
+ {"int":{"value": 200}},
+ {"int":{"value": 50}},
+ {"int":{"value": 4}}
+ ]
+ },
+ {
+ "tag_families": [
+ {
+ "tags": [
+ {"str":{"value": "1"}},
+ {"str":{"value": "minute"}}
+ ]
+ }
+ ],
+ "fields": [
+ {"int":{"value": 150}},
+ {"int":{"value": 300}},
+ {"int":{"value": 5}}
+ ]
+ }
+]
\ No newline at end of file
diff --git a/banyand/measure/testdata/write_data.json b/banyand/measure/testdata/write_data.json
new file mode 100644
index 0000000..e4f4e6d
--- /dev/null
+++ b/banyand/measure/testdata/write_data.json
@@ -0,0 +1,47 @@
+[
+ {
+ "tag_families": [
+ {
+ "tags": [
+ {"str":{"value": "1"}},
+ {"str":{"value": "minute"}}
+ ]
+ }
+ ],
+ "fields": [
+ {"int":{"value": 100}},
+ {"int":{"value": 100}},
+ {"int":{"value": 1}}
+ ]
+ },
+ {
+ "tag_families": [
+ {
+ "tags": [
+ {"str":{"value": "1"}},
+ {"str":{"value": "minute"}}
+ ]
+ }
+ ],
+ "fields": [
+ {"int":{"value": 100}},
+ {"int":{"value": 100}},
+ {"int":{"value": 1}}
+ ]
+ },
+ {
+ "tag_families": [
+ {
+ "tags": [
+ {"str":{"value": "1"}},
+ {"str":{"value": "minute"}}
+ ]
+ }
+ ],
+ "fields": [
+ {"int":{"value": 100}},
+ {"int":{"value": 100}},
+ {"int":{"value": 1}}
+ ]
+ }
+]
\ No newline at end of file
diff --git a/banyand/stream/stream_write.go b/banyand/measure/write.go
similarity index 54%
copy from banyand/stream/stream_write.go
copy to banyand/measure/write.go
index 8afed59..b1dc724 100644
--- a/banyand/stream/stream_write.go
+++ b/banyand/measure/write.go
@@ -15,18 +15,23 @@
// specific language governing permissions and limitations
// under the License.
-package stream
+package measure
import (
+ "bytes"
+
"github.com/pkg/errors"
"google.golang.org/protobuf/proto"
"github.com/apache/skywalking-banyandb/api/common"
- streamv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/stream/v1"
+ databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+ measurev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/measure/v1"
+ modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
"github.com/apache/skywalking-banyandb/banyand/tsdb"
+ "github.com/apache/skywalking-banyandb/banyand/tsdb/index"
"github.com/apache/skywalking-banyandb/pkg/bus"
+ "github.com/apache/skywalking-banyandb/pkg/convert"
"github.com/apache/skywalking-banyandb/pkg/logger"
- "github.com/apache/skywalking-banyandb/pkg/partition"
pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
)
@@ -34,7 +39,11 @@ var (
ErrMalformedElement = errors.New("element is malformed")
)
-func (s *stream) Write(value *streamv1.ElementValue) error {
+const (
+ TagFlag byte = iota
+)
+
+func (s *measure) Write(value *measurev1.DataPointValue) error {
entity, shardID, err := s.entityLocator.Locate(value.GetTagFamilies(), s.schema.GetOpts().GetShardNum())
if err != nil {
return err
@@ -51,7 +60,7 @@ func (s *stream) Write(value *streamv1.ElementValue) error {
return nil
}
-func (s *stream) write(shardID common.ShardID, seriesHashKey []byte, value *streamv1.ElementValue, cb callbackFn) error {
+func (s *measure) write(shardID common.ShardID, seriesHashKey []byte, value *measurev1.DataPointValue, cb index.CallbackFn) error {
sm := s.schema
fLen := len(value.GetTagFamilies())
if fLen < 1 {
@@ -85,7 +94,7 @@ func (s *stream) write(shardID common.ShardID, seriesHashKey []byte, value *stre
}
for ti, tag := range family.GetTags() {
tagSpec := familySpec.GetTags()[ti]
- tType, isNull := tagValueTypeConv(tag)
+ tType, isNull := pbv1.TagValueTypeConv(tag)
if isNull {
continue
}
@@ -97,9 +106,26 @@ func (s *stream) write(shardID common.ShardID, seriesHashKey []byte, value *stre
if errMarshal != nil {
return nil, errMarshal
}
- builder.Family(sm.GetTagFamilies()[fi].GetName(), bb)
+ builder.Family(familyIdentity(sm.GetTagFamilies()[fi].GetName(), TagFlag), bb)
+ }
+ if len(value.GetFields()) > len(sm.GetFields()) {
+ return nil, errors.Wrap(ErrMalformedElement, "fields number is more than expected")
+ }
+ for fi, fieldValue := range value.GetFields() {
+ fieldSpec := sm.GetFields()[fi]
+ fType, isNull := pbv1.FieldValueTypeConv(fieldValue)
+ if isNull {
+ continue
+ }
+ if fType != fieldSpec.GetFieldType() {
+ return nil, errors.Wrapf(ErrMalformedElement, "field %s type is unexpected", fieldSpec.GetName())
+ }
+ data := encodeFieldValue(fieldValue)
+ if data == nil {
+ continue
+ }
+ builder.Family(familyIdentity(sm.GetFields()[fi].GetName(), encoderFieldFlag(fieldSpec)), data)
}
- builder.Val([]byte(value.GetElementId()))
writer, errWrite := builder.Build()
if errWrite != nil {
return nil, errWrite
@@ -119,52 +145,25 @@ func (s *stream) write(shardID common.ShardID, seriesHashKey []byte, value *stre
_ = wp.Close()
return err
}
- m := indexMessage{
- localWriter: writer,
- value: value,
- blockCloser: wp,
- cb: cb,
+ m := index.Message{
+ LocalWriter: writer,
+ Value: index.Value{
+ TagFamilies: value.GetTagFamilies(),
+ Timestamp: value.GetTimestamp().AsTime(),
+ },
+ BlockCloser: wp,
+ Cb: cb,
}
- go func(m indexMessage) {
- defer func() {
- if recover() != nil {
- _ = m.blockCloser.Close()
- }
- }()
- s.indexCh <- m
- }(m)
+ s.indexWriter.Write(m)
return err
}
-func getIndexValue(ruleIndex indexRule, value *streamv1.ElementValue) (val []byte, isInt bool, err error) {
- val = make([]byte, 0, len(ruleIndex.tagIndices))
- var existInt bool
- for _, tIndex := range ruleIndex.tagIndices {
- tag, err := partition.GetTagByOffset(value.GetTagFamilies(), tIndex.FamilyOffset, tIndex.TagOffset)
- if err != nil {
- return nil, false, errors.WithMessagef(err, "index rule:%v", ruleIndex.rule.Metadata)
- }
- if tag.GetInt() != nil {
- existInt = true
- }
- v, err := pbv1.MarshalIndexFieldValue(tag)
- if err != nil {
- return nil, false, err
- }
- val = append(val, v...)
- }
- if len(ruleIndex.tagIndices) == 1 && existInt {
- return val, true, nil
- }
- return val, false, nil
-}
-
type writeCallback struct {
l *logger.Logger
- schemaMap map[string]*stream
+ schemaMap map[string]*measure
}
-func setUpWriteCallback(l *logger.Logger, schemaMap map[string]*stream) *writeCallback {
+func setUpWriteCallback(l *logger.Logger, schemaMap map[string]*measure) *writeCallback {
wcb := &writeCallback{
l: l,
schemaMap: schemaMap,
@@ -173,16 +172,50 @@ func setUpWriteCallback(l *logger.Logger, schemaMap map[string]*stream) *writeCa
}
func (w *writeCallback) Rev(message bus.Message) (resp bus.Message) {
- writeEvent, ok := message.Data().(*streamv1.InternalWriteRequest)
+ writeEvent, ok := message.Data().(*measurev1.InternalWriteRequest)
if !ok {
w.l.Warn().Msg("invalid event data type")
return
}
sm := writeEvent.GetRequest().GetMetadata()
- id := formatStreamID(sm.GetName(), sm.GetGroup())
- err := w.schemaMap[id].write(common.ShardID(writeEvent.GetShardId()), writeEvent.GetSeriesHash(), writeEvent.GetRequest().GetElement(), nil)
+ id := formatMeasureID(sm.GetName(), sm.GetGroup())
+ err := w.schemaMap[id].write(common.ShardID(writeEvent.GetShardId()), writeEvent.GetSeriesHash(), writeEvent.GetRequest().GetDataPoint(), nil)
if err != nil {
w.l.Debug().Err(err)
}
return
}
+
+func familyIdentity(name string, flag byte) []byte {
+ return bytes.Join([][]byte{[]byte(name), {flag}}, nil)
+}
+
+func encodeFieldValue(fieldValue *modelv1.FieldValue) []byte {
+ switch fieldValue.GetValue().(type) {
+ case *modelv1.FieldValue_Int:
+ return convert.Int64ToBytes(fieldValue.GetInt().GetValue())
+ case *modelv1.FieldValue_Str:
+ return []byte(fieldValue.GetStr().Value)
+ case *modelv1.FieldValue_BinaryData:
+ return fieldValue.GetBinaryData()
+ }
+ return nil
+}
+
+func decodeFieldValue(fieldValue []byte, fieldSpec *databasev1.FieldSpec) *modelv1.FieldValue {
+ switch fieldSpec.GetFieldType() {
+ case databasev1.FieldType_FIELD_TYPE_STRING:
+ return &modelv1.FieldValue{Value: &modelv1.FieldValue_Str{Str: &modelv1.Str{Value: string(fieldValue)}}}
+ case databasev1.FieldType_FIELD_TYPE_INT:
+ return &modelv1.FieldValue{Value: &modelv1.FieldValue_Int{Int: &modelv1.Int{Value: convert.BytesToInt64(fieldValue)}}}
+ case databasev1.FieldType_FIELD_TYPE_DATA_BINARY:
+ return &modelv1.FieldValue{Value: &modelv1.FieldValue_BinaryData{BinaryData: fieldValue}}
+ }
+ return &modelv1.FieldValue{Value: &modelv1.FieldValue_Null{}}
+}
+
+func encoderFieldFlag(fieldSpec *databasev1.FieldSpec) byte {
+ encodingMethod := byte(fieldSpec.GetEncodingMethod().Number())
+ compressionMethod := byte(fieldSpec.GetCompressionMethod().Number())
+ return encodingMethod<<4 | compressionMethod
+}
diff --git a/banyand/measure/write_test.go b/banyand/measure/write_test.go
new file mode 100644
index 0000000..3141c5a
--- /dev/null
+++ b/banyand/measure/write_test.go
@@ -0,0 +1,113 @@
+// 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 measure
+
+import (
+ "context"
+ "embed"
+ "encoding/json"
+ "os"
+ "testing"
+ "time"
+
+ "github.com/golang/protobuf/jsonpb"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+ "google.golang.org/protobuf/types/known/timestamppb"
+
+ commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
+ measurev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/measure/v1"
+ "github.com/apache/skywalking-banyandb/banyand/metadata"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ "github.com/apache/skywalking-banyandb/pkg/test"
+ testmeasure "github.com/apache/skywalking-banyandb/pkg/test/measure"
+)
+
+func Test_Measure_Write(t *testing.T) {
+ s, deferFunc := setup(t)
+ defer deferFunc()
+ writeData(t, "write_data.json", s)
+}
+
+func setup(t *testing.T) (*measure, func()) {
+ req := require.New(t)
+ req.NoError(logger.Init(logger.Logging{
+ Env: "dev",
+ Level: "info",
+ }))
+ tempDir, deferFunc := test.Space(req)
+
+ mService, err := metadata.NewService(context.TODO())
+ req.NoError(err)
+
+ etcdRootDir := testmeasure.RandomTempDir()
+ err = mService.FlagSet().Parse([]string{"--metadata-root-path=" + etcdRootDir})
+ req.NoError(err)
+
+ err = mService.PreRun()
+ req.NoError(err)
+
+ err = testmeasure.PreloadSchema(mService.SchemaRegistry())
+ req.NoError(err)
+
+ sa, err := mService.MeasureRegistry().GetMeasure(context.TODO(), &commonv1.Metadata{
+ Name: "cpm",
+ Group: "default",
+ })
+ req.NoError(err)
+ iRules, err := mService.IndexRules(context.TODO(), sa.Metadata)
+ req.NoError(err)
+ sSpec := measureSpec{
+ schema: sa,
+ indexRules: iRules,
+ }
+ s, err := openMeasure(tempDir, sSpec, logger.GetLogger("test"))
+ req.NoError(err)
+ return s, func() {
+ _ = s.Close()
+ mService.GracefulStop()
+ deferFunc()
+ _ = os.RemoveAll(etcdRootDir)
+ }
+}
+
+//go:embed testdata/*.json
+var dataFS embed.FS
+
+func writeData(testing *testing.T, dataFile string, measure *measure) (baseTime time.Time) {
+ t := assert.New(testing)
+ r := require.New(testing)
+ var templates []interface{}
+ baseTime = time.Now()
+ content, err := dataFS.ReadFile("testdata/" + dataFile)
+ r.NoError(err)
+ r.NoError(json.Unmarshal(content, &templates))
+ now := time.Now()
+ for i, template := range templates {
+ rawDataPointValue, errMarshal := json.Marshal(template)
+ r.NoError(errMarshal)
+ dataPointValue := &measurev1.DataPointValue{}
+ if dataPointValue.Timestamp == nil {
+ dataPointValue.Timestamp = timestamppb.New(now.Add(time.Duration(i) * time.Minute))
+ }
+ r.NoError(jsonpb.UnmarshalString(string(rawDataPointValue), dataPointValue))
+ errInner := measure.Write(dataPointValue)
+ t.NoError(errInner)
+ }
+ return baseTime
+}
diff --git a/banyand/metadata/metadata_test.go b/banyand/metadata/metadata_test.go
index 6abc9be..3684dc2 100644
--- a/banyand/metadata/metadata_test.go
+++ b/banyand/metadata/metadata_test.go
@@ -26,7 +26,7 @@ import (
commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
- "github.com/apache/skywalking-banyandb/pkg/test"
+ test "github.com/apache/skywalking-banyandb/pkg/test/stream"
)
func Test_service_RulesBySubject(t *testing.T) {
diff --git a/banyand/metadata/schema/etcd_test.go b/banyand/metadata/schema/etcd_test.go
index cacb5d2..5751ba5 100644
--- a/banyand/metadata/schema/etcd_test.go
+++ b/banyand/metadata/schema/etcd_test.go
@@ -162,7 +162,7 @@ func Test_Etcd_Entity_Get(t *testing.T) {
},
{
name: "Get unknown Measure",
- meta: &commonv1.Metadata{Name: "unknown-measure", Group: "default"},
+ meta: &commonv1.Metadata{Name: "unknown-stream", Group: "default"},
get: func(r Registry, meta *commonv1.Metadata) (HasMetadata, error) {
e, innerErr := registry.GetMeasure(context.TODO(), meta)
if innerErr != nil {
diff --git a/banyand/query/processor.go b/banyand/query/processor.go
index c6217de..b8c38df 100644
--- a/banyand/query/processor.go
+++ b/banyand/query/processor.go
@@ -62,7 +62,7 @@ func (q *queryProcessor) Rev(message bus.Message) (resp bus.Message) {
meta := queryCriteria.GetMetadata()
ec, err := q.streamService.Stream(meta)
if err != nil {
- q.logger.Error().Err(err).Msg("fail to get stream execution context")
+ q.logger.Error().Err(err).Msg("fail to get measure execution context")
return
}
diff --git a/banyand/query/processor_test.go b/banyand/query/processor_test.go
index 2320317..651ca8f 100644
--- a/banyand/query/processor_test.go
+++ b/banyand/query/processor_test.go
@@ -45,6 +45,7 @@ import (
pb "github.com/apache/skywalking-banyandb/pkg/pb/v1"
"github.com/apache/skywalking-banyandb/pkg/query/logical"
"github.com/apache/skywalking-banyandb/pkg/test"
+ teststream "github.com/apache/skywalking-banyandb/pkg/test/stream"
)
var (
@@ -95,7 +96,7 @@ func setupServices(tester *require.Assertions) (stream.Service, queue.Queue, fun
streamSvc, err := stream.NewService(context.TODO(), metadataSvc, repo, pipeline)
tester.NoError(err)
- etcdRootDir := test.RandomTempDir()
+ etcdRootDir := teststream.RandomTempDir()
err = metadataSvc.FlagSet().Parse([]string{"--metadata-root-path=" + etcdRootDir})
tester.NoError(err)
@@ -108,13 +109,13 @@ func setupServices(tester *require.Assertions) (stream.Service, queue.Queue, fun
// :PreRun:
// 1) metadata
- // 2) stream
+ // 2) measure
// 3) query
// 4) liaison
err = metadataSvc.PreRun()
tester.NoError(err)
- err = test.PreloadSchema(metadataSvc.SchemaRegistry())
+ err = teststream.PreloadSchema(metadataSvc.SchemaRegistry())
tester.NoError(err)
err = streamSvc.PreRun()
diff --git a/banyand/stream/service.go b/banyand/stream/service.go
index 6af8634..38a8ef2 100644
--- a/banyand/stream/service.go
+++ b/banyand/stream/service.go
@@ -128,7 +128,7 @@ func (s *service) Serve() error {
TagOffset: uint32(tagLocator.TagOffset),
})
}
- _, err := s.repo.Publish(event.TopicEntityEvent, bus.NewMessage(bus.MessageID(now), &databasev1.EntityEvent{
+ _, err := s.repo.Publish(event.StreamTopicEntityEvent, bus.NewMessage(bus.MessageID(now), &databasev1.EntityEvent{
Subject: &commonv1.Metadata{
Name: sMeta.name,
Group: sMeta.group,
@@ -141,7 +141,7 @@ func (s *service) Serve() error {
return err
}
for i := 0; i < int(sMeta.schema.GetOpts().GetShardNum()); i++ {
- _, errShard := s.repo.Publish(event.TopicShardEvent, bus.NewMessage(bus.MessageID(now), &databasev1.ShardEvent{
+ _, errShard := s.repo.Publish(event.StreamTopicShardEvent, bus.NewMessage(bus.MessageID(now), &databasev1.ShardEvent{
Shard: &databasev1.Shard{
Id: uint64(i),
Total: sMeta.schema.GetOpts().GetShardNum(),
diff --git a/banyand/stream/stream.go b/banyand/stream/stream.go
index 444a483..8eab183 100644
--- a/banyand/stream/stream.go
+++ b/banyand/stream/stream.go
@@ -21,8 +21,8 @@ import (
"context"
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/tsdb"
+ "github.com/apache/skywalking-banyandb/banyand/tsdb/index"
"github.com/apache/skywalking-banyandb/pkg/encoding"
"github.com/apache/skywalking-banyandb/pkg/logger"
"github.com/apache/skywalking-banyandb/pkg/partition"
@@ -31,26 +31,19 @@ import (
// a chunk is 1MB
const chunkSize = 1 << 20
-type indexRule struct {
- rule *databasev1.IndexRule
- tagIndices []partition.TagLocator
-}
-
type stream struct {
- name string
- group string
- l *logger.Logger
- schema *databasev1.Stream
- db tsdb.Database
- entityLocator partition.EntityLocator
- indexRules []*databasev1.IndexRule
- indexRuleIndex []indexRule
-
- indexCh chan indexMessage
+ name string
+ group string
+ l *logger.Logger
+ schema *databasev1.Stream
+ db tsdb.Database
+ entityLocator partition.EntityLocator
+ indexRules []*databasev1.IndexRule
+ indexWriter *index.Writer
}
func (s *stream) Close() error {
- close(s.indexCh)
+ _ = s.indexWriter.Close()
return s.db.Close()
}
@@ -58,33 +51,7 @@ func (s *stream) parseSchema() {
sm := s.schema
meta := sm.GetMetadata()
s.name, s.group = meta.GetName(), meta.GetGroup()
- for _, tagInEntity := range sm.Entity.GetTagNames() {
- fIndex, tIndex, tag := s.findTagByName(tagInEntity)
- if tag != nil {
- s.entityLocator = append(s.entityLocator, partition.TagLocator{FamilyOffset: fIndex, TagOffset: tIndex})
- }
- }
- for _, rule := range s.indexRules {
- tagIndices := make([]partition.TagLocator, 0, len(rule.GetTags()))
- for _, tagInIndex := range rule.GetTags() {
- fIndex, tIndex, tag := s.findTagByName(tagInIndex)
- if tag != nil {
- tagIndices = append(tagIndices, partition.TagLocator{FamilyOffset: fIndex, TagOffset: tIndex})
- }
- }
- s.indexRuleIndex = append(s.indexRuleIndex, indexRule{rule: rule, tagIndices: tagIndices})
- }
-}
-
-func (s *stream) findTagByName(tagName string) (int, int, *databasev1.TagSpec) {
- for fi, family := range s.schema.GetTagFamilies() {
- for ti, tag := range family.Tags {
- if tagName == tag.GetName() {
- return fi, ti, tag
- }
- }
- }
- return 0, 0, nil
+ s.entityLocator = partition.NewEntityLocator(sm.TagFamilies, sm.Entity)
}
type streamSpec struct {
@@ -97,11 +64,11 @@ func openStream(root string, spec streamSpec, l *logger.Logger) (*stream, error)
schema: spec.schema,
indexRules: spec.indexRules,
l: l,
- indexCh: make(chan indexMessage),
}
sm.parseSchema()
+ ctx := context.WithValue(context.Background(), logger.ContextKey, l)
db, err := tsdb.OpenDatabase(
- context.WithValue(context.Background(), logger.ContextKey, l),
+ ctx,
tsdb.DatabaseOpts{
Location: root,
ShardNum: sm.schema.GetOpts().GetShardNum(),
@@ -115,28 +82,15 @@ func openStream(root string, spec streamSpec, l *logger.Logger) (*stream, error)
return nil, err
}
sm.db = db
- sm.bootIndexGenerator()
+ sm.indexWriter = index.NewWriter(ctx, index.WriterOptions{
+ DB: db,
+ ShardNum: spec.schema.GetOpts().ShardNum,
+ Families: spec.schema.TagFamilies,
+ IndexRules: spec.indexRules,
+ })
return sm, nil
}
func formatStreamID(name, group string) string {
return name + ":" + group
}
-
-func tagValueTypeConv(tagValue *modelv1.TagValue) (tagType databasev1.TagType, isNull bool) {
- switch tagValue.GetValue().(type) {
- case *modelv1.TagValue_Int:
- return databasev1.TagType_TAG_TYPE_INT, false
- case *modelv1.TagValue_Str:
- return databasev1.TagType_TAG_TYPE_STRING, false
- case *modelv1.TagValue_IntArray:
- return databasev1.TagType_TAG_TYPE_INT_ARRAY, false
- case *modelv1.TagValue_StrArray:
- return databasev1.TagType_TAG_TYPE_STRING_ARRAY, false
- case *modelv1.TagValue_BinaryData:
- return databasev1.TagType_TAG_TYPE_DATA_BINARY, false
- case *modelv1.TagValue_Null:
- return databasev1.TagType_TAG_TYPE_UNSPECIFIED, true
- }
- return databasev1.TagType_TAG_TYPE_UNSPECIFIED, false
-}
diff --git a/banyand/stream/stream_write.go b/banyand/stream/stream_write.go
index 8afed59..a600bce 100644
--- a/banyand/stream/stream_write.go
+++ b/banyand/stream/stream_write.go
@@ -24,9 +24,9 @@ import (
"github.com/apache/skywalking-banyandb/api/common"
streamv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/stream/v1"
"github.com/apache/skywalking-banyandb/banyand/tsdb"
+ "github.com/apache/skywalking-banyandb/banyand/tsdb/index"
"github.com/apache/skywalking-banyandb/pkg/bus"
"github.com/apache/skywalking-banyandb/pkg/logger"
- "github.com/apache/skywalking-banyandb/pkg/partition"
pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
)
@@ -51,7 +51,7 @@ func (s *stream) Write(value *streamv1.ElementValue) error {
return nil
}
-func (s *stream) write(shardID common.ShardID, seriesHashKey []byte, value *streamv1.ElementValue, cb callbackFn) error {
+func (s *stream) write(shardID common.ShardID, seriesHashKey []byte, value *streamv1.ElementValue, cb index.CallbackFn) error {
sm := s.schema
fLen := len(value.GetTagFamilies())
if fLen < 1 {
@@ -85,7 +85,7 @@ func (s *stream) write(shardID common.ShardID, seriesHashKey []byte, value *stre
}
for ti, tag := range family.GetTags() {
tagSpec := familySpec.GetTags()[ti]
- tType, isNull := tagValueTypeConv(tag)
+ tType, isNull := pbv1.TagValueTypeConv(tag)
if isNull {
continue
}
@@ -97,7 +97,7 @@ func (s *stream) write(shardID common.ShardID, seriesHashKey []byte, value *stre
if errMarshal != nil {
return nil, errMarshal
}
- builder.Family(sm.GetTagFamilies()[fi].GetName(), bb)
+ builder.Family([]byte(sm.GetTagFamilies()[fi].GetName()), bb)
}
builder.Val([]byte(value.GetElementId()))
writer, errWrite := builder.Build()
@@ -111,7 +111,7 @@ func (s *stream) write(shardID common.ShardID, seriesHashKey []byte, value *stre
Interface("data", value).
Uint64("series_id", uint64(series.ID())).
Int("shard_id", int(shardID)).
- Msg("write stream")
+ Msg("write measure")
return writer, errWrite
}
writer, err := writeFn()
@@ -119,46 +119,19 @@ func (s *stream) write(shardID common.ShardID, seriesHashKey []byte, value *stre
_ = wp.Close()
return err
}
- m := indexMessage{
- localWriter: writer,
- value: value,
- blockCloser: wp,
- cb: cb,
+ m := index.Message{
+ LocalWriter: writer,
+ Value: index.Value{
+ TagFamilies: value.GetTagFamilies(),
+ Timestamp: value.GetTimestamp().AsTime(),
+ },
+ BlockCloser: wp,
+ Cb: cb,
}
- go func(m indexMessage) {
- defer func() {
- if recover() != nil {
- _ = m.blockCloser.Close()
- }
- }()
- s.indexCh <- m
- }(m)
+ s.indexWriter.Write(m)
return err
}
-func getIndexValue(ruleIndex indexRule, value *streamv1.ElementValue) (val []byte, isInt bool, err error) {
- val = make([]byte, 0, len(ruleIndex.tagIndices))
- var existInt bool
- for _, tIndex := range ruleIndex.tagIndices {
- tag, err := partition.GetTagByOffset(value.GetTagFamilies(), tIndex.FamilyOffset, tIndex.TagOffset)
- if err != nil {
- return nil, false, errors.WithMessagef(err, "index rule:%v", ruleIndex.rule.Metadata)
- }
- if tag.GetInt() != nil {
- existInt = true
- }
- v, err := pbv1.MarshalIndexFieldValue(tag)
- if err != nil {
- return nil, false, err
- }
- val = append(val, v...)
- }
- if len(ruleIndex.tagIndices) == 1 && existInt {
- return val, true, nil
- }
- return val, false, nil
-}
-
type writeCallback struct {
l *logger.Logger
schemaMap map[string]*stream
diff --git a/banyand/stream/stream_write_test.go b/banyand/stream/stream_write_test.go
index da484d7..e4a58fa 100644
--- a/banyand/stream/stream_write_test.go
+++ b/banyand/stream/stream_write_test.go
@@ -33,6 +33,7 @@ import (
"github.com/apache/skywalking-banyandb/banyand/metadata"
"github.com/apache/skywalking-banyandb/pkg/logger"
"github.com/apache/skywalking-banyandb/pkg/test"
+ teststream "github.com/apache/skywalking-banyandb/pkg/test/stream"
)
func Test_Stream_Write(t *testing.T) {
@@ -194,14 +195,14 @@ func setup(t *testing.T) (*stream, func()) {
mService, err := metadata.NewService(context.TODO())
req.NoError(err)
- etcdRootDir := test.RandomTempDir()
+ etcdRootDir := teststream.RandomTempDir()
err = mService.FlagSet().Parse([]string{"--metadata-root-path=" + etcdRootDir})
req.NoError(err)
err = mService.PreRun()
req.NoError(err)
- err = test.PreloadSchema(mService.SchemaRegistry())
+ err = teststream.PreloadSchema(mService.SchemaRegistry())
req.NoError(err)
sa, err := mService.StreamRegistry().GetStream(context.TODO(), &commonv1.Metadata{
diff --git a/banyand/stream/index.go b/banyand/tsdb/index/writer.go
similarity index 50%
rename from banyand/stream/index.go
rename to banyand/tsdb/index/writer.go
index c1b5e7b..44ef340 100644
--- a/banyand/stream/index.go
+++ b/banyand/tsdb/index/writer.go
@@ -15,64 +15,122 @@
// specific language governing permissions and limitations
// under the License.
-package stream
+package index
import (
+ "context"
"io"
+ "time"
+ "github.com/pkg/errors"
"go.uber.org/multierr"
"github.com/apache/skywalking-banyandb/api/common"
databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
- streamv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/stream/v1"
+ modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
"github.com/apache/skywalking-banyandb/banyand/tsdb"
"github.com/apache/skywalking-banyandb/pkg/index"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
"github.com/apache/skywalking-banyandb/pkg/partition"
+ pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
)
-type callbackFn func()
-type indexMessage struct {
- localWriter tsdb.Writer
- blockCloser io.Closer
- value *streamv1.ElementValue
- cb callbackFn
+type CallbackFn func()
+
+type Message struct {
+ Value Value
+ LocalWriter tsdb.Writer
+ BlockCloser io.Closer
+ Cb CallbackFn
+}
+
+type Value struct {
+ TagFamilies []*modelv1.TagFamilyForWrite
+ Timestamp time.Time
+}
+
+type WriterOptions struct {
+ ShardNum uint32
+ Families []*databasev1.TagFamilySpec
+ IndexRules []*databasev1.IndexRule
+ DB tsdb.Database
+}
+
+type Writer struct {
+ l *logger.Logger
+ db tsdb.Database
+ shardNum uint32
+ ch chan Message
+ indexRuleIndex []*partition.IndexRuleLocator
+}
+
+func NewWriter(ctx context.Context, options WriterOptions) *Writer {
+ w := new(Writer)
+ parentLogger := ctx.Value(logger.ContextKey)
+ if parentLogger != nil {
+ if pl, ok := parentLogger.(*logger.Logger); ok {
+ w.l = pl.Named("index-writer")
+ }
+ }
+ w.shardNum = options.ShardNum
+ w.db = options.DB
+ w.indexRuleIndex = partition.ParseIndexRuleLocators(options.Families, options.IndexRules)
+ w.ch = make(chan Message)
+ w.bootIndexGenerator()
+ return w
}
-func (s *stream) bootIndexGenerator() {
+func (s *Writer) Write(value Message) {
+ go func(m Message) {
+ defer func() {
+ if recover() != nil {
+ _ = m.BlockCloser.Close()
+ }
+ }()
+ s.ch <- m
+ }(value)
+}
+
+func (s *Writer) Close() error {
+ close(s.ch)
+ return nil
+}
+
+func (s *Writer) bootIndexGenerator() {
go func() {
for {
- m, more := <-s.indexCh
+ m, more := <-s.ch
if !more {
return
}
var err error
for _, ruleIndex := range s.indexRuleIndex {
- rule := ruleIndex.rule
+ rule := ruleIndex.Rule
switch rule.GetLocation() {
case databasev1.IndexRule_LOCATION_SERIES:
- err = multierr.Append(err, writeLocalIndex(m.localWriter, ruleIndex, m.value))
+ err = multierr.Append(err, writeLocalIndex(m.LocalWriter, ruleIndex, m.Value))
case databasev1.IndexRule_LOCATION_GLOBAL:
- err = multierr.Append(err, s.writeGlobalIndex(ruleIndex, m.localWriter.ItemID(), m.value))
+ err = multierr.Append(err, s.writeGlobalIndex(ruleIndex, m.LocalWriter.ItemID(), m.Value))
}
}
- err = multierr.Append(err, m.blockCloser.Close())
+ err = multierr.Append(err, m.BlockCloser.Close())
if err != nil {
s.l.Error().Err(err).Msg("encounter some errors when generating indices")
}
- if m.cb != nil {
- m.cb()
+ if m.Cb != nil {
+ m.Cb()
}
}
}()
}
//TODO: should listen to pipeline in a distributed cluster
-func (s *stream) writeGlobalIndex(ruleIndex indexRule, ref tsdb.GlobalItemID, value *streamv1.ElementValue) error {
+func (s *Writer) writeGlobalIndex(ruleIndex *partition.IndexRuleLocator, ref tsdb.GlobalItemID, value Value) error {
val, _, err := getIndexValue(ruleIndex, value)
if err != nil {
return err
}
- indexShardID, err := partition.ShardID(val, s.schema.GetOpts().GetShardNum())
+ indexShardID, err := partition.ShardID(val, s.shardNum)
if err != nil {
return err
}
@@ -83,12 +141,12 @@ func (s *stream) writeGlobalIndex(ruleIndex indexRule, ref tsdb.GlobalItemID, va
builder := shard.Index().WriterBuilder()
indexWriter, err := builder.
GlobalItemID(ref).
- Time(value.GetTimestamp().AsTime()).
+ Time(value.Timestamp).
Build()
if err != nil {
return err
}
- rule := ruleIndex.rule
+ rule := ruleIndex.Rule
switch rule.GetType() {
case databasev1.IndexRule_TYPE_INVERTED:
return indexWriter.WriteInvertedIndex(index.Field{
@@ -108,12 +166,12 @@ func (s *stream) writeGlobalIndex(ruleIndex indexRule, ref tsdb.GlobalItemID, va
return err
}
-func writeLocalIndex(writer tsdb.Writer, ruleIndex indexRule, value *streamv1.ElementValue) (err error) {
+func writeLocalIndex(writer tsdb.Writer, ruleIndex *partition.IndexRuleLocator, value Value) (err error) {
val, _, err := getIndexValue(ruleIndex, value)
if err != nil {
return err
}
- rule := ruleIndex.rule
+ rule := ruleIndex.Rule
switch rule.GetType() {
case databasev1.IndexRule_TYPE_INVERTED:
return writer.WriteInvertedIndex(index.Field{
@@ -132,3 +190,26 @@ func writeLocalIndex(writer tsdb.Writer, ruleIndex indexRule, value *streamv1.El
}
return err
}
+
+func getIndexValue(ruleIndex *partition.IndexRuleLocator, value Value) (val []byte, isInt bool, err error) {
+ val = make([]byte, 0, len(ruleIndex.TagIndices))
+ var existInt bool
+ for _, tIndex := range ruleIndex.TagIndices {
+ tag, err := partition.GetTagByOffset(value.TagFamilies, tIndex.FamilyOffset, tIndex.TagOffset)
+ if err != nil {
+ return nil, false, errors.WithMessagef(err, "index rule:%v", ruleIndex.Rule.Metadata)
+ }
+ if tag.GetInt() != nil {
+ existInt = true
+ }
+ v, err := pbv1.MarshalIndexFieldValue(tag)
+ if err != nil {
+ return nil, false, err
+ }
+ val = append(val, v...)
+ }
+ if len(ruleIndex.TagIndices) == 1 && existInt {
+ return val, true, nil
+ }
+ return val, false, nil
+}
diff --git a/banyand/tsdb/series_write.go b/banyand/tsdb/series_write.go
index 8232648..ec14e33 100644
--- a/banyand/tsdb/series_write.go
+++ b/banyand/tsdb/series_write.go
@@ -29,7 +29,7 @@ import (
)
type WriterBuilder interface {
- Family(name string, val []byte) WriterBuilder
+ Family(name []byte, val []byte) WriterBuilder
Time(ts time.Time) WriterBuilder
Val(val []byte) WriterBuilder
Build() (Writer, error)
@@ -54,11 +54,11 @@ type writerBuilder struct {
seriesIDBytes []byte
}
-func (w *writerBuilder) Family(name string, val []byte) WriterBuilder {
+func (w *writerBuilder) Family(name []byte, val []byte) WriterBuilder {
w.values = append(w.values, struct {
family []byte
val []byte
- }{family: []byte(name), val: val})
+ }{family: name, val: val})
return w
}
diff --git a/pkg/encoding/int.go b/pkg/encoding/int.go
index 9571ec4..b8887ec 100644
--- a/pkg/encoding/int.go
+++ b/pkg/encoding/int.go
@@ -72,7 +72,7 @@ func (ie *intEncoder) Append(ts uint64, value []byte) {
ie.bw.WriteBool(false)
ie.num++
}
- ie.bw.WriteBool(true)
+ ie.bw.WriteBool(len(value) > 0)
ie.values.Write(binary.LittleEndian.Uint64(value))
ie.num++
}
diff --git a/pkg/partition/entity.go b/pkg/partition/entity.go
index 682d961..5f5f452 100644
--- a/pkg/partition/entity.go
+++ b/pkg/partition/entity.go
@@ -21,6 +21,7 @@ import (
"github.com/pkg/errors"
"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/tsdb"
pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
@@ -37,6 +38,17 @@ type TagLocator struct {
TagOffset int
}
+func NewEntityLocator(families []*databasev1.TagFamilySpec, entity *databasev1.Entity) EntityLocator {
+ locator := make(EntityLocator, 0, len(entity.GetTagNames()))
+ for _, tagInEntity := range entity.GetTagNames() {
+ fIndex, tIndex, tag := pbv1.FindTagByName(families, tagInEntity)
+ if tag != nil {
+ locator = append(locator, TagLocator{FamilyOffset: fIndex, TagOffset: tIndex})
+ }
+ }
+ return locator
+}
+
func (e EntityLocator) Find(value []*modelv1.TagFamilyForWrite) (tsdb.Entity, error) {
entity := make(tsdb.Entity, len(e))
for i, index := range e {
diff --git a/api/event/discovery.go b/pkg/partition/index.go
similarity index 50%
rename from api/event/discovery.go
rename to pkg/partition/index.go
index b419442..a6f8b81 100644
--- a/api/event/discovery.go
+++ b/pkg/partition/index.go
@@ -15,23 +15,28 @@
// specific language governing permissions and limitations
// under the License.
-package event
+package partition
import (
- "github.com/apache/skywalking-banyandb/api/common"
- "github.com/apache/skywalking-banyandb/pkg/bus"
+ databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+ pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
)
-var (
- ShardEventKindVersion = common.KindVersion{
- Version: "v1",
- Kind: "event-shard",
- }
- TopicShardEvent = bus.UniTopic(ShardEventKindVersion.String())
+type IndexRuleLocator struct {
+ Rule *databasev1.IndexRule
+ TagIndices []TagLocator
+}
- EntityEventKindVersion = common.KindVersion{
- Version: "v1",
- Kind: "event-entity",
+func ParseIndexRuleLocators(families []*databasev1.TagFamilySpec, indexRules []*databasev1.IndexRule) (locators []*IndexRuleLocator) {
+ for _, rule := range indexRules {
+ tagIndices := make([]TagLocator, 0, len(rule.GetTags()))
+ for _, tagInIndex := range rule.GetTags() {
+ fIndex, tIndex, tag := pbv1.FindTagByName(families, tagInIndex)
+ if tag != nil {
+ tagIndices = append(tagIndices, TagLocator{FamilyOffset: fIndex, TagOffset: tIndex})
+ }
+ }
+ locators = append(locators, &IndexRuleLocator{Rule: rule, TagIndices: tagIndices})
}
- TopicEntityEvent = bus.UniTopic(EntityEventKindVersion.String())
-)
+ return locators
+}
diff --git a/pkg/pb/v1/metadata.go b/pkg/pb/v1/metadata.go
new file mode 100644
index 0000000..61f93c2
--- /dev/null
+++ b/pkg/pb/v1/metadata.go
@@ -0,0 +1,66 @@
+// 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 v1
+
+import (
+ databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+ modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
+)
+
+func FindTagByName(families []*databasev1.TagFamilySpec, tagName string) (int, int, *databasev1.TagSpec) {
+ for fi, family := range families {
+ for ti, tag := range family.Tags {
+ if tagName == tag.GetName() {
+ return fi, ti, tag
+ }
+ }
+ }
+ return 0, 0, nil
+}
+
+func TagValueTypeConv(tagValue *modelv1.TagValue) (tagType databasev1.TagType, isNull bool) {
+ switch tagValue.GetValue().(type) {
+ case *modelv1.TagValue_Int:
+ return databasev1.TagType_TAG_TYPE_INT, false
+ case *modelv1.TagValue_Str:
+ return databasev1.TagType_TAG_TYPE_STRING, false
+ case *modelv1.TagValue_IntArray:
+ return databasev1.TagType_TAG_TYPE_INT_ARRAY, false
+ case *modelv1.TagValue_StrArray:
+ return databasev1.TagType_TAG_TYPE_STRING_ARRAY, false
+ case *modelv1.TagValue_BinaryData:
+ return databasev1.TagType_TAG_TYPE_DATA_BINARY, false
+ case *modelv1.TagValue_Null:
+ return databasev1.TagType_TAG_TYPE_UNSPECIFIED, true
+ }
+ return databasev1.TagType_TAG_TYPE_UNSPECIFIED, false
+}
+
+func FieldValueTypeConv(tagValue *modelv1.FieldValue) (tagType databasev1.FieldType, isNull bool) {
+ switch tagValue.GetValue().(type) {
+ case *modelv1.FieldValue_Int:
+ return databasev1.FieldType_FIELD_TYPE_INT, false
+ case *modelv1.FieldValue_Str:
+ return databasev1.FieldType_FIELD_TYPE_STRING, false
+ case *modelv1.FieldValue_BinaryData:
+ return databasev1.FieldType_FIELD_TYPE_DATA_BINARY, false
+ case *modelv1.FieldValue_Null:
+ return databasev1.FieldType_FIELD_TYPE_UNSPECIFIED, true
+ }
+ return databasev1.FieldType_FIELD_TYPE_UNSPECIFIED, false
+}
diff --git a/pkg/query/logical/analyzer_test.go b/pkg/query/logical/analyzer_test.go
index 58cdb8c..dd976a8 100644
--- a/pkg/query/logical/analyzer_test.go
+++ b/pkg/query/logical/analyzer_test.go
@@ -31,11 +31,11 @@ import (
"github.com/apache/skywalking-banyandb/banyand/tsdb"
pb "github.com/apache/skywalking-banyandb/pkg/pb/v1"
"github.com/apache/skywalking-banyandb/pkg/query/logical"
- "github.com/apache/skywalking-banyandb/pkg/test"
+ teststream "github.com/apache/skywalking-banyandb/pkg/test/stream"
)
// setUpAnalyzer creates a default analyzer for testing.
-// You have to close the underlying metadata after test
+// You have to close the underlying metadata after teststream
func setUpAnalyzer() (*logical.Analyzer, func(), error) {
metadataService, err := metadata.NewService(context.TODO())
if err != nil {
@@ -43,7 +43,7 @@ func setUpAnalyzer() (*logical.Analyzer, func(), error) {
}, err
}
- rootDir := test.RandomTempDir()
+ rootDir := teststream.RandomTempDir()
err = metadataService.FlagSet().Parse([]string{"--metadata-root-path=" + rootDir})
if err != nil {
@@ -57,7 +57,7 @@ func setUpAnalyzer() (*logical.Analyzer, func(), error) {
}, err
}
- err = test.PreloadSchema(metadataService.SchemaRegistry())
+ err = teststream.PreloadSchema(metadataService.SchemaRegistry())
if err != nil {
return nil, func() {
}, err
diff --git a/pkg/query/logical/common_test.go b/pkg/query/logical/common_test.go
index a947207..3e3c40e 100644
--- a/pkg/query/logical/common_test.go
+++ b/pkg/query/logical/common_test.go
@@ -39,6 +39,7 @@ import (
"github.com/apache/skywalking-banyandb/banyand/stream"
"github.com/apache/skywalking-banyandb/pkg/logger"
"github.com/apache/skywalking-banyandb/pkg/test"
+ teststream "github.com/apache/skywalking-banyandb/pkg/test/stream"
)
//go:embed testdata/*.json
@@ -90,7 +91,7 @@ func setup(t *require.Assertions) (stream.Stream, metadata.Service, func()) {
metadataSvc, err := metadata.NewService(context.TODO())
t.NoError(err)
- etcdRootDir := test.RandomTempDir()
+ etcdRootDir := teststream.RandomTempDir()
err = metadataSvc.FlagSet().Parse([]string{"--metadata-root-path=" + etcdRootDir})
t.NoError(err)
@@ -101,7 +102,7 @@ func setup(t *require.Assertions) (stream.Stream, metadata.Service, func()) {
err = metadataSvc.PreRun()
t.NoError(err)
- err = test.PreloadSchema(metadataSvc.SchemaRegistry())
+ err = teststream.PreloadSchema(metadataSvc.SchemaRegistry())
t.NoError(err)
err = streamSvc.FlagSet().Parse([]string{"--root-path=" + tempDir})
diff --git a/pkg/test/etcd.go b/pkg/test/measure/etcd.go
similarity index 92%
copy from pkg/test/etcd.go
copy to pkg/test/measure/etcd.go
index 5e2b1d3..958e29d 100644
--- a/pkg/test/etcd.go
+++ b/pkg/test/measure/etcd.go
@@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-package test
+package measure
import (
"context"
@@ -39,8 +39,8 @@ var (
indexRuleStore embed.FS
//go:embed testdata/index_rule_binding.json
indexRuleBindingJSON string
- //go:embed testdata/stream.json
- streamJSON string
+ //go:embed testdata/measure.json
+ measureJSON string
)
func PreloadSchema(e schema.Registry) error {
@@ -48,11 +48,11 @@ func PreloadSchema(e schema.Registry) error {
return err
}
- s := &databasev1.Stream{}
- if err := protojson.Unmarshal([]byte(streamJSON), s); err != nil {
+ s := &databasev1.Measure{}
+ if err := protojson.Unmarshal([]byte(measureJSON), s); err != nil {
return err
}
- err := e.UpdateStream(context.Background(), s)
+ err := e.UpdateMeasure(context.Background(), s)
if err != nil {
return err
}
diff --git a/pkg/test/measure/testdata/index_rule_binding.json b/pkg/test/measure/testdata/index_rule_binding.json
new file mode 100644
index 0000000..544e27f
--- /dev/null
+++ b/pkg/test/measure/testdata/index_rule_binding.json
@@ -0,0 +1,16 @@
+{
+ "metadata": {
+ "name": "cpm-index-rule-binding",
+ "group": "default"
+ },
+ "rules": [
+ "entity_id"
+ ],
+ "subject":{
+ "catalog": "CATALOG_MEASURE",
+ "name": "cpm"
+ },
+ "begin_at": "2021-04-15T01:30:15.01Z",
+ "expire_at": "2121-04-15T01:30:15.01Z",
+ "updated_at": "2021-04-15T01:30:15.01Z"
+}
diff --git a/pkg/test/testdata/index_rules/db.instance.json b/pkg/test/measure/testdata/index_rules/entity_id.json
similarity index 80%
copy from pkg/test/testdata/index_rules/db.instance.json
copy to pkg/test/measure/testdata/index_rules/entity_id.json
index d6bcbb1..bc8fbbe 100644
--- a/pkg/test/testdata/index_rules/db.instance.json
+++ b/pkg/test/measure/testdata/index_rules/entity_id.json
@@ -1,11 +1,11 @@
{
"metadata": {
"id": 1,
- "name": "db.instance",
+ "name": "entity_id",
"group": "default"
},
"tags": [
- "db.instance"
+ "entity_id"
],
"type": "TYPE_INVERTED",
"location": "LOCATION_SERIES",
diff --git a/pkg/test/measure/testdata/measure.json b/pkg/test/measure/testdata/measure.json
new file mode 100644
index 0000000..125a44b
--- /dev/null
+++ b/pkg/test/measure/testdata/measure.json
@@ -0,0 +1,71 @@
+{
+ "metadata": {
+ "name": "cpm",
+ "group": "default"
+ },
+ "tag_families": [
+ {
+ "name": "default",
+ "tags": [
+ {
+ "name": "entity_id",
+ "type": "TAG_TYPE_STRING"
+ },
+ {
+ "name": "scope",
+ "type": "TAG_TYPE_STRING"
+ }
+ ]
+ }
+ ],
+ "fields": [
+ {
+ "name": "summation",
+ "field_type": "FIELD_TYPE_INT",
+ "encoding_method": "ENCODING_METHOD_GORILLA",
+ "compression_method": "COMPRESSION_METHOD_ZSTD"
+ },
+ {
+ "name": "count",
+ "field_type": "FIELD_TYPE_INT",
+ "encoding_method": "ENCODING_METHOD_GORILLA",
+ "compression_method": "COMPRESSION_METHOD_ZSTD"
+ },
+ {
+ "name": "value",
+ "field_type": "FIELD_TYPE_INT",
+ "encoding_method": "ENCODING_METHOD_GORILLA",
+ "compression_method": "COMPRESSION_METHOD_ZSTD"
+ }
+ ],
+ "entity": {
+ "tag_names": [
+ "entity_id"
+ ]
+ },
+ "interval_rules": [
+ {
+ "tag_name": "scope",
+ "str": "minute",
+ "interval": "1m"
+ },
+ {
+ "tag_name": "scope",
+ "str": "hour",
+ "interval": "1h"
+ },
+ {
+ "tag_name": "scope",
+ "str": "day",
+ "interval": "1d"
+ }
+ ],
+ "opts": {
+ "shard_num": 2,
+ "ttl": {
+ "val": 7,
+ "unit": "DURATION_UNIT_DAY"
+ }
+ },
+ "updated_at_nanoseconds": "2021-04-15T01:30:15.01Z"
+}
\ No newline at end of file
diff --git a/pkg/test/etcd.go b/pkg/test/stream/etcd.go
similarity index 99%
rename from pkg/test/etcd.go
rename to pkg/test/stream/etcd.go
index 5e2b1d3..7a6ec38 100644
--- a/pkg/test/etcd.go
+++ b/pkg/test/stream/etcd.go
@@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-package test
+package stream
import (
"context"
diff --git a/pkg/test/testdata/index_rule_binding.json b/pkg/test/stream/testdata/index_rule_binding.json
similarity index 100%
rename from pkg/test/testdata/index_rule_binding.json
rename to pkg/test/stream/testdata/index_rule_binding.json
diff --git a/pkg/test/testdata/index_rules/db.instance.json b/pkg/test/stream/testdata/index_rules/db.instance.json
similarity index 100%
rename from pkg/test/testdata/index_rules/db.instance.json
rename to pkg/test/stream/testdata/index_rules/db.instance.json
diff --git a/pkg/test/testdata/index_rules/db.type.json b/pkg/test/stream/testdata/index_rules/db.type.json
similarity index 100%
rename from pkg/test/testdata/index_rules/db.type.json
rename to pkg/test/stream/testdata/index_rules/db.type.json
diff --git a/pkg/test/testdata/index_rules/duration.json b/pkg/test/stream/testdata/index_rules/duration.json
similarity index 100%
rename from pkg/test/testdata/index_rules/duration.json
rename to pkg/test/stream/testdata/index_rules/duration.json
diff --git a/pkg/test/testdata/index_rules/endpoint_id.json b/pkg/test/stream/testdata/index_rules/endpoint_id.json
similarity index 100%
rename from pkg/test/testdata/index_rules/endpoint_id.json
rename to pkg/test/stream/testdata/index_rules/endpoint_id.json
diff --git a/pkg/test/testdata/index_rules/http.method.json b/pkg/test/stream/testdata/index_rules/http.method.json
similarity index 100%
rename from pkg/test/testdata/index_rules/http.method.json
rename to pkg/test/stream/testdata/index_rules/http.method.json
diff --git a/pkg/test/testdata/index_rules/mq.broker.json b/pkg/test/stream/testdata/index_rules/mq.broker.json
similarity index 100%
rename from pkg/test/testdata/index_rules/mq.broker.json
rename to pkg/test/stream/testdata/index_rules/mq.broker.json
diff --git a/pkg/test/testdata/index_rules/mq.queue.json b/pkg/test/stream/testdata/index_rules/mq.queue.json
similarity index 100%
rename from pkg/test/testdata/index_rules/mq.queue.json
rename to pkg/test/stream/testdata/index_rules/mq.queue.json
diff --git a/pkg/test/testdata/index_rules/mq.topic.json b/pkg/test/stream/testdata/index_rules/mq.topic.json
similarity index 100%
rename from pkg/test/testdata/index_rules/mq.topic.json
rename to pkg/test/stream/testdata/index_rules/mq.topic.json
diff --git a/pkg/test/testdata/index_rules/status_code.json b/pkg/test/stream/testdata/index_rules/status_code.json
similarity index 100%
rename from pkg/test/testdata/index_rules/status_code.json
rename to pkg/test/stream/testdata/index_rules/status_code.json
diff --git a/pkg/test/testdata/index_rules/trace_id.json b/pkg/test/stream/testdata/index_rules/trace_id.json
similarity index 100%
rename from pkg/test/testdata/index_rules/trace_id.json
rename to pkg/test/stream/testdata/index_rules/trace_id.json
diff --git a/pkg/test/testdata/stream.json b/pkg/test/stream/testdata/stream.json
similarity index 100%
rename from pkg/test/testdata/stream.json
rename to pkg/test/stream/testdata/stream.json