shifted SP and FED opcodes to enum. Created InstructionType to consolidate all types.

Author: aarnaty

src/main/java/org/apache/sysds/runtime/instructions/spark/BinaryFrameFrameSPInstruction.java

@@ -33,7 +33,7 @@
 
 public class BinaryFrameFrameSPInstruction extends BinarySPInstruction {
 	protected BinaryFrameFrameSPInstruction(Operator op, CPOperand in1, CPOperand in2, CPOperand out,
-			String opcode, String istr) {
+											String opcode, String istr) {
 		super(SPType.Binary, op, in1, in2, out, opcode, istr);
 	}
 
@@ -45,7 +45,7 @@ public void processInstruction(ExecutionContext ec) {
 		// Get input RDDs
 		JavaPairRDD<Long, FrameBlock> in1 = sec.getFrameBinaryBlockRDDHandleForVariable(input1.getName());
 		JavaPairRDD<Long, FrameBlock> out = null;
-		
+
 		if(getOpcode().equals(Opcodes.DROPINVALIDTYPE.toString())) {
 			// get schema frame-block
 			Broadcast<FrameBlock> fb = sec.getSparkContext().broadcast(sec.getFrameInput(input2.getName()));
@@ -60,18 +60,25 @@ else if(getOpcode().equals(Opcodes.VALUESWAP.toString())) {
 			// Attach result frame with FrameBlock associated with output_name
 			sec.releaseFrameInput(input2.getName());
 		}
+		else if(getOpcode().equals(Opcodes.APPLYSCHEMA.toString())){
+			Broadcast<FrameBlock> fb = sec.getSparkContext().broadcast(sec.getFrameInput(input2.getName()));
+			out = in1.mapValues(new applySchema(fb.getValue()));
+			sec.releaseFrameInput(input2.getName());
+		}
 		else {
 			JavaPairRDD<Long, FrameBlock> in2 = sec.getFrameBinaryBlockRDDHandleForVariable(input2.getName());
 			// create output frame
 			BinaryOperator dop = (BinaryOperator) _optr;
 			// check for binary operations
 			out = in1.join(in2).mapValues(new FrameComparison(dop));
 		}
-		
+
 		//set output RDD and maintain dependencies
 		sec.setRDDHandleForVariable(output.getName(), out);
 		sec.addLineageRDD(output.getName(), input1.getName());
-		if( !getOpcode().equals(Opcodes.DROPINVALIDTYPE.toString())  && !getOpcode().equals(Opcodes.VALUESWAP.toString()))
+		if(!getOpcode().equals(Opcodes.DROPINVALIDTYPE.toString()) && //
+				!getOpcode().equals(Opcodes.VALUESWAP.toString()) && //
+				!getOpcode().equals(Opcodes.APPLYSCHEMA.toString()))
 			sec.addLineageRDD(output.getName(), input2.getName());
 	}
 
@@ -117,4 +124,20 @@ public FrameBlock call(FrameBlock arg0) throws Exception {
 			return arg0.valueSwap(schema_frame);
 		}
 	}
-}
+
+
+	private static class applySchema implements Function<FrameBlock, FrameBlock>{
+		private static final long serialVersionUID = 58504021316402L;
+
+		private FrameBlock schema;
+
+		public applySchema(FrameBlock schema ) {
+			this.schema = schema;
+		}
+
+		@Override
+		public FrameBlock call(FrameBlock arg0) throws Exception {
+			return arg0.applySchema(schema);
+		}
+	}
+}