Add support for exact `median` aggregate function

[andygrove]

Is your feature request related to a problem or challenge? Please describe what you are trying to do.
Needed for h20 benchmark group by query 6.

Describe the solution you'd like
Add support for median aggregate function

Describe alternatives you've considered
We have approx_median but that would be cheating.

Additional context
None

[andygrove]

So this is a bit more involved than I had hoped.

Per the discussion in #1486 (comment)

TLDR is I think median is quite complicated to get both correct and high performance.

The key problem is that you basically need to have buffered the entire input stream before you know what the output is

Starting with an approximation of median would be fine and I suspect other users of DataFusion would find it valuable.

To implement an exact median, you could implement an Aggregator like we have for other aggregates that buffers up all the input, sorts it, and produces the median value. It would have terrible memory consumption but is probably reasonably straightforward to implement using existing compute kernels as @matthewmturner suggests

If you wanted to try and reuse existing operators (e.g. LogicalPlan and ExecutionPlan), I think that would be tough.

The mapping of SQL --> LogicalPlan is formulaic but non trivial. All aggregate functions are treated the same (create the same LogicalPlan nodes, with different expr lists, so if we special case median with a special operator we'll have to sort out how to handle queries like SELECT sum(foo), median(foo) from bar GROUP BY baz which I think will be complicated

Another challenge with median is that there is no way to partially aggregate intermediate results. DataFusion currently makes this pattern (which is useful when calculating sums, for example, because you can calculate partial sums and then sum them together)

                        ┌──────────┐                     
                        │          │                     
                        │Aggregator│                     
                        │ (Final)  │                     
                        │          │                     
                        │          │                     
                        └──────────┘                     
                              ▲                          
                              │                          
      ┌────────────────┬──────┴────────────────────┐     
      │                │                           │     
      │                │                           │     
      │                │                           │     
┌──────────┐     ┌──────────┐                ┌──────────┐
│          │     │          │                │          │
│Aggregator│     │Aggregator│                │Aggregator│
│(Partial) │     │(Partial) │         ...    │(Partial) │
│          │     │          │                │          │
│          │     │          │                │          │
└──────────┘     └──────────┘                └──────────┘
      ▲                ▲                           ▲     
      │                │                           │     
      └────────────────┴──────┬────────────────────┘     
                              │                          
                              │                          
                        ┌───────────┐                    
                        │           │                    
                        │Repartition│                    
                        │           │                    
                        │           │                    
                        └───────────┘                    
                              ▲                          
                              │                          
                              │                          
                        ┌───────────┐                    
                        │Input      │                    
                        │           │                    
                        └───────────┘                    

[andygrove]

I think that the first step here would be to change the Accumulator::state function to be able to return either ArrayRef or ScalarValue.

[andygrove]

I am making good progress on this and will have a PR up in the next week or two.

[jonmmease]

I'm interested in using this to add median aggregation support to VegaFusion, so I'd be happy to help test/review when you're ready. Thanks for taking this on!

[andygrove]

Thanks @jonmmease. I just submitted a PR with the functionality. I need to clean this up, but I think it is ready for testing.