@Override
  public synchronized FunctionSource getFunction(final List<SqlType> argTypeList) {

    final List<SqlArgument> args = argTypeList.stream()
            .map((type) -> type == null ? null : SqlArgument.of(type))
            .collect(Collectors.toList());

    final UdafFactoryInvoker creator = udfIndex.getFunction(args);
    if (creator == null) {
      throw new KsqlException("There is no aggregate function with name='" + getName()
          + "' that has arguments of type="
          + argTypeList.stream()
          .map(SqlType::baseType)
          .map(Objects::toString)
          .collect(Collectors.joining(",")));
    }

    final boolean isFactoryVariadic = creator.literalParams().stream()
            .anyMatch(ParameterInfo::isVariadic);

    /* There can only be one variadic argument, so we know either the column args are bounded
    or the initial args are bounded. */
    final int numInitArgs;
    final int numSignatureInitArgs = creator.literalParams().size();
    if (isFactoryVariadic) {
      numInitArgs = argTypeList.size() - (creator.parameterInfo().size() - numSignatureInitArgs);
    } else {
      numInitArgs = numSignatureInitArgs;
    }

    return new FunctionSource(
            numInitArgs,
            (initArgs) -> creator.createFunction(initArgs, args)
    );
  }
@Test
  public void shouldHandleNullLiteralParams() {
    // When:
    AggregateFunctionFactory.FunctionSource result = functionFactory.getFunction(
            Arrays.asList(SqlTypes.STRING, null, SqlTypes.INTEGER,
                    SqlTypes.BIGINT, SqlTypes.DOUBLE, SqlTypes.STRING)
    );

    int initArgs = result.initArgs;

    result.source.apply(new AggregateFunctionInitArguments(
            Collections.singletonList(0),
            ImmutableMap.of(),
            Arrays.asList(null, 5, 4L, 2.3d, "s")
    ));

    // Then:
    assertEquals(5, initArgs);
    verify(functionIndex).getFunction(Arrays.asList(SqlArgument.of(SqlTypes.STRING), null,
            SqlArgument.of(SqlTypes.INTEGER), SqlArgument.of(SqlTypes.BIGINT),
            SqlArgument.of(SqlTypes.DOUBLE), SqlArgument.of(SqlTypes.STRING)));
  }