Aggregation ORDER BY & DISTINCT spilling

[sachdevs]

This PR implements ORDER BY and DISTINCT spilling for use in aggregation functions. Will be publishing docs on implementation details and updating this PR in the future.

== RELEASE NOTES ==

General Changes
* Add local disk spilling support for aggregation functions with `ORDER BY` or `DISTINCT` syntax.

[highker]

As discussed offline, let's spill RowBlocks for intermediate states.

[highker]

The overall idea is on the right direction. Here is the high-level design suggestion to unify the solution for both ordering and distincting cases:

1. Both Ordering and Distincting (Grouped)Accumulators do not have xxxIntermediateXxx() interfaces implemented. Because they do not support partial aggregation. So let's create a new class called FinalOnly(Grouped)Accumulator. This class should implement all xxxIntermediateXxx() interfaces by throwing UnsupportedOperationException. Have both Ordering and Distincting (Grouped)Accumulators inheriting FinalOnly(Grouped)Accumulator.
2. Create new classes inheriting (Grouped)Accumulators called SpillableFinalOnly(Grouped)Accumulator. The classes take FinalOnly(Grouped)Accumulators as a delegate. SpillableFinalOnly(Grouped)Accumulator is responsible for maintaining the hashtable state. Use ObjectBigArray for the hashtable. (Check my inline comment).
3. SpillableFinalOnly(Grouped)Accumulator should implement xxxIntermediateXxx() interfaces. Together with the addInput interface, they should build the hashtable for RowBlock <-> Page mapping. As we have discussed offline. The hashtable is only to accumulate the original data by grouping data into different group Ids so that we can spill by groups. (We do not lose information at this step though it may use quite a lot of memory).
4. SpillableFinalOnly(Grouped)Accumulator prepare/evalFinal interfaces should call addInput/prepareFinal/evalFinal interfaces for its FinalOnly(Grouped)Accumulator delegate.
5. Directly create corresponding FinalOnly(Grouped)Accumulator if spilling is not enabled. Otherwise, create SpillableFinalOnly(Grouped)Accumulator wrapping FinalOnly(Grouped)Accumulator.

cc @arhimondr in case there is better design.

[highker]

Let's clean up the PR with the design proposed. That will make the review easier.

[sachdevs]

@highker planning on doing a full refactor with proposed design once distinct is working. I pushed my work so far, having some trouble with array_agg, will update when it works.

[sachdevs]

Distinct and order by spilling work now. There are one main glaring issue to iron out before we continue, we need to figure out how to reliably detect if unspill is happening. Due to the current logic, things like array_agg(DISTINCT x ORDER BY y) doesn't work as well as simple things like, "SELECT count(distinct x), y FROM t GROUP BY y" since we look at the last block of the unspilt page to see if it is an array block. See the attached comment for details.

[highker]

Did a fast skim through DistinctingGroupedAccumulator. The logic looks legit. (Didn't get into details line by line).

In terms of if (page.getBlock(page.getChannelCount() - 1) instanceof ArrayBlock) {, We might be able to add a new interface for GroupedAccumulator to hint the input is a new input or an unspilled input. But I would hold this new interface until we have the abstraction ready. Then we can better evaluate how to have the interface change.

[sachdevs]

Summarized the main issues with the design so far.

[sachdevs]

Will address these comments + look into page compaction for next iteration.

EDIT as discussed offline, page compaction/ memory reduction for distinct has been made into its own task.

[highker]

minor comments only

[highker]

Is this part necessary?

[sachdevs]

Yeah since memory usage is updated in updateMemory() to be set to the size of the empty hash agg builder. This is not correct since we never spilt in the first place in startMemoryRevoke.

[sachdevs]

Adding comments.

[highker]

If I understand correctly, the original logic can only happen if an operator that has never spilled and just started to build final results; however, a revoke request comes in. Am I right? If that is the case, shall we make a comment here?

[sachdevs]

Yeah so essentially we are declining memory revoking in the case that the hashaggbuilder has already completed. This is only set to true when InMemoryHashAggregationBuilder.buildResult is called - NOT InMemoryHashAggregationBuilder.buildHashSortedResult. This is because after buildResult, spilling should be impossible, because it can no longer process any more input anyway.

[highker]

I would be surprised if spillEnabled cannot be tunneled from LocalExecutionPlanner... Can we give it a try and see what will happen?

[sachdevs]

Last update should fix any failing checks.

[highker]

Is this right? the function is to create IntermediateAccumulator but here it delegates to GroupedAccumulator and then use createDefaultGroupedAccumulator instead of createDefaultGroupedIntermediateAccumulator?

[sachdevs]

Yes since we do the same logic in either case since we do not care about the underlying accumulator working on intermediate values in our FinalOnlyGroupedAccumulator/SpillableFinalOnlyGroupedAccumulator delegate. If I was to separate these functions the resulting logic should be the same. It originally was separate but I noticed it could be simplified.

[highker]

minor comments

[highker]

Actually, is it even possible we need createSpillableGroupedIntermediateAccumulator()? Can we check if the following is good enough?

    @Override
    public GroupedAccumulator createGroupedIntermediateAccumulator()
    {
        checkState(!hasDistinct() || !hasOrderBy(), "distinct or order by cannot have partial aggregation");
        
        try {
            return groupedAccumulatorConstructor.newInstance(stateDescriptors, ImmutableList.of(), Optional.empty(), lambdaProviders);
        }
        catch (InstantiationException | IllegalAccessException | InvocationTargetException e) {
            throw new RuntimeException(e);
        }
    }

[sachdevs]

We do need this since before:

1. createGroupedIntermediateAccumulator used to be never called (in case of ORDER BY or DISTINCT) since order by and distinct did not have support for intermediate results. Hence, when spill is enabled, we use createGroupedIntermediateAccumulator to recreate the intermediate version of the accumulator with the spillable wrapper when unspilling and creating intermediate accumulators.

2. checkState(!hasDistinct() || !hasOrderBy()) not sure what this does exactly in this context since having distinct AND orderby shouldnt be a state failure. createGroupedIntermediateAccumulator can be called with hasDistinct/hasOrderBy set to true.

3. return groupedAccumulatorConstructor.newInstance would be a bug since this references the underlying accumulator (NOT order by or distinct, but the accumulator inside of order by or distinct). This means that during intermediate accumulation we would get non-distinct non-ordered values.

I actually tried writing this code without this section in a previous iteration but I realized we need the spillable wrapper in the intermediate case.

[highker]

lgtm