You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by GitBox <gi...@apache.org> on 2023/01/10 10:19:59 UTC

[GitHub] [skywalking-banyandb] lujiajing1126 opened a new pull request, #239: Unify TagFilter for Streaming scenario

lujiajing1126 opened a new pull request, #239:
URL: https://github.com/apache/skywalking-banyandb/pull/239

   Closes https://github.com/apache/skywalking/issues/9770


-- 
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: notifications-unsubscribe@skywalking.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [skywalking-banyandb] lujiajing1126 commented on a diff in pull request #239: Unify TagFilter for Streaming scenario

Posted by "lujiajing1126 (via GitHub)" <gi...@apache.org>.
lujiajing1126 commented on code in PR #239:
URL: https://github.com/apache/skywalking-banyandb/pull/239#discussion_r1083442123


##########
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:
   > 1. Move `processorManager.start` to the `Serve` phase. That means you should iterate all measures to start their embedded `processorManager`
   
   I take the first approach which is straight forward.
   
   > 2. Introduce Etcd's transactional operation. You could leverage a transaction to create topN resource and its derived measure by moving the creation operation to the `metadata` module instead of `liaison`.
   
   We should support transactional op which would be helpful for https://github.com/apache/skywalking/issues/10010. The current client impl may cause incomplete metadata registration in some edge cases.



-- 
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: notifications-unsubscribe@skywalking.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [skywalking-banyandb] lujiajing1126 commented on a diff in pull request #239: Unify TagFilter for Streaming scenario

Posted by GitBox <gi...@apache.org>.
lujiajing1126 commented on code in PR #239:
URL: https://github.com/apache/skywalking-banyandb/pull/239#discussion_r1082600211


##########
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:
   I think we should avoid creating new measures while loading.......



-- 
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: notifications-unsubscribe@skywalking.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [skywalking-banyandb] lujiajing1126 commented on a diff in pull request #239: Unify TagFilter for Streaming scenario

Posted by "lujiajing1126 (via GitHub)" <gi...@apache.org>.
lujiajing1126 commented on code in PR #239:
URL: https://github.com/apache/skywalking-banyandb/pull/239#discussion_r1083442123


##########
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:
   > 1. Move `processorManager.start` to the `Serve` phase. That means you should iterate all measures to start their embedded `processorManager`
   
   I take the first approach which is straight forward.
   
   > 2. Introduce Etcd's transactional operation. You could leverage a transaction to create topN resource and its derived measure by moving the creation operation to the `metadata` module instead of `liaison`.
   
   We should support transactional op which would be helpful for https://github.com/apache/skywalking/issues/10010. The current client impl could cause incomplete metadata registration in some edge cases.



-- 
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: notifications-unsubscribe@skywalking.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [skywalking-banyandb] lujiajing1126 commented on a diff in pull request #239: Unify TagFilter for Streaming scenario

Posted by GitBox <gi...@apache.org>.
lujiajing1126 commented on code in PR #239:
URL: https://github.com/apache/skywalking-banyandb/pull/239#discussion_r1082603413


##########
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:
   Removed



-- 
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: notifications-unsubscribe@skywalking.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [skywalking-banyandb] lujiajing1126 commented on a diff in pull request #239: Unify TagFilter for Streaming scenario

Posted by GitBox <gi...@apache.org>.
lujiajing1126 commented on code in PR #239:
URL: https://github.com/apache/skywalking-banyandb/pull/239#discussion_r1082598524


##########
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:
   We cannot do this since `topNProcessorManager` will start in the `PreRun` stage, but the measure notifier is registered in the `Serve` stage which will not be called if new measures are create.....Any idea?



##########
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:
   We cannot do this since `topNProcessorManager` will start in the `PreRun` stage, but the measure notifier is registered in the `Serve` stage which will not be called if new measures are create.....Any idea? @hanahmily 



