[SYSTEMDS-3787] Fix missing transformencode update federated backend

This patch adds support for the new transform encode w/o metadata
output also to the federated backend, and fixes the robustness for
handling empty frame blocks.

Author: mboehm7

src/main/java/org/apache/sysds/runtime/frame/data/FrameBlock.java

@@ -925,9 +925,10 @@ public boolean isShallowSerialize() {
 	public boolean isShallowSerialize(boolean inclConvert) {
 		// shallow serialize if non-string schema because a frame block
 		// is always dense but strings have large array overhead per cell
-		for(int j = 0; j < _schema.length; j++)
-			if(!_coldata[j].isShallowSerialize())
-				return false;
+		if( _schema != null )
+			for(int j = 0; j < _schema.length; j++)
+				if(!_coldata[j].isShallowSerialize())
+					return false;
 		return true;
 	}
 

src/main/java/org/apache/sysds/runtime/instructions/cp/MultiReturnParameterizedBuiltinCPInstruction.java

@@ -123,4 +123,8 @@ public Pair<String, LineageItem>[] getLineageItems(ExecutionContext ec) {
 		}
 		return ret; 
 	}
+
+	public boolean getMetaReturn() {
+		return _metaReturn;
+	}
 }

src/main/java/org/apache/sysds/runtime/instructions/fed/MultiReturnParameterizedBuiltinFEDInstruction.java

@@ -70,10 +70,12 @@
 
 public class MultiReturnParameterizedBuiltinFEDInstruction extends ComputationFEDInstruction {
 	protected final List<CPOperand> _outputs;
-
+	protected final boolean _metaReturn;
+	
 	private MultiReturnParameterizedBuiltinFEDInstruction(Operator op, CPOperand input1, CPOperand input2,
-		List<CPOperand> outputs, String opcode, String istr) {
+		List<CPOperand> outputs, boolean metaReturn, String opcode, String istr) {
 		super(FEDType.MultiReturnParameterizedBuiltin, op, input1, input2, null, opcode, istr);
+		_metaReturn = metaReturn;
 		_outputs = outputs;
 	}
 
@@ -104,13 +106,13 @@ public static MultiReturnParameterizedBuiltinFEDInstruction parseInstruction(
 	private static MultiReturnParameterizedBuiltinFEDInstruction parseInstruction(
 		MultiReturnParameterizedBuiltinCPInstruction instr) {
 		return new MultiReturnParameterizedBuiltinFEDInstruction(instr.getOperator(), instr.input1, instr.input2,
-			instr.getOutputs(), instr.getOpcode(), instr.getInstructionString());
+			instr.getOutputs(), instr.getMetaReturn(), instr.getOpcode(), instr.getInstructionString());
 	}
 
 	private static MultiReturnParameterizedBuiltinFEDInstruction parseInstruction(
 		MultiReturnParameterizedBuiltinSPInstruction instr) {
 		return new MultiReturnParameterizedBuiltinFEDInstruction(instr.getOperator(), instr.input1, instr.input2,
-			instr.getOutputs(), instr.getOpcode(), instr.getInstructionString());
+			instr.getOutputs(), instr.getMetaReturn(), instr.getOpcode(), instr.getInstructionString());
 	}
 
 	public static MultiReturnParameterizedBuiltinFEDInstruction parseInstruction(String str) {
@@ -122,9 +124,15 @@ public static MultiReturnParameterizedBuiltinFEDInstruction parseInstruction(Str
 			// one input and two outputs
 			CPOperand in1 = new CPOperand(parts[1]);
 			CPOperand in2 = new CPOperand(parts[2]);
-			outputs.add(new CPOperand(parts[3], Types.ValueType.FP64, Types.DataType.MATRIX));
-			outputs.add(new CPOperand(parts[4], Types.ValueType.STRING, Types.DataType.FRAME));
-			return new MultiReturnParameterizedBuiltinFEDInstruction(null, in1, in2, outputs, opcode, str);
+			int pos = 3;
+			boolean metaReturn = true;
+			System.out.println(Arrays.toString(parts));
+			if( parts.length == 7 ) //no need for meta data
+				metaReturn = new CPOperand(parts[pos++]).getLiteral().getBooleanValue();
+			outputs.add(new CPOperand(parts[pos], Types.ValueType.FP64, Types.DataType.MATRIX));
+			outputs.add(new CPOperand(parts[pos+1], Types.ValueType.STRING, Types.DataType.FRAME));
+			return new MultiReturnParameterizedBuiltinFEDInstruction(
+				null, in1, in2, outputs, metaReturn, opcode, str);
 		}
 		else {
 			throw new DMLRuntimeException("Invalid opcode in MultiReturnBuiltin instruction: " + opcode);
@@ -198,7 +206,7 @@ public void processInstruction(ExecutionContext ec) {
 		encodeFederatedFrames(fedMapping, globalEncoder, ec.getMatrixObject(getOutput(0)));
 
 		// release input and outputs
-		ec.setFrameOutput(getOutput(1).getName(), meta);
+		ec.setFrameOutput(getOutput(1).getName(), _metaReturn ? meta : new FrameBlock());
 	}
 
 	private class EncoderColnames {

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

@@ -468,4 +468,8 @@ else if(_encoder.getMethod(colix) == MVMethod.CONSTANT) {
 			return ret.iterator();
 		}
 	}
+
+	public boolean getMetaReturn() {
+		return _metaReturn;
+	}
 }