coral-beam/src/main/java/com/linkedin/beam/planner/CodeGenerator.java
private List<Statement> initPigUDFs(List<MemberDeclaration> memberDeclarations, Set<Expression> resourceFilesExprSet) {
    final List<Statement> udfInits = new ArrayList<>();
    final Set<Class> exceptions = new HashSet<>();

    for (PigUserDefinedFunction pigUDF : pigUdfMap.values()) {
      final Class clazz = ((ScalarFunctionImpl) pigUDF.getFunction()).method.getDeclaringClass();
      if (EXCLUDED_PIG_UDFS.contains(clazz)) {
        continue;
      }

      List<ConstantExpression> constructorArgs = new ArrayList<>();
      if (pigUDF.funcSpec.getCtorArgs() != null) {
        final String[] ctorArgs = pigUDF.funcSpec.getCtorArgs();
        final Class[] ctorTypes = new Class[ctorArgs.length];
        for (int i = 0; i < ctorArgs.length; i++) {
          final String filename = ctorArgs[i].substring(ctorArgs[i].lastIndexOf("/") + 1);
          constructorArgs.add(Expressions.constant(filename, String.class));
          ctorTypes[i] = String.class;
        }
        try {
          Collections.addAll(exceptions, clazz.getConstructor(ctorTypes).getExceptionTypes());
        } catch (NoSuchMethodException e) {
          e.printStackTrace();
          StringBuilder message = new StringBuilder("Cannot find constructor for Pig UDF ").append(clazz.getName())
              .append(" with arguments: ")
              .append(Arrays.asList(ctorTypes));
          throw new RuntimeException(message.toString());
        }
        if (clazz == JythonFunction.class) {
          // If Python UDF, only need to copy the python source file, which is the first argument
          resourceFilesExprSet.add(constructorArgs.get(0));
        } else {
          for (ConstantExpression arg : constructorArgs) {
            resourceFilesExprSet.add(arg);
          }
        }
      }
      final ParameterExpression udfName = PigUDFUtils.getPigUDFParamExpr(pigUDF);
      memberDeclarations.add(Expressions.fieldDecl(Modifier.PRIVATE | Modifier.STATIC, udfName));
      udfInits.add(Expressions.statement(Expressions.assign(udfName, Expressions.new_(clazz, constructorArgs))));
    }

    if (exceptions.isEmpty()) {
      return udfInits;
    }
    return RexBeamUtils.wrapTryCatch(udfInits, exceptions).statements;
  }
/**
   * Initializes Pig UDF and capture resource files needed to initialize these UDF objects.
   * @param memberDeclarations Output param for UDF object declaration
   * @param resourceFilesExprSet Output param for resource files needed
   */
Initializes Pig UDF and capture resource files needed to initialize these UDF objects.
@param memberDeclarations Output param for UDF object declaration
@param resourceFilesExprSet Output param for resource files needed
Initializes Pig UDF and capture resource files needed to initialize these UDF objects . @param memberDeclarations Output param for UDF object declaration @param resourceFilesExprSet Output param for resource files needed
private List<Statement> initPigUDFs(List<MemberDeclaration> memberDeclarations, Set<Expression> resourceFilesExprSet) {
    final List<Statement> udfInits = new ArrayList<>();
    final Set<Class> exceptions = new HashSet<>();
    for (PigUserDefinedFunction pigUDF : pigUdfMap.values()) {
      final Class clazz = ((ScalarFunctionImpl) pigUDF.getFunction()).method.getDeclaringClass();
      if (EXCLUDED_PIG_UDFS.contains(clazz)) {
        continue;
      }
      List<ConstantExpression> constructorArgs = new ArrayList<>();
      if (pigUDF.funcSpec.getCtorArgs() != null) {
        final String[] ctorArgs = pigUDF.funcSpec.getCtorArgs();
        final Class[] ctorTypes = new Class[ctorArgs.length];
        for (int i = 0; i < ctorArgs.length; i++) {
          final String filename = ctorArgs[i].substring(ctorArgs[i].lastIndexOf("/") + 1);
          constructorArgs.add(Expressions.constant(filename, String.class));
          ctorTypes[i] = String.class;
        }
        try {
          Collections.addAll(exceptions, clazz.getConstructor(ctorTypes).getExceptionTypes());
        } catch (NoSuchMethodException e) {
          e.printStackTrace();
          StringBuilder message = new StringBuilder("Cannot find constructor for Pig UDF ").append(clazz.getName())
              .append(" with arguments: ")
              .append(Arrays.asList(ctorTypes));
          throw new RuntimeException(message.toString());
        }
        if (clazz == JythonFunction.class) {
          resourceFilesExprSet.add(constructorArgs.get(0));
        } else {
          for (ConstantExpression arg : constructorArgs) {
            resourceFilesExprSet.add(arg);
          }
        }
      }
      final ParameterExpression udfName = PigUDFUtils.getPigUDFParamExpr(pigUDF);
      memberDeclarations.add(Expressions.fieldDecl(Modifier.PRIVATE | Modifier.STATIC, udfName));
      udfInits.add(Expressions.statement(Expressions.assign(udfName, Expressions.new_(clazz, constructorArgs))));
    }
    if (exceptions.isEmpty()) {
      return udfInits;
    }
    return RexBeamUtils.wrapTryCatch(udfInits, exceptions).statements;
  }
