Release hash aggregation memory on output

[pettyjamesm]

Description

Incrementally releases memory from FlatGroupByHash when HashAggregationOperator starts producing output. Previously, the entire hash table contents were kept in memory until output data was produced from hash aggregations.

Release notes

( ) This is not user-visible or is docs only, and no release notes are required.
( ) Release notes are required. Please propose a release note for me.
(x) Release notes are required, with the following suggested text:

## General
* Reduce memory usage of aggregations by incrementally releasing memory as output rows are are produced ({issue}`25879`)

[Copilot]

Pull Request Overview

This PR introduces incremental memory release for hash aggregations by freeing portions of the hash table as output is produced. Key changes include:

• New test (testReleaseMemoryOnOutput) verifying memory reduction and restricted operations after output release.
• Implementation of startReleasingOutput in FlatHash and FlatGroupByHash, which updates internal state to release memory.
• Adjustments in InMemoryHashAggregationBuilder and other GroupByHash implementations to support and integrate the new memory release behavior.

Reviewed Changes

Copilot reviewed 9 out of 9 changed files in this pull request and generated 1 comment.

Show a summary per file

File	Description
core/trino-main/src/test/java/io/trino/operator/TestGroupByHash.java	Added tests for incremental memory release during output production.
core/trino-main/src/test/java/io/trino/operator/CyclingGroupByHash.java	Introduced a stub for startReleasingOutput throwing an UnsupportedOperationException.
core/trino-main/src/main/java/io/trino/operator/aggregation/builder/InMemoryHashAggregationBuilder.java	Updated aggregation builder to utilize incremental memory release.
core/trino-main/src/main/java/io/trino/operator/FlatHash.java	Modified memory handling by nullifying arrays to free memory and added extra checks for the releasing state.
core/trino-main/src/main/java/io/trino/operator/FlatGroupByHash.java	Updated to delegate startReleasingOutput to FlatHash.
core/trino-main/src/main/java/io/trino/operator/BigintGroupByHash.java	Implemented a no-operation for startReleasingOutput.
core/trino-main/src/main/java/io/trino/operator/AppendOnlyVariableWidthData.java	Enhanced chunk-freeing logic with safe null checks.
core/trino-main/src/main/java/io/trino/operator/GroupByHash.java	Updated interface to include startReleasingOutput.

Comments suppressed due to low confidence (2)

core/trino-main/src/test/java/io/trino/operator/CyclingGroupByHash.java:58

• Consider adding a comment explaining that startReleasingOutput is not supported in CyclingGroupByHash and will be implemented later, so that readers understand the rationale behind the UnsupportedOperationException.

public void startReleasingOutput()

core/trino-main/src/main/java/io/trino/operator/GroupByHash.java:112

• Consider adding Javadoc for the startReleasingOutput method to clarify the expected behavior and usage for implementations, helping future maintainers understand the contract behind memory release.

void startReleasingOutput();

[sopel39]

Does it matter in practice? Without this condition the code could be simplified and startReleasingOutput could be removed

[pettyjamesm]

1. I think you need startReleasingOutput() no matter what since there are structures in the GroupByHash that can be released immediately before any groupId is emitted.
2. Avoiding releasing memory in partial aggregations can help avoid lock contention in the memory tracking system for high parallelism operations (partial aggregations are likely to appear on top of table scans). It may not matter much in practice but it seemed like incrementally releasing there was unlikely to be helpful since we're flushing anyway and the memory released is expected to be negligible

[sopel39]

lgtm. Would be good if @dain review too