### Question:
HashAggregationOperator implements Operator { @Override public Page getOutput() { if (finished) { return null; } if (unfinishedWork != null) { boolean workDone = unfinishedWork.process(); aggregationBuilder.updateMemory(); if (!workDone) { return null; } unfinishedWork = null; } if (outputPages == null) { if (finishing) { if (!inputProcessed && produceDefaultOutput) { finished = true; return getGlobalAggregationOutput(); } if (aggregationBuilder == null) { finished = true; return null; } } if (!finishing && (aggregationBuilder == null || !aggregationBuilder.isFull())) { return null; } outputPages = aggregationBuilder.buildResult(); } if (!outputPages.process()) { return null; } if (outputPages.isFinished()) { closeAggregationBuilder(); return null; } return outputPages.getResult(); } HashAggregationOperator(
            OperatorContext operatorContext,
            List<Type> groupByTypes,
            List<Integer> groupByChannels,
            List<Integer> globalAggregationGroupIds,
            Step step,
            boolean produceDefaultOutput,
            List<AccumulatorFactory> accumulatorFactories,
            Optional<Integer> hashChannel,
            Optional<Integer> groupIdChannel,
            int expectedGroups,
            Optional<DataSize> maxPartialMemory,
            boolean spillEnabled,
            DataSize memoryLimitForMerge,
            DataSize memoryLimitForMergeWithMemory,
            SpillerFactory spillerFactory,
            JoinCompiler joinCompiler,
            BlockTypeOperators blockTypeOperators,
            boolean useSystemMemory); @Override OperatorContext getOperatorContext(); @Override void finish(); @Override boolean isFinished(); @Override boolean needsInput(); @Override void addInput(Page page); @Override ListenableFuture<?> startMemoryRevoke(); @Override void finishMemoryRevoke(); @Override Page getOutput(); @Override void close(); @VisibleForTesting HashAggregationBuilder getAggregationBuilder();  }### Answer:
@Test(dataProvider = "dataType") public void testMemoryReservationYield(Type type) { List<Page> input = createPagesWithDistinctHashKeys(type, 6_000, 600); OperatorFactory operatorFactory = new HashAggregationOperatorFactory( 0, new PlanNodeId("test"), ImmutableList.of(type), ImmutableList.of(0), ImmutableList.of(), Step.SINGLE, ImmutableList.of(COUNT.bind(ImmutableList.of(0), Optional.empty())), Optional.of(1), Optional.empty(), 1, Optional.of(DataSize.of(16, MEGABYTE)), joinCompiler, blockTypeOperators, false); GroupByHashYieldResult result; result = finishOperatorWithYieldingGroupByHash(input, type, operatorFactory, this::getHashCapacity, 1_400_000); assertGreaterThan(result.getYieldCount(), 5); assertGreaterThan(result.getMaxReservedBytes(), 20L << 20); int count = 0; for (Page page : result.getOutput()) { assertEquals(page.getChannelCount(), 3); for (int i = 0; i < page.getPositionCount(); i++) { assertEquals(page.getBlock(2).getLong(i, 0), 1); count++; } } assertEquals(count, 6_000 * 600); }