private List < Statement > initPigUDFs ( List < MemberDeclaration > memberDeclarations , Set < Expression > resourceFilesExprSet ) { final List < Statement > udfInits = new ArrayList < > ( ) ; final Set < Class > exceptions = new HashSet < > ( ) ; for ( PigUserDefinedFunction pigUDF : pigUdfMap . values ( ) ) { final Class clazz = ( ( ScalarFunctionImpl ) pigUDF . getFunction ( ) ) . method . getDeclaringClass ( ) ; if ( EXCLUDED_PIG_UDFS . contains ( clazz ) ) { continue ; } List < ConstantExpression > constructorArgs = new ArrayList < > ( ) ; if ( pigUDF . funcSpec . getCtorArgs ( ) != null ) { final String [ ] ctorArgs = pigUDF . funcSpec . getCtorArgs ( ) ; final Class [ ] ctorTypes = new Class [ ctorArgs . length ] ; for ( int i = 0 ; i < ctorArgs . length ; i ++ ) { final String filename = ctorArgs [ i ] . substring ( ctorArgs [ i ] . lastIndexOf ( "/" ) + 1 ) ; constructorArgs . add ( Expressions . constant ( filename , String . class ) ) ; ctorTypes [ i ] = String . class ; } try { Collections . addAll ( exceptions , clazz . getConstructor ( ctorTypes ) . getExceptionTypes ( ) ) ; } catch ( NoSuchMethodException e ) { e . printStackTrace ( ) ; StringBuilder message = new StringBuilder ( "Cannot find constructor for Pig UDF " ) . append ( clazz . getName ( ) ) . append ( " with arguments: " ) . append ( Arrays . asList ( ctorTypes ) ) ; throw new RuntimeException ( message . toString ( ) ) ; } if ( clazz == JythonFunction . class ) { resourceFilesExprSet . add ( constructorArgs . get ( 0 ) ) ; } else { for ( ConstantExpression arg : constructorArgs ) { resourceFilesExprSet . add ( arg ) ; } } } final ParameterExpression udfName = PigUDFUtils . getPigUDFParamExpr ( pigUDF ) ; memberDeclarations . add ( Expressions . fieldDecl ( Modifier . PRIVATE | Modifier . STATIC , udfName ) ) ; udfInits . add ( Expressions . statement ( Expressions . assign ( udfName , Expressions . new_ ( clazz , constructorArgs ) ) ) ) ; } if ( exceptions . isEmpty ( ) ) { return udfInits ; } return RexBeamUtils . wrapTryCatch ( udfInits , exceptions ) . statements ; }
Initializes Pig UDF and capture resource files needed to initialize these UDF objects.
Initializes Pig UDF and capture resource files needed to initialize these UDF objects .
// If Python UDF, only need to copy the python source file, which is the first argument