public static FunctionTypeInfo getFunctionTypeInfo(
      final ExpressionTypeManager expressionTypeManager,
      final FunctionCall functionCall,
      final UdfFactory udfFactory,
      final Map<String, SqlType> lambdaMapping
  ) {
    // CHECKSTYLE_RULES.ON: CyclomaticComplexity
    final List<Expression> arguments = functionCall.getArguments();
    final List<SqlArgument> functionArgumentTypes = firstPassOverFunctionArguments(
        arguments,
        expressionTypeManager,
        lambdaMapping
    );

    final KsqlScalarFunction function = udfFactory.getFunction(functionArgumentTypes);
    final SqlType returnSchema;
    final List<ArgumentInfo> argumentInfoForFunction = new ArrayList<>();

    if (!functionCall.hasLambdaFunctionCallArguments()) {
      returnSchema = function.getReturnType(functionArgumentTypes);
      return FunctionTypeInfo.of(
          functionArgumentTypes.stream()
              .map(argument -> ArgumentInfo.of(argument, new HashMap<>(lambdaMapping)))
              .collect(Collectors.toList()),
          returnSchema,
          function
      );
    } else {
      final List<ParamType> paramTypes = function.parameters();
      final Map<GenericType, SqlType> reservedGenerics = new HashMap<>();
      final List<SqlArgument> functionArgumentTypesWithResolvedLambdaType = new ArrayList<>();

      // second pass over the function arguments to properly do lambda type checking
      for (int i = 0; i < arguments.size(); i++) {
        final Expression expression = arguments.get(i);
        final ParamType parameter = paramTypes.get(i);
        if (expression instanceof LambdaFunctionCall) {
          // the function returned from the UDF factory should have lambda
          // at this index in the function arguments if there's a
          // lambda node at this index in the function node argument list
          if (!(parameter instanceof LambdaType)) {
            throw new RuntimeException(String.format("Error while processing lambda function."
                + "Expected lambda parameter but was %s"
                + "This is most likely an internal error and a "
                + "Github issue should be filed for debugging. "
                + "Include the function name, the parameters passed in, the expected "
                + "signature, and any other relevant information.", parameter.toString()));
          }

          final ArrayList<SqlType> lambdaSqlTypes = new ArrayList<>();
          final Map<String, SqlType> variableTypeMapping = mapLambdaParametersToTypes(
              (LambdaFunctionCall) expression,
              (LambdaType) parameter,
              reservedGenerics,
              lambdaSqlTypes
          );

          final Map<String,SqlType> updateLambdaMapping =
              LambdaMappingUtil.resolveOldAndNewLambdaMapping(variableTypeMapping, lambdaMapping);

          final SqlType resolvedLambdaReturnType =
              expressionTypeManager.getExpressionSqlType(expression, updateLambdaMapping);
          final SqlArgument lambdaArgument = SqlArgument.of(
              SqlLambdaResolved.of(lambdaSqlTypes, resolvedLambdaReturnType));

          functionArgumentTypesWithResolvedLambdaType.add(lambdaArgument);
          argumentInfoForFunction.add(
              ArgumentInfo.of(
                  lambdaArgument,
                  new HashMap<>(updateLambdaMapping)));
        } else {
          functionArgumentTypesWithResolvedLambdaType.add(functionArgumentTypes.get(i));
          argumentInfoForFunction.add(
              ArgumentInfo.of(
                  functionArgumentTypes.get(i),
                  new HashMap<>(lambdaMapping)));
        }

        if (GenericsUtil.hasGenerics(parameter)) {
          final Pair<Boolean, Optional<KsqlException>> success = GenericsUtil.reserveGenerics(
              parameter,
              functionArgumentTypesWithResolvedLambdaType.get(i),
              reservedGenerics
          );
          if (!success.getLeft() && success.getRight().isPresent()) {
            throw success.getRight().get();
          }
        }
      }

      returnSchema = function.getReturnType(functionArgumentTypesWithResolvedLambdaType);
      return new FunctionTypeInfo(
          argumentInfoForFunction,
          returnSchema,
          function
      );
    }
  }
