@Override
        public <U> RelNode visit(CalculatedQueryOperation<U> calculatedTable) {
            DataType resultType = fromLegacyInfoToDataType(calculatedTable.getResultType());
            TableFunction<?> tableFunction = calculatedTable.getTableFunction();
            String[] fieldNames = calculatedTable.getTableSchema().getFieldNames();

            TypedFlinkTableFunction function = new TypedFlinkTableFunction(
                    tableFunction, fieldNames, resultType);

            FlinkTypeFactory typeFactory = relBuilder.getTypeFactory();

            TableSqlFunction sqlFunction = new TableSqlFunction(
                    FunctionIdentifier.of(tableFunction.functionIdentifier()),
                    tableFunction.toString(),
                    tableFunction,
                    resultType,
                    typeFactory,
                    function,
                    scala.Option.empty());

            List<RexNode> parameters = convertToRexNodes(calculatedTable.getParameters());

            return LogicalTableFunctionScan.create(
                    relBuilder.peek().getCluster(),
                    Collections.emptyList(),
                    relBuilder.call(sqlFunction, parameters),
                    function.getElementType(null),
                    function.getRowType(typeFactory, null, null),
                    null);
        }
Get the {@link AggCall} correspond to the aggregate or table aggregate expression.
core/src/main/java/org/apache/flink/table/planner/plan/QueryOperationConverter.java