class HashAggregate extends ForwardingLogicalPlan {

    private static final String MERGE_PHASE_NAME = "mergeOnHandler";
    final List<Function> aggregates;

    HashAggregate(LogicalPlan source, List<Function> aggregates) {
        super(source);
        this.aggregates = aggregates;
    }

    @Override
    public ExecutionPlan build(DependencyCarrier executor,
                               PlannerContext plannerContext,
                               Set<PlanHint> planHints,
                               ProjectionBuilder projectionBuilder,
                               int limit,
                               int offset,
                               @Nullable OrderBy order,
                               @Nullable Integer pageSizeHint,
                               Row params,
                               SubQueryResults subQueryResults) {
        // Avoid source look-ups for performance reasons. Global aggregations are a pipeline breaker using all data.
        // So use column store instead, because it is likely more efficient.
        if (planHints.contains(PlanHint.PREFER_SOURCE_LOOKUP)) {
            planHints = new HashSet<>(planHints);
            planHints.remove(PlanHint.PREFER_SOURCE_LOOKUP);
        }
        ExecutionPlan executionPlan = source.build(
            executor, plannerContext, planHints, projectionBuilder, NO_LIMIT, 0, null, null, params, subQueryResults);

        AggregationOutputValidator.validateOutputs(aggregates);
        var paramBinder = new SubQueryAndParamBinder(params, subQueryResults);

        var sourceOutputs = source.outputs();
        if (executionPlan.resultDescription().hasRemainingLimitOrOffset()) {
            executionPlan = Merge.ensureOnHandler(executionPlan, plannerContext);
        }
        if (ExecutionPhases.executesOnHandler(plannerContext.handlerNode(), executionPlan.resultDescription().nodeIds())) {
            if (source.preferShardProjections()) {
                executionPlan.addProjection(
                    projectionBuilder.aggregationProjection(
                        sourceOutputs,
                        aggregates,
                        paramBinder,
                        AggregateMode.ITER_PARTIAL,
                        RowGranularity.SHARD,
                        plannerContext.transactionContext().sessionSettings().searchPath()
                    )
                );
                executionPlan.addProjection(
                    projectionBuilder.aggregationProjection(
                        aggregates,
                        aggregates,
                        paramBinder,
                        AggregateMode.PARTIAL_FINAL,
                        RowGranularity.CLUSTER,
                        plannerContext.transactionContext().sessionSettings().searchPath()
                    )
                );
                return executionPlan;
            }
            AggregationProjection fullAggregation = projectionBuilder.aggregationProjection(
                sourceOutputs,
                aggregates,
                paramBinder,
                AggregateMode.ITER_FINAL,
                RowGranularity.CLUSTER,
                plannerContext.transactionContext().sessionSettings().searchPath()
            );
            executionPlan.addProjection(fullAggregation);
            return executionPlan;
        }
        AggregationProjection toPartial = projectionBuilder.aggregationProjection(
            sourceOutputs,
            aggregates,
            paramBinder,
            AggregateMode.ITER_PARTIAL,
            source.preferShardProjections() ? RowGranularity.SHARD : RowGranularity.NODE,
            plannerContext.transactionContext().sessionSettings().searchPath()
        );
        executionPlan.addProjection(toPartial);

        AggregationProjection toFinal = projectionBuilder.aggregationProjection(
            aggregates,
            aggregates,
            paramBinder,
            AggregateMode.PARTIAL_FINAL,
            RowGranularity.CLUSTER,
            plannerContext.transactionContext().sessionSettings().searchPath()
        );
        return new Merge(
            executionPlan,
            new MergePhase(
                plannerContext.jobId(),
                plannerContext.nextExecutionPhaseId(),
                MERGE_PHASE_NAME,
                executionPlan.resultDescription().nodeIds().size(),
                1,
                Collections.singletonList(plannerContext.handlerNode()),
                executionPlan.resultDescription().streamOutputs(),
                Collections.singletonList(toFinal),
                DistributionInfo.DEFAULT_BROADCAST,
                null
            ),
            NO_LIMIT,
            0,
            aggregates.size(),
            1,
            null
        );
    }

    public List<Function> aggregates() {
        return aggregates;
    }

    @Override
    public List<Symbol> outputs() {
        return new ArrayList<>(aggregates);
    }

    @Override
    public LogicalPlan replaceSources(List<LogicalPlan> sources) {
        return new HashAggregate(Lists.getOnlyElement(sources), aggregates);
    }

    @Override
    public LogicalPlan pruneOutputsExcept(SequencedCollection<Symbol> outputsToKeep) {<FILL_FUNCTION_BODY>}

    @Override
    public <C, R> R accept(LogicalPlanVisitor<C, R> visitor, C context) {
        return visitor.visitHashAggregate(this, context);
    }

    private static class OutputValidatorContext {
        private boolean insideAggregation = false;
    }

    public static class AggregationOutputValidator extends SymbolVisitor<OutputValidatorContext, Void> {

        private static final AggregationOutputValidator INSTANCE = new AggregationOutputValidator();

        public static void validateOutputs(Collection<? extends Symbol> outputs) {
            OutputValidatorContext ctx = new OutputValidatorContext();
            for (Symbol output : outputs) {
                ctx.insideAggregation = false;
                output.accept(INSTANCE, ctx);
            }
        }

        @Override
        public Void visitFunction(Function symbol, OutputValidatorContext context) {
            context.insideAggregation =
                context.insideAggregation || symbol.signature().getKind().equals(FunctionType.AGGREGATE);
            for (Symbol argument : symbol.arguments()) {
                argument.accept(this, context);
            }
            context.insideAggregation = false;
            return null;
        }

        @Override
        public Void visitReference(Reference symbol, OutputValidatorContext context) {
            if (context.insideAggregation) {
                IndexType indexType = symbol.indexType();
                if (indexType == IndexType.FULLTEXT) {
                    throw new IllegalArgumentException(Symbols.format(
                        "Cannot select analyzed column '%s' within grouping or aggregations", symbol));
                }
            }
            return null;
        }

        @Override
        protected Void visitSymbol(Symbol symbol, OutputValidatorContext context) {
            return null;
        }
    }
}

        ArrayList<Function> newAggregates = new ArrayList<>();
        for (Symbol outputToKeep : outputsToKeep) {
            SymbolVisitors.intersection(outputToKeep, aggregates, newAggregates::add);
        }
        LinkedHashSet<Symbol> toKeep = new LinkedHashSet<>();
        for (Function newAggregate : newAggregates) {
            SymbolVisitors.intersection(newAggregate, source.outputs(), toKeep::add);
        }
        LogicalPlan newSource = source.pruneOutputsExcept(toKeep);
        if (source == newSource && newAggregates == aggregates) {
            return this;
        }
        return new HashAggregate(newSource, newAggregates);
    
<methods>public void <init>(io.crate.planner.operators.LogicalPlan) ,public Map<io.crate.planner.operators.LogicalPlan,io.crate.expression.symbol.SelectSymbol> dependencies() ,public List<io.crate.expression.symbol.Symbol> outputs() ,public io.crate.planner.operators.LogicalPlan pruneOutputsExcept(SequencedCollection<io.crate.expression.symbol.Symbol>) ,public List<io.crate.metadata.RelationName> relationNames() ,public io.crate.planner.operators.LogicalPlan source() ,public List<io.crate.planner.operators.LogicalPlan> sources() ,public boolean supportsDistributedReads() <variables>final non-sealed io.crate.planner.operators.LogicalPlan source,private final non-sealed List<io.crate.planner.operators.LogicalPlan> sources