-- 
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: notifications-unsubscribe@skywalking.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [skywalking-banyandb] codecov-commenter commented on pull request #239: Unify TagFilter for Streaming scenario

Posted by GitBox <gi...@apache.org>.
codecov-commenter commented on PR #239:
URL: https://github.com/apache/skywalking-banyandb/pull/239#issuecomment-1377395858

   # [Codecov](https://codecov.io/gh/apache/skywalking-banyandb/pull/239?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
   > Merging [#239](https://codecov.io/gh/apache/skywalking-banyandb/pull/239?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (415342c) into [main](https://codecov.io/gh/apache/skywalking-banyandb/commit/1c19243df23f8350ea5c1542fd914c49e8698960?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (1c19243) will **increase** coverage by `0.81%`.
   > The diff coverage is `49.09%`.
   
   ```diff
   @@            Coverage Diff             @@
   ##             main     #239      +/-   ##
   ==========================================
   + Coverage   45.23%   46.04%   +0.81%     
   ==========================================
     Files          87       87              
     Lines        8879     8793      -86     
   ==========================================
   + Hits         4016     4049      +33     
   + Misses       4484     4364     -120     
   - Partials      379      380       +1     
   ```
   
   
   | [Impacted Files](https://codecov.io/gh/apache/skywalking-banyandb/pull/239?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
   |---|---|---|
   | [banyand/measure/measure\_topn.go](https://codecov.io/gh/apache/skywalking-banyandb/pull/239?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-YmFueWFuZC9tZWFzdXJlL21lYXN1cmVfdG9wbi5nbw==) | `30.06% <0.00%> (+6.67%)` | :arrow_up: |
   | [pkg/query/logical/measure/measure\_analyzer.go](https://codecov.io/gh/apache/skywalking-banyandb/pull/239?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGtnL3F1ZXJ5L2xvZ2ljYWwvbWVhc3VyZS9tZWFzdXJlX2FuYWx5emVyLmdv) | `0.00% <ø> (ø)` | |
   | [.../query/logical/measure/measure\_plan\_aggregation.go](https://codecov.io/gh/apache/skywalking-banyandb/pull/239?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGtnL3F1ZXJ5L2xvZ2ljYWwvbWVhc3VyZS9tZWFzdXJlX3BsYW5fYWdncmVnYXRpb24uZ28=) | `0.00% <0.00%> (ø)` | |
   | [pkg/query/logical/measure/measure\_plan\_top.go](https://codecov.io/gh/apache/skywalking-banyandb/pull/239?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGtnL3F1ZXJ5L2xvZ2ljYWwvbWVhc3VyZS9tZWFzdXJlX3BsYW5fdG9wLmdv) | `0.00% <0.00%> (ø)` | |
   | [banyand/measure/schema.go](https://codecov.io/gh/apache/skywalking-banyandb/pull/239?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-YmFueWFuZC9tZWFzdXJlL3NjaGVtYS5nbw==) | `12.06% <11.76%> (ø)` | |
   | [banyand/measure/measure.go](https://codecov.io/gh/apache/skywalking-banyandb/pull/239?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-YmFueWFuZC9tZWFzdXJlL21lYXN1cmUuZ28=) | `86.07% <89.28%> (+5.30%)` | :arrow_up: |
   
   :mega: We’re building smart automated test selection to slash your CI/CD build times. [Learn more](https://about.codecov.io/iterative-testing/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
   


-- 
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: notifications-unsubscribe@skywalking.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [skywalking-banyandb] lujiajing1126 commented on a diff in pull request #239: Unify TagFilter for Streaming scenario

Posted by GitBox <gi...@apache.org>.
lujiajing1126 commented on code in PR #239:
URL: https://github.com/apache/skywalking-banyandb/pull/239#discussion_r1082550935


##########
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:
   This part is moved to `measure` pkg. PTAL



-- 
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: notifications-unsubscribe@skywalking.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [skywalking-banyandb] hanahmily commented on a diff in pull request #239: Unify TagFilter for Streaming scenario

Posted by "hanahmily (via GitHub)" <gi...@apache.org>.
hanahmily commented on code in PR #239:
URL: https://github.com/apache/skywalking-banyandb/pull/239#discussion_r1083372899


##########
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:
   Placing the creation in the loading phase ensures the measure always exists. Restarting the server will create the target measure if it's missing. As I mentioned, banyandb doesn't support transactional operations. Creating more than one resource in a task is too dangerous.
   
   There are two options here:
   
   1. Move `processorManager.start` to the `Serve` phase. That means you should iterate all measures to start their embedded `processorManager`
   2. Introduce Etcd's transactional operation. Move the creation operation to the `metadata` module instead of `liaison`.



-- 
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: notifications-unsubscribe@skywalking.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [skywalking-banyandb] hanahmily commented on a diff in pull request #239: Unify TagFilter for Streaming scenario

Posted by "hanahmily (via GitHub)" <gi...@apache.org>.
hanahmily commented on code in PR #239:
URL: https://github.com/apache/skywalking-banyandb/pull/239#discussion_r1083372899


##########
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:
   Placing the creation in the loading phase ensures the measure always exists. Restarting the server will create the target measure if it's missing. As I mentioned, banyandb doesn't support transactional operations. Creating more than one resource in a task is too dangerous.
   
   There are two options here:
   
   1. Move `processorManager.start` to the `Serve` phase. That means you should iterate all measures to start their embedded `processorManager`
   2. Introduce Etcd's transactional operation. You could leverage a transaction to create topN resource and its derived measure by moving the creation operation to the `metadata` module instead of `liaison`.



-- 
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: notifications-unsubscribe@skywalking.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [skywalking-banyandb] hanahmily commented on a diff in pull request #239: Unify TagFilter for Streaming scenario

Posted by "hanahmily (via GitHub)" <gi...@apache.org>.
hanahmily commented on code in PR #239:
URL: https://github.com/apache/skywalking-banyandb/pull/239#discussion_r1083444222


##########
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:
   We might support the transactional operations on fixing 10010



-- 
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: notifications-unsubscribe@skywalking.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [skywalking-banyandb] lujiajing1126 commented on a diff in pull request #239: Unify TagFilter for Streaming scenario

Posted by GitBox <gi...@apache.org>.
lujiajing1126 commented on code in PR #239:
URL: https://github.com/apache/skywalking-banyandb/pull/239#discussion_r1081016963


##########
banyand/measure/measure_write.go:
##########
@@ -60,7 +61,7 @@ func (s *measure) write(md *commonv1.Metadata, shardID common.ShardID, entity []
 	if err != nil {
 		return err
 	}
-	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	ctx, cancel := context.WithTimeout(context.TODO(), 5*time.Second)

Review Comment:
   There is no difference between `TODO` and `Background` from the implementation. Anyway, reverted.



-- 
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: notifications-unsubscribe@skywalking.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [skywalking-banyandb] hanahmily commented on a diff in pull request #239: Unify TagFilter for Streaming scenario

Posted by GitBox <gi...@apache.org>.
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: notifications-unsubscribe@skywalking.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [skywalking-banyandb] lujiajing1126 commented on pull request #239: Unify TagFilter for Streaming scenario

Posted by GitBox <gi...@apache.org>.
lujiajing1126 commented on PR #239:
URL: https://github.com/apache/skywalking-banyandb/pull/239#issuecomment-1383046938

   I've reverted changes made to the `BuildSchema()` in the `logical/measure` package, but extracted the underlying `TagSpecRegistry` to allow simply using the fine-grained component. This helps overcome the cyclic import issue without breaking the current design.
   
   Further, I've polished the write part by sharing the measure write method among different cases,
   
    - write **real** measure entities,
    - write **virtual** measure entities, i.e. `FlowRecord` generated by `TopNAggregation`, by transforming to a `DataPointForWrite` entity.
   
   The latter would help further optimization, e.g. fully separate `topN` to another package. But this required much work to be done, for example, the current `Repository` is high coupled with a so-called `ResourceSupplier` which is unexpected in the original design.
   
   So I think this PR is sufficient to resolve the current issue. We could stop here.


-- 
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: notifications-unsubscribe@skywalking.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [skywalking-banyandb] hanahmily commented on a diff in pull request #239: Unify TagFilter for Streaming scenario

Posted by GitBox <gi...@apache.org>.
hanahmily commented on code in PR #239:
URL: https://github.com/apache/skywalking-banyandb/pull/239#discussion_r1072158568


##########
banyand/measure/measure_write.go:
##########
@@ -121,6 +122,9 @@ func (s *measure) write(md *commonv1.Metadata, shardID common.ShardID, entity []
 		_ = wp.Close()
 		return err
 	}
+	if isVirtualMeasure {

Review Comment:
   You don't have to add this flag. The `measure` stores topn's results don't refer to any `indexRules`. It won't generate any indices. 
   
   



##########
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:
   Injecting a faked measure is a terrible design that will mess up the writing process. 
   You had better build and register a new `measure` to write topN's result. 



##########
banyand/measure/measure_write.go:
##########
@@ -60,7 +61,7 @@ func (s *measure) write(md *commonv1.Metadata, shardID common.ShardID, entity []
 	if err != nil {
 		return err
 	}
-	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	ctx, cancel := context.WithTimeout(context.TODO(), 5*time.Second)

Review Comment:
   Why do you change it to `TODO`? This context is the top-level, `Background` is better here. 



-- 
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: notifications-unsubscribe@skywalking.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [skywalking-banyandb] lujiajing1126 commented on a diff in pull request #239: Unify TagFilter for Streaming scenario

Posted by GitBox <gi...@apache.org>.
lujiajing1126 commented on code in PR #239:
URL: https://github.com/apache/skywalking-banyandb/pull/239#discussion_r1082550935


##########
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:
   This part is moved to `measure` pkg. PTAL



##########
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:
   Removed



-- 
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: notifications-unsubscribe@skywalking.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [skywalking-banyandb] lujiajing1126 commented on pull request #239: Unify TagFilter for Streaming scenario

Posted by GitBox <gi...@apache.org>.
lujiajing1126 commented on PR #239:
URL: https://github.com/apache/skywalking-banyandb/pull/239#issuecomment-1396703190

   I've refactored the wring path of `TopN Aggregation` results by creating new internal writing request to the pipeline. And every TopN schema is now bounded to a real Measure schema which is used to listen and write TopN entities.
   
   The virtual/real measure concepts are still there since the measure of `TopN` is incomplete.


-- 
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: notifications-unsubscribe@skywalking.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [skywalking-banyandb] hanahmily merged pull request #239: Unify TagFilter for Streaming scenario

Posted by "hanahmily (via GitHub)" <gi...@apache.org>.
hanahmily merged PR #239:
URL: https://github.com/apache/skywalking-banyandb/pull/239


-- 
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: notifications-unsubscribe@skywalking.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [skywalking-banyandb] lujiajing1126 commented on a diff in pull request #239: Unify TagFilter for Streaming scenario

Posted by GitBox <gi...@apache.org>.
lujiajing1126 commented on code in PR #239:
URL: https://github.com/apache/skywalking-banyandb/pull/239#discussion_r1082550614


##########
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:
   Fixed



-- 
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: notifications-unsubscribe@skywalking.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [skywalking-banyandb] lujiajing1126 commented on a diff in pull request #239: Unify TagFilter for Streaming scenario

Posted by GitBox <gi...@apache.org>.
lujiajing1126 commented on code in PR #239:
URL: https://github.com/apache/skywalking-banyandb/pull/239#discussion_r1082551208


##########
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:
   Use `derived` now



-- 
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: notifications-unsubscribe@skywalking.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org