hanahmily commented on code in PR #924:
URL:
https://github.com/apache/skywalking-banyandb/pull/924#discussion_r2665137541
##########
pkg/query/logical/measure/measure_plan_distributed.go:
##########
@@ -514,3 +534,52 @@ func (s *pushedDownAggregatedIterator) Current()
[]*measurev1.DataPoint {
func (s *pushedDownAggregatedIterator) Close() error {
return nil
}
+
+// deduplicateAggregatedDataPoints removes duplicate aggregated results from
multiple replicas
+// by keeping only one data point per group. Since replicas hold identical
data, aggregates
+// for the same group are identical across replicas.
+func deduplicateAggregatedDataPoints(dataPoints []*measurev1.DataPoint,
groupByTagsRefs [][]*logical.TagRef) ([]*measurev1.DataPoint, error) {
+ if len(groupByTagsRefs) == 0 {
+ return dataPoints, nil
+ }
+ groupMap := make(map[uint64]*measurev1.DataPoint)
+ result := make([]*measurev1.DataPoint, 0)
+ for _, dp := range dataPoints {
+ key, err := formatGroupByKey(dp, groupByTagsRefs)
+ if err != nil {
+ return nil, err
+ }
+ existing, exists := groupMap[key]
+ if !exists {
+ groupMap[key] = dp
+ result = append(result, dp)
+ } else if !groupByTagsEqual(dp, existing, groupByTagsRefs) {
+ // Hash collision: different groups have the same hash,
keep both
Review Comment:
The hash collision rate is very low, so there is no need to solve it. Remove
groupByTagsEqual.
--
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]
