coral-beam/src/main/java/com/linkedin/beam/utils/RexToBeamConverter.java
private List<Expression> wrapPigOperands(List<Expression> exOperands) {
    final List<Expression> newOperands = new ArrayList<>();
    for (Expression operand : exOperands) {
      if (operand.getType() == GenericArray.class) {
        newOperands.add(declareLocalVar(null, Expressions.new_(AvroBagWrapper.class, operand)));
      } else if (operand.getType() == GenericRecord.class) {
        // TODO: GenericRecord to Pig tuple not tested
        newOperands.add(declareLocalVar(null, Expressions.new_(AvroTupleWrapper.class, operand)));
      } else if (operand.getType() == ByteBuffer.class) {
        final Expression nullSafeExpr = RexBeamUtils.buildNullSafeExpr(ImmutableList.of(operand));
        final Expression byteBufferArrayExpr =
            Expressions.new_(DataByteArray.class, Expressions.call(operand, MethodNames.ARRAY));
        final Expression dataByteArrayExpr =
            Expressions.condition(nullSafeExpr, byteBufferArrayExpr, Expressions.constant(null));
        newOperands.add(declareLocalVar(null, dataByteArrayExpr));
      } else {
        newOperands.add(operand);
      }
    }
    return newOperands;
  }
/**
   * If input for buildTuple() is in Avro types, need to wrap it into Pig types
   */
If input for buildTuple() is in Avro types, need to wrap it into Pig types
If input for buildTuple () is in Avro types need to wrap it into Pig types
private List<Expression> wrapPigOperands(List<Expression> exOperands) {
    final List<Expression> newOperands = new ArrayList<>();
    for (Expression operand : exOperands) {
      if (operand.getType() == GenericArray.class) {
        newOperands.add(declareLocalVar(null, Expressions.new_(AvroBagWrapper.class, operand)));
      } else if (operand.getType() == GenericRecord.class) {
        newOperands.add(declareLocalVar(null, Expressions.new_(AvroTupleWrapper.class, operand)));
      } else if (operand.getType() == ByteBuffer.class) {
        final Expression nullSafeExpr = RexBeamUtils.buildNullSafeExpr(ImmutableList.of(operand));
        final Expression byteBufferArrayExpr =
            Expressions.new_(DataByteArray.class, Expressions.call(operand, MethodNames.ARRAY));
        final Expression dataByteArrayExpr =
            Expressions.condition(nullSafeExpr, byteBufferArrayExpr, Expressions.constant(null));
        newOperands.add(declareLocalVar(null, dataByteArrayExpr));
      } else {
        newOperands.add(operand);
      }
    }
    return newOperands;
  }
private List < Expression > wrapPigOperands ( List < Expression > exOperands ) { final List < Expression > newOperands = new ArrayList < > ( ) ; for ( Expression operand : exOperands ) { if ( operand . getType ( ) == GenericArray . class ) { newOperands . add ( declareLocalVar ( null , Expressions . new_ ( AvroBagWrapper . class , operand ) ) ) ; } else if ( operand . getType ( ) == GenericRecord . class ) { newOperands . add ( declareLocalVar ( null , Expressions . new_ ( AvroTupleWrapper . class , operand ) ) ) ; } else if ( operand . getType ( ) == ByteBuffer . class ) { final Expression nullSafeExpr = RexBeamUtils . buildNullSafeExpr ( ImmutableList . of ( operand ) ) ; final Expression byteBufferArrayExpr = Expressions . new_ ( DataByteArray . class , Expressions . call ( operand , MethodNames . ARRAY ) ) ; final Expression dataByteArrayExpr = Expressions . condition ( nullSafeExpr , byteBufferArrayExpr , Expressions . constant ( null ) ) ; newOperands . add ( declareLocalVar ( null , dataByteArrayExpr ) ) ; } else { newOperands . add ( operand ) ; } } return newOperands ; }
If input for buildTuple() is in Avro types, need to wrap it into Pig types
If input for buildTuple () is in Avro types need to wrap it into Pig types
// TODO: GenericRecord to Pig tuple not tested