(skywalking-banyandb) 01/01: Fix the data race

[ha...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

hanahmily pushed a commit to branch data-race
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git

commit 9bbf506c91306e112e016600a3d64ed7ecbddfb7
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Thu Feb 20 16:29:34 2025 +0800

    Fix the data race
    
    Signed-off-by: Gao Hongtao <ha...@gmail.com>
---
 pkg/query/logical/measure/topn_plan_localscan.go | 30 +++++++++---------------
 1 file changed, 11 insertions(+), 19 deletions(-)

diff --git a/pkg/query/logical/measure/topn_plan_localscan.go b/pkg/query/logical/measure/topn_plan_localscan.go
index 0eba422f..92421573 100644
--- a/pkg/query/logical/measure/topn_plan_localscan.go
+++ b/pkg/query/logical/measure/topn_plan_localscan.go
@@ -152,9 +152,7 @@ func (i *localScan) Execute(ctx context.Context) (mit executor.MIterator, err er
 		return nil, fmt.Errorf("failed to query measure: %w", err)
 	}
 	return &topNMIterator{
-		result:    result,
-		topNValue: measure.GenerateTopNValue(),
-		decoder:   generateTopNValuesDecoder(),
+		result: result,
 	}, nil
 }
 
@@ -173,11 +171,9 @@ func (i *localScan) Schema() logical.Schema {
 }
 
 type topNMIterator struct {
-	result    model.MeasureQueryResult
-	err       error
-	topNValue *measure.TopNValue
-	decoder   *encoding.BytesBlockDecoder
-	current   []*measurev1.DataPoint
+	result  model.MeasureQueryResult
+	err     error
+	current []*measurev1.DataPoint
 }
 
 func (ei *topNMIterator) Next() bool {
@@ -194,6 +190,10 @@ func (ei *topNMIterator) Next() bool {
 		return false
 	}
 	ei.current = ei.current[:0]
+	topNValue := measure.GenerateTopNValue()
+	defer measure.ReleaseTopNValue(topNValue)
+	decoder := generateTopNValuesDecoder()
+	defer releaseTopNValuesDecoder(decoder)
 
 	for i := range r.Timestamps {
 		fv := r.Fields[0].Values[i]
@@ -203,13 +203,13 @@ func (ei *topNMIterator) Next() bool {
 			return false
 		}
 		ts := timestamppb.New(time.Unix(0, r.Timestamps[i]))
-		ei.topNValue.Reset()
-		err := ei.topNValue.Unmarshal(bd, ei.decoder)
+		topNValue.Reset()
+		err := topNValue.Unmarshal(bd, decoder)
 		if err != nil {
 			ei.err = multierr.Append(ei.err, errors.WithMessagef(err, "failed to unmarshal topN values[%d]:[%s]%s", i, ts, hex.EncodeToString(fv.GetBinaryData())))
 			continue
 		}
-		fieldName, entityNames, values, entities := ei.topNValue.Values()
+		fieldName, entityNames, values, entities := topNValue.Values()
 		for j := range entities {
 			dp := &measurev1.DataPoint{
 				Timestamp: ts,
@@ -251,14 +251,6 @@ func (ei *topNMIterator) Close() error {
 		ei.result.Release()
 		ei.result = nil
 	}
-	if ei.decoder != nil {
-		releaseTopNValuesDecoder(ei.decoder)
-		ei.decoder = nil
-	}
-	if ei.topNValue != nil {
-		measure.ReleaseTopNValue(ei.topNValue)
-		ei.topNValue = nil
-	}
 	return ei.err
 }