@Test
  public void shouldResolveGenericsInLambdaFunction() {
    // Given:
    givenUdfWithNameAndReturnType("ComplexFunction", SqlTypes.DOUBLE);
    when(function.parameters()).thenReturn(
        ImmutableList.of(
            ArrayType.of(GenericType.of("X")),
            MapType.of(GenericType.of("K"), GenericType.of("V")),
            GenericType.of("Q"),
            LambdaType.of(ImmutableList.of(GenericType.of("K"), GenericType.of("V"), GenericType.of("Q")), GenericType.of("K")),
            LambdaType.of(ImmutableList.of(GenericType.of("Q"), GenericType.of("V")), GenericType.of("X"))));

    final FunctionCall expression = givenFunctionCallWithMultipleLambdas();


    // When:
    final FunctionTypeInfo argumentsAndContexts = 
        FunctionArgumentsUtil.getFunctionTypeInfo(expressionTypeManager, expression, udfFactory, Collections.emptyMap());

    // Then:
    assertThat(argumentsAndContexts.getReturnType(), is(SqlTypes.DOUBLE));
    assertThat(argumentsAndContexts.getArgumentInfos().size(), is(5));

    assertThat(argumentsAndContexts.getArgumentInfos().get(3).getSqlArgument(), is(SqlArgument.of(SqlLambdaResolved.of(ImmutableList.of(SqlTypes.BIGINT, SqlTypes.DOUBLE, SqlTypes.STRING), SqlTypes.BIGINT))));
    assertThat(argumentsAndContexts.getArgumentInfos().get(3).getLambdaSqlTypeMapping().get("X"), is(SqlTypes.BIGINT));
    assertThat(argumentsAndContexts.getArgumentInfos().get(3).getLambdaSqlTypeMapping().get("Y"), is(SqlTypes.DOUBLE));
    assertThat(argumentsAndContexts.getArgumentInfos().get(3).getLambdaSqlTypeMapping().get("Z"), is(SqlTypes.STRING));

    assertThat(argumentsAndContexts.getArgumentInfos().get(4).getSqlArgument(), is(SqlArgument.of(SqlLambdaResolved.of(ImmutableList.of(SqlTypes.STRING, SqlTypes.DOUBLE), SqlTypes.DOUBLE))));
    assertThat(argumentsAndContexts.getArgumentInfos().get(4).getLambdaSqlTypeMapping().get("A"), is(SqlTypes.STRING));
    assertThat(argumentsAndContexts.getArgumentInfos().get(4).getLambdaSqlTypeMapping().get("B"), is(SqlTypes.DOUBLE));

    // in the first pass we should have
    verify(udfFactory).getFunction(
        ImmutableList.of(
            SqlArgument.of(SqlTypes.array(SqlTypes.DOUBLE)),
            SqlArgument.of(SqlTypes.map(SqlTypes.BIGINT, SqlTypes.DOUBLE)),
            SqlArgument.of(SqlTypes.STRING),
            SqlArgument.of(SqlLambda.of(3)),
            SqlArgument.of(SqlLambda.of(2))
        )
    );
    verify(function).getReturnType(
        ImmutableList.of(
            SqlArgument.of(SqlTypes.array(SqlTypes.DOUBLE)),
            SqlArgument.of(SqlTypes.map(SqlTypes.BIGINT, SqlTypes.DOUBLE)),
            SqlArgument.of(SqlTypes.STRING),
            SqlArgument.of(SqlLambdaResolved.of(ImmutableList.of(SqlTypes.BIGINT, SqlTypes.DOUBLE, SqlTypes.STRING), SqlTypes.BIGINT)),
            SqlArgument.of(SqlLambdaResolved.of(ImmutableList.of(SqlTypes.STRING, SqlTypes.DOUBLE), SqlTypes.DOUBLE))
        )
    );
  }