You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by "OmCheeLin (via GitHub)" <gi...@apache.org> on 2026/02/11 01:13:01 UTC
[PR] Map-Reduce Aggregation Redesign [skywalking-banyandb]
OmCheeLin opened a new pull request, #970:
URL: https://github.com/apache/skywalking-banyandb/pull/970
### Map-Reduce Aggregation Redesign
- [ ] If this is non-trivial feature, paste the links/URLs to the design doc.
- [ ] Update the documentation to include this new feature.
- [x] Tests(including UT, IT, E2E) are added to verify the new feature.
- [ ] If it's UI related, attach the screenshots below.
- [x] If this pull request closes/resolves/fixes an existing issue, replace the issue number. Fixes apache/skywalking#13694.
- [ ] Update the [`CHANGES` log](https://github.com/apache/skywalking-banyandb/blob/main/CHANGES.md).
--
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
Re: [PR] Map-Reduce Aggregation Redesign [skywalking-banyandb]
Posted by "hanahmily (via GitHub)" <gi...@apache.org>.
hanahmily merged PR #970:
URL: https://github.com/apache/skywalking-banyandb/pull/970
--
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
Re: [PR] Map-Reduce Aggregation Redesign [skywalking-banyandb]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #970:
URL: https://github.com/apache/skywalking-banyandb/pull/970#issuecomment-3884271926
## [Codecov](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/970?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `0%` with `175 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 29.04%. Comparing base ([`3530dd9`](https://app.codecov.io/gh/apache/skywalking-banyandb/commit/3530dd9b5cefaa4fef60eec91ecbfbdfcaf243c5?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`9682517`](https://app.codecov.io/gh/apache/skywalking-banyandb/commit/9682517f882680c3d0cfe8c627cc517fa34a4876?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 131 commits behind head on main.
| [Files with missing lines](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/970?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [.../query/logical/measure/measure\_plan\_aggregation.go](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/970?src=pr&el=tree&filepath=pkg%2Fquery%2Flogical%2Fmeasure%2Fmeasure_plan_aggregation.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGtnL3F1ZXJ5L2xvZ2ljYWwvbWVhc3VyZS9tZWFzdXJlX3BsYW5fYWdncmVnYXRpb24uZ28=) | 0.00% | [134 Missing :warning: ](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/970?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [.../query/logical/measure/measure\_plan\_distributed.go](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/970?src=pr&el=tree&filepath=pkg%2Fquery%2Flogical%2Fmeasure%2Fmeasure_plan_distributed.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGtnL3F1ZXJ5L2xvZ2ljYWwvbWVhc3VyZS9tZWFzdXJlX3BsYW5fZGlzdHJpYnV0ZWQuZ28=) | 0.00% | [31 Missing :warning: ](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/970?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [pkg/query/logical/measure/measure\_analyzer.go](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/970?src=pr&el=tree&filepath=pkg%2Fquery%2Flogical%2Fmeasure%2Fmeasure_analyzer.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGtnL3F1ZXJ5L2xvZ2ljYWwvbWVhc3VyZS9tZWFzdXJlX2FuYWx5emVyLmdv) | 0.00% | [7 Missing :warning: ](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/970?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [pkg/query/logical/measure/topn\_analyzer.go](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/970?src=pr&el=tree&filepath=pkg%2Fquery%2Flogical%2Fmeasure%2Ftopn_analyzer.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGtnL3F1ZXJ5L2xvZ2ljYWwvbWVhc3VyZS90b3BuX2FuYWx5emVyLmdv) | 0.00% | [3 Missing :warning: ](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/970?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
> :exclamation: There is a different number of reports uploaded between BASE (3530dd9) and HEAD (9682517). Click for more details.
>
> <details><summary>HEAD has 1 upload less than BASE</summary>
>
>| Flag | BASE (3530dd9) | HEAD (9682517) |
>|------|------|------|
>||1|0|
></details>
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## main #970 +/- ##
===========================================
- Coverage 45.97% 29.04% -16.93%
===========================================
Files 328 118 -210
Lines 55505 12403 -43102
===========================================
- Hits 25520 3603 -21917
+ Misses 27909 8460 -19449
+ Partials 2076 340 -1736
```
| [Flag](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/970/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [pkg](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/970/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `29.04% <0.00%> (?)` | |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
</details>
[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/970?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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
Re: [PR] Map-Reduce Aggregation Redesign [skywalking-banyandb]
Posted by "hanahmily (via GitHub)" <gi...@apache.org>.
hanahmily commented on code in PR #970:
URL: https://github.com/apache/skywalking-banyandb/pull/970#discussion_r2793211926
##########
api/proto/banyandb/measure/v1/query.proto:
##########
@@ -138,6 +138,8 @@ message QueryRequest {
bool trace = 13;
// stages is used to specify the stage of the data points in the lifecycle
repeated string stages = 14;
- // rewriteAggTopNResult will rewrite agg result to raw data
+ // rewrite_agg_top_n_result will rewrite agg result to raw data
bool rewrite_agg_top_n_result = 15;
+ // agg_return_partial when true asks data nodes to return aggregation partials (for reduce at liaison)
+ bool agg_return_partial = 16;
Review Comment:
Move this field to "InternalQueryRequest"
##########
pkg/query/logical/measure/measure_plan_aggregation.go:
##########
@@ -127,29 +153,37 @@ func (g *aggregationPlan[N]) Execute(ec context.Context) (executor.MIterator, er
if err != nil {
return nil, err
}
+ if g.useReduceMode {
Review Comment:
The "reduce" should support aggregation without grouping.
##########
pkg/query/logical/measure/measure_plan_aggregation.go:
##########
@@ -282,3 +349,114 @@ func (ami *aggAllIterator[N]) Current() []*measurev1.InternalDataPoint {
func (ami *aggAllIterator[N]) Close() error {
return ami.prev.Close()
}
+
+type reduceGroupIterator[N aggregation.Number] struct {
+ prev executor.MIterator
+ aggregationFieldRef *logical.FieldRef
+ reduceFunc aggregation.Reduce[N]
+ err error
+ groupMap map[uint64][]*measurev1.InternalDataPoint
+ groupByTagsRefs [][]*logical.TagRef
+ groupKeys []uint64
+ aggrType modelv1.AggregationFunction
+ index int
+}
+
+func newReduceGroupIterator[N aggregation.Number](
+ prev executor.MIterator,
+ aggregationFieldRef *logical.FieldRef,
+ reduceFunc aggregation.Reduce[N],
+ aggrType modelv1.AggregationFunction,
+ groupByTagsRefs [][]*logical.TagRef,
+) executor.MIterator {
+ return &reduceGroupIterator[N]{
+ prev: prev,
+ aggregationFieldRef: aggregationFieldRef,
+ reduceFunc: reduceFunc,
+ aggrType: aggrType,
+ groupByTagsRefs: groupByTagsRefs,
+ groupMap: make(map[uint64][]*measurev1.InternalDataPoint),
+ }
+}
+
+func (rgi *reduceGroupIterator[N]) loadGroups() bool {
+ if rgi.groupKeys != nil {
+ return true
+ }
+ for rgi.prev.Next() {
+ group := rgi.prev.Current()
+ for _, idp := range group {
+ key, keyErr := formatGroupByKey(idp.GetDataPoint(), rgi.groupByTagsRefs)
Review Comment:
The data is grouped by the measure_plan_groupby.go. Refer to aggGroupIterator to implement a reduce version.
--
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
Re: [PR] Map-Reduce Aggregation Redesign [skywalking-banyandb]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #970:
URL: https://github.com/apache/skywalking-banyandb/pull/970#discussion_r2793129562
##########
pkg/query/logical/measure/measure_plan_aggregation.go:
##########
@@ -282,3 +349,114 @@ func (ami *aggAllIterator[N]) Current() []*measurev1.InternalDataPoint {
func (ami *aggAllIterator[N]) Close() error {
return ami.prev.Close()
}
+
+type reduceGroupIterator[N aggregation.Number] struct {
+ prev executor.MIterator
+ aggregationFieldRef *logical.FieldRef
+ reduceFunc aggregation.Reduce[N]
+ err error
+ groupMap map[uint64][]*measurev1.InternalDataPoint
+ groupByTagsRefs [][]*logical.TagRef
+ groupKeys []uint64
+ aggrType modelv1.AggregationFunction
+ index int
+}
+
+func newReduceGroupIterator[N aggregation.Number](
+ prev executor.MIterator,
+ aggregationFieldRef *logical.FieldRef,
+ reduceFunc aggregation.Reduce[N],
+ aggrType modelv1.AggregationFunction,
+ groupByTagsRefs [][]*logical.TagRef,
+) executor.MIterator {
+ return &reduceGroupIterator[N]{
+ prev: prev,
+ aggregationFieldRef: aggregationFieldRef,
+ reduceFunc: reduceFunc,
+ aggrType: aggrType,
+ groupByTagsRefs: groupByTagsRefs,
+ groupMap: make(map[uint64][]*measurev1.InternalDataPoint),
+ }
+}
+
+func (rgi *reduceGroupIterator[N]) loadGroups() bool {
+ if rgi.groupKeys != nil {
+ return true
+ }
+ for rgi.prev.Next() {
+ group := rgi.prev.Current()
+ for _, idp := range group {
+ key, keyErr := formatGroupByKey(idp.GetDataPoint(), rgi.groupByTagsRefs)
+ if keyErr != nil {
+ rgi.err = keyErr
+ return false
+ }
+ rgi.groupMap[key] = append(rgi.groupMap[key], idp)
+ }
+ }
+ if closeErr := rgi.prev.Close(); closeErr != nil {
+ rgi.err = closeErr
+ return false
+ }
Review Comment:
reduceGroupIterator.loadGroups closes rgi.prev, but reduceGroupIterator.Close() also calls rgi.prev.Close() again. If the underlying iterator Close is not idempotent, this can cause double-close errors or unintended side effects. Consider removing the close from loadGroups (and only closing in Close), or track a closed flag and make Close a no-op after the early close.
```suggestion
```
##########
pkg/query/logical/measure/measure_plan_aggregation.go:
##########
@@ -189,16 +223,30 @@ func (ami *aggGroupIterator[N]) Current() []*measurev1.InternalDataPoint {
if resultDp == nil {
return nil
}
- val, err := aggregation.ToFieldValue(ami.aggrFunc.Val())
- if err != nil {
- ami.err = err
- return nil
- }
- resultDp.Fields = []*measurev1.DataPoint_Field{
- {
- Name: ami.aggregationFieldRef.Field.Name,
- Value: val,
- },
+ if ami.emitPartial {
+ part := ami.mapFunc.Partial()
+ fvs, err := aggregation.PartialToFieldValues(ami.aggrType, part)
+ if err != nil {
+ ami.err = err
+ return nil
+ }
+ resultDp.Fields = make([]*measurev1.DataPoint_Field, len(fvs))
+ for i, fv := range fvs {
+ name := ami.aggregationFieldRef.Field.Name
+ if i > 0 {
+ name = "__agg_count"
+ }
+ resultDp.Fields[i] = &measurev1.DataPoint_Field{Name: name, Value: fv}
+ }
Review Comment:
Partial aggregation encodes the extra MEAN count as a synthetic field named "__agg_count". This introduces an undocumented reserved field name and duplicates a magic string in multiple places, which can collide with real field names and is hard to maintain. Consider defining a constant for this reserved name and documenting/enforcing that it cannot conflict with user fields (or carry partials out-of-band instead of as normal fields).
##########
pkg/query/logical/measure/measure_plan_aggregation.go:
##########
@@ -282,3 +349,114 @@ func (ami *aggAllIterator[N]) Current() []*measurev1.InternalDataPoint {
func (ami *aggAllIterator[N]) Close() error {
return ami.prev.Close()
}
+
+type reduceGroupIterator[N aggregation.Number] struct {
+ prev executor.MIterator
+ aggregationFieldRef *logical.FieldRef
+ reduceFunc aggregation.Reduce[N]
+ err error
+ groupMap map[uint64][]*measurev1.InternalDataPoint
+ groupByTagsRefs [][]*logical.TagRef
+ groupKeys []uint64
+ aggrType modelv1.AggregationFunction
+ index int
+}
+
+func newReduceGroupIterator[N aggregation.Number](
+ prev executor.MIterator,
+ aggregationFieldRef *logical.FieldRef,
+ reduceFunc aggregation.Reduce[N],
+ aggrType modelv1.AggregationFunction,
+ groupByTagsRefs [][]*logical.TagRef,
+) executor.MIterator {
+ return &reduceGroupIterator[N]{
+ prev: prev,
+ aggregationFieldRef: aggregationFieldRef,
+ reduceFunc: reduceFunc,
+ aggrType: aggrType,
+ groupByTagsRefs: groupByTagsRefs,
+ groupMap: make(map[uint64][]*measurev1.InternalDataPoint),
+ }
+}
+
+func (rgi *reduceGroupIterator[N]) loadGroups() bool {
+ if rgi.groupKeys != nil {
+ return true
+ }
+ for rgi.prev.Next() {
+ group := rgi.prev.Current()
+ for _, idp := range group {
+ key, keyErr := formatGroupByKey(idp.GetDataPoint(), rgi.groupByTagsRefs)
+ if keyErr != nil {
+ rgi.err = keyErr
+ return false
+ }
+ rgi.groupMap[key] = append(rgi.groupMap[key], idp)
+ }
+ }
+ if closeErr := rgi.prev.Close(); closeErr != nil {
+ rgi.err = closeErr
+ return false
+ }
+ rgi.groupKeys = make([]uint64, 0, len(rgi.groupMap))
+ for k := range rgi.groupMap {
+ rgi.groupKeys = append(rgi.groupKeys, k)
+ }
Review Comment:
reduceGroupIterator builds rgi.groupKeys by ranging over a map, which yields a nondeterministic iteration order. This can make aggregation result ordering unstable and potentially flaky across runs. Consider preserving first-seen order (like groupBy.hash does via groupLst) or explicitly sorting groupKeys before iteration.
##########
pkg/query/logical/measure/measure_plan_aggregation.go:
##########
@@ -282,3 +349,114 @@ func (ami *aggAllIterator[N]) Current() []*measurev1.InternalDataPoint {
func (ami *aggAllIterator[N]) Close() error {
return ami.prev.Close()
}
+
+type reduceGroupIterator[N aggregation.Number] struct {
+ prev executor.MIterator
+ aggregationFieldRef *logical.FieldRef
+ reduceFunc aggregation.Reduce[N]
+ err error
+ groupMap map[uint64][]*measurev1.InternalDataPoint
+ groupByTagsRefs [][]*logical.TagRef
+ groupKeys []uint64
+ aggrType modelv1.AggregationFunction
+ index int
+}
+
+func newReduceGroupIterator[N aggregation.Number](
+ prev executor.MIterator,
+ aggregationFieldRef *logical.FieldRef,
+ reduceFunc aggregation.Reduce[N],
+ aggrType modelv1.AggregationFunction,
+ groupByTagsRefs [][]*logical.TagRef,
+) executor.MIterator {
+ return &reduceGroupIterator[N]{
+ prev: prev,
+ aggregationFieldRef: aggregationFieldRef,
+ reduceFunc: reduceFunc,
+ aggrType: aggrType,
+ groupByTagsRefs: groupByTagsRefs,
+ groupMap: make(map[uint64][]*measurev1.InternalDataPoint),
+ }
+}
+
+func (rgi *reduceGroupIterator[N]) loadGroups() bool {
+ if rgi.groupKeys != nil {
+ return true
+ }
+ for rgi.prev.Next() {
+ group := rgi.prev.Current()
+ for _, idp := range group {
+ key, keyErr := formatGroupByKey(idp.GetDataPoint(), rgi.groupByTagsRefs)
+ if keyErr != nil {
+ rgi.err = keyErr
+ return false
+ }
+ rgi.groupMap[key] = append(rgi.groupMap[key], idp)
+ }
Review Comment:
reduceGroupIterator.loadGroups buffers all incoming datapoints into rgi.groupMap before returning any results. This prevents streaming and can cause high memory usage for large aggregations (even if results are partially aggregated). Consider reducing per-group as data arrives (streaming combine) or at least bounding memory usage.
##########
pkg/query/logical/measure/measure_plan_distributed.go:
##########
@@ -310,7 +311,7 @@ func (t *distributedPlan) Execute(ctx context.Context) (mi executor.MIterator, e
span.Tagf("response_count", "%d", responseCount)
span.Tagf("data_point_count", "%d", dataPointCount)
}
- if t.needCompletePushDownAgg {
+ if t.pushDownAgg {
deduplicatedDps, dedupErr := deduplicateAggregatedDataPointsWithShard(pushedDownAggDps, t.groupByTagsRefs)
if dedupErr != nil {
Review Comment:
When pushDownAgg is enabled and there is no GroupBy, t.groupByTagsRefs is empty and deduplicateAggregatedDataPointsWithShard currently performs no deduplication. In a replicated setup this can cause partials from multiple replicas of the same shard to be combined, producing incorrect aggregated results (double counting, etc.). Consider deduplicating by shard ID even when groupByTagsRefs is empty (e.g., treat groupKey as 0 and still hash with shard_id, or add a dedicated path for no-group-by aggregation).
--
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