Release hash aggregation memory on output

Incrementally releases memory from FlatGroupByHash when
HashAggregationOperator starts producing output.

Author: pettyjamesm

core/trino-main/src/main/java/io/trino/operator/AppendOnlyVariableWidthData.java

@@ -138,6 +138,27 @@ public byte[] getChunk(byte[] pointer, int pointerOffset)
         return chunks.get(chunkIndex);
     }
 
+    public void freeChunksBefore(byte[] pointer, int pointerOffset)
+    {
+        int chunkIndex = getChunkIndex(pointer, pointerOffset);
+        if (chunks.isEmpty()) {
+            verify(chunkIndex == 0);
+            return;
+        }
+        checkIndex(chunkIndex, chunks.size());
+        // Release any previous chunks until a null chunk is encountered, which means it and any previous
+        // batches have already been released
+        int releaseIndex = chunkIndex - 1;
+        while (releaseIndex >= 0) {
+            byte[] releaseChunk = chunks.set(releaseIndex, null);
+            if (releaseChunk == null) {
+                break;
+            }
+            chunksRetainedSizeInBytes -= releaseChunk.length;
+            releaseIndex--;
+        }
+    }
+
     // growth factor for each chunk doubles up to 512KB, then increases by 1.5x for each chunk after that
     private static long nextChunkSize(long previousChunkSize)
     {

core/trino-main/src/main/java/io/trino/operator/BigintGroupByHash.java

@@ -126,6 +126,12 @@ public int getGroupCount()
         return nextGroupId;
     }
 
+    @Override
+    public void startReleasingOutput()
+    {
+        // NOOP
+    }
+
     @Override
     public void appendValuesTo(int groupId, PageBuilder pageBuilder)
     {

core/trino-main/src/main/java/io/trino/operator/FlatGroupByHash.java

@@ -129,6 +129,12 @@ public int getGroupCount()
         return flatHash.size();
     }
 
+    @Override
+    public void startReleasingOutput()
+    {
+        flatHash.startReleasingOutput();
+    }
+
     @Override
     public void appendValuesTo(int groupId, PageBuilder pageBuilder)
     {

core/trino-main/src/main/java/io/trino/operator/FlatHash.java

@@ -124,8 +124,8 @@ public FlatHash(FlatHash other)
         this.mask = other.mask;
         this.nextGroupId = other.nextGroupId;
         this.maxFill = other.maxFill;
-        this.control = Arrays.copyOf(other.control, other.control.length);
-        this.groupIdsByHash = Arrays.copyOf(other.groupIdsByHash, other.groupIdsByHash.length);
+        this.control = other.control == null ? null : Arrays.copyOf(other.control, other.control.length);
+        this.groupIdsByHash = other.groupIdsByHash == null ? null : Arrays.copyOf(other.groupIdsByHash, other.groupIdsByHash.length);
         this.fixedSizeRecords = Arrays.stream(other.fixedSizeRecords)
                 .map(fixedSizeRecords -> fixedSizeRecords == null ? null : Arrays.copyOf(fixedSizeRecords, fixedSizeRecords.length))
                 .toArray(byte[][]::new);
@@ -153,11 +153,28 @@ public int getCapacity()
         return capacity;
     }
 
