Multi-pass aggregation support

[polyfractal]

Aggregations perform a single pass over the data today, reducing the shard-results at the coordinating node and then applying pipeline aggregations on the reduced results. This is ideal for aggregation latency and scalability, but does limit the types of algorithms we can implement.

We would like to start investigating extending the agg framework to multiple passes over the data. In particular, we want to be able to run multiple map-reduce cycles, not just multiple shard-local passes. This is needed when global information (e.g. a global mean) is required for the second pass over the data.

Multi-pass should unblock a number of interesting aggs and enhancements:

• Spatial Stats spatial_stats aggregation #14727
• Conflation Conflation Aggregator #11460
• Numeric and Geo Clustering (2D and low dimension nD, k-means), Add K-means clustering feature #5512
• "other" bucket for Terms agg Terms agg: calculate aggs on 'other' bucket #12411
• Accurate counts on terms agg

Technicals and Open Questions

• We should probably reuse the existing SearchPhase mechanisms in place. This will limit the amount of new code that we need to write, and should (hopefully) play nicer with mechanisms like CCS
  • One approach is adding a new phase that is executed after AggregationPhase, which can recursively keep calling itself for the next phase to perform multiple passes
  • Alternatively, we could implement a new phase after AggregationPhase which deals with the n+1 passes
• Should we limit to two passes? n passes, but limited to low number?
• How do we pass back state data to the nth passes? Global map of state which all aggs can fetch from (based on known ordinal or something?) Pass down specific state to relevant sub-agg?
• Are multi-pass aggs a new type of agg that encompasses logic for each stage? Or is a multi-pass agg a regular agg for first pass, and a new type of agg for second/etc passes?
• Do multi-pass agg share the same syntax as existing aggs, in the same tree?

There are some implications to multi-pass and async search which need resolving. Perhaps multi-pass is implemented on a per-"feature" basis (e.g. a dedicated "cluster endpoint" that does kmeans clustering, instead of trying to modify the agg framework more generically)

Probably a lot more points to consider, just wanted to get an initial braindump down :)

[elasticmachine]

Pinging @elastic/es-analytics-geo (:Analytics/Aggregations)

[polyfractal]

Did some thinking on this last week. The technically easiest route is a dedicated endpoint-per-functionality (_analytics/cluster, etc), since we can build whatever needs to be built without regards to other aggs. This is also the least useful to users, since they'll have to construct "alternate universe" aggs with a different API.

The best user situation is a unified DSL where aggs "just work" regardless of how many passes are needed. This is difficult technically because all existing aggs would need to learn how to deal with intermediate state, interact with multiple passes, etc.

As a middle ground, I'd like to suggest the following:

• MP aggs share the same DSL, and are specified in the "aggs" portion of a request as usual
• MP aggs must be top-level
• MP aggs may not have non-MP sub-aggs. E.g. they do not "interact" with regular aggs at all, cannot be used with pipelines, etc
  • For a first iteration, we might decide that MP aggs can't even have MP sub-aggs, since the intermediate state could explode in size/complexity
• MP aggs derive from a new set of base classes that understand how to (de)-serialize intermediate state and perform a second pass over the data
• MP agg and regular agg Internal results are mergeable. Since they do not interact, and do not overlap, we should be able to cleanly merge the results of both agg types during a final reduction. This also means that async search can largely "just work" because we can just continue to wait on pending MP shard results and then merge them in

From a technical point of view, the first pass over the data could work as normal. When the coordinator receives intermediate results, it would need to identify all MP shard results, merge them together, and then execute the secondary phase returning back to data nodes. How this is handled on the data nodes is unclear, maybe a new phase? Or we treat it like a regular agg phase and with a different set of aggregators?