Support for single phase and two-phase distributed hash aggregation

[siddharthteotia]

I am not entirely sure to what extent this is already supported -- so apologies if this is something that is already done or something not applicable

Idea:

GROUP BY processing using hash based aggregation in a distributed query engine can be done using two different methods and each one has an impact on performance depending on cardinality

(1) Two-phase hash aggregation: In the first phase, each node will do a local group by from the data (segment) on that node -- essentially computing partial resultset from a single thread query execution for a given segment at a given node.

The physical plan or execution plan will have a shuffle above the aggregation group by operator -- the shuffle operator will use the partial aggregates and send them around such that each node now ends up with all the data (partial aggregates) corresponding to group by key(s).

The second phase of hash agg will now work on the partial aggregates and compute the final resultset for a given set of key(s) per node. Each node can now send its result set to broker. Broker only needs to union as opposed to doing any aggregation related processing.

(2) Single-phase hash aggregation: The above described two phase method proves to be useful if the cardinality of group by key columns is low -- since doing local hash agg first achieves a higher degree of reduction and thus reduces the data sent around. However, if cardinality is very high then it is better to do single phase aggregation. In this case, below the aggregation operator, there will be a shuffle operator. Again, after hash agg, each node can send the final result set to broker and it just needs to union.

[kishoreg]

What we have in Pinot is very similar to two-phase hash aggregation but we merge groups across segments within the same node. We call this Combine Operator - analogous to the map-combine-reduce concept in M/R. We have S (number of segments) mappers, N (number of servers) Combiners, 1 Reducer.

I think what you are suggesting S mappers, N reducers and 1 accumulator (union operator). I am trying to come up with a scenario where this would be better than the one we currently have in Pinot.

[siddharthteotia]

@kishoreg , thanks for your response. My understanding was that there was no reduction happening until data from each processing node was sent to the broker. I mistakenly thought that Combine operator doesn't come into picture up until broker. Thanks for clarifying that.

So the broker (1 reducer) will do the final aggregation processing by combining the data across nodes. Right?

Another reason behind thinking of two-phase was that for low cardinality but high volume data set, if we can reduce everything at the segment server level (with a shuffle), then less data will be sent to broker. All of the execution specific operator logic can be kept in the segment server and broker need not worry about running out of memory and crashing while doing the final processing -- it can still while doing the union but there we can just use limit. Similarly, the segment servers can support spilling etc.

[mayankshriv]

@siddharthteotia Yes, the broker (1 reducer) will perform the final aggregation across nodes.

If the cardinality is low, then the individual server responses will be small and single broker reducer should be able to handle easily. For scalability, we don't want too many nodes to serve the same query, so the number of nodes is also limited.

If the cardinality is high, then we shouldn't shuffle anyways.

So I have the same question as @kishoreg, if we can come up with a case where shuffle would be better than broker side reduce (for limited number of servers per query), then we can discuss in that context.