public static KsqlAggregateFunction<?, ?, ?> resolveAggregateFunction(
      final FunctionRegistry functionRegistry,
      final FunctionCall functionCall,
      final LogicalSchema schema,
      final KsqlConfig config
  ) {
    try {
      final ExpressionTypeManager expressionTypeManager =
          new ExpressionTypeManager(schema, functionRegistry);

      final List<SqlType> args = functionCall.getArguments().stream()
              .map(expressionTypeManager::getExpressionSqlType)
              .collect(Collectors.toList());

      final AggregateFunctionFactory.FunctionSource func = functionRegistry
              .getAggregateFactory(functionCall.getName())
              .getFunction(args);

      final int totalArgs = functionCall.getArguments().size();

      // All non-constant UDAF arguments must be column references
      final List<Integer> argIndices = functionCall.getArguments().stream()
              .limit(totalArgs - func.initArgs)
              .map((arg) -> {
                final Optional<Column> column;

                if (arg instanceof UnqualifiedColumnReferenceExp) {
                  final UnqualifiedColumnReferenceExp colRef =
                          (UnqualifiedColumnReferenceExp) arg;
                  column = schema.findValueColumn(colRef.getColumnName());
                } else {
                  // assume that it is a column reference with no alias
                  column = schema.findValueColumn(ColumnName.of(arg.toString()));
                }

                return column.orElseThrow(
                        () -> new KsqlException("Could not find column for expression: " + arg)
                );
              }).map(Column::index).collect(Collectors.toList());

      return func.source.apply(createAggregateFunctionInitArgs(
              func.initArgs,
              argIndices,
              functionCall,
              config
      ));
    } catch (final Exception e) {
      throw new KsqlException("Failed to create aggregate function: " + functionCall, e);
    }
  }
@Test
  public void shouldThrowIfSecondParamIsColArgAndNotACol() {
    // Given:
    when(functionCall.getArguments()).thenReturn(ImmutableList.of(
            new UnqualifiedColumnReferenceExp(ColumnName.of("FOO")),
            new StringLiteral("Not good!"),
            new StringLiteral("No issue here")
    ));

    // When:
    final Exception e = assertThrows(
            KsqlException.class,
            () -> UdafUtil.resolveAggregateFunction(
                    functionRegistry,
                    functionCall,
                    SCHEMA,
                    KsqlConfig.empty()
            )
    );

    // Then:
    assertThat(e.getMessage(), is("Failed to create aggregate function: functionCall"));
  }