hanahmily commented on code in PR #239:
URL:
https://github.com/apache/skywalking-banyandb/pull/239#discussion_r1081318960
##########
banyand/liaison/grpc/registry.go:
##########
@@ -435,7 +437,37 @@ type topNAggregationRegistryServer struct {
func (ts *topNAggregationRegistryServer) Create(ctx context.Context,
req *databasev1.TopNAggregationRegistryServiceCreateRequest,
) (*databasev1.TopNAggregationRegistryServiceCreateResponse, error) {
- if err :=
ts.schemaRegistry.TopNAggregationRegistry().CreateTopNAggregation(ctx,
req.GetTopNAggregation()); err != nil {
+ topNSchema := req.GetTopNAggregation()
Review Comment:
Move the measure creation operation into `topNProcessorManager.start`.
Liaison is a proxy and should not implement complicated functions.
If there is no such measure, the manager should create a new one. The
manager should check the on-disk measure's specs. If it's different from
expectation, it could get updated.
##########
banyand/liaison/grpc/registry.go:
##########
@@ -435,7 +437,37 @@ type topNAggregationRegistryServer struct {
func (ts *topNAggregationRegistryServer) Create(ctx context.Context,
req *databasev1.TopNAggregationRegistryServiceCreateRequest,
) (*databasev1.TopNAggregationRegistryServiceCreateResponse, error) {
- if err :=
ts.schemaRegistry.TopNAggregationRegistry().CreateTopNAggregation(ctx,
req.GetTopNAggregation()); err != nil {
+ topNSchema := req.GetTopNAggregation()
+ if err :=
ts.schemaRegistry.TopNAggregationRegistry().CreateTopNAggregation(ctx,
topNSchema); err != nil {
+ return nil, err
+ }
+
+ sourceMeasure, err :=
ts.schemaRegistry.MeasureRegistry().GetMeasure(ctx,
topNSchema.GetSourceMeasure())
+ if err != nil {
+ return nil, err
+ }
+ // create a "virtual" measure for TopN result
+ topNMeasure := &databasev1.Measure{
+ Metadata: topNSchema.GetMetadata(),
+ Interval: sourceMeasure.GetInterval(),
+ TagFamilies: []*databasev1.TagFamilySpec{
+ {
+ Name: measure.TopNTagFamily,
+ Tags: []*databasev1.TagSpec{
+ {
+ Name: "measureID",
Review Comment:
use a snake case here: `measure_id`
##########
banyand/liaison/grpc/registry.go:
##########
@@ -453,12 +485,13 @@ func (ts *topNAggregationRegistryServer) Update(ctx
context.Context,
func (ts *topNAggregationRegistryServer) Delete(ctx context.Context,
req *databasev1.TopNAggregationRegistryServiceDeleteRequest,
) (*databasev1.TopNAggregationRegistryServiceDeleteResponse, error) {
- ok, err :=
ts.schemaRegistry.TopNAggregationRegistry().DeleteTopNAggregation(ctx,
req.GetMetadata())
- if err != nil {
+ okTopNDeleted, errDelTopN :=
ts.schemaRegistry.TopNAggregationRegistry().DeleteTopNAggregation(ctx,
req.GetMetadata())
+ okMeasureDeleted, errDelMeasure :=
ts.schemaRegistry.MeasureRegistry().DeleteMeasure(ctx, req.GetMetadata())
Review Comment:
We don't have to support the cascade deletion. We don't introduce
transactional operations. The operation is too dangerous.
##########
banyand/liaison/grpc/registry.go:
##########
@@ -435,7 +437,37 @@ type topNAggregationRegistryServer struct {
func (ts *topNAggregationRegistryServer) Create(ctx context.Context,
req *databasev1.TopNAggregationRegistryServiceCreateRequest,
) (*databasev1.TopNAggregationRegistryServiceCreateResponse, error) {
- if err :=
ts.schemaRegistry.TopNAggregationRegistry().CreateTopNAggregation(ctx,
req.GetTopNAggregation()); err != nil {
+ topNSchema := req.GetTopNAggregation()
+ if err :=
ts.schemaRegistry.TopNAggregationRegistry().CreateTopNAggregation(ctx,
topNSchema); err != nil {
+ return nil, err
+ }
+
+ sourceMeasure, err :=
ts.schemaRegistry.MeasureRegistry().GetMeasure(ctx,
topNSchema.GetSourceMeasure())
+ if err != nil {
+ return nil, err
+ }
+ // create a "virtual" measure for TopN result
Review Comment:
"virtual" is a common notion. How about using `derived` or `corresponding`
instead?
##########
banyand/measure/measure_write.go:
##########
@@ -39,17 +39,18 @@ import (
var errMalformedElement = errors.New("element is malformed")
-func (s *measure) write(md *commonv1.Metadata, shardID common.ShardID, entity
[]byte, entityValues tsdb.EntityValues, value *measurev1.DataPointValue) error {
+func (s *measure) write(sm *databasev1.Measure, shardID common.ShardID, entity
[]byte, entityValues tsdb.EntityValues,
Review Comment:
You had better remove the first parameter `sm`. We could get it from `s`
directly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
