class BeamSqlUdfExpression extends BeamSqlExpression {

  @Override
  public BeamSqlPrimitive evaluate(
      Row inputRow, BoundedWindow window, BeamSqlExpressionEnvironment env) {
    if (method == null) {
      reConstructMethod();
    }
    try {
      List<Object> paras = new ArrayList<>();
      for (BeamSqlExpression e : getOperands()) {
        paras.add(e.evaluate(inputRow, window, env).getValue());
      }

      return BeamSqlPrimitive.of(
          getOutputType(), method.invoke(udfIns, paras.toArray(new Object[] {})));
    } catch (Exception ex) {
      throw new RuntimeException(ex);
    }
  }

  public  BeamSqlUdfExpression(
      Method method, List<BeamSqlExpression> subExps, SqlTypeName sqlTypeName);

}