Remove unnecessary aggregator creations

Avoids eager, unnecessary creation of aggregator instances just to
determine their output type inside of the HashAggregationOperator
constructor.

Author: pettyjamesm

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

@@ -256,8 +256,6 @@ public OperatorFactory duplicate()
     private final FlatHashStrategyCompiler flatHashStrategyCompiler;
     private final AggregationMetrics aggregationMetrics = new AggregationMetrics();
 
-    private final List<Type> types;
-
     private HashAggregationBuilder aggregationBuilder;
     private final LocalMemoryContext memoryContext;
     private WorkProcessor<Page> outputPages;
@@ -305,7 +303,6 @@ private HashAggregationOperator(
         this.produceDefaultOutput = produceDefaultOutput;
         this.expectedGroups = expectedGroups;
         this.maxPartialMemory = requireNonNull(maxPartialMemory, "maxPartialMemory is null");
-        this.types = toTypes(groupByTypes, aggregatorFactories);
         this.spillEnabled = spillEnabled;
         this.memoryLimitForMerge = requireNonNull(memoryLimitForMerge, "memoryLimitForMerge is null");
         this.memoryLimitForMergeWithMemory = requireNonNull(memoryLimitForMergeWithMemory, "memoryLimitForMergeWithMemory is null");
@@ -540,7 +537,7 @@ private Page getGlobalAggregationOutput()
     {
         // global aggregation output page will only be constructed once,
         // so a new PageBuilder is constructed (instead of using PageBuilder.reset)
-        PageBuilder output = new PageBuilder(globalAggregationGroupIds.size(), types);
+        PageBuilder output = new PageBuilder(globalAggregationGroupIds.size(), toTypes(groupByTypes, aggregatorFactories));
 
         for (int groupId : globalAggregationGroupIds) {
             output.declarePosition();

core/trino-main/src/main/java/io/trino/operator/aggregation/AggregatorFactory.java

@@ -58,6 +58,12 @@ public AggregatorFactory(
         checkArgument(step.isInputRaw() || inputChannels.size() == 1, "expected 1 input channel for intermediate aggregation");
     }
 
+    public Type getOutputType()
+    {
+        // Note: this must match Aggregator#getType() and GroupedAggregator#getType()
+        return step.isOutputPartial() ? intermediateType : finalType;
+    }
+
     public Aggregator createAggregator(AggregationMetrics metrics)
     {
         Accumulator accumulator;

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

@@ -349,7 +349,7 @@ public static List<Type> toTypes(List<? extends Type> groupByType, List<Aggregat
         ImmutableList.Builder<Type> types = ImmutableList.builderWithExpectedSize(groupByType.size() + factories.size());
         types.addAll(groupByType);
         for (AggregatorFactory factory : factories) {
-            types.add(factory.createAggregator(new AggregationMetrics()).getType());
+            types.add(factory.getOutputType());
         }
         return types.build();
     }