@Override
            public RelBuilder.AggCall visit(CallExpression call) {
                FunctionDefinition def = call.getFunctionDefinition();
                if (BuiltInFunctionDefinitions.DISTINCT == def) {
                    Expression innerAgg = call.getChildren().get(0);
                    return innerAgg.accept(new AggCallVisitor(relBuilder, expressionConverter, name, true));
                } else {
                    SqlAggFunction sqlAggFunction = call.accept(sqlAggFunctionVisitor);
                    return relBuilder.aggregateCall(
                            sqlAggFunction,
                            isDistinct,
                            false,
                            null,
                            name,
                            call.getChildren().stream().map(expr -> expr.accept(expressionConverter))
                                    .collect(Collectors.toList()));
                }
            }
Get the {@link AggCall} correspond to the aggregate or table aggregate expression.
core/src/main/java/org/apache/flink/table/planner/plan/QueryOperationConverter.java