Spill large blocks to separate spill files in distinct aggregates

[pgupta2]

Distinct Aggregates spilling logic compacts all values corresponding
to the same groupId into a single array block and put it in the output
block of the aggregator. This is done for each aggregate function
in the HashAggregationBuilder. If multiple distinct aggregate functions
are present, then we perform same kind of compaction into
corresponding blocks in the same position within a page.

Due to this, a single row within the page can become really huge and
when this page is spilled, it can fail with 'integer overflow' error during
serialization. Instead of compressing all values within a single array
block, we can spill them into a separate spill file, if the block size goes
beyond a certain threshold.

When large block spill is enabled, we use a hybrid approach for prepping
page for spill. If an array block for a groupId goes beyond a threshold, we
spill it into a separate spill file and store the serialized file handle in the
output block. If the size is within the threshold, we will directly store the
values in the output block. This approach enables us to perform a spill
only iff a block is huge. Since, this will happen rarely, most of the time,
we will be storing the data directly in the output block of the aggregate.

Test plan -

• Unit Tests
• Tested ~350 production pipelines that have distinct aggregates for correctness and performance.

== RELEASE NOTES ==

General Changes
* Add a new configuration property ``experimental.distinct-aggregation-large-block-spill-enabled`` to enable spilling of blocks that are larger than ``experimental.distinct-aggregation-large-block-size-threshold`` bytes into a separate spill file.  This can be overridden by ``distinct_aggregation_large_block_spill_enabled`` session property.
* Add a new configuration property ``experimental.distinct-aggregation-large-block-size-threshold`` to define the threshold size beyond which the block will be spilled into a separate spill file.  This can be overridden by ``distinct_aggregation_large_block_size_threshold`` session property.

[highker]

The PR is on my radar. I will review it next week.

[highker]

• High-level comments + nits only. Didn't dig deep into logic details.
• Do we have unit tests coverage?

[highker]

Generic comment: this class has become way too big. It would be good to split it into multiple files in a followup PR

[highker]

remove else {

[highker]

s/page -> page.getPositionCount()/Page::getPositionCount

[highker]

Would be good to add a javadoc to explain the difference between this and Spiller

[highker]

Would recommend to use a class to represent file handle like SpillerHandle or SpilledFileHandle or whatever the name making sense. byte[] could be confusing. My first glance on it made me think it's the returning stream.

[highker]

nit: TempStorageStandaloneSpiller

[highker]

ioException == null is always true

[highker]

Can we reuse or create spilling errors?

[highker]

directly return

[highker]

Do we need this line?

[pgupta2]

Before exiting this method, finally will be called and we will try to close tempDataSink. If we get an exception during this operation, then we catch the exception and throw it again. We do need this line to throw the exception else it will be swallowed. Right?

[pgupta2]

@highker : Sorry for getting late on this PR. I was busy with other stuff. I have addressed the comments so far.

Do we have unit tests coverage?

TestSpilledAggregationWithLargeBlockSpillingEnabled.java enables large block spilling and some of the queries trigger the large block spill path as well. So, we do have unit test coverage for this.

[highker]

Isn't this interface going to take SerializedTempStorageHandle?

[highker]

Call this SerializedStorageHandle?