@Override
        public AggCall visit(CallExpression unresolvedCall) {
            if (unresolvedCall.getFunctionDefinition() == AS) {
                String aggregateName = extractValue(unresolvedCall.getChildren().get(1), String.class)
                        .orElseThrow(() -> new TableException("Unexpected name."));

                Expression aggregate = unresolvedCall.getChildren().get(0);
                if (isFunctionOfKind(aggregate, AGGREGATE)) {
                    return aggregate.accept(callResolver).accept(
                            new AggCallVisitor(relBuilder, expressionConverter, aggregateName, false));
                }
            }
            throw new TableException("Expected named aggregate. Got: " + unresolvedCall);
        }
Get the {@link AggCall} correspond to the aggregate or table aggregate expression.
core/src/main/java/org/apache/flink/table/planner/plan/QueryOperationConverter.java