presto-main/src/main/java/io/prestosql/operator/aggregation/AggregationMetadata.java/AggregationMetadata/verifyInputFunctionSignature
class AggregationMetadata:    private static void verifyInputFunctionSignature(MethodHandle method, List<ParameterMetadata> dataChannelMetadata, List<Class<?>> lambdaInterfaces, List<AccumulatorStateDescriptor> stateDescriptors)
    {
        Class<?>[] parameters = method.type().parameterArray();
        checkArgument(parameters.length > 0, "Aggregation input function must have at least one parameter");
        checkArgument(parameters.length == dataChannelMetadata.size() + lambdaInterfaces.size(), "Wenlei TODO...");
        checkArgument(dataChannelMetadata.stream().filter(m -> m.getParameterType() == STATE).count() == stateDescriptors.size(), "Number of state parameter in input function must be the same as size of stateDescriptors");
        checkArgument(dataChannelMetadata.get(0).getParameterType() == STATE, "First parameter must be state");

        // verify data channels
        int stateIndex = 0;
        for (int i = 0; i < dataChannelMetadata.size(); i++) {
            ParameterMetadata metadata = dataChannelMetadata.get(i);
            switch (metadata.getParameterType()) {
                case STATE:
                    checkArgument(stateDescriptors.get(stateIndex).getStateInterface() == parameters[i], "State argument must be of type %s", stateDescriptors.get(stateIndex).getStateInterface());
                    stateIndex++;
                    break;
                case BLOCK_INPUT_CHANNEL:
                case NULLABLE_BLOCK_INPUT_CHANNEL:
                    checkArgument(parameters[i] == Block.class, "Parameter must be Block if it has @BlockPosition");
                    break;
                case INPUT_CHANNEL:
                    checkArgument(SUPPORTED_PARAMETER_TYPES.contains(parameters[i]), "Unsupported type: %s", parameters[i].getSimpleName());
                    verifyMethodParameterType(method, i, metadata.getSqlType().getJavaType(), metadata.getSqlType().getDisplayName());
                    break;
                case BLOCK_INDEX:
                    checkArgument(parameters[i] == int.class, "Block index parameter must be an int");
                    break;
                default:
                    throw new IllegalArgumentException("Unsupported parameter: " + metadata.getParameterType());
            }
        }
        checkArgument(stateIndex == stateDescriptors.size(), "Input function only has %s states, expected: %s", stateIndex, stateDescriptors.size());

        // verify lambda channels
        for (int i = 0; i < lambdaInterfaces.size(); i++) {
            verifyMethodParameterType(method, i + dataChannelMetadata.size(), lambdaInterfaces.get(i), "function");
        }
    }