+    public void startReleasingOutput()
+    {
+        if (isReleasingOutput()) {
+            throw new IllegalStateException("already releasing output");
+        }
+        control = null;
+        groupIdsByHash = null;
+    }
+
+    public boolean isReleasingOutput()
+    {
+        return control == null;
+    }
+
     public long hashPosition(int groupId)
     {
         if (groupId < 0) {
             throw new IllegalArgumentException("groupId is negative");
         }
+        if (isReleasingOutput()) {
+            throw new IllegalStateException("already releasing output");
+        }
         byte[] fixedSizeRecords = getFixedSizeRecords(groupId);
         int fixedRecordOffset = getFixedRecordOffset(groupId);
         if (cacheHashValue) {
@@ -182,7 +199,8 @@ public void appendTo(int groupId, BlockBuilder[] blockBuilders)
     {
         checkArgument(groupId < nextGroupId, "groupId out of range");
 
-        byte[] fixedSizeRecords = getFixedSizeRecords(groupId);
+        int recordGroupIndex = recordGroupIndexForGroupId(groupId);
+        byte[] fixedSizeRecords = this.fixedSizeRecords[recordGroupIndex];
         int recordOffset = getFixedRecordOffset(groupId);
 
         byte[] variableWidthChunk = null;
@@ -202,6 +220,18 @@ public void appendTo(int groupId, BlockBuilder[] blockBuilders)
         if (hasPrecomputedHash) {
             BIGINT.writeLong(blockBuilders[blockBuilders.length - 1], (long) LONG_HANDLE.get(fixedSizeRecords, recordOffset));
         }
+        // Release memory from the previous fixed size records batch
+        if (isReleasingOutput() && recordOffset == 0 && recordGroupIndex > 0) {
+            byte[] releasedRecords = this.fixedSizeRecords[recordGroupIndex - 1];
+            this.fixedSizeRecords[recordGroupIndex - 1] = null;
+            if (releasedRecords == null) {
+                throw new IllegalStateException("already released previous record batch");
+            }
+            fixedRecordGroupsRetainedSize -= sizeOf(releasedRecords);
+            if (variableWidthData != null) {
+                variableWidthData.freeChunksBefore(fixedSizeRecords, recordOffset + variableWidthOffset);
+            }
+        }
     }
 
     public void computeHashes(Block[] blocks, long[] hashes, int offset, int length)
@@ -251,6 +281,9 @@ public int putIfAbsent(Block[] blocks, int position, long hash)
 
     private int getIndex(Block[] blocks, int position, long hash)
     {
+        if (isReleasingOutput()) {
+            throw new IllegalStateException("already releasing output");
+        }
         byte hashPrefix = (byte) (hash & 0x7F | 0x80);
         int bucket = bucket((int) (hash >> 7));
 
@@ -351,6 +384,9 @@ private void setControl(int index, byte hashPrefix)
 
     public boolean ensureAvailableCapacity(int batchSize)
     {
+        if (isReleasingOutput()) {
+            throw new IllegalStateException("already releasing output");
+        }
         long requiredMaxFill = nextGroupId + batchSize;
         if (requiredMaxFill >= maxFill) {
             long minimumRequiredCapacity = (requiredMaxFill + 1) * 16 / 15;

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

@@ -109,6 +109,8 @@ static GroupByHash createGroupByHash(
 
     void appendValuesTo(int groupId, PageBuilder pageBuilder);
 
+    void startReleasingOutput();
+
     Work<?> addPage(Page page);
 
     /**

core/trino-main/src/main/java/io/trino/operator/NoChannelGroupByHash.java

@@ -50,6 +50,12 @@ public void appendValuesTo(int groupId, PageBuilder pageBuilder)
         throw new UnsupportedOperationException("NoChannelGroupByHash does not support appendValuesTo");
     }
 
+    @Override
+    public void startReleasingOutput()
+    {
+        throw new UnsupportedOperationException("NoChannelGroupByHash does not support startReleasingOutput");
+    }
+
     @Override
     public Work<?> addPage(Page page)
     {

core/trino-main/src/main/java/io/trino/operator/aggregation/builder/InMemoryHashAggregationBuilder.java

@@ -248,12 +248,18 @@ public WorkProcessor<Page> buildResult()
         for (GroupedAggregator groupedAggregator : groupedAggregators) {
             groupedAggregator.prepareFinal();
         }
-        return buildResult(consecutiveGroupIds());
+        // Only incrementally release memory for final aggregations, since partial aggregations have a fixed
+        // memory limit and can be expected to fully flush and release their output quickly
+        boolean releaseOutputMemory = !partial;
+        if (releaseOutputMemory) {
+            groupByHash.startReleasingOutput();
+        }
+        return buildResult(consecutiveGroupIds(), releaseOutputMemory);
     }
 
     public WorkProcessor<Page> buildHashSortedResult()
     {
-        return buildResult(hashSortedGroupIds());
+        return buildResult(hashSortedGroupIds(), false);
     }
 
     public List<Type> buildSpillTypes()
@@ -271,7 +277,7 @@ public int getCapacity()
         return groupByHash.getCapacity();
     }
 
-    private WorkProcessor<Page> buildResult(IntIterator groupIds)
+    private WorkProcessor<Page> buildResult(IntIterator groupIds, boolean releaseMemoryOnOutput)
     {
         PageBuilder pageBuilder = new PageBuilder(buildTypes());
         return WorkProcessor.create(() -> {
@@ -294,6 +300,11 @@ private WorkProcessor<Page> buildResult(IntIterator groupIds)
                 }
             }
 
+            // Update memory usage after producing each page of output
+            if (releaseMemoryOnOutput) {
+                updateMemory();
+            }
+
             return ProcessState.ofResult(pageBuilder.build());
         });
     }

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

@@ -54,6 +54,12 @@ public int getGroupCount()
         return maxGroupId + 1;
     }
 
+    @Override
+    public void startReleasingOutput()
+    {
+        throw new UnsupportedOperationException("Not yet supported");
+    }
+
     @Override
     public void appendValuesTo(int groupId, PageBuilder pageBuilder)
     {

core/trino-main/src/test/java/io/trino/operator/TestGroupByHash.java

@@ -46,6 +46,7 @@
 import static io.trino.spi.type.VarcharType.VARCHAR;
 import static io.trino.type.TypeTestUtils.getHashBlock;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 public class TestGroupByHash
 {
@@ -333,6 +334,54 @@ public void testUpdateMemoryBigint()
         assertThat(rehashCount.get()).isEqualTo(2 * BIGINT_EXPECTED_REHASH);
     }
 
+    @Test
+    public void testReleaseMemoryOnOutput()
+    {
+        Type type = VARCHAR;
+        // values expands into multiple FlatGroupByHash fixed record groups
+        Block valuesBlock = createStringSequenceBlock(0, 1_000_000);
+
+        GroupByHash groupByHash = createGroupByHash(ImmutableList.of(type), selectGroupByHashMode(false, false, ImmutableList.of(type)), 10_000, false, new FlatHashStrategyCompiler(new TypeOperators()), () -> true);
+        assertThat(groupByHash.addPage(new Page(valuesBlock)).process()).isTrue();
+        assertThat(groupByHash.getGroupCount()).isEqualTo(valuesBlock.getPositionCount());
+
+        long memoryUsageAfterInput = groupByHash.getEstimatedSize();
+        groupByHash.startReleasingOutput();
+        // memory usage should have decreased from dropping the hash table
+        long memoryUsageAfterReleasingOutput = groupByHash.getEstimatedSize();
+        // single immediate release of memory for the control and groupId by hash values
+        assertThat(memoryUsageAfterReleasingOutput).isLessThan(memoryUsageAfterInput);
+
+        // no more inputs accepted after switching to releasing output
+        assertThatThrownBy(() -> groupByHash.addPage(new Page(valuesBlock)).process())
+                .isInstanceOf(IllegalStateException.class)
+                .hasMessage("already releasing output");
+
+        PageBuilder pageBuilder = new PageBuilder(ImmutableList.of(type));
+        int groupId = 0;
+        // FlatGroupByHash first 1024 records are within the first record group
+        for (; groupId < 1024; groupId++) {
+            groupByHash.appendValuesTo(groupId, pageBuilder);
+            pageBuilder.declarePosition();
+        }
+        pageBuilder.build();
+        // No memory released yet after completing the first group
+        assertThat(groupByHash.getEstimatedSize()).isEqualTo(memoryUsageAfterReleasingOutput);
+
+        groupByHash.appendValuesTo(groupId++, pageBuilder);
+        pageBuilder.declarePosition();
+        // Memory released
+        long memoryUsageAfterFirstRelease = groupByHash.getEstimatedSize();
+        assertThat(memoryUsageAfterFirstRelease).isLessThan(memoryUsageAfterReleasingOutput);
+
+        for (; groupId < valuesBlock.getPositionCount(); groupId++) {
+            groupByHash.appendValuesTo(groupId, pageBuilder);
+            pageBuilder.declarePosition();
+        }
+        // More memory released
+        assertThat(groupByHash.getEstimatedSize()).isLessThan(memoryUsageAfterFirstRelease);
+    }
+
     @Test
     public void testMemoryReservationYield()
     {