@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 shouldHandleInitParamsWhenFactoryIsVariadic() {
    // When:
    AggregateFunctionFactory.FunctionSource result = variadicFunctionFactory.getFunction(
            ImmutableList.of(SqlTypes.STRING, SqlTypes.INTEGER, SqlTypes.BOOLEAN,
                    SqlTypes.BOOLEAN, SqlTypes.BOOLEAN, SqlTypes.BOOLEAN)
    );

    int initArgs = result.initArgs;

    result.source.apply(new AggregateFunctionInitArguments(
            Collections.singletonList(0),
            ImmutableMap.of(),
            ImmutableList.of(2, true, false, true, true)
    ));

    // Then: did not throw.
    assertEquals(5, initArgs);
  }