physicalplan: add support for multi-stage execution of aggregate functions

[yuzefovich]

Most aggregate functions support multi-stage distributed execution in which on every node we perform a local aggregation, then we route all the intermediate results to all of the nodes (according to the grouping columns), and then we perform a final aggregation that gives us the result. For some aggregate functions the decomposition into local and final stages is trivial - e.g. in order to calculate sum we need to perform local sum and then the final sum; for other functions it is a bit more involved - e.g. in order to compute avg we calculate partial sum and count locally, then we finalize sum and count at the final stage and we render the result as final_sum / final_count.

(I believe Postgres calls the multi-stage execution a partial mode.)

The decomposition of supported functions is defined here

cockroach/pkg/sql/physicalplan/aggregator_funcs.go

Line 87 in 9f678e7

var DistAggregationTable = map[execinfrapb.AggregatorSpec_Func]DistAggregationInfo{

We are currently missing the following functions:

• sqrdiff (I think it should be possible to decompose it)
• corr
• var_pop
• stddev_pop
• covar_pop
• covar_samp
• regr_intercept
• regr_r2
• regr_slope
• regr_sxx
• regr_syy
• regr_sxy
• regr_count
• regr_avgx
• regr_avgy

avg and variance are good examples of how more complicated decompositions are handled.

cc @mneverov in case you're interested

Jira issue: CRDB-3407

[mneverov]

@yuzefovich, definitely, that would be awesome to work on this, thanks!
Although, from the conversation:

This doesn't look right. I think this entire thing only makes sense for single-arg operators anyway,

I got the impression that it is impossible to have merge operation for multi-arg operators.

I'll take a look next year, so if anyone can start working immediately - please go ahead.

[yuzefovich]

I think that conversation is not directly applicable to this issue since here we are deciding on how to execute a particular logical plan (meaning after the optimizer has already chosen the plan) whereas that convo was about some "transformations" of the logical plan in order to get an equivalent one that has supposedly less cost.

Possibly, my usage of a word "decomposition" is a bit misleading since AggregatesCanMerge also talks about "decomposition".

[mneverov]

Ok, need to dig into the code :).

[piao88]

Hi @yuzefovich
I'm new to the community and would like to pick this up.

[mneverov]

hi @piao88, have you already started working on this? May I take it?

[piao88]

hi @mneverov , I have already started.

[yuzefovich]

@piao88 @mneverov I think this issue is big enough for both of you to work on it, just make sure to communicate which functions you are working on.

[piao88]

hi @mneverov @yuzefovich I have already finished var_pop and stddev_pop

[henrjan]

hi, is this issue still ongoing? can i work on this?

[yuzefovich]

Hi @henrjan, thanks for the interest! Let's check in with @piao88 who have already opened up a PR (#59174) that addresses this issue.

@piao88 are you still interested in pushing #59174 forward or would you be ok if someone took over this issue and possibly your PR?

[piao88]

Hi@henrjan ,Of course you can .

[alimoosavi]

@yuzefovich
I am new to cockroachdb project and this issue seems attractive to me , please assign this issue to me.

[yuzefovich]

@mneverov are you still interested in working on this issue? Looks like @piao88 made some progress in #59174, but that PR seems abandoned, so I'd assign the issue to @mneverov if they are still interested.

[piao88]

@yuzefovich Sorry, I am no longer interested in this issue, you can give it to someone else.

[mneverov]

thanks @yuzefovich , i'll take a look