You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@systemds.apache.org by ba...@apache.org on 2022/08/19 16:36:29 UTC
[systemds] branch main updated: [SYSTEMDS-3386] Refactor replacement of CP instructions
This is an automated email from the ASF dual-hosted git repository.
baunsgaard pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/systemds.git
The following commit(s) were added to refs/heads/main by this push:
new 94f8c7f2e6 [SYSTEMDS-3386] Refactor replacement of CP instructions
94f8c7f2e6 is described below
commit 94f8c7f2e6b8a8615c3ec233db235e4c58ce158f
Author: Kevin Innerebner <ke...@yahoo.com>
AuthorDate: Fri Aug 19 18:34:23 2022 +0200
[SYSTEMDS-3386] Refactor replacement of CP instructions
- Refactor replacement with FED instructions
- Fix FED output flag removal
Closes #1628
---
.../controlprogram/federated/FederationUtils.java | 3 +-
.../runtime/instructions/InstructionUtils.java | 3 +-
.../instructions/cp/CtableCPInstruction.java | 16 ++
.../instructions/cp/IndexingCPInstruction.java | 16 ++
.../instructions/cp/MMChainCPInstruction.java | 4 +
.../instructions/cp/MMTSJCPInstruction.java | 4 +
.../instructions/cp/QuantilePickCPInstruction.java | 10 +-
.../instructions/cp/QuantileSortCPInstruction.java | 4 +
.../instructions/cp/QuaternaryCPInstruction.java | 2 +-
.../instructions/cp/ReshapeCPInstruction.java | 16 ++
.../instructions/cp/SpoofCPInstruction.java | 4 +
.../instructions/cp/UnaryScalarCPInstruction.java | 4 +-
.../fed/AggregateBinaryFEDInstruction.java | 14 +-
.../fed/AggregateTernaryFEDInstruction.java | 12 ++
.../fed/AggregateUnaryFEDInstruction.java | 13 ++
.../instructions/fed/AppendFEDInstruction.java | 28 ++--
.../instructions/fed/BinaryFEDInstruction.java | 2 +-
.../fed/BinaryMatrixMatrixFEDInstruction.java | 15 ++
.../fed/BinaryMatrixScalarFEDInstruction.java | 15 ++
.../instructions/fed/CastFEDInstruction.java | 6 +
.../fed/CentralMomentFEDInstruction.java | 18 +--
.../instructions/fed/CovarianceFEDInstruction.java | 20 +--
.../instructions/fed/CtableFEDInstruction.java | 35 +++--
.../fed/CumulativeOffsetFEDInstruction.java | 6 +
.../instructions/fed/FEDInstructionUtils.java | 166 +++++++++++----------
.../instructions/fed/IndexingFEDInstruction.java | 12 ++
.../instructions/fed/MMChainFEDInstruction.java | 6 +
.../runtime/instructions/fed/MMFEDInstruction.java | 15 +-
...tiReturnParameterizedBuiltinFEDInstruction.java | 18 ++-
.../fed/ParameterizedBuiltinFEDInstruction.java | 16 +-
.../fed/QuantilePickFEDInstruction.java | 14 +-
.../fed/QuantileSortFEDInstruction.java | 12 ++
.../instructions/fed/QuaternaryFEDInstruction.java | 38 ++++-
.../fed/QuaternaryWCeMMFEDInstruction.java | 14 ++
.../fed/QuaternaryWDivMMFEDInstruction.java | 14 ++
.../fed/QuaternaryWSLossFEDInstruction.java | 14 ++
.../fed/QuaternaryWSigmoidFEDInstruction.java | 15 ++
.../fed/QuaternaryWUMMFEDInstruction.java | 15 ++
.../instructions/fed/ReblockFEDInstruction.java | 13 +-
.../instructions/fed/ReorgFEDInstruction.java | 12 ++
.../instructions/fed/ReshapeFEDInstruction.java | 13 ++
.../instructions/fed/SpoofFEDInstruction.java | 14 ++
.../instructions/fed/TernaryFEDInstruction.java | 12 ++
.../fed/TernaryFrameScalarFEDInstruction.java | 14 ++
.../instructions/fed/TsmmFEDInstruction.java | 8 +-
.../fed/UnaryMatrixFEDInstruction.java | 12 ++
.../instructions/fed/VariableFEDInstruction.java | 1 +
.../spark/AggregateBinarySPInstruction.java | 33 ++++
.../spark/AppendGAlignedSPInstruction.java | 9 +-
.../instructions/spark/AppendGSPInstruction.java | 9 +-
.../instructions/spark/AppendMSPInstruction.java | 8 +-
.../instructions/spark/AppendRSPInstruction.java | 10 +-
.../instructions/spark/AppendSPInstruction.java | 40 +++++
.../spark/BinaryMatrixBVectorSPInstruction.java | 4 +-
.../instructions/spark/CpmmSPInstruction.java | 2 +-
.../instructions/spark/CtableSPInstruction.java | 17 +++
.../spark/CumulativeOffsetSPInstruction.java | 8 +
.../instructions/spark/IndexingSPInstruction.java | 16 ++
.../instructions/spark/MapmmSPInstruction.java | 2 +-
.../spark/MatrixReshapeSPInstruction.java | 14 +-
...ltiReturnParameterizedBuiltinSPInstruction.java | 4 +
.../spark/ParameterizedBuiltinSPInstruction.java | 4 +
.../spark/QuantilePickSPInstruction.java | 6 +-
.../spark/QuaternarySPInstruction.java | 4 +
.../instructions/spark/ReblockSPInstruction.java | 8 +
.../instructions/spark/RmmSPInstruction.java | 2 +-
.../instructions/spark/SpoofSPInstruction.java | 4 +
67 files changed, 750 insertions(+), 192 deletions(-)
diff --git a/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederationUtils.java b/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederationUtils.java
index 82f78fe176..269d91e9c5 100644
--- a/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederationUtils.java
+++ b/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederationUtils.java
@@ -152,6 +152,7 @@ public class FederationUtils {
}
public static FederatedRequest callInstruction(String inst, CPOperand varOldOut, long outputId, CPOperand[] varOldIn, long[] varNewIn, ExecType type, boolean rmFedOutputFlag) {
+ boolean isFedInstr = inst.startsWith(ExecType.FED.name() + Lop.OPERAND_DELIMITOR);
String linst = InstructionUtils.replaceOperand(inst, 0, type.name());
linst = linst.replace(Lop.OPERAND_DELIMITOR+varOldOut.getName()+Lop.DATATYPE_PREFIX, Lop.OPERAND_DELIMITOR+outputId+Lop.DATATYPE_PREFIX);
for(int i=0; i<varOldIn.length; i++)
@@ -161,7 +162,7 @@ public class FederationUtils {
Lop.OPERAND_DELIMITOR+(varNewIn[i])+Lop.DATATYPE_PREFIX);
linst = linst.replace("="+varOldIn[i].getName(), "="+(varNewIn[i])); //parameterized
}
- if(rmFedOutputFlag)
+ if(rmFedOutputFlag && isFedInstr)
linst = InstructionUtils.removeFEDOutputFlag(linst);
return new FederatedRequest(RequestType.EXEC_INST, outputId, linst);
}
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/InstructionUtils.java b/src/main/java/org/apache/sysds/runtime/instructions/InstructionUtils.java
index 4d7028f217..2c8468955f 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/InstructionUtils.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/InstructionUtils.java
@@ -1135,10 +1135,11 @@ public class InstructionUtils
* @return instruction string prepared for federated request
*/
public static String instructionStringFEDPrepare(String inst, CPOperand varOldOut, long id, CPOperand[] varOldIn, long[] varNewIn, boolean rmFederatedOutput){
+ boolean isFedInstr = inst.startsWith(ExecType.FED.name() + Lop.OPERAND_DELIMITOR);
String linst = replaceExecTypeWithCP(inst);
linst = replaceOutputOperand(linst, varOldOut, id);
linst = replaceInputOperand(linst, varOldIn, varNewIn);
- if(rmFederatedOutput)
+ if(rmFederatedOutput && isFedInstr)
linst = removeFEDOutputFlag(linst);
return linst;
}
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/cp/CtableCPInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/cp/CtableCPInstruction.java
index f44aeb4b58..647b8e9c24 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/cp/CtableCPInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/cp/CtableCPInstruction.java
@@ -185,4 +185,20 @@ public class CtableCPInstruction extends ComputationCPInstruction {
LineageItemUtils.getLineage(ec, input1, input2, input3);
return Pair.of(output.getName(), new LineageItem(getOpcode(), linputs));
}
+
+ public CPOperand getOutDim1() {
+ return _outDim1;
+ }
+
+ public CPOperand getOutDim2() {
+ return _outDim2;
+ }
+
+ public boolean getIsExpand() {
+ return _isExpand;
+ }
+
+ public boolean getIgnoreZeros() {
+ return _ignoreZeros;
+ }
}
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/cp/IndexingCPInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/cp/IndexingCPInstruction.java
index 09939cf839..86bb2bef9e 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/cp/IndexingCPInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/cp/IndexingCPInstruction.java
@@ -56,6 +56,22 @@ public abstract class IndexingCPInstruction extends UnaryCPInstruction {
(int)(ec.getScalarInput(colUpper).getLongValue()-1));
}
+ public CPOperand getRowLower() {
+ return rowLower;
+ }
+
+ public CPOperand getRowUpper() {
+ return rowUpper;
+ }
+
+ public CPOperand getColLower() {
+ return colLower;
+ }
+
+ public CPOperand getColUpper() {
+ return colUpper;
+ }
+
public static IndexingCPInstruction parseInstruction ( String str ) {
String[] parts = InstructionUtils.getInstructionPartsWithValueType(str);
String opcode = parts[0];
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/cp/MMChainCPInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/cp/MMChainCPInstruction.java
index dcff65bf70..03d646c1f0 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/cp/MMChainCPInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/cp/MMChainCPInstruction.java
@@ -41,6 +41,10 @@ public class MMChainCPInstruction extends UnaryCPInstruction {
return _type;
}
+ public int getNumThreads() {
+ return _numThreads;
+ }
+
public static MMChainCPInstruction parseInstruction ( String str ) {
//parse instruction parts (without exec type)
String[] parts = InstructionUtils.getInstructionPartsWithValueType( str );
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/cp/MMTSJCPInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/cp/MMTSJCPInstruction.java
index 4c866b0f9e..f1913d7817 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/cp/MMTSJCPInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/cp/MMTSJCPInstruction.java
@@ -70,4 +70,8 @@ public class MMTSJCPInstruction extends UnaryCPInstruction {
{
return _type;
}
+
+ public int getNumThreads() {
+ return _numThreads;
+ }
}
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/cp/QuantilePickCPInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/cp/QuantilePickCPInstruction.java
index ee4c6d273a..2006529c1a 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/cp/QuantilePickCPInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/cp/QuantilePickCPInstruction.java
@@ -65,7 +65,7 @@ public class QuantilePickCPInstruction extends BinaryCPInstruction {
CPOperand out = new CPOperand(parts[2]);
OperationTypes ptype = OperationTypes.valueOf(parts[3]);
boolean inmem = Boolean.parseBoolean(parts[4]);
- return new QuantilePickCPInstruction(null, in1, new CPOperand(), out, ptype, inmem, opcode, str);
+ return new QuantilePickCPInstruction(null, in1, out, ptype, inmem, opcode, str);
}
else if( parts.length == 6 ) {
CPOperand in1 = new CPOperand(parts[1]);
@@ -127,4 +127,12 @@ public class QuantilePickCPInstruction extends BinaryCPInstruction {
throw new DMLRuntimeException("Unsupported qpick operation type: "+_type);
}
}
+
+ public OperationTypes getOperationType() {
+ return _type;
+ }
+
+ public boolean isInMem() {
+ return _inmem;
+ }
}
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/cp/QuantileSortCPInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/cp/QuantileSortCPInstruction.java
index 3123af55bd..176d6ca7ea 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/cp/QuantileSortCPInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/cp/QuantileSortCPInstruction.java
@@ -119,4 +119,8 @@ public class QuantileSortCPInstruction extends UnaryCPInstruction {
//set and release output
ec.setMatrixOutput(output.getName(), resultBlock);
}
+
+ public int getNumThreads() {
+ return _numThreads;
+ }
}
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/cp/QuaternaryCPInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/cp/QuaternaryCPInstruction.java
index 7f8ec4da4f..137038dd49 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/cp/QuaternaryCPInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/cp/QuaternaryCPInstruction.java
@@ -135,5 +135,5 @@ public class QuaternaryCPInstruction extends ComputationCPInstruction {
}
ec.setMatrixOutput(output.getName(), out);
}
- }
+ }
}
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/cp/ReshapeCPInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/cp/ReshapeCPInstruction.java
index f3fa3fb552..9cf30acb37 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/cp/ReshapeCPInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/cp/ReshapeCPInstruction.java
@@ -113,4 +113,20 @@ public class ReshapeCPInstruction extends UnaryCPInstruction {
return Pair.of(output.getName(), new LineageItem(getOpcode(),
LineageItemUtils.getLineage(ec, input1, _opRows, _opCols, _opDims, _opByRow)));
}
+
+ public CPOperand getOpRows() {
+ return _opRows;
+ }
+
+ public CPOperand getOpCols() {
+ return _opCols;
+ }
+
+ public CPOperand getOpDims() {
+ return _opDims;
+ }
+
+ public CPOperand getOpByRow() {
+ return _opByRow;
+ }
}
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/cp/SpoofCPInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/cp/SpoofCPInstruction.java
index 15d35b7cce..1bbf064d94 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/cp/SpoofCPInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/cp/SpoofCPInstruction.java
@@ -54,6 +54,10 @@ public class SpoofCPInstruction extends ComputationCPInstruction {
_in = in;
}
+ public SpoofOperator getSpoofOperator() {
+ return _op;
+ }
+
public Class<?> getOperatorClass() {
return _class;
}
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/cp/UnaryScalarCPInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/cp/UnaryScalarCPInstruction.java
index b3757cd386..ecced1f13d 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/cp/UnaryScalarCPInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/cp/UnaryScalarCPInstruction.java
@@ -26,10 +26,10 @@ import org.apache.sysds.runtime.controlprogram.context.ExecutionContext;
import org.apache.sysds.runtime.matrix.operators.Operator;
import org.apache.sysds.runtime.matrix.operators.UnaryOperator;
-public class UnaryScalarCPInstruction extends UnaryMatrixCPInstruction {
+public class UnaryScalarCPInstruction extends UnaryCPInstruction {
protected UnaryScalarCPInstruction(Operator op, CPOperand in, CPOperand out, String opcode, String instr) {
- super(op, in, out, opcode, instr);
+ super(CPType.Unary, op, in, out, opcode, instr);
}
@Override
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/fed/AggregateBinaryFEDInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/fed/AggregateBinaryFEDInstruction.java
index abbe17d09e..6128df20e0 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/fed/AggregateBinaryFEDInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/fed/AggregateBinaryFEDInstruction.java
@@ -36,7 +36,9 @@ import org.apache.sysds.runtime.controlprogram.federated.FederationMap;
import org.apache.sysds.runtime.controlprogram.federated.FederationUtils;
import org.apache.sysds.runtime.controlprogram.federated.MatrixLineagePair;
import org.apache.sysds.runtime.instructions.InstructionUtils;
+import org.apache.sysds.runtime.instructions.cp.AggregateBinaryCPInstruction;
import org.apache.sysds.runtime.instructions.cp.CPOperand;
+import org.apache.sysds.runtime.instructions.spark.AggregateBinarySPInstruction;
import org.apache.sysds.runtime.matrix.data.MatrixBlock;
import org.apache.sysds.runtime.matrix.operators.Operator;
@@ -52,7 +54,17 @@ public class AggregateBinaryFEDInstruction extends BinaryFEDInstruction {
String opcode, String istr, FederatedOutput fedOut) {
super(FEDType.AggregateBinary, op, in1, in2, out, opcode, istr, fedOut);
}
-
+
+ public static AggregateBinaryFEDInstruction parseInstruction(AggregateBinaryCPInstruction instr) {
+ return new AggregateBinaryFEDInstruction(instr.getOperator(), instr.input1, instr.input2, instr.output,
+ instr.getOpcode(), instr.getInstructionString(), FederatedOutput.NONE);
+ }
+
+ public static AggregateBinaryFEDInstruction parseInstruction(AggregateBinarySPInstruction instr) {
+ return new AggregateBinaryFEDInstruction(instr.getOperator(), instr.input1, instr.input2, instr.output,
+ instr.getOpcode(), instr.getInstructionString(), FederatedOutput.NONE);
+ }
+
public static AggregateBinaryFEDInstruction parseInstruction(String str) {
String[] parts = InstructionUtils.getInstructionPartsWithValueType(str);
String opcode = parts[0];
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/fed/AggregateTernaryFEDInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/fed/AggregateTernaryFEDInstruction.java
index d98c3c061a..5a54fc6374 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/fed/AggregateTernaryFEDInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/fed/AggregateTernaryFEDInstruction.java
@@ -31,9 +31,11 @@ import org.apache.sysds.runtime.controlprogram.federated.FederatedResponse;
import org.apache.sysds.runtime.controlprogram.federated.FederationUtils;
import org.apache.sysds.runtime.controlprogram.federated.MatrixLineagePair;
import org.apache.sysds.runtime.instructions.InstructionUtils;
+import org.apache.sysds.runtime.instructions.cp.AggregateTernaryCPInstruction;
import org.apache.sysds.runtime.instructions.cp.CPOperand;
import org.apache.sysds.runtime.instructions.cp.DoubleObject;
import org.apache.sysds.runtime.instructions.cp.ScalarObject;
+import org.apache.sysds.runtime.instructions.spark.AggregateTernarySPInstruction;
import org.apache.sysds.runtime.matrix.operators.AggregateTernaryOperator;
import org.apache.sysds.runtime.matrix.operators.AggregateUnaryOperator;
import org.apache.sysds.runtime.matrix.operators.Operator;
@@ -46,6 +48,16 @@ public class AggregateTernaryFEDInstruction extends ComputationFEDInstruction {
super(FEDType.AggregateTernary, op, in1, in2, in3, out, opcode, istr, fedOut);
}
+ public static AggregateTernaryFEDInstruction parseInstruction(AggregateTernaryCPInstruction instr) {
+ return new AggregateTernaryFEDInstruction(instr.getOperator(), instr.input1, instr.input2, instr.input3,
+ instr.output, instr.getOpcode(), instr.getInstructionString(), FederatedOutput.NONE);
+ }
+
+ public static AggregateTernaryFEDInstruction parseInstruction(AggregateTernarySPInstruction instr) {
+ return new AggregateTernaryFEDInstruction(instr.getOperator(), instr.input1, instr.input2, instr.input3,
+ instr.output, instr.getOpcode(), instr.getInstructionString(), FederatedOutput.NONE);
+ }
+
public static AggregateTernaryFEDInstruction parseInstruction(String str) {
String[] parts = InstructionUtils.getInstructionPartsWithValueType(str);
String opcode = parts[0];
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/fed/AggregateUnaryFEDInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/fed/AggregateUnaryFEDInstruction.java
index 1dfc6ed3b9..b4b729a96d 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/fed/AggregateUnaryFEDInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/fed/AggregateUnaryFEDInstruction.java
@@ -33,8 +33,10 @@ import org.apache.sysds.runtime.controlprogram.federated.FederatedResponse;
import org.apache.sysds.runtime.controlprogram.federated.FederationMap;
import org.apache.sysds.runtime.controlprogram.federated.FederationUtils;
import org.apache.sysds.runtime.instructions.InstructionUtils;
+import org.apache.sysds.runtime.instructions.cp.AggregateUnaryCPInstruction;
import org.apache.sysds.runtime.instructions.cp.CPOperand;
import org.apache.sysds.runtime.instructions.cp.ScalarObject;
+import org.apache.sysds.runtime.instructions.spark.AggregateUnarySPInstruction;
import org.apache.sysds.runtime.matrix.operators.AggregateUnaryOperator;
import org.apache.sysds.runtime.matrix.operators.Operator;
import org.apache.sysds.runtime.meta.DataCharacteristics;
@@ -66,6 +68,17 @@ public class AggregateUnaryFEDInstruction extends UnaryFEDInstruction {
super(FEDType.AggregateUnary, op, in1, in2, in3, out, opcode, istr);
}
+ public static AggregateUnaryFEDInstruction parseInstruction(AggregateUnaryCPInstruction instr) {
+ return new AggregateUnaryFEDInstruction(instr.getOperator(), instr.input1, instr.input2, instr.input3,
+ instr.output, instr.getOpcode(), instr.getInstructionString());
+ }
+
+ public static AggregateUnaryFEDInstruction parseInstruction(AggregateUnarySPInstruction instr) {
+ // TODO: during processing the NONE-flag of AggregateUnarySPInstruction (SparkAggType) will be removed, making the instruction unparseable
+ return new AggregateUnaryFEDInstruction(instr.getOperator(), instr.input1, instr.input2, instr.input3,
+ instr.output, instr.getOpcode(), instr.getInstructionString());
+ }
+
public static AggregateUnaryFEDInstruction parseInstruction(String str) {
String[] parts = InstructionUtils.getInstructionPartsWithValueType(str);
String opcode = parts[0];
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/fed/AppendFEDInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/fed/AppendFEDInstruction.java
index 8e672447b4..0b915c7436 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/fed/AppendFEDInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/fed/AppendFEDInstruction.java
@@ -30,11 +30,10 @@ import org.apache.sysds.runtime.controlprogram.federated.FederationMap;
import org.apache.sysds.runtime.controlprogram.federated.FederationUtils;
import org.apache.sysds.runtime.controlprogram.federated.MatrixLineagePair;
import org.apache.sysds.runtime.functionobjects.OffsetColumnIndex;
-import org.apache.sysds.runtime.instructions.Instruction;
import org.apache.sysds.runtime.instructions.InstructionUtils;
-import org.apache.sysds.runtime.instructions.cp.CPInstruction;
+import org.apache.sysds.runtime.instructions.cp.AppendCPInstruction;
import org.apache.sysds.runtime.instructions.cp.CPOperand;
-import org.apache.sysds.runtime.instructions.spark.SPInstruction;
+import org.apache.sysds.runtime.instructions.spark.AppendSPInstruction;
import org.apache.sysds.runtime.matrix.operators.Operator;
import org.apache.sysds.runtime.matrix.operators.ReorgOperator;
import org.apache.sysds.runtime.meta.DataCharacteristics;
@@ -56,22 +55,15 @@ public class AppendFEDInstruction extends BinaryFEDInstruction {
_cbind = cbind;
}
- public static AppendFEDInstruction parseInstruction(Instruction inst){
- if ( inst instanceof CPInstruction || inst instanceof SPInstruction ){
- String instStr = inst.getInstructionString();
- String[] parts = InstructionUtils.getInstructionPartsWithValueType(instStr);
- InstructionUtils.checkNumFields(parts, 6, 5, 4);
-
- String opcode = parts[0];
- CPOperand in1 = new CPOperand(parts[1]);
- CPOperand in2 = new CPOperand(parts[2]);
- CPOperand out = new CPOperand(parts[parts.length - 2]);
- boolean cbind = Boolean.parseBoolean(parts[parts.length - 1]);
+ public static AppendFEDInstruction parseInstruction(AppendCPInstruction instr) {
+ return new AppendFEDInstruction(instr.getOperator(), instr.input1, instr.input2, instr.output,
+ instr.getAppendType().equals(AppendCPInstruction.AppendType.CBIND), instr.getOpcode(),
+ instr.getInstructionString(), FederatedOutput.NONE);
+ }
- Operator op = new ReorgOperator(OffsetColumnIndex.getOffsetColumnIndexFnObject(-1));
- return new AppendFEDInstruction(op, in1, in2, out, cbind, opcode, instStr);
- }
- else return parseInstruction(inst.getInstructionString());
+ public static AppendFEDInstruction parseInstruction(AppendSPInstruction instr) {
+ return new AppendFEDInstruction(instr.getOperator(), instr.input1, instr.input2, instr.output, instr.getCBind(),
+ instr.getOpcode(), instr.getInstructionString(), FederatedOutput.NONE);
}
public static AppendFEDInstruction parseInstruction(String str) {
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/fed/BinaryFEDInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/fed/BinaryFEDInstruction.java
index 5e1d35fb0f..20b378ac8e 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/fed/BinaryFEDInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/fed/BinaryFEDInstruction.java
@@ -104,7 +104,7 @@ public abstract class BinaryFEDInstruction extends ComputationFEDInstruction {
" and " + in2.getName() + " must produce a matrix, which " + out.getName() + " is not");
}
- private static String rewriteSparkInstructionToCP(String inst_str) {
+ protected static String rewriteSparkInstructionToCP(String inst_str) {
// rewrite the spark instruction to a cp instruction
inst_str = inst_str.replace(ExecType.SPARK.name(), ExecType.CP.name());
inst_str = inst_str.replace(Lop.OPERAND_DELIMITOR + "map", Lop.OPERAND_DELIMITOR);
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/fed/BinaryMatrixMatrixFEDInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/fed/BinaryMatrixMatrixFEDInstruction.java
index 529233ac24..250b5d193a 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/fed/BinaryMatrixMatrixFEDInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/fed/BinaryMatrixMatrixFEDInstruction.java
@@ -28,7 +28,10 @@ import org.apache.sysds.runtime.controlprogram.federated.FederatedRequest;
import org.apache.sysds.runtime.controlprogram.federated.FederationMap;
import org.apache.sysds.runtime.controlprogram.federated.FederationUtils;
import org.apache.sysds.runtime.controlprogram.federated.MatrixLineagePair;
+import org.apache.sysds.runtime.instructions.InstructionUtils;
+import org.apache.sysds.runtime.instructions.cp.BinaryMatrixMatrixCPInstruction;
import org.apache.sysds.runtime.instructions.cp.CPOperand;
+import org.apache.sysds.runtime.instructions.spark.BinaryMatrixMatrixSPInstruction;
import org.apache.sysds.runtime.matrix.operators.BinaryOperator;
import org.apache.sysds.runtime.matrix.operators.Operator;
@@ -39,6 +42,18 @@ public class BinaryMatrixMatrixFEDInstruction extends BinaryFEDInstruction
super(FEDType.Binary, op, in1, in2, out, opcode, istr, fedOut);
}
+ public static BinaryMatrixMatrixFEDInstruction parseInstruction(BinaryMatrixMatrixCPInstruction instr) {
+ return new BinaryMatrixMatrixFEDInstruction(instr.getOperator(), instr.input1, instr.input2, instr.output,
+ instr.getOpcode(), instr.getInstructionString(), FederatedOutput.NONE);
+ }
+
+ public static BinaryMatrixMatrixFEDInstruction parseInstruction(BinaryMatrixMatrixSPInstruction instr) {
+ String instrStr = rewriteSparkInstructionToCP(instr.getInstructionString());
+ String opcode = InstructionUtils.getInstructionPartsWithValueType(instrStr)[0];
+ return new BinaryMatrixMatrixFEDInstruction(instr.getOperator(), instr.input1, instr.input2, instr.output,
+ opcode, instrStr, FederatedOutput.NONE);
+ }
+
@Override
public void processInstruction(ExecutionContext ec) {
MatrixLineagePair mo1 = ec.getMatrixLineagePair(input1);
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/fed/BinaryMatrixScalarFEDInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/fed/BinaryMatrixScalarFEDInstruction.java
index 5edcbd3b5c..9330f6dfd2 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/fed/BinaryMatrixScalarFEDInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/fed/BinaryMatrixScalarFEDInstruction.java
@@ -23,7 +23,10 @@ import org.apache.sysds.runtime.controlprogram.caching.MatrixObject;
import org.apache.sysds.runtime.controlprogram.context.ExecutionContext;
import org.apache.sysds.runtime.controlprogram.federated.FederatedRequest;
import org.apache.sysds.runtime.controlprogram.federated.FederationUtils;
+import org.apache.sysds.runtime.instructions.InstructionUtils;
+import org.apache.sysds.runtime.instructions.cp.BinaryMatrixScalarCPInstruction;
import org.apache.sysds.runtime.instructions.cp.CPOperand;
+import org.apache.sysds.runtime.instructions.spark.BinaryMatrixScalarSPInstruction;
import org.apache.sysds.runtime.matrix.operators.Operator;
public class BinaryMatrixScalarFEDInstruction extends BinaryFEDInstruction
@@ -33,6 +36,18 @@ public class BinaryMatrixScalarFEDInstruction extends BinaryFEDInstruction
super(FEDType.Binary, op, in1, in2, out, opcode, istr, fedOut);
}
+ public static BinaryMatrixScalarFEDInstruction parseInstruction(BinaryMatrixScalarCPInstruction instr) {
+ return new BinaryMatrixScalarFEDInstruction(instr.getOperator(), instr.input1, instr.input2, instr.output,
+ instr.getOpcode(), instr.getInstructionString(), FederatedOutput.NONE);
+ }
+
+ public static BinaryMatrixScalarFEDInstruction parseInstruction(BinaryMatrixScalarSPInstruction instr) {
+ String instrStr = rewriteSparkInstructionToCP(instr.getInstructionString());
+ String opcode = InstructionUtils.getInstructionPartsWithValueType(instrStr)[0];
+ return new BinaryMatrixScalarFEDInstruction(instr.getOperator(), instr.input1, instr.input2, instr.output,
+ opcode, instrStr, FederatedOutput.NONE);
+ }
+
@Override
public void processInstruction(ExecutionContext ec) {
CPOperand matrix = input1.isMatrix() ? input1 : input2;
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/fed/CastFEDInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/fed/CastFEDInstruction.java
index 5c29d13128..df2fe11e12 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/fed/CastFEDInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/fed/CastFEDInstruction.java
@@ -39,6 +39,7 @@ import org.apache.sysds.runtime.controlprogram.federated.FederationMap;
import org.apache.sysds.runtime.controlprogram.federated.FederationUtils;
import org.apache.sysds.runtime.instructions.InstructionUtils;
import org.apache.sysds.runtime.instructions.cp.CPOperand;
+import org.apache.sysds.runtime.instructions.spark.CastSPInstruction;
import org.apache.sysds.runtime.matrix.operators.Operator;
import org.apache.sysds.runtime.meta.MatrixCharacteristics;
@@ -48,6 +49,11 @@ public class CastFEDInstruction extends UnaryFEDInstruction {
super(FEDInstruction.FEDType.Cast, op, in, out, opcode, istr);
}
+ public static CastFEDInstruction parseInstruction(CastSPInstruction spInstruction) {
+ return new CastFEDInstruction(spInstruction.getOperator(), spInstruction.input1, spInstruction.output,
+ spInstruction.getOpcode(), spInstruction.getInstructionString());
+ }
+
public static CastFEDInstruction parseInstruction ( String str ) {
String[] parts = InstructionUtils.getInstructionPartsWithValueType(str);
InstructionUtils.checkNumFields(parts, 2);
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/fed/CentralMomentFEDInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/fed/CentralMomentFEDInstruction.java
index 7180fec353..0d6ee0ded8 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/fed/CentralMomentFEDInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/fed/CentralMomentFEDInstruction.java
@@ -32,7 +32,6 @@ import org.apache.sysds.runtime.controlprogram.federated.FederatedResponse;
import org.apache.sysds.runtime.controlprogram.federated.FederatedUDF;
import org.apache.sysds.runtime.controlprogram.federated.FederationMap;
import org.apache.sysds.runtime.controlprogram.federated.FederationUtils;
-import org.apache.sysds.runtime.instructions.Instruction;
import org.apache.sysds.runtime.instructions.InstructionUtils;
import org.apache.sysds.runtime.instructions.cp.CM_COV_Object;
import org.apache.sysds.runtime.instructions.cp.CPOperand;
@@ -40,7 +39,7 @@ import org.apache.sysds.runtime.instructions.cp.CentralMomentCPInstruction;
import org.apache.sysds.runtime.instructions.cp.Data;
import org.apache.sysds.runtime.instructions.cp.DoubleObject;
import org.apache.sysds.runtime.instructions.cp.ScalarObject;
-import org.apache.sysds.runtime.instructions.spark.SPInstruction;
+import org.apache.sysds.runtime.instructions.spark.CentralMomentSPInstruction;
import org.apache.sysds.runtime.lineage.LineageItem;
import org.apache.sysds.runtime.matrix.data.MatrixBlock;
import org.apache.sysds.runtime.matrix.operators.CMOperator;
@@ -64,18 +63,13 @@ public class CentralMomentFEDInstruction extends AggregateUnaryFEDInstruction {
return fedInst;
}
- public static CentralMomentFEDInstruction parseInstruction(Instruction inst){
- if ( inst instanceof CentralMomentCPInstruction)
- return parseInstruction((CentralMomentCPInstruction) inst);
- else if ( inst instanceof SPInstruction )
- return parseInstruction(CentralMomentCPInstruction.parseInstruction(inst.getInstructionString()));
- else
- return parseInstruction(inst.getInstructionString());
+ public static CentralMomentFEDInstruction parseInstruction(CentralMomentCPInstruction inst) {
+ return new CentralMomentFEDInstruction(inst.getOperator(), inst.input1, inst.input2, inst.input3, inst.output,
+ inst.getOpcode(), inst.getInstructionString());
}
- public static CentralMomentFEDInstruction parseInstruction(CentralMomentCPInstruction inst) {
- return new CentralMomentFEDInstruction(inst.getOperator(),
- inst.input1, inst.input2, inst.input3, inst.output,
+ public static CentralMomentFEDInstruction parseInstruction(CentralMomentSPInstruction inst) {
+ return new CentralMomentFEDInstruction(inst.getOperator(), inst.input1, inst.input2, inst.input3, inst.output,
inst.getOpcode(), inst.getInstructionString());
}
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/fed/CovarianceFEDInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/fed/CovarianceFEDInstruction.java
index 2c41ee2e62..9e79b024a2 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/fed/CovarianceFEDInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/fed/CovarianceFEDInstruction.java
@@ -37,7 +37,6 @@ import org.apache.sysds.runtime.controlprogram.federated.FederatedUDF;
import org.apache.sysds.runtime.controlprogram.federated.FederationMap;
import org.apache.sysds.runtime.controlprogram.federated.FederationUtils;
import org.apache.sysds.runtime.controlprogram.federated.MatrixLineagePair;
-import org.apache.sysds.runtime.instructions.Instruction;
import org.apache.sysds.runtime.instructions.InstructionUtils;
import org.apache.sysds.runtime.instructions.cp.CM_COV_Object;
import org.apache.sysds.runtime.instructions.cp.CPOperand;
@@ -45,7 +44,7 @@ import org.apache.sysds.runtime.instructions.cp.CovarianceCPInstruction;
import org.apache.sysds.runtime.instructions.cp.Data;
import org.apache.sysds.runtime.instructions.cp.DoubleObject;
import org.apache.sysds.runtime.instructions.cp.ScalarObject;
-import org.apache.sysds.runtime.instructions.spark.SPInstruction;
+import org.apache.sysds.runtime.instructions.spark.CovarianceSPInstruction;
import org.apache.sysds.runtime.lineage.LineageItem;
import org.apache.sysds.runtime.matrix.data.MatrixBlock;
import org.apache.sysds.runtime.matrix.operators.COVOperator;
@@ -68,21 +67,16 @@ public class CovarianceFEDInstruction extends BinaryFEDInstruction {
return fedInst;
}
- public static CovarianceFEDInstruction parseInstruction(Instruction inst){
- if ( inst instanceof CovarianceCPInstruction )
- return parseInstruction((CovarianceCPInstruction) inst);
- else if ( inst instanceof SPInstruction )
- return parseInstruction(CovarianceCPInstruction.parseInstruction(inst.getInstructionString()));
- else
- return parseInstruction(inst.getInstructionString());
+ public static CovarianceFEDInstruction parseInstruction(CovarianceCPInstruction inst) {
+ return new CovarianceFEDInstruction(inst.getOperator(), inst.input1, inst.input2, inst.input3, inst.output,
+ inst.getOpcode(), inst.getInstructionString());
}
- public static CovarianceFEDInstruction parseInstruction(CovarianceCPInstruction inst) {
- return new CovarianceFEDInstruction(inst.getOperator(),
- inst.input1, inst.input2, inst.input3, inst.output,
+ public static CovarianceFEDInstruction parseInstruction(CovarianceSPInstruction inst) {
+ return new CovarianceFEDInstruction(inst.getOperator(), inst.input1, inst.input2, inst.input3, inst.output,
inst.getOpcode(), inst.getInstructionString());
}
-
+
@Override
public void processInstruction(ExecutionContext ec) {
MatrixObject mo1 = ec.getMatrixObject(input1);
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/fed/CtableFEDInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/fed/CtableFEDInstruction.java
index 3b5274e8f5..3f87668492 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/fed/CtableFEDInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/fed/CtableFEDInstruction.java
@@ -45,8 +45,10 @@ import org.apache.sysds.runtime.controlprogram.federated.MatrixLineagePair;
import org.apache.sysds.runtime.instructions.Instruction;
import org.apache.sysds.runtime.instructions.InstructionUtils;
import org.apache.sysds.runtime.instructions.cp.CPOperand;
+import org.apache.sysds.runtime.instructions.cp.CtableCPInstruction;
import org.apache.sysds.runtime.instructions.cp.Data;
import org.apache.sysds.runtime.instructions.cp.ScalarObject;
+import org.apache.sysds.runtime.instructions.spark.CtableSPInstruction;
import org.apache.sysds.runtime.lineage.LineageItem;
import org.apache.sysds.runtime.matrix.data.MatrixBlock;
import org.apache.sysds.runtime.matrix.operators.BinaryOperator;
@@ -57,13 +59,25 @@ public class CtableFEDInstruction extends ComputationFEDInstruction {
//private final boolean _isExpand;
//private final boolean _ignoreZeros;
- private CtableFEDInstruction(CPOperand in1, CPOperand in2, CPOperand in3, CPOperand out, String outputDim1, boolean dim1Literal, String outputDim2, boolean dim2Literal, boolean isExpand,
- boolean ignoreZeros, String opcode, String istr) {
+ private CtableFEDInstruction(CPOperand in1, CPOperand in2, CPOperand in3, CPOperand out, CPOperand outDim1,
+ CPOperand outDim2, boolean isExpand, boolean ignoreZeros, String opcode, String istr) {
super(FEDType.Ctable, null, in1, in2, in3, out, opcode, istr);
- _outDim1 = new CPOperand(outputDim1, ValueType.FP64, DataType.SCALAR, dim1Literal);
- _outDim2 = new CPOperand(outputDim2, ValueType.FP64, DataType.SCALAR, dim2Literal);
- //_isExpand = isExpand;
- //_ignoreZeros = ignoreZeros;
+ _outDim1 = outDim1;
+ _outDim2 = outDim2;
+ // _isExpand = isExpand;
+ // _ignoreZeros = ignoreZeros;
+ }
+
+ public static CtableFEDInstruction parseInstruction(CtableCPInstruction instr) {
+ return new CtableFEDInstruction(instr.input1, instr.input2, instr.input3, instr.output, instr.getOutDim1(),
+ instr.getOutDim2(), instr.getIsExpand(), instr.getIgnoreZeros(), instr.getOpcode(),
+ instr.getInstructionString());
+ }
+
+ public static CtableFEDInstruction parseInstruction(CtableSPInstruction instr) {
+ return new CtableFEDInstruction(instr.input1, instr.input2, instr.input3, instr.output, instr.getOutDim1(),
+ instr.getOutDim2(), instr.getIsExpand(), instr.getIgnoreZeros(), instr.getOpcode(),
+ instr.getInstructionString());
}
public static CtableFEDInstruction parseInstruction(String inst) {
@@ -88,11 +102,14 @@ public class CtableFEDInstruction extends ComputationFEDInstruction {
CPOperand out = new CPOperand(parts[6]);
boolean ignoreZeros = Boolean.parseBoolean(parts[7]);
-
+
+ boolean dim1Literal = Boolean.parseBoolean(dim1Fields[1]);
+ CPOperand outDim1 = new CPOperand(dim1Fields[0], ValueType.FP64, DataType.SCALAR, dim1Literal);
+ boolean dim2Literal = Boolean.parseBoolean(dim2Fields[1]);
+ CPOperand outDim2 = new CPOperand(dim2Fields[0], ValueType.FP64, DataType.SCALAR, dim2Literal);
// ctable does not require any operator, so we simply pass-in a dummy operator with null functionobject
return new CtableFEDInstruction(in1,
- in2, in3, out, dim1Fields[0], Boolean.parseBoolean(dim1Fields[1]),
- dim2Fields[0], Boolean.parseBoolean(dim2Fields[1]), false, ignoreZeros, opcode, inst);
+ in2, in3, out, outDim1, outDim2, false, ignoreZeros, opcode, inst);
}
@Override
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/fed/CumulativeOffsetFEDInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/fed/CumulativeOffsetFEDInstruction.java
index 0112c63859..58b58c58f5 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/fed/CumulativeOffsetFEDInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/fed/CumulativeOffsetFEDInstruction.java
@@ -34,6 +34,7 @@ import org.apache.sysds.runtime.controlprogram.federated.MatrixLineagePair;
import org.apache.sysds.runtime.functionobjects.Builtin;
import org.apache.sysds.runtime.instructions.InstructionUtils;
import org.apache.sysds.runtime.instructions.cp.CPOperand;
+import org.apache.sysds.runtime.instructions.spark.CumulativeOffsetSPInstruction;
import org.apache.sysds.runtime.matrix.data.MatrixBlock;
import org.apache.sysds.runtime.matrix.operators.Operator;
import org.apache.sysds.runtime.matrix.operators.UnaryOperator;
@@ -59,6 +60,11 @@ public class CumulativeOffsetFEDInstruction extends BinaryFEDInstruction
_uop = new UnaryOperator(Builtin.getBuiltinFnObject("ucummax"));
}
+ public static CumulativeOffsetFEDInstruction parseInstruction(CumulativeOffsetSPInstruction instr) {
+ return new CumulativeOffsetFEDInstruction(instr.getOperator(), instr.input1, instr.input2, instr.output,
+ instr.getInitValue(), instr.getBroadcast(), instr.getOpcode(), instr.getInstructionString());
+ }
+
public static CumulativeOffsetFEDInstruction parseInstruction ( String str ) {
String[] parts = InstructionUtils.getInstructionPartsWithValueType( str );
InstructionUtils.checkNumFields(parts, 5);
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/fed/FEDInstructionUtils.java b/src/main/java/org/apache/sysds/runtime/instructions/fed/FEDInstructionUtils.java
index 5e3b7a9f8b..eb93ff9d59 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/fed/FEDInstructionUtils.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/fed/FEDInstructionUtils.java
@@ -29,15 +29,18 @@ import org.apache.sysds.runtime.codegen.SpoofRowwise;
import org.apache.sysds.runtime.controlprogram.caching.CacheableData;
import org.apache.sysds.runtime.controlprogram.caching.FrameObject;
import org.apache.sysds.runtime.controlprogram.caching.MatrixObject;
-import org.apache.sysds.runtime.controlprogram.caching.TensorObject;
import org.apache.sysds.runtime.controlprogram.context.ExecutionContext;
import org.apache.sysds.runtime.instructions.Instruction;
import org.apache.sysds.runtime.instructions.InstructionUtils;
import org.apache.sysds.runtime.instructions.cp.AggregateBinaryCPInstruction;
import org.apache.sysds.runtime.instructions.cp.AggregateTernaryCPInstruction;
import org.apache.sysds.runtime.instructions.cp.AggregateUnaryCPInstruction;
+import org.apache.sysds.runtime.instructions.cp.AppendCPInstruction;
import org.apache.sysds.runtime.instructions.cp.BinaryCPInstruction;
+import org.apache.sysds.runtime.instructions.cp.BinaryMatrixMatrixCPInstruction;
+import org.apache.sysds.runtime.instructions.cp.BinaryMatrixScalarCPInstruction;
import org.apache.sysds.runtime.instructions.cp.CentralMomentCPInstruction;
+import org.apache.sysds.runtime.instructions.cp.CovarianceCPInstruction;
import org.apache.sysds.runtime.instructions.cp.CtableCPInstruction;
import org.apache.sysds.runtime.instructions.cp.Data;
import org.apache.sysds.runtime.instructions.cp.IndexingCPInstruction;
@@ -45,8 +48,11 @@ import org.apache.sysds.runtime.instructions.cp.MMChainCPInstruction;
import org.apache.sysds.runtime.instructions.cp.MMTSJCPInstruction;
import org.apache.sysds.runtime.instructions.cp.MultiReturnParameterizedBuiltinCPInstruction;
import org.apache.sysds.runtime.instructions.cp.ParameterizedBuiltinCPInstruction;
+import org.apache.sysds.runtime.instructions.cp.QuantilePickCPInstruction;
+import org.apache.sysds.runtime.instructions.cp.QuantileSortCPInstruction;
import org.apache.sysds.runtime.instructions.cp.QuaternaryCPInstruction;
import org.apache.sysds.runtime.instructions.cp.ReorgCPInstruction;
+import org.apache.sysds.runtime.instructions.cp.ReshapeCPInstruction;
import org.apache.sysds.runtime.instructions.cp.SpoofCPInstruction;
import org.apache.sysds.runtime.instructions.cp.TernaryCPInstruction;
import org.apache.sysds.runtime.instructions.cp.TernaryFrameScalarCPInstruction;
@@ -54,21 +60,20 @@ import org.apache.sysds.runtime.instructions.cp.UnaryCPInstruction;
import org.apache.sysds.runtime.instructions.cp.UnaryMatrixCPInstruction;
import org.apache.sysds.runtime.instructions.cp.VariableCPInstruction;
import org.apache.sysds.runtime.instructions.cp.VariableCPInstruction.VariableOperationCode;
-import org.apache.sysds.runtime.instructions.fed.FEDInstruction.FederatedOutput;
+import org.apache.sysds.runtime.instructions.spark.AggregateBinarySPInstruction;
import org.apache.sysds.runtime.instructions.spark.AggregateTernarySPInstruction;
import org.apache.sysds.runtime.instructions.spark.AggregateUnarySPInstruction;
import org.apache.sysds.runtime.instructions.spark.AppendGAlignedSPInstruction;
import org.apache.sysds.runtime.instructions.spark.AppendGSPInstruction;
import org.apache.sysds.runtime.instructions.spark.AppendMSPInstruction;
import org.apache.sysds.runtime.instructions.spark.AppendRSPInstruction;
-import org.apache.sysds.runtime.instructions.spark.BinaryMatrixBVectorSPInstruction;
+import org.apache.sysds.runtime.instructions.spark.AppendSPInstruction;
import org.apache.sysds.runtime.instructions.spark.BinaryMatrixMatrixSPInstruction;
import org.apache.sysds.runtime.instructions.spark.BinaryMatrixScalarSPInstruction;
import org.apache.sysds.runtime.instructions.spark.BinarySPInstruction;
-import org.apache.sysds.runtime.instructions.spark.BinaryTensorTensorBroadcastSPInstruction;
-import org.apache.sysds.runtime.instructions.spark.BinaryTensorTensorSPInstruction;
import org.apache.sysds.runtime.instructions.spark.CastSPInstruction;
import org.apache.sysds.runtime.instructions.spark.CentralMomentSPInstruction;
+import org.apache.sysds.runtime.instructions.spark.CovarianceSPInstruction;
import org.apache.sysds.runtime.instructions.spark.CpmmSPInstruction;
import org.apache.sysds.runtime.instructions.spark.CtableSPInstruction;
import org.apache.sysds.runtime.instructions.spark.CumulativeOffsetSPInstruction;
@@ -91,7 +96,7 @@ import org.apache.sysds.runtime.instructions.spark.WriteSPInstruction;
public class FEDInstructionUtils {
- private static String[] PARAM_BUILTINS = new String[]{
+ private static final String[] PARAM_BUILTINS = new String[]{
"replace", "rmempty", "lowertri", "uppertri", "transformdecode", "transformapply", "tokenize"};
public static boolean noFedRuntimeConversion = false;
@@ -120,8 +125,7 @@ public class FEDInstructionUtils {
if ( (mo1.isFederated(FType.ROW) && mo1.isFederatedExcept(FType.BROADCAST))
|| (mo2.isFederated(FType.ROW) && mo2.isFederatedExcept(FType.BROADCAST))
|| (mo1.isFederated(FType.COL) && mo1.isFederatedExcept(FType.BROADCAST))) {
- fedinst = AggregateBinaryFEDInstruction.parseInstruction(
- InstructionUtils.concatOperands(inst.getInstructionString(), FederatedOutput.NONE.name()));
+ fedinst = AggregateBinaryFEDInstruction.parseInstruction(instruction);
}
}
}
@@ -129,14 +133,14 @@ public class FEDInstructionUtils {
MMChainCPInstruction linst = (MMChainCPInstruction) inst;
MatrixObject mo = ec.getMatrixObject(linst.input1);
if( mo.isFederated(FType.ROW) )
- fedinst = MMChainFEDInstruction.parseInstruction(linst.getInstructionString());
+ fedinst = MMChainFEDInstruction.parseInstruction(linst);
}
else if( inst instanceof MMTSJCPInstruction ) {
MMTSJCPInstruction linst = (MMTSJCPInstruction) inst;
MatrixObject mo = ec.getMatrixObject(linst.input1);
if( (mo.isFederated(FType.ROW) && mo.isFederatedExcept(FType.BROADCAST) && linst.getMMTSJType().isLeft()) ||
(mo.isFederated(FType.COL) && mo.isFederatedExcept(FType.BROADCAST) && linst.getMMTSJType().isRight()))
- fedinst = TsmmFEDInstruction.parseInstruction(linst.getInstructionString());
+ fedinst = TsmmFEDInstruction.parseInstruction(linst);
}
else if (inst instanceof UnaryCPInstruction && ! (inst instanceof IndexingCPInstruction)) {
UnaryCPInstruction instruction = (UnaryCPInstruction) inst;
@@ -147,62 +151,65 @@ public class FEDInstructionUtils {
if((mo instanceof MatrixObject || mo instanceof FrameObject)
&& mo.isFederatedExcept(FType.BROADCAST) )
- fedinst = ReorgFEDInstruction.parseInstruction(
- InstructionUtils.concatOperands(rinst.getInstructionString(),FederatedOutput.NONE.name()));
+ fedinst = ReorgFEDInstruction.parseInstruction(rinst);
}
else if(instruction.input1 != null && instruction.input1.isMatrix()
&& ec.containsVariable(instruction.input1)) {
MatrixObject mo1 = ec.getMatrixObject(instruction.input1);
if( mo1.isFederatedExcept(FType.BROADCAST) ) {
- if(instruction.getOpcode().equalsIgnoreCase("cm"))
- fedinst = CentralMomentFEDInstruction.parseInstruction(inst);
- else if(inst.getOpcode().equalsIgnoreCase("qsort")) {
+ if(instruction instanceof CentralMomentCPInstruction)
+ fedinst = CentralMomentFEDInstruction.parseInstruction((CentralMomentCPInstruction) inst);
+ else if(inst instanceof QuantileSortCPInstruction) {
if(mo1.isFederated(FType.ROW) || mo1.getFedMapping().getFederatedRanges().length == 1 && mo1.isFederated(FType.COL))
- fedinst = QuantileSortFEDInstruction.parseInstruction(inst.getInstructionString(), false);
+ fedinst = QuantileSortFEDInstruction.parseInstruction((QuantileSortCPInstruction) inst);
}
- else if(inst.getOpcode().equalsIgnoreCase("rshape"))
- fedinst = ReshapeFEDInstruction.parseInstruction(inst.getInstructionString());
+ else if(inst instanceof ReshapeCPInstruction)
+ fedinst = ReshapeFEDInstruction.parseInstruction((ReshapeCPInstruction) inst);
else if(inst instanceof AggregateUnaryCPInstruction &&
((AggregateUnaryCPInstruction) instruction).getAUType() == AggregateUnaryCPInstruction.AUType.DEFAULT)
- fedinst = AggregateUnaryFEDInstruction.parseInstruction(
- InstructionUtils.concatOperands(inst.getInstructionString(),FederatedOutput.NONE.name()));
+ fedinst = AggregateUnaryFEDInstruction.parseInstruction((AggregateUnaryCPInstruction) inst);
else if(inst instanceof UnaryMatrixCPInstruction) {
if(UnaryMatrixFEDInstruction.isValidOpcode(inst.getOpcode()) &&
!(inst.getOpcode().equalsIgnoreCase("ucumk+*") && mo1.isFederated(FType.COL)))
- fedinst = UnaryMatrixFEDInstruction.parseInstruction(inst.getInstructionString());
+ fedinst = UnaryMatrixFEDInstruction.parseInstruction((UnaryMatrixCPInstruction) inst);
}
}
}
}
else if (inst instanceof BinaryCPInstruction) {
BinaryCPInstruction instruction = (BinaryCPInstruction) inst;
- if( (instruction.input1.isMatrix() && ec.getMatrixObject(instruction.input1).isFederatedExcept(FType.BROADCAST))
- || (instruction.input2.isMatrix() && ec.getMatrixObject(instruction.input2).isFederatedExcept(FType.BROADCAST))) {
- if(instruction.getOpcode().equals("append") )
- fedinst = AppendFEDInstruction.parseInstruction(inst);
- else if(instruction.getOpcode().equals("qpick"))
- fedinst = QuantilePickFEDInstruction.parseInstruction(inst);
- else if("cov".equals(instruction.getOpcode()) && (ec.getMatrixObject(instruction.input1).isFederated(FType.ROW) ||
- ec.getMatrixObject(instruction.input2).isFederated(FType.ROW)))
- fedinst = CovarianceFEDInstruction.parseInstruction(inst);
- else
- fedinst = BinaryFEDInstruction.parseInstruction(
- InstructionUtils.concatOperands(inst.getInstructionString(),FederatedOutput.NONE.name()));
+ if((instruction.input1.isMatrix() &&
+ ec.getMatrixObject(instruction.input1).isFederatedExcept(FType.BROADCAST)) ||
+ (instruction.input2 != null && instruction.input2.isMatrix() &&
+ ec.getMatrixObject(instruction.input2).isFederatedExcept(FType.BROADCAST))) {
+ if(instruction instanceof AppendCPInstruction)
+ fedinst = AppendFEDInstruction.parseInstruction((AppendCPInstruction) inst);
+ else if(instruction instanceof QuantilePickCPInstruction)
+ fedinst = QuantilePickFEDInstruction.parseInstruction((QuantilePickCPInstruction) inst);
+ else if(instruction instanceof CovarianceCPInstruction &&
+ (ec.getMatrixObject(instruction.input1).isFederated(FType.ROW) ||
+ ec.getMatrixObject(instruction.input2).isFederated(FType.ROW)))
+ fedinst = CovarianceFEDInstruction.parseInstruction((CovarianceCPInstruction) inst);
+ else if(instruction instanceof BinaryMatrixMatrixCPInstruction)
+ fedinst = BinaryMatrixMatrixFEDInstruction
+ .parseInstruction((BinaryMatrixMatrixCPInstruction) inst);
+ else if(instruction instanceof BinaryMatrixScalarCPInstruction)
+ fedinst = BinaryMatrixScalarFEDInstruction
+ .parseInstruction((BinaryMatrixScalarCPInstruction) inst);
}
}
else if( inst instanceof ParameterizedBuiltinCPInstruction ) {
ParameterizedBuiltinCPInstruction pinst = (ParameterizedBuiltinCPInstruction) inst;
if( ArrayUtils.contains(PARAM_BUILTINS, pinst.getOpcode()) && pinst.getTarget(ec).isFederatedExcept(FType.BROADCAST) )
- fedinst = ParameterizedBuiltinFEDInstruction.parseInstruction(pinst.getInstructionString());
+ fedinst = ParameterizedBuiltinFEDInstruction.parseInstruction(pinst);
}
else if (inst instanceof MultiReturnParameterizedBuiltinCPInstruction) {
MultiReturnParameterizedBuiltinCPInstruction minst = (MultiReturnParameterizedBuiltinCPInstruction) inst;
if(minst.getOpcode().equals("transformencode") && minst.input1.isFrame()) {
CacheableData<?> fo = ec.getCacheableData(minst.input1);
if(fo.isFederatedExcept(FType.BROADCAST)) {
- fedinst = MultiReturnParameterizedBuiltinFEDInstruction
- .parseInstruction(minst.getInstructionString());
+ fedinst = MultiReturnParameterizedBuiltinFEDInstruction.parseInstruction(minst);
}
}
}
@@ -211,7 +218,7 @@ public class FEDInstructionUtils {
IndexingCPInstruction minst = (IndexingCPInstruction) inst;
if((minst.input1.isMatrix() || minst.input1.isFrame())
&& ec.getCacheableData(minst.input1).isFederatedExcept(FType.BROADCAST)) {
- fedinst = IndexingFEDInstruction.parseInstruction(minst.getInstructionString());
+ fedinst = IndexingFEDInstruction.parseInstruction(minst);
}
}
else if(inst instanceof TernaryCPInstruction) {
@@ -221,13 +228,12 @@ public class FEDInstructionUtils {
long margin = ec.getScalarInput(tinst.input3).getLongValue();
FrameObject fo = ec.getFrameObject(tinst.input1);
if(margin == 0 || (fo.isFederated(FType.ROW) && margin == 1) || (fo.isFederated(FType.COL) && margin == 2))
- fedinst = TernaryFrameScalarFEDInstruction
- .parseInstruction(InstructionUtils.concatOperands(inst.getInstructionString(),FederatedOutput.NONE.name()));
+ fedinst = TernaryFrameScalarFEDInstruction.parseInstruction((TernaryFrameScalarCPInstruction) inst);
}
else if((tinst.input1.isMatrix() && ec.getCacheableData(tinst.input1).isFederatedExcept(FType.BROADCAST))
|| (tinst.input2.isMatrix() && ec.getCacheableData(tinst.input2).isFederatedExcept(FType.BROADCAST))
|| (tinst.input3.isMatrix() && ec.getCacheableData(tinst.input3).isFederatedExcept(FType.BROADCAST))) {
- fedinst = TernaryFEDInstruction.parseInstruction(tinst.getInstructionString());
+ fedinst = TernaryFEDInstruction.parseInstruction(tinst);
}
}
else if(inst instanceof VariableCPInstruction ){
@@ -252,14 +258,14 @@ public class FEDInstructionUtils {
AggregateTernaryCPInstruction ins = (AggregateTernaryCPInstruction) inst;
if(ins.input1.isMatrix() && ec.getCacheableData(ins.input1).isFederatedExcept(FType.BROADCAST)
&& ins.input2.isMatrix() && ec.getCacheableData(ins.input2).isFederatedExcept(FType.BROADCAST)) {
- fedinst = AggregateTernaryFEDInstruction.parseInstruction(ins.getInstructionString());
+ fedinst = AggregateTernaryFEDInstruction.parseInstruction(ins);
}
}
else if(inst instanceof QuaternaryCPInstruction) {
QuaternaryCPInstruction instruction = (QuaternaryCPInstruction) inst;
Data data = ec.getVariable(instruction.input1);
if(data instanceof MatrixObject && ((MatrixObject) data).isFederatedExcept(FType.BROADCAST))
- fedinst = QuaternaryFEDInstruction.parseInstruction(instruction.getInstructionString());
+ fedinst = QuaternaryFEDInstruction.parseInstruction(instruction);
}
else if(inst instanceof SpoofCPInstruction) {
SpoofCPInstruction ins = (SpoofCPInstruction) inst;
@@ -267,7 +273,7 @@ public class FEDInstructionUtils {
if(((scla == SpoofCellwise.class || scla == SpoofMultiAggregate.class || scla == SpoofOuterProduct.class)
&& SpoofFEDInstruction.isFederated(ec, ins.getInputs(), scla))
|| (scla == SpoofRowwise.class && SpoofFEDInstruction.isFederated(ec, FType.ROW, ins.getInputs(), scla))) {
- fedinst = SpoofFEDInstruction.parseInstruction(ins.getInstructionString());
+ fedinst = SpoofFEDInstruction.parseInstruction(ins);
}
}
else if(inst instanceof CtableCPInstruction) {
@@ -276,7 +282,7 @@ public class FEDInstructionUtils {
&& ( ec.getCacheableData(cinst.input1).isFederated(FType.ROW)
|| (cinst.input2.isMatrix() && ec.getCacheableData(cinst.input2).isFederated(FType.ROW))
|| (cinst.input3.isMatrix() && ec.getCacheableData(cinst.input3).isFederated(FType.ROW))))
- fedinst = CtableFEDInstruction.parseInstruction(cinst.getInstructionString());
+ fedinst = CtableFEDInstruction.parseInstruction(cinst);
}
//set thread id for federated context management
@@ -296,7 +302,7 @@ public class FEDInstructionUtils {
if((ins.getOpcode().equalsIgnoreCase(OpOp1.CAST_AS_FRAME.toString())
|| ins.getOpcode().equalsIgnoreCase(OpOp1.CAST_AS_MATRIX.toString()))
&& ins.input1.isMatrix() && ec.getCacheableData(ins.input1).isFederatedExcept(FType.BROADCAST)){
- fedinst = CastFEDInstruction.parseInstruction(ins.getInstructionString());
+ fedinst = CastFEDInstruction.parseInstruction(ins);
}
}
else if (inst instanceof WriteSPInstruction) {
@@ -312,7 +318,7 @@ public class FEDInstructionUtils {
QuaternarySPInstruction instruction = (QuaternarySPInstruction) inst;
Data data = ec.getVariable(instruction.input1);
if(data instanceof MatrixObject && ((MatrixObject) data).isFederated())
- fedinst = QuaternaryFEDInstruction.parseInstruction(instruction.getInstructionString());
+ fedinst = QuaternaryFEDInstruction.parseInstruction(instruction);
}
else if(inst instanceof SpoofSPInstruction) {
SpoofSPInstruction ins = (SpoofSPInstruction) inst;
@@ -320,7 +326,7 @@ public class FEDInstructionUtils {
if(((scla == SpoofCellwise.class || scla == SpoofMultiAggregate.class || scla == SpoofOuterProduct.class)
&& SpoofFEDInstruction.isFederated(ec, ins.getInputs(), scla))
|| (scla == SpoofRowwise.class && SpoofFEDInstruction.isFederated(ec, FType.ROW, ins.getInputs(), scla))) {
- fedinst = SpoofFEDInstruction.parseInstruction(inst.getInstructionString());
+ fedinst = SpoofFEDInstruction.parseInstruction(ins);
}
}
else if (inst instanceof UnarySPInstruction && ! (inst instanceof IndexingSPInstruction)) {
@@ -329,12 +335,12 @@ public class FEDInstructionUtils {
CentralMomentSPInstruction cinstruction = (CentralMomentSPInstruction) inst;
Data data = ec.getVariable(cinstruction.input1);
if (data instanceof MatrixObject && ((MatrixObject) data).isFederated() && ((MatrixObject) data).isFederatedExcept(FType.BROADCAST))
- fedinst = CentralMomentFEDInstruction.parseInstruction(inst);
+ fedinst = CentralMomentFEDInstruction.parseInstruction(cinstruction);
} else if (inst instanceof QuantileSortSPInstruction) {
QuantileSortSPInstruction qinstruction = (QuantileSortSPInstruction) inst;
Data data = ec.getVariable(qinstruction.input1);
if (data instanceof MatrixObject && ((MatrixObject) data).isFederated() && ((MatrixObject) data).isFederatedExcept(FType.BROADCAST))
- fedinst = QuantileSortFEDInstruction.parseInstruction(inst.getInstructionString(), false);
+ fedinst = QuantileSortFEDInstruction.parseInstruction(qinstruction);
}
else if (inst instanceof AggregateUnarySPInstruction) {
AggregateUnarySPInstruction auinstruction = (AggregateUnarySPInstruction) inst;
@@ -342,25 +348,23 @@ public class FEDInstructionUtils {
if(data instanceof MatrixObject && ((MatrixObject) data).isFederated() && ((MatrixObject) data).isFederatedExcept(FType.BROADCAST))
if(ArrayUtils.contains(new String[]{"uarimin", "uarimax"}, auinstruction.getOpcode())) {
if(((MatrixObject) data).getFedMapping().getType() == FType.ROW)
- fedinst = AggregateUnaryFEDInstruction.parseInstruction(
- InstructionUtils.concatOperands(inst.getInstructionString(),FederatedOutput.NONE.name()));
+ fedinst = AggregateUnaryFEDInstruction.parseInstruction(auinstruction);
}
else
- fedinst = AggregateUnaryFEDInstruction.parseInstruction(InstructionUtils.concatOperands(inst.getInstructionString(),FederatedOutput.NONE.name()));
+ fedinst = AggregateUnaryFEDInstruction.parseInstruction(auinstruction);
}
else if(inst instanceof ReorgSPInstruction && (inst.getOpcode().equals("r'") || inst.getOpcode().equals("rdiag")
|| inst.getOpcode().equals("rev"))) {
ReorgSPInstruction rinst = (ReorgSPInstruction) inst;
CacheableData<?> mo = ec.getCacheableData(rinst.input1);
- if((mo instanceof MatrixObject || mo instanceof FrameObject) && mo.isFederated() && ((MatrixObject) mo).isFederatedExcept(FType.BROADCAST))
- fedinst = ReorgFEDInstruction.parseInstruction(
- InstructionUtils.concatOperands(rinst.getInstructionString(), FederatedOutput.NONE.name()));
+ if((mo instanceof MatrixObject || mo instanceof FrameObject) && mo.isFederated() && mo.isFederatedExcept(FType.BROADCAST))
+ fedinst = ReorgFEDInstruction.parseInstruction(rinst);
}
else if(inst instanceof ReblockSPInstruction && instruction.input1 != null && (instruction.input1.isFrame() || instruction.input1.isMatrix())) {
ReblockSPInstruction rinst = (ReblockSPInstruction) instruction;
CacheableData<?> data = ec.getCacheableData(rinst.input1);
if(data.isFederatedExcept(FType.BROADCAST))
- fedinst = ReblockFEDInstruction.parseInstruction(inst.getInstructionString());
+ fedinst = ReblockFEDInstruction.parseInstruction((ReblockSPInstruction) inst);
}
else if(instruction.input1 != null && instruction.input1.isMatrix() && ec.containsVariable(instruction.input1)) {
MatrixObject mo1 = ec.getMatrixObject(instruction.input1);
@@ -376,7 +380,7 @@ public class FEDInstructionUtils {
}
else if(inst instanceof UnaryMatrixSPInstruction) {
if(UnaryMatrixFEDInstruction.isValidOpcode(inst.getOpcode()))
- fedinst = UnaryMatrixFEDInstruction.parseInstruction(inst.getInstructionString());
+ fedinst = UnaryMatrixFEDInstruction.parseInstruction((UnaryMatrixSPInstruction) inst);
}
}
}
@@ -386,7 +390,7 @@ public class FEDInstructionUtils {
if (inst instanceof MapmmSPInstruction || inst instanceof CpmmSPInstruction || inst instanceof RmmSPInstruction) {
Data data = ec.getVariable(instruction.input1);
if (data instanceof MatrixObject && ((MatrixObject) data).isFederatedExcept(FType.BROADCAST)) {
- fedinst = MMFEDInstruction.parseInstruction(instruction.getInstructionString());
+ fedinst = MMFEDInstruction.parseInstruction((AggregateBinarySPInstruction) instruction);
}
}
else
@@ -394,7 +398,7 @@ public class FEDInstructionUtils {
QuantilePickSPInstruction qinstruction = (QuantilePickSPInstruction) inst;
Data data = ec.getVariable(qinstruction.input1);
if(data instanceof MatrixObject && ((MatrixObject) data).isFederatedExcept(FType.BROADCAST))
- fedinst = QuantilePickFEDInstruction.parseInstruction(inst);
+ fedinst = QuantilePickFEDInstruction.parseInstruction(qinstruction);
}
else if (inst instanceof AppendGAlignedSPInstruction || inst instanceof AppendGSPInstruction
|| inst instanceof AppendMSPInstruction || inst instanceof AppendRSPInstruction) {
@@ -403,44 +407,45 @@ public class FEDInstructionUtils {
Data data2 = ec.getVariable(ainstruction.input2);
if ((data1 instanceof MatrixObject && ((MatrixObject) data1).isFederatedExcept(FType.BROADCAST))
|| (data2 instanceof MatrixObject && ((MatrixObject) data2).isFederatedExcept(FType.BROADCAST))) {
- fedinst = AppendFEDInstruction.parseInstruction(instruction);
+ fedinst = AppendFEDInstruction.parseInstruction((AppendSPInstruction) instruction);
+ }
+ }
+ else if (inst instanceof BinaryMatrixScalarSPInstruction) {
+ Data data = ec.getVariable(instruction.input1);
+ if(data instanceof MatrixObject && ((MatrixObject)data).isFederatedExcept(FType.BROADCAST)) {
+ fedinst = BinaryMatrixScalarFEDInstruction.parseInstruction((BinaryMatrixScalarSPInstruction) inst);
}
}
- else if (inst instanceof BinaryMatrixScalarSPInstruction
- || inst instanceof BinaryMatrixMatrixSPInstruction
- || inst instanceof BinaryMatrixBVectorSPInstruction
- || inst instanceof BinaryTensorTensorSPInstruction
- || inst instanceof BinaryTensorTensorBroadcastSPInstruction) {
+ else if (inst instanceof BinaryMatrixMatrixSPInstruction) {
Data data = ec.getVariable(instruction.input1);
- if((data instanceof MatrixObject && ((MatrixObject)data).isFederatedExcept(FType.BROADCAST))
- || (data instanceof TensorObject && ((TensorObject)data).isFederatedExcept(FType.BROADCAST))) {
- fedinst = BinaryFEDInstruction.parseInstruction(
- InstructionUtils.concatOperands(inst.getInstructionString(),FederatedOutput.NONE.name()));
+ if(data instanceof MatrixObject && ((MatrixObject)data).isFederatedExcept(FType.BROADCAST)) {
+ fedinst = BinaryMatrixMatrixFEDInstruction.parseInstruction((BinaryMatrixMatrixSPInstruction) inst);
}
}
else if( (instruction.input1.isMatrix() && ec.getCacheableData(instruction.input1).isFederatedExcept(FType.BROADCAST))
|| (instruction.input2.isMatrix() && ec.getMatrixObject(instruction.input2).isFederatedExcept(FType.BROADCAST))) {
- if("cov".equals(instruction.getOpcode()) && (ec.getMatrixObject(instruction.input1)
+ if(inst instanceof CovarianceSPInstruction && (ec.getMatrixObject(instruction.input1)
.isFederated(FType.ROW) || ec.getMatrixObject(instruction.input2).isFederated(FType.ROW)))
- fedinst = CovarianceFEDInstruction.parseInstruction(inst);
+ fedinst = CovarianceFEDInstruction.parseInstruction((CovarianceSPInstruction) inst);
else if(inst instanceof CumulativeOffsetSPInstruction) {
- fedinst = CumulativeOffsetFEDInstruction.parseInstruction(inst.getInstructionString());
+ fedinst = CumulativeOffsetFEDInstruction.parseInstruction((CumulativeOffsetSPInstruction) inst);
}
else
- fedinst = BinaryFEDInstruction.parseInstruction(InstructionUtils.concatOperands(inst.getInstructionString(), FederatedOutput.NONE.name()));
+ fedinst = BinaryFEDInstruction.parseInstruction(InstructionUtils
+ .concatOperands(inst.getInstructionString(), FEDInstruction.FederatedOutput.NONE.name()));
}
}
else if( inst instanceof ParameterizedBuiltinSPInstruction) {
ParameterizedBuiltinSPInstruction pinst = (ParameterizedBuiltinSPInstruction) inst;
if( pinst.getOpcode().equalsIgnoreCase("replace") && pinst.getTarget(ec).isFederatedExcept(FType.BROADCAST) )
- fedinst = ParameterizedBuiltinFEDInstruction.parseInstruction(pinst.getInstructionString());
+ fedinst = ParameterizedBuiltinFEDInstruction.parseInstruction(pinst);
}
else if (inst instanceof MultiReturnParameterizedBuiltinSPInstruction) {
MultiReturnParameterizedBuiltinSPInstruction minst = (MultiReturnParameterizedBuiltinSPInstruction) inst;
if(minst.getOpcode().equals("transformencode") && minst.input1.isFrame()) {
CacheableData<?> fo = ec.getCacheableData(minst.input1);
if(fo.isFederatedExcept(FType.BROADCAST)) {
- fedinst = MultiReturnParameterizedBuiltinFEDInstruction.parseInstruction(minst.getInstructionString());
+ fedinst = MultiReturnParameterizedBuiltinFEDInstruction.parseInstruction(minst);
}
}
}
@@ -449,7 +454,7 @@ public class FEDInstructionUtils {
IndexingSPInstruction minst = (IndexingSPInstruction) inst;
if((minst.input1.isMatrix() || minst.input1.isFrame())
&& ec.getCacheableData(minst.input1).isFederatedExcept(FType.BROADCAST)) {
- fedinst = IndexingFEDInstruction.parseInstruction(minst.getInstructionString());
+ fedinst = IndexingFEDInstruction.parseInstruction(minst);
}
}
else if(inst instanceof TernarySPInstruction) {
@@ -459,19 +464,18 @@ public class FEDInstructionUtils {
long margin = ec.getScalarInput(tinst.input3).getLongValue();
FrameObject fo = ec.getFrameObject(tinst.input1);
if(margin == 0 || (fo.isFederated(FType.ROW) && margin == 1) || (fo.isFederated(FType.COL) && margin == 2))
- fedinst = TernaryFrameScalarFEDInstruction
- .parseInstruction(InstructionUtils.concatOperands(inst.getInstructionString(),FederatedOutput.NONE.name()));
+ fedinst = TernaryFrameScalarFEDInstruction.parseInstruction((TernaryFrameScalarSPInstruction) tinst);
} else if((tinst.input1.isMatrix() && ec.getCacheableData(tinst.input1).isFederatedExcept(FType.BROADCAST))
|| (tinst.input2.isMatrix() && ec.getCacheableData(tinst.input2).isFederatedExcept(FType.BROADCAST))
|| (tinst.input3.isMatrix() && ec.getCacheableData(tinst.input3).isFederatedExcept(FType.BROADCAST))) {
- fedinst = TernaryFEDInstruction.parseInstruction(tinst.getInstructionString());
+ fedinst = TernaryFEDInstruction.parseInstruction(tinst);
}
}
else if(inst instanceof AggregateTernarySPInstruction){
AggregateTernarySPInstruction ins = (AggregateTernarySPInstruction) inst;
if(ins.input1.isMatrix() && ec.getCacheableData(ins.input1).isFederatedExcept(FType.BROADCAST) && ins.input2.isMatrix() &&
ec.getCacheableData(ins.input2).isFederatedExcept(FType.BROADCAST)) {
- fedinst = AggregateTernaryFEDInstruction.parseInstruction(ins.getInstructionString());
+ fedinst = AggregateTernaryFEDInstruction.parseInstruction(ins);
}
}
else if(inst instanceof CtableSPInstruction) {
@@ -480,7 +484,7 @@ public class FEDInstructionUtils {
&& ( ec.getCacheableData(cinst.input1).isFederated(FType.ROW)
|| (cinst.input2.isMatrix() && ec.getCacheableData(cinst.input2).isFederated(FType.ROW))
|| (cinst.input3.isMatrix() && ec.getCacheableData(cinst.input3).isFederated(FType.ROW))))
- fedinst = CtableFEDInstruction.parseInstruction(cinst.getInstructionString());
+ fedinst = CtableFEDInstruction.parseInstruction(cinst);
}
//set thread id for federated context management
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/fed/IndexingFEDInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/fed/IndexingFEDInstruction.java
index 6fc8c24a7e..a544e966ac 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/fed/IndexingFEDInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/fed/IndexingFEDInstruction.java
@@ -46,8 +46,10 @@ import org.apache.sysds.runtime.controlprogram.federated.FederationMap;
import org.apache.sysds.runtime.controlprogram.federated.FederationUtils;
import org.apache.sysds.runtime.instructions.InstructionUtils;
import org.apache.sysds.runtime.instructions.cp.CPOperand;
+import org.apache.sysds.runtime.instructions.cp.IndexingCPInstruction;
import org.apache.sysds.runtime.instructions.cp.ScalarObject;
import org.apache.sysds.runtime.instructions.cp.VariableCPInstruction;
+import org.apache.sysds.runtime.instructions.spark.IndexingSPInstruction;
import org.apache.sysds.runtime.meta.MatrixCharacteristics;
import org.apache.sysds.runtime.util.IndexRange;
@@ -80,6 +82,16 @@ public final class IndexingFEDInstruction extends UnaryFEDInstruction {
(int) (ec.getScalarInput(colUpper).getLongValue() - 1));
}
+ public static IndexingFEDInstruction parseInstruction(IndexingCPInstruction instr) {
+ return new IndexingFEDInstruction(instr.input1, instr.input2, instr.getRowLower(), instr.getRowUpper(),
+ instr.getColLower(), instr.getColUpper(), instr.output, instr.getOpcode(), instr.getInstructionString());
+ }
+
+ public static IndexingFEDInstruction parseInstruction(IndexingSPInstruction instr) {
+ return new IndexingFEDInstruction(instr.input1, instr.input2, instr.getRowLower(), instr.getRowUpper(),
+ instr.getColLower(), instr.getColUpper(), instr.output, instr.getOpcode(), instr.getInstructionString());
+ }
+
public static IndexingFEDInstruction parseInstruction(String str) {
String[] parts = InstructionUtils.getInstructionPartsWithValueType(str);
String opcode = parts[0];
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/fed/MMChainFEDInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/fed/MMChainFEDInstruction.java
index 306c06be0e..5ddc46d899 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/fed/MMChainFEDInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/fed/MMChainFEDInstruction.java
@@ -31,6 +31,7 @@ import org.apache.sysds.runtime.controlprogram.federated.FederatedRequest.Reques
import org.apache.sysds.runtime.controlprogram.federated.MatrixLineagePair;
import org.apache.sysds.runtime.instructions.InstructionUtils;
import org.apache.sysds.runtime.instructions.cp.CPOperand;
+import org.apache.sysds.runtime.instructions.cp.MMChainCPInstruction;
import org.apache.sysds.runtime.matrix.data.MatrixBlock;
import java.util.concurrent.Future;
@@ -49,6 +50,11 @@ public class MMChainFEDInstruction extends UnaryFEDInstruction {
return _type;
}
+ public static MMChainFEDInstruction parseInstruction(MMChainCPInstruction instr) {
+ return new MMChainFEDInstruction(instr.input1, instr.input2, instr.input3, instr.output, instr.getMMChainType(),
+ instr.getNumThreads(), instr.getOpcode(), instr.getInstructionString());
+ }
+
public static MMChainFEDInstruction parseInstruction ( String str ) {
//parse instruction parts (without exec type)
String[] parts = InstructionUtils.getInstructionPartsWithValueType( str );
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/fed/MMFEDInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/fed/MMFEDInstruction.java
index 73b20a2758..b0769671fb 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/fed/MMFEDInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/fed/MMFEDInstruction.java
@@ -24,7 +24,6 @@ import java.util.concurrent.Future;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.sysds.common.Types.DataType;
import org.apache.sysds.common.Types.ExecType;
-import org.apache.sysds.hops.AggBinaryOp;
import org.apache.sysds.hops.fedplanner.FTypes.AlignType;
import org.apache.sysds.hops.fedplanner.FTypes.FType;
import org.apache.sysds.lops.MapMult;
@@ -40,6 +39,7 @@ import org.apache.sysds.runtime.controlprogram.federated.FederationUtils;
import org.apache.sysds.runtime.controlprogram.federated.MatrixLineagePair;
import org.apache.sysds.runtime.instructions.InstructionUtils;
import org.apache.sysds.runtime.instructions.cp.CPOperand;
+import org.apache.sysds.runtime.instructions.spark.AggregateBinarySPInstruction;
import org.apache.sysds.runtime.matrix.data.MatrixBlock;
import org.apache.sysds.runtime.matrix.operators.AggregateBinaryOperator;
import org.apache.sysds.runtime.matrix.operators.Operator;
@@ -47,11 +47,15 @@ import org.apache.sysds.runtime.meta.MatrixCharacteristics;
public class MMFEDInstruction extends BinaryFEDInstruction
{
- private MMFEDInstruction(Operator op, CPOperand in1, CPOperand in2, CPOperand out, MapMult.CacheType type,
- boolean outputEmpty, AggBinaryOp.SparkAggType aggtype, String opcode, String istr) {
+ private MMFEDInstruction(Operator op, CPOperand in1, CPOperand in2, CPOperand out, String opcode, String istr) {
super(FEDType.MAPMM, op, in1, in2, out, opcode, istr);
}
+ public static MMFEDInstruction parseInstruction(AggregateBinarySPInstruction instr) {
+ return new MMFEDInstruction(instr.getOperator(), instr.input1, instr.input2, instr.output, instr.getOpcode(),
+ instr.getInstructionString());
+ }
+
public static MMFEDInstruction parseInstruction( String str ) {
String parts[] = InstructionUtils.getInstructionPartsWithValueType(str);
String opcode = parts[0];
@@ -62,12 +66,9 @@ public class MMFEDInstruction extends BinaryFEDInstruction
CPOperand in1 = new CPOperand(parts[1]);
CPOperand in2 = new CPOperand(parts[2]);
CPOperand out = new CPOperand(parts[3]);
- MapMult.CacheType type = MapMult.CacheType.valueOf(parts[4]);
- boolean outputEmpty = Boolean.parseBoolean(parts[5]);
- AggBinaryOp.SparkAggType aggtype = AggBinaryOp.SparkAggType.valueOf(parts[6]);
AggregateBinaryOperator aggbin = InstructionUtils.getMatMultOperator(1);
- return new MMFEDInstruction(aggbin, in1, in2, out, type, outputEmpty, aggtype, opcode, str);
+ return new MMFEDInstruction(aggbin, in1, in2, out, opcode, str);
}
public void processInstruction(ExecutionContext ec) {
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/fed/MultiReturnParameterizedBuiltinFEDInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/fed/MultiReturnParameterizedBuiltinFEDInstruction.java
index da51164012..16471a1497 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/fed/MultiReturnParameterizedBuiltinFEDInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/fed/MultiReturnParameterizedBuiltinFEDInstruction.java
@@ -50,6 +50,8 @@ import org.apache.sysds.runtime.controlprogram.federated.FederationUtils;
import org.apache.sysds.runtime.instructions.InstructionUtils;
import org.apache.sysds.runtime.instructions.cp.CPOperand;
import org.apache.sysds.runtime.instructions.cp.Data;
+import org.apache.sysds.runtime.instructions.cp.MultiReturnParameterizedBuiltinCPInstruction;
+import org.apache.sysds.runtime.instructions.spark.MultiReturnParameterizedBuiltinSPInstruction;
import org.apache.sysds.runtime.lineage.LineageItem;
import org.apache.sysds.runtime.lineage.LineageItemUtils;
import org.apache.sysds.runtime.matrix.data.FrameBlock;
@@ -64,10 +66,10 @@ import org.apache.sysds.runtime.transform.encode.MultiColumnEncoder;
import org.apache.sysds.runtime.util.IndexRange;
public class MultiReturnParameterizedBuiltinFEDInstruction extends ComputationFEDInstruction {
- protected final ArrayList<CPOperand> _outputs;
+ protected final List<CPOperand> _outputs;
private MultiReturnParameterizedBuiltinFEDInstruction(Operator op, CPOperand input1, CPOperand input2,
- ArrayList<CPOperand> outputs, String opcode, String istr) {
+ List<CPOperand> outputs, String opcode, String istr) {
super(FEDType.MultiReturnParameterizedBuiltin, op, input1, input2, null, opcode, istr);
_outputs = outputs;
}
@@ -76,6 +78,18 @@ public class MultiReturnParameterizedBuiltinFEDInstruction extends ComputationFE
return _outputs.get(i);
}
+ public static MultiReturnParameterizedBuiltinFEDInstruction parseInstruction(
+ MultiReturnParameterizedBuiltinCPInstruction instr) {
+ return new MultiReturnParameterizedBuiltinFEDInstruction(instr.getOperator(), instr.input1, instr.input2,
+ instr.getOutputs(), instr.getOpcode(), instr.getInstructionString());
+ }
+
+ public static MultiReturnParameterizedBuiltinFEDInstruction parseInstruction(
+ MultiReturnParameterizedBuiltinSPInstruction instr) {
+ return new MultiReturnParameterizedBuiltinFEDInstruction(instr.getOperator(), instr.input1, instr.input2,
+ instr.getOutputs(), instr.getOpcode(), instr.getInstructionString());
+ }
+
public static MultiReturnParameterizedBuiltinFEDInstruction parseInstruction(String str) {
String[] parts = InstructionUtils.getInstructionPartsWithValueType(str);
ArrayList<CPOperand> outputs = new ArrayList<>();
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/fed/ParameterizedBuiltinFEDInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/fed/ParameterizedBuiltinFEDInstruction.java
index 2a1335426f..c0b60c557a 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/fed/ParameterizedBuiltinFEDInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/fed/ParameterizedBuiltinFEDInstruction.java
@@ -62,7 +62,9 @@ import org.apache.sysds.runtime.functionobjects.ValueFunction;
import org.apache.sysds.runtime.instructions.InstructionUtils;
import org.apache.sysds.runtime.instructions.cp.CPOperand;
import org.apache.sysds.runtime.instructions.cp.Data;
+import org.apache.sysds.runtime.instructions.cp.ParameterizedBuiltinCPInstruction;
import org.apache.sysds.runtime.instructions.cp.ScalarObject;
+import org.apache.sysds.runtime.instructions.spark.ParameterizedBuiltinSPInstruction;
import org.apache.sysds.runtime.lineage.LineageItem;
import org.apache.sysds.runtime.lineage.LineageItemUtils;
import org.apache.sysds.runtime.matrix.data.FrameBlock;
@@ -80,9 +82,9 @@ import org.apache.sysds.runtime.transform.encode.MultiColumnEncoder;
import org.apache.sysds.runtime.util.UtilFunctions;
public class ParameterizedBuiltinFEDInstruction extends ComputationFEDInstruction {
- protected final LinkedHashMap<String, String> params;
+ protected final HashMap<String, String> params;
- protected ParameterizedBuiltinFEDInstruction(Operator op, LinkedHashMap<String, String> paramsMap, CPOperand out,
+ protected ParameterizedBuiltinFEDInstruction(Operator op, HashMap<String, String> paramsMap, CPOperand out,
String opcode, String istr) {
super(FEDType.ParameterizedBuiltin, op, null, null, out, opcode, istr);
params = paramsMap;
@@ -110,6 +112,16 @@ public class ParameterizedBuiltinFEDInstruction extends ComputationFEDInstructio
return paramMap;
}
+ public static ParameterizedBuiltinFEDInstruction parseInstruction(ParameterizedBuiltinCPInstruction instr) {
+ return new ParameterizedBuiltinFEDInstruction(instr.getOperator(), instr.getParameterMap(), instr.output,
+ instr.getOpcode(), instr.getInstructionString());
+ }
+
+ public static ParameterizedBuiltinFEDInstruction parseInstruction(ParameterizedBuiltinSPInstruction instr) {
+ return new ParameterizedBuiltinFEDInstruction(instr.getOperator(), instr.getParameterMap(), instr.output,
+ instr.getOpcode(), instr.getInstructionString());
+ }
+
public static ParameterizedBuiltinFEDInstruction parseInstruction(String str) {
String[] parts = InstructionUtils.getInstructionPartsWithValueType(str);
// first part is always the opcode
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/fed/QuantilePickFEDInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/fed/QuantilePickFEDInstruction.java
index 2fa9c30e26..f3eee5ce4a 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/fed/QuantilePickFEDInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/fed/QuantilePickFEDInstruction.java
@@ -34,7 +34,6 @@ import org.apache.commons.lang3.tuple.ImmutablePair;
import org.apache.commons.lang3.tuple.ImmutableTriple;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.sysds.hops.fedplanner.FTypes.FType;
-import org.apache.sysds.lops.Lop;
import org.apache.sysds.lops.PickByCount.OperationTypes;
import org.apache.sysds.runtime.DMLRuntimeException;
import org.apache.sysds.runtime.controlprogram.caching.CacheableData;
@@ -48,12 +47,13 @@ import org.apache.sysds.runtime.controlprogram.federated.FederatedResponse;
import org.apache.sysds.runtime.controlprogram.federated.FederatedUDF;
import org.apache.sysds.runtime.controlprogram.federated.FederationMap;
import org.apache.sysds.runtime.controlprogram.federated.FederationUtils;
-import org.apache.sysds.runtime.instructions.Instruction;
import org.apache.sysds.runtime.instructions.InstructionUtils;
import org.apache.sysds.runtime.instructions.cp.CPOperand;
import org.apache.sysds.runtime.instructions.cp.Data;
import org.apache.sysds.runtime.instructions.cp.DoubleObject;
+import org.apache.sysds.runtime.instructions.cp.QuantilePickCPInstruction;
import org.apache.sysds.runtime.instructions.cp.ScalarObject;
+import org.apache.sysds.runtime.instructions.spark.QuantilePickSPInstruction;
import org.apache.sysds.runtime.lineage.LineageItem;
import org.apache.sysds.runtime.matrix.data.FrameBlock;
import org.apache.sysds.runtime.matrix.data.MatrixBlock;
@@ -80,8 +80,14 @@ public class QuantilePickFEDInstruction extends BinaryFEDInstruction {
this(op, in, in2, out, type, inmem, opcode, istr, FederatedOutput.NONE);
}
- public static QuantilePickFEDInstruction parseInstruction( Instruction inst ){
- return parseInstruction(inst.getInstructionString() + Lop.OPERAND_DELIMITOR + FederatedOutput.NONE);
+ public static QuantilePickFEDInstruction parseInstruction(QuantilePickCPInstruction instr) {
+ return new QuantilePickFEDInstruction(instr.getOperator(), instr.input1, instr.input2, instr.output,
+ instr.getOperationType(), instr.isInMem(), instr.getOpcode(), instr.getInstructionString());
+ }
+
+ public static QuantilePickFEDInstruction parseInstruction(QuantilePickSPInstruction instr) {
+ return new QuantilePickFEDInstruction(instr.getOperator(), instr.input1, instr.input2, instr.output,
+ instr.getOperationType(), false, instr.getOpcode(), instr.getInstructionString());
}
public static QuantilePickFEDInstruction parseInstruction ( String str ) {
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/fed/QuantileSortFEDInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/fed/QuantileSortFEDInstruction.java
index ece09392f6..f817c4c2a6 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/fed/QuantileSortFEDInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/fed/QuantileSortFEDInstruction.java
@@ -35,6 +35,8 @@ import org.apache.sysds.runtime.controlprogram.federated.MatrixLineagePair;
import org.apache.sysds.runtime.instructions.InstructionUtils;
import org.apache.sysds.runtime.instructions.cp.CPOperand;
import org.apache.sysds.runtime.instructions.cp.Data;
+import org.apache.sysds.runtime.instructions.cp.QuantileSortCPInstruction;
+import org.apache.sysds.runtime.instructions.spark.QuantileSortSPInstruction;
import org.apache.sysds.runtime.lineage.LineageItem;
import org.apache.sysds.runtime.matrix.data.MatrixBlock;
@@ -51,6 +53,16 @@ public class QuantileSortFEDInstruction extends UnaryFEDInstruction {
_numThreads = k;
}
+ public static QuantileSortFEDInstruction parseInstruction(QuantileSortCPInstruction instr) {
+ return new QuantileSortFEDInstruction(instr.input1, instr.input2, instr.output, instr.getOpcode(),
+ instr.getInstructionString(), instr.getNumThreads());
+ }
+
+ public static QuantileSortFEDInstruction parseInstruction(QuantileSortSPInstruction instr) {
+ return new QuantileSortFEDInstruction(instr.input1, instr.input2, instr.output, instr.getOpcode(),
+ instr.getInstructionString(), 1);
+ }
+
private static void parseInstruction(String instr, CPOperand in1, CPOperand in2, CPOperand out) {
String[] parts = InstructionUtils.getInstructionPartsWithValueType(instr);
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/fed/QuaternaryFEDInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/fed/QuaternaryFEDInstruction.java
index 0868901341..9b5014e6d7 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/fed/QuaternaryFEDInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/fed/QuaternaryFEDInstruction.java
@@ -39,6 +39,8 @@ import org.apache.sysds.runtime.controlprogram.caching.MatrixObject;
import org.apache.sysds.runtime.controlprogram.context.ExecutionContext;
import org.apache.sysds.runtime.instructions.InstructionUtils;
import org.apache.sysds.runtime.instructions.cp.CPOperand;
+import org.apache.sysds.runtime.instructions.cp.QuaternaryCPInstruction;
+import org.apache.sysds.runtime.instructions.spark.QuaternarySPInstruction;
import org.apache.sysds.runtime.matrix.operators.Operator;
import org.apache.sysds.runtime.matrix.operators.QuaternaryOperator;
@@ -56,6 +58,38 @@ public abstract class QuaternaryFEDInstruction extends ComputationFEDInstruction
_input4 = in4;
}
+ public static QuaternaryFEDInstruction parseInstruction(QuaternaryCPInstruction instr) {
+ QuaternaryOperator qop = (QuaternaryOperator) instr.getOperator();
+ if(qop.wtype1 != null)
+ return QuaternaryWSLossFEDInstruction.parseInstruction(instr);
+ else if(qop.wtype2 != null)
+ return QuaternaryWSigmoidFEDInstruction.parseInstruction(instr);
+ else if(qop.wtype3 != null)
+ return QuaternaryWDivMMFEDInstruction.parseInstruction(instr);
+ else if(qop.wtype4 != null)
+ return QuaternaryWCeMMFEDInstruction.parseInstruction(instr);
+ else if(qop.wtype5 != null)
+ return QuaternaryWUMMFEDInstruction.parseInstruction(instr);
+ // unreachable
+ return null;
+ }
+
+ public static QuaternaryFEDInstruction parseInstruction(QuaternarySPInstruction instr) {
+ QuaternaryOperator qop = (QuaternaryOperator) instr.getOperator();
+ if(qop.wtype1 != null)
+ return QuaternaryWSLossFEDInstruction.parseInstruction(instr);
+ else if(qop.wtype2 != null)
+ return QuaternaryWSigmoidFEDInstruction.parseInstruction(instr);
+ else if(qop.wtype3 != null)
+ return QuaternaryWDivMMFEDInstruction.parseInstruction(instr);
+ else if(qop.wtype4 != null)
+ return QuaternaryWCeMMFEDInstruction.parseInstruction(instr);
+ else if(qop.wtype5 != null)
+ return QuaternaryWUMMFEDInstruction.parseInstruction(instr);
+ // unreachable
+ return null;
+ }
+
public static QuaternaryFEDInstruction parseInstruction(String str) {
if(str.startsWith(ExecType.SPARK.name())) {
// rewrite the spark instruction to a cp instruction
@@ -143,7 +177,9 @@ public abstract class QuaternaryFEDInstruction extends ComputationFEDInstruction
return false;
}
- private static String rewriteSparkInstructionToCP(String inst_str) {
+ protected static String rewriteSparkInstructionToCP(String inst_str) {
+ // TODO: don't perform replacement over the whole instruction string, possibly changing string literals,
+ // instead only at positions of ExecType and Opcode
// rewrite the spark instruction to a cp instruction
inst_str = inst_str.replace(ExecType.SPARK.name(), ExecType.CP.name());
if(inst_str.contains(WeightedCrossEntropy.OPCODE))
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/fed/QuaternaryWCeMMFEDInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/fed/QuaternaryWCeMMFEDInstruction.java
index ce9fcb5111..4352bd9476 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/fed/QuaternaryWCeMMFEDInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/fed/QuaternaryWCeMMFEDInstruction.java
@@ -21,6 +21,8 @@ package org.apache.sysds.runtime.instructions.fed;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.sysds.runtime.controlprogram.federated.FederatedRequest.RequestType;
+import org.apache.sysds.runtime.instructions.cp.QuaternaryCPInstruction;
+import org.apache.sysds.runtime.instructions.spark.QuaternarySPInstruction;
import org.apache.sysds.runtime.matrix.operators.AggregateUnaryOperator;
import org.apache.sysds.runtime.instructions.InstructionUtils;
import org.apache.sysds.common.Types.DataType;
@@ -56,6 +58,18 @@ public class QuaternaryWCeMMFEDInstruction extends QuaternaryFEDInstruction
super(FEDType.Quaternary, operator, in1, in2, in3, in4, out, opcode, instruction_str);
}
+ public static QuaternaryWCeMMFEDInstruction parseInstruction(QuaternaryCPInstruction instr) {
+ return new QuaternaryWCeMMFEDInstruction(instr.getOperator(), instr.input1, instr.input2, instr.input3,
+ instr.getInput4(), instr.output, instr.getOpcode(), instr.getInstructionString());
+ }
+
+ public static QuaternaryWCeMMFEDInstruction parseInstruction(QuaternarySPInstruction instr) {
+ String instrStr = rewriteSparkInstructionToCP(instr.getInstructionString());
+ String opcode = InstructionUtils.getInstructionPartsWithValueType(instrStr)[0];
+ return new QuaternaryWCeMMFEDInstruction(instr.getOperator(), instr.input1, instr.input2, instr.input3,
+ instr.getInput4(), instr.output, opcode, instrStr);
+ }
+
@Override
public void processInstruction(ExecutionContext ec)
{
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/fed/QuaternaryWDivMMFEDInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/fed/QuaternaryWDivMMFEDInstruction.java
index 35e84586c9..fa5344b4f9 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/fed/QuaternaryWDivMMFEDInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/fed/QuaternaryWDivMMFEDInstruction.java
@@ -21,6 +21,8 @@ package org.apache.sysds.runtime.instructions.fed;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.sysds.runtime.instructions.InstructionUtils;
+import org.apache.sysds.runtime.instructions.cp.QuaternaryCPInstruction;
+import org.apache.sysds.runtime.instructions.spark.QuaternarySPInstruction;
import org.apache.sysds.runtime.matrix.operators.AggregateUnaryOperator;
import org.apache.sysds.common.Types.DataType;
import org.apache.sysds.hops.fedplanner.FTypes.AlignType;
@@ -71,6 +73,18 @@ public class QuaternaryWDivMMFEDInstruction extends QuaternaryFEDInstruction
_qop = operator;
}
+ public static QuaternaryWDivMMFEDInstruction parseInstruction(QuaternaryCPInstruction instr) {
+ return new QuaternaryWDivMMFEDInstruction((QuaternaryOperator) instr.getOperator(), instr.input1, instr.input2,
+ instr.input3, instr.getInput4(), instr.output, instr.getOpcode(), instr.getInstructionString());
+ }
+
+ public static QuaternaryWDivMMFEDInstruction parseInstruction(QuaternarySPInstruction instr) {
+ String instrStr = rewriteSparkInstructionToCP(instr.getInstructionString());
+ String opcode = InstructionUtils.getInstructionPartsWithValueType(instrStr)[0];
+ return new QuaternaryWDivMMFEDInstruction((QuaternaryOperator) instr.getOperator(), instr.input1, instr.input2,
+ instr.input3, instr.getInput4(), instr.output, opcode, instrStr);
+ }
+
@Override
public void processInstruction(ExecutionContext ec)
{
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/fed/QuaternaryWSLossFEDInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/fed/QuaternaryWSLossFEDInstruction.java
index 8e58d266a6..8d47b4239f 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/fed/QuaternaryWSLossFEDInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/fed/QuaternaryWSLossFEDInstruction.java
@@ -33,6 +33,8 @@ import org.apache.sysds.runtime.controlprogram.federated.MatrixLineagePair;
import org.apache.sysds.runtime.DMLRuntimeException;
import org.apache.sysds.runtime.instructions.cp.CPOperand;
import org.apache.sysds.runtime.instructions.InstructionUtils;
+import org.apache.sysds.runtime.instructions.cp.QuaternaryCPInstruction;
+import org.apache.sysds.runtime.instructions.spark.QuaternarySPInstruction;
import org.apache.sysds.runtime.matrix.operators.AggregateUnaryOperator;
import org.apache.sysds.runtime.matrix.operators.Operator;
import org.apache.sysds.runtime.matrix.operators.QuaternaryOperator;
@@ -61,6 +63,18 @@ public class QuaternaryWSLossFEDInstruction extends QuaternaryFEDInstruction {
super(FEDType.Quaternary, operator, in1, in2, in3, in4, out, opcode, instruction_str);
}
+ public static QuaternaryWSLossFEDInstruction parseInstruction(QuaternaryCPInstruction instr) {
+ return new QuaternaryWSLossFEDInstruction(instr.getOperator(), instr.input1, instr.input2, instr.input3,
+ instr.getInput4(), instr.output, instr.getOpcode(), instr.getInstructionString());
+ }
+
+ public static QuaternaryWSLossFEDInstruction parseInstruction(QuaternarySPInstruction instr) {
+ String instrStr = rewriteSparkInstructionToCP(instr.getInstructionString());
+ String opcode = InstructionUtils.getInstructionPartsWithValueType(instrStr)[0];
+ return new QuaternaryWSLossFEDInstruction(instr.getOperator(), instr.input1, instr.input2, instr.input3,
+ instr.getInput4(), instr.output, opcode, instrStr);
+ }
+
@Override
public void processInstruction(ExecutionContext ec) {
QuaternaryOperator qop = (QuaternaryOperator) _optr;
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/fed/QuaternaryWSigmoidFEDInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/fed/QuaternaryWSigmoidFEDInstruction.java
index 0d614703d7..1fd3c82fa5 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/fed/QuaternaryWSigmoidFEDInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/fed/QuaternaryWSigmoidFEDInstruction.java
@@ -31,7 +31,10 @@ import org.apache.sysds.runtime.controlprogram.federated.FederatedRequest;
import org.apache.sysds.runtime.controlprogram.federated.FederationMap;
import org.apache.sysds.runtime.controlprogram.federated.FederationUtils;
import org.apache.sysds.runtime.controlprogram.federated.MatrixLineagePair;
+import org.apache.sysds.runtime.instructions.InstructionUtils;
import org.apache.sysds.runtime.instructions.cp.CPOperand;
+import org.apache.sysds.runtime.instructions.cp.QuaternaryCPInstruction;
+import org.apache.sysds.runtime.instructions.spark.QuaternarySPInstruction;
import org.apache.sysds.runtime.matrix.operators.Operator;
public class QuaternaryWSigmoidFEDInstruction extends QuaternaryFEDInstruction {
@@ -54,6 +57,18 @@ public class QuaternaryWSigmoidFEDInstruction extends QuaternaryFEDInstruction {
super(FEDType.Quaternary, operator, in1, in2, in3, out, opcode, instruction_str);
}
+ public static QuaternaryWSigmoidFEDInstruction parseInstruction(QuaternaryCPInstruction instr) {
+ return new QuaternaryWSigmoidFEDInstruction(instr.getOperator(), instr.input1, instr.input2, instr.input3,
+ instr.getOutput(), instr.getOpcode(), instr.getInstructionString());
+ }
+
+ public static QuaternaryWSigmoidFEDInstruction parseInstruction(QuaternarySPInstruction instr) {
+ String instrStr = rewriteSparkInstructionToCP(instr.getInstructionString());
+ String opcode = InstructionUtils.getInstructionPartsWithValueType(instrStr)[0];
+ return new QuaternaryWSigmoidFEDInstruction(instr.getOperator(), instr.input1, instr.input2, instr.input3,
+ instr.output, opcode, instrStr);
+ }
+
@Override
public void processInstruction(ExecutionContext ec) {
MatrixObject X = ec.getMatrixObject(input1);
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/fed/QuaternaryWUMMFEDInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/fed/QuaternaryWUMMFEDInstruction.java
index df17e6c256..40851e502f 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/fed/QuaternaryWUMMFEDInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/fed/QuaternaryWUMMFEDInstruction.java
@@ -31,7 +31,10 @@ import org.apache.sysds.runtime.controlprogram.federated.FederatedRequest;
import org.apache.sysds.runtime.controlprogram.federated.FederationMap;
import org.apache.sysds.runtime.controlprogram.federated.FederationUtils;
import org.apache.sysds.runtime.controlprogram.federated.MatrixLineagePair;
+import org.apache.sysds.runtime.instructions.InstructionUtils;
import org.apache.sysds.runtime.instructions.cp.CPOperand;
+import org.apache.sysds.runtime.instructions.cp.QuaternaryCPInstruction;
+import org.apache.sysds.runtime.instructions.spark.QuaternarySPInstruction;
import org.apache.sysds.runtime.matrix.operators.Operator;
public class QuaternaryWUMMFEDInstruction extends QuaternaryFEDInstruction {
@@ -55,6 +58,18 @@ public class QuaternaryWUMMFEDInstruction extends QuaternaryFEDInstruction {
super(FEDType.Quaternary, operator, in1, in2, in3, out, opcode, instruction_str);
}
+ public static QuaternaryWUMMFEDInstruction parseInstruction(QuaternaryCPInstruction instr) {
+ return new QuaternaryWUMMFEDInstruction(instr.getOperator(), instr.input1, instr.input2, instr.input3,
+ instr.output, instr.getOpcode(), instr.getInstructionString());
+ }
+
+ public static QuaternaryWUMMFEDInstruction parseInstruction(QuaternarySPInstruction instr) {
+ String instrStr = rewriteSparkInstructionToCP(instr.getInstructionString());
+ String opcode = InstructionUtils.getInstructionPartsWithValueType(instrStr)[0];
+ return new QuaternaryWUMMFEDInstruction(instr.getOperator(), instr.input1, instr.input2, instr.input3,
+ instr.output, opcode, instrStr);
+ }
+
@Override
public void processInstruction(ExecutionContext ec) {
MatrixObject X = ec.getMatrixObject(input1);
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/fed/ReblockFEDInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/fed/ReblockFEDInstruction.java
index 40feddb0a1..557f2d8fbd 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/fed/ReblockFEDInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/fed/ReblockFEDInstruction.java
@@ -28,6 +28,7 @@ import org.apache.sysds.runtime.controlprogram.federated.FederatedRequest;
import org.apache.sysds.runtime.controlprogram.federated.FederationUtils;
import org.apache.sysds.runtime.instructions.InstructionUtils;
import org.apache.sysds.runtime.instructions.cp.CPOperand;
+import org.apache.sysds.runtime.instructions.spark.ReblockSPInstruction;
import org.apache.sysds.runtime.matrix.operators.Operator;
import org.apache.sysds.runtime.meta.DataCharacteristics;
import org.apache.sysds.runtime.meta.MatrixCharacteristics;
@@ -36,11 +37,15 @@ import org.apache.sysds.runtime.meta.MetaDataFormat;
public class ReblockFEDInstruction extends UnaryFEDInstruction {
private int blen;
- private ReblockFEDInstruction(Operator op, CPOperand in, CPOperand out, int br, int bc, boolean emptyBlocks,
+ private ReblockFEDInstruction(Operator op, CPOperand in, CPOperand out, int blen, boolean emptyBlocks,
String opcode, String instr) {
super(FEDInstruction.FEDType.Reblock, op, in, out, opcode, instr);
- blen = br;
- blen = bc;
+ this.blen = blen;
+ }
+
+ public static ReblockFEDInstruction parseInstruction(ReblockSPInstruction instr) {
+ return new ReblockFEDInstruction(instr.getOperator(), instr.input1, instr.output, instr.getBlockLength(),
+ instr.getOutputEmptyBlocks(), instr.getOpcode(), instr.getInstructionString());
}
public static ReblockFEDInstruction parseInstruction(String str) {
@@ -57,7 +62,7 @@ public class ReblockFEDInstruction extends UnaryFEDInstruction {
boolean outputEmptyBlocks = Boolean.parseBoolean(parts[4]);
Operator op = null; // no operator for ReblockFEDInstruction
- return new ReblockFEDInstruction(op, in, out, blen, blen, outputEmptyBlocks, opcode, str);
+ return new ReblockFEDInstruction(op, in, out, blen, outputEmptyBlocks, opcode, str);
}
@Override
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/fed/ReorgFEDInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/fed/ReorgFEDInstruction.java
index 2f9e26a2a1..bf3632f1dd 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/fed/ReorgFEDInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/fed/ReorgFEDInstruction.java
@@ -47,6 +47,8 @@ import org.apache.sysds.runtime.functionobjects.SwapIndex;
import org.apache.sysds.runtime.instructions.InstructionUtils;
import org.apache.sysds.runtime.instructions.cp.CPOperand;
import org.apache.sysds.runtime.instructions.cp.Data;
+import org.apache.sysds.runtime.instructions.cp.ReorgCPInstruction;
+import org.apache.sysds.runtime.instructions.spark.ReorgSPInstruction;
import org.apache.sysds.runtime.lineage.LineageItem;
import org.apache.sysds.runtime.lineage.LineageItemUtils;
import org.apache.sysds.runtime.matrix.data.MatrixBlock;
@@ -64,6 +66,16 @@ public class ReorgFEDInstruction extends UnaryFEDInstruction {
super(FEDType.Reorg, op, in1, out, opcode, istr);
}
+ public static ReorgFEDInstruction parseInstruction(ReorgCPInstruction rinst) {
+ return new ReorgFEDInstruction(rinst.getOperator(), rinst.input1, rinst.output, rinst.getOpcode(),
+ rinst.getInstructionString(), FederatedOutput.NONE);
+ }
+
+ public static ReorgFEDInstruction parseInstruction(ReorgSPInstruction rinst) {
+ return new ReorgFEDInstruction(rinst.getOperator(), rinst.input1, rinst.output, rinst.getOpcode(),
+ rinst.getInstructionString(), FederatedOutput.NONE);
+ }
+
public static ReorgFEDInstruction parseInstruction ( String str ) {
CPOperand in = new CPOperand("", Types.ValueType.UNKNOWN, Types.DataType.UNKNOWN);
CPOperand out = new CPOperand("", Types.ValueType.UNKNOWN, Types.DataType.UNKNOWN);
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/fed/ReshapeFEDInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/fed/ReshapeFEDInstruction.java
index ef7ac4c1c6..3d355cd1dd 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/fed/ReshapeFEDInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/fed/ReshapeFEDInstruction.java
@@ -35,6 +35,8 @@ import org.apache.sysds.runtime.controlprogram.federated.FederationUtils;
import org.apache.sysds.runtime.instructions.InstructionUtils;
import org.apache.sysds.runtime.instructions.cp.BooleanObject;
import org.apache.sysds.runtime.instructions.cp.CPOperand;
+import org.apache.sysds.runtime.instructions.cp.ReshapeCPInstruction;
+import org.apache.sysds.runtime.instructions.spark.MatrixReshapeSPInstruction;
import org.apache.sysds.runtime.lineage.LineageItem;
import org.apache.sysds.runtime.lineage.LineageItemUtils;
import org.apache.sysds.runtime.matrix.operators.Operator;
@@ -54,6 +56,17 @@ public class ReshapeFEDInstruction extends UnaryFEDInstruction {
_opByRow = in5;
}
+ public static ReshapeFEDInstruction parseInstruction(ReshapeCPInstruction instr) {
+ return new ReshapeFEDInstruction(instr.getOperator(), instr.input1, instr.getOpRows(), instr.getOpCols(),
+ instr.getOpDims(), instr.getOpByRow(), instr.output, instr.getOpcode(), instr.getInstructionString());
+ }
+
+ public static ReshapeFEDInstruction parseInstruction(MatrixReshapeSPInstruction instr) {
+ // TODO: add dims argument (for tensors) to MatrixReshapeSPInstruction
+ return new ReshapeFEDInstruction(instr.getOperator(), instr.input1, instr.getOpRows(), instr.getOpCols(), null,
+ instr.getOpByRow(), instr.output, instr.getOpcode(), instr.getInstructionString());
+ }
+
public static ReshapeFEDInstruction parseInstruction(String str) {
String[] parts = InstructionUtils.getInstructionPartsWithValueType(str);
InstructionUtils.checkNumFields(parts, 6, 7);
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/fed/SpoofFEDInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/fed/SpoofFEDInstruction.java
index 6bd5aeb985..e5af25ef02 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/fed/SpoofFEDInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/fed/SpoofFEDInstruction.java
@@ -50,6 +50,8 @@ import org.apache.sysds.runtime.instructions.InstructionUtils;
import org.apache.sysds.runtime.instructions.cp.CPOperand;
import org.apache.sysds.runtime.instructions.cp.Data;
import org.apache.sysds.runtime.instructions.cp.ScalarObject;
+import org.apache.sysds.runtime.instructions.cp.SpoofCPInstruction;
+import org.apache.sysds.runtime.instructions.spark.SpoofSPInstruction;
import org.apache.sysds.runtime.matrix.data.MatrixBlock;
import org.apache.sysds.runtime.matrix.operators.AggregateUnaryOperator;
@@ -67,6 +69,17 @@ public class SpoofFEDInstruction extends FEDInstruction
_output = out;
}
+ public static SpoofFEDInstruction parseInstruction(SpoofCPInstruction instr) {
+ return new SpoofFEDInstruction(instr.getSpoofOperator(), instr.getInputs(), instr.getOutput(),
+ instr.getOpcode(), instr.getInstructionString());
+ }
+
+ public static SpoofFEDInstruction parseInstruction(SpoofSPInstruction instr) {
+ SpoofOperator op = CodegenUtils.createInstance(instr.getOperatorClass());
+ return new SpoofFEDInstruction(op, instr.getInputs(), instr.getOutput(), instr.getOpcode(),
+ instr.getInstructionString());
+ }
+
public static SpoofFEDInstruction parseInstruction(String str) {
String[] parts = InstructionUtils.getInstructionPartsWithValueType(str);
@@ -482,6 +495,7 @@ public class SpoofFEDInstruction extends FEDInstruction
}
public static boolean isFederated(ExecutionContext ec, CPOperand[] inputs, Class<?> scla) {
+ // TODO: use the `SpoofOperator` instead of the `Class<?>`, as it is more robust to inheritance hierarchy change
return isFederated(ec, null, inputs, scla);
}
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/fed/TernaryFEDInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/fed/TernaryFEDInstruction.java
index c9d518f226..342faf3296 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/fed/TernaryFEDInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/fed/TernaryFEDInstruction.java
@@ -33,6 +33,8 @@ import org.apache.sysds.runtime.controlprogram.federated.FederationUtils;
import org.apache.sysds.runtime.controlprogram.federated.MatrixLineagePair;
import org.apache.sysds.runtime.instructions.InstructionUtils;
import org.apache.sysds.runtime.instructions.cp.CPOperand;
+import org.apache.sysds.runtime.instructions.cp.TernaryCPInstruction;
+import org.apache.sysds.runtime.instructions.spark.TernarySPInstruction;
import org.apache.sysds.runtime.matrix.operators.TernaryOperator;
import org.apache.sysds.runtime.meta.MatrixCharacteristics;
@@ -44,6 +46,16 @@ public class TernaryFEDInstruction extends ComputationFEDInstruction {
super(FEDInstruction.FEDType.Ternary, op, in1, in2, in3, out, opcode, str, fedOut);
}
+ public static TernaryFEDInstruction parseInstruction(TernaryCPInstruction instr) {
+ return new TernaryFEDInstruction((TernaryOperator) instr.getOperator(), instr.input1, instr.input2,
+ instr.input3, instr.output, instr.getOpcode(), instr.getInstructionString(), FederatedOutput.NONE);
+ }
+
+ public static TernaryFEDInstruction parseInstruction(TernarySPInstruction instr) {
+ return new TernaryFEDInstruction((TernaryOperator) instr.getOperator(), instr.input1, instr.input2,
+ instr.input3, instr.output, instr.getOpcode(), instr.getInstructionString(), FederatedOutput.NONE);
+ }
+
public static TernaryFEDInstruction parseInstruction(String str) {
String[] parts = InstructionUtils.getInstructionPartsWithValueType(str);
String opcode = parts[0];
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/fed/TernaryFrameScalarFEDInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/fed/TernaryFrameScalarFEDInstruction.java
index 6205d2067b..72d038acb6 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/fed/TernaryFrameScalarFEDInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/fed/TernaryFrameScalarFEDInstruction.java
@@ -25,6 +25,8 @@ import org.apache.sysds.runtime.controlprogram.federated.FederatedRequest;
import org.apache.sysds.runtime.controlprogram.federated.FederationMap;
import org.apache.sysds.runtime.controlprogram.federated.FederationUtils;
import org.apache.sysds.runtime.instructions.cp.CPOperand;
+import org.apache.sysds.runtime.instructions.cp.TernaryFrameScalarCPInstruction;
+import org.apache.sysds.runtime.instructions.spark.TernaryFrameScalarSPInstruction;
import org.apache.sysds.runtime.matrix.operators.TernaryOperator;
public class TernaryFrameScalarFEDInstruction extends TernaryFEDInstruction
@@ -34,6 +36,18 @@ public class TernaryFrameScalarFEDInstruction extends TernaryFEDInstruction
super(op, in1, in2, in3, out, opcode, istr, fedOut);
}
+ public static TernaryFrameScalarFEDInstruction parseInstruction(TernaryFrameScalarCPInstruction instr) {
+ return new TernaryFrameScalarFEDInstruction((TernaryOperator) instr.getOperator(), instr.input1, instr.input2,
+ instr.input3, instr.output, instr.getOpcode(), instr.getInstructionString(),
+ FEDInstruction.FederatedOutput.NONE);
+ }
+
+ public static TernaryFrameScalarFEDInstruction parseInstruction(TernaryFrameScalarSPInstruction instr) {
+ return new TernaryFrameScalarFEDInstruction((TernaryOperator) instr.getOperator(), instr.input1, instr.input2,
+ instr.input3, instr.output, instr.getOpcode(), instr.getInstructionString(),
+ FEDInstruction.FederatedOutput.NONE);
+ }
+
@Override
public void processInstruction(ExecutionContext ec) {
// get input frames
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/fed/TsmmFEDInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/fed/TsmmFEDInstruction.java
index 5ebe7f6295..3b15b273db 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/fed/TsmmFEDInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/fed/TsmmFEDInstruction.java
@@ -33,6 +33,7 @@ import org.apache.sysds.runtime.controlprogram.federated.FederationMap;
import org.apache.sysds.runtime.controlprogram.federated.FederationUtils;
import org.apache.sysds.runtime.instructions.InstructionUtils;
import org.apache.sysds.runtime.instructions.cp.CPOperand;
+import org.apache.sysds.runtime.instructions.cp.MMTSJCPInstruction;
import org.apache.sysds.runtime.matrix.data.MatrixBlock;
public class TsmmFEDInstruction extends BinaryFEDInstruction {
@@ -49,7 +50,12 @@ public class TsmmFEDInstruction extends BinaryFEDInstruction {
public TsmmFEDInstruction(CPOperand in, CPOperand out, MMTSJType type, int k, String opcode, String istr) {
this(in, out, type, k, opcode, istr, FederatedOutput.NONE);
}
-
+
+ public static TsmmFEDInstruction parseInstruction(MMTSJCPInstruction instr) {
+ return new TsmmFEDInstruction(instr.input1, instr.getOutput(), instr.getMMTSJType(), instr.getNumThreads(),
+ instr.getOpcode(), instr.getInstructionString());
+ }
+
public static TsmmFEDInstruction parseInstruction(String str) {
String[] parts = InstructionUtils.getInstructionPartsWithValueType(str);
String opcode = parts[0];
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/fed/UnaryMatrixFEDInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/fed/UnaryMatrixFEDInstruction.java
index 615e94a8ff..c3c2111641 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/fed/UnaryMatrixFEDInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/fed/UnaryMatrixFEDInstruction.java
@@ -36,6 +36,8 @@ import org.apache.sysds.runtime.functionobjects.Builtin;
import org.apache.sysds.runtime.functionobjects.ValueFunction;
import org.apache.sysds.runtime.instructions.InstructionUtils;
import org.apache.sysds.runtime.instructions.cp.CPOperand;
+import org.apache.sysds.runtime.instructions.cp.UnaryMatrixCPInstruction;
+import org.apache.sysds.runtime.instructions.spark.UnaryMatrixSPInstruction;
import org.apache.sysds.runtime.matrix.data.LibCommonsMath;
import org.apache.sysds.runtime.matrix.data.MatrixBlock;
import org.apache.sysds.runtime.matrix.operators.Operator;
@@ -51,6 +53,16 @@ public class UnaryMatrixFEDInstruction extends UnaryFEDInstruction {
return !LibCommonsMath.isSupportedUnaryOperation(opcode);
}
+ public static UnaryMatrixFEDInstruction parseInstruction(UnaryMatrixCPInstruction instr) {
+ return new UnaryMatrixFEDInstruction(instr.getOperator(), instr.input1, instr.output, instr.getOpcode(),
+ instr.getInstructionString());
+ }
+
+ public static UnaryMatrixFEDInstruction parseInstruction(UnaryMatrixSPInstruction instr) {
+ return new UnaryMatrixFEDInstruction(instr.getOperator(), instr.input1, instr.output, instr.getOpcode(),
+ instr.getInstructionString());
+ }
+
public static UnaryMatrixFEDInstruction parseInstruction(String str) {
CPOperand in = new CPOperand("", ValueType.UNKNOWN, DataType.UNKNOWN);
CPOperand out = new CPOperand("", ValueType.UNKNOWN, DataType.UNKNOWN);
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/fed/VariableFEDInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/fed/VariableFEDInstruction.java
index 197ba43cf7..4a51f49083 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/fed/VariableFEDInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/fed/VariableFEDInstruction.java
@@ -44,6 +44,7 @@ import org.apache.sysds.runtime.instructions.cp.VariableCPInstruction.VariableOp
import org.apache.sysds.runtime.lineage.LineageItem;
import org.apache.sysds.runtime.lineage.LineageTraceable;
+// TODO: merge with `CastFEDInstruction`
public class VariableFEDInstruction extends FEDInstruction implements LineageTraceable {
private static final Log LOG = LogFactory.getLog(VariableFEDInstruction.class.getName());
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/spark/AggregateBinarySPInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/spark/AggregateBinarySPInstruction.java
new file mode 100644
index 0000000000..80de732505
--- /dev/null
+++ b/src/main/java/org/apache/sysds/runtime/instructions/spark/AggregateBinarySPInstruction.java
@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.sysds.runtime.instructions.spark;
+
+import org.apache.sysds.runtime.instructions.cp.CPOperand;
+import org.apache.sysds.runtime.matrix.operators.Operator;
+
+/**
+ * Class to group the different MM <code>SPInstruction</code>s together.
+ */
+public abstract class AggregateBinarySPInstruction extends BinarySPInstruction {
+ protected AggregateBinarySPInstruction(SPType type, Operator op, CPOperand in1, CPOperand in2, CPOperand out,
+ String opcode, String istr) {
+ super(type, op, in1, in2, out, opcode, istr);
+ }
+}
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/spark/AppendGAlignedSPInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/spark/AppendGAlignedSPInstruction.java
index bd230bb01b..949a078025 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/spark/AppendGAlignedSPInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/spark/AppendGAlignedSPInstruction.java
@@ -34,13 +34,10 @@ import org.apache.sysds.runtime.matrix.operators.ReorgOperator;
import org.apache.sysds.runtime.meta.DataCharacteristics;
import scala.Tuple2;
-public class AppendGAlignedSPInstruction extends BinarySPInstruction {
- private boolean _cbind = true;
-
+public class AppendGAlignedSPInstruction extends AppendSPInstruction {
private AppendGAlignedSPInstruction(Operator op, CPOperand in1, CPOperand in2, CPOperand in3, CPOperand out,
- boolean cbind, String opcode, String istr) {
- super(SPType.GAppend, op, in1, in2, out, opcode, istr);
- _cbind = cbind;
+ boolean cbind, String opcode, String istr) {
+ super(SPType.GAppend, op, in1, in2, out, cbind, opcode, istr);
}
public static AppendGAlignedSPInstruction parseInstruction ( String str ) {
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/spark/AppendGSPInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/spark/AppendGSPInstruction.java
index 0a3209cfae..693dd6ed47 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/spark/AppendGSPInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/spark/AppendGSPInstruction.java
@@ -40,13 +40,10 @@ import scala.Tuple2;
import java.util.ArrayList;
import java.util.Iterator;
-public class AppendGSPInstruction extends BinarySPInstruction {
- private boolean _cbind = true;
-
+public class AppendGSPInstruction extends AppendSPInstruction {
private AppendGSPInstruction(Operator op, CPOperand in1, CPOperand in2, CPOperand offset, CPOperand offset2,
- CPOperand out, boolean cbind, String opcode, String istr) {
- super(SPType.GAppend, op, in1, in2, out, opcode, istr);
- _cbind = cbind;
+ CPOperand out, boolean cbind, String opcode, String istr) {
+ super(SPType.GAppend, op, in1, in2, out, cbind, opcode, istr);
}
public static AppendGSPInstruction parseInstruction ( String str ) {
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/spark/AppendMSPInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/spark/AppendMSPInstruction.java
index e0e4f0973b..58e446ed1e 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/spark/AppendMSPInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/spark/AppendMSPInstruction.java
@@ -27,15 +27,13 @@ import org.apache.sysds.runtime.instructions.cp.CPOperand;
import org.apache.sysds.runtime.matrix.operators.Operator;
import org.apache.sysds.runtime.matrix.operators.ReorgOperator;
-public abstract class AppendMSPInstruction extends BinarySPInstruction {
+public abstract class AppendMSPInstruction extends AppendSPInstruction {
protected CPOperand _offset = null;
- protected boolean _cbind = true;
protected AppendMSPInstruction(Operator op, CPOperand in1, CPOperand in2, CPOperand offset, CPOperand out,
- boolean cbind, String opcode, String istr) {
- super(SPType.MAppend, op, in1, in2, out, opcode, istr);
+ boolean cbind, String opcode, String istr) {
+ super(SPType.MAppend, op, in1, in2, out, cbind, opcode, istr);
_offset = offset;
- _cbind = cbind;
}
public static AppendMSPInstruction parseInstruction( String str ) {
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/spark/AppendRSPInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/spark/AppendRSPInstruction.java
index 966fb4b2f1..4a0cd61430 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/spark/AppendRSPInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/spark/AppendRSPInstruction.java
@@ -26,12 +26,10 @@ import org.apache.sysds.runtime.instructions.cp.CPOperand;
import org.apache.sysds.runtime.matrix.operators.Operator;
import org.apache.sysds.runtime.matrix.operators.ReorgOperator;
-public abstract class AppendRSPInstruction extends BinarySPInstruction {
- protected boolean _cbind = true;
-
- protected AppendRSPInstruction(Operator op, CPOperand in1, CPOperand in2, CPOperand out, boolean cbind, String opcode, String istr) {
- super(SPType.RAppend, op, in1, in2, out, opcode, istr);
- _cbind = cbind;
+public abstract class AppendRSPInstruction extends AppendSPInstruction {
+ protected AppendRSPInstruction(Operator op, CPOperand in1, CPOperand in2, CPOperand out, boolean cbind,
+ String opcode, String istr) {
+ super(SPType.RAppend, op, in1, in2, out, cbind, opcode, istr);
}
public static AppendRSPInstruction parseInstruction ( String str ) {
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/spark/AppendSPInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/spark/AppendSPInstruction.java
new file mode 100644
index 0000000000..4a768df9c9
--- /dev/null
+++ b/src/main/java/org/apache/sysds/runtime/instructions/spark/AppendSPInstruction.java
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.sysds.runtime.instructions.spark;
+
+import org.apache.sysds.runtime.instructions.cp.CPOperand;
+import org.apache.sysds.runtime.matrix.operators.Operator;
+
+/**
+ * Class to group the different append <code>SPInstruction</code>s together.
+ */
+public abstract class AppendSPInstruction extends BinarySPInstruction {
+ protected boolean _cbind;
+
+ protected AppendSPInstruction(SPType type, Operator op, CPOperand input1, CPOperand input2, CPOperand output,
+ boolean cbind, String opcode, String instructionString) {
+ super(type, op, input1, input2, output, opcode, instructionString);
+ _cbind = cbind;
+ }
+
+ public boolean getCBind() {
+ return _cbind;
+ }
+}
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/spark/BinaryMatrixBVectorSPInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/spark/BinaryMatrixBVectorSPInstruction.java
index 99eb2927d3..9d816705d1 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/spark/BinaryMatrixBVectorSPInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/spark/BinaryMatrixBVectorSPInstruction.java
@@ -24,12 +24,12 @@ import org.apache.sysds.runtime.controlprogram.context.ExecutionContext;
import org.apache.sysds.runtime.instructions.cp.CPOperand;
import org.apache.sysds.runtime.matrix.operators.Operator;
-public class BinaryMatrixBVectorSPInstruction extends BinarySPInstruction {
+public class BinaryMatrixBVectorSPInstruction extends BinaryMatrixMatrixSPInstruction {
private VectorType _vtype = null;
protected BinaryMatrixBVectorSPInstruction(Operator op, CPOperand in1, CPOperand in2, CPOperand out,
VectorType vtype, String opcode, String istr) {
- super(SPType.Binary, op, in1, in2, out, opcode, istr);
+ super(op, in1, in2, out, opcode, istr);
_vtype = vtype;
}
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/spark/CpmmSPInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/spark/CpmmSPInstruction.java
index d5c0225489..f9b3130ccc 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/spark/CpmmSPInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/spark/CpmmSPInstruction.java
@@ -54,7 +54,7 @@ import scala.Tuple2;
* this would result in a degree of parallelism of 1.
*
*/
-public class CpmmSPInstruction extends BinarySPInstruction {
+public class CpmmSPInstruction extends AggregateBinarySPInstruction {
private final boolean _outputEmptyBlocks;
private final SparkAggType _aggtype;
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/spark/CtableSPInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/spark/CtableSPInstruction.java
index 038a55c242..f79b86ac61 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/spark/CtableSPInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/spark/CtableSPInstruction.java
@@ -23,6 +23,7 @@ import org.apache.commons.lang3.ArrayUtils;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFlatMapFunction;
+import org.apache.sysds.common.Types;
import org.apache.sysds.common.Types.ValueType;
import org.apache.sysds.lops.Ctable;
import org.apache.sysds.runtime.DMLRuntimeException;
@@ -195,6 +196,22 @@ public class CtableSPInstruction extends ComputationSPInstruction {
SparkUtils.postprocessUltraSparseOutput(sec.getMatrixObject(output), mcOut);
}
+ public CPOperand getOutDim1() {
+ return new CPOperand(_outDim1, ValueType.FP64, Types.DataType.SCALAR, _dim1Literal);
+ }
+
+ public CPOperand getOutDim2() {
+ return new CPOperand(_outDim1, ValueType.FP64, Types.DataType.SCALAR, _dim1Literal);
+ }
+
+ public boolean getIsExpand() {
+ return _isExpand;
+ }
+
+ public boolean getIgnoreZeros() {
+ return _ignoreZeros;
+ }
+
private static class CTableFunction implements PairFlatMapFunction<Iterator<Tuple2<MatrixIndexes, MatrixBlock[]>>, MatrixIndexes, MatrixBlock>
{
private static final long serialVersionUID = 5348127596473232337L;
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/spark/CumulativeOffsetSPInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/spark/CumulativeOffsetSPInstruction.java
index ef79595b3d..ba424486cc 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/spark/CumulativeOffsetSPInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/spark/CumulativeOffsetSPInstruction.java
@@ -122,6 +122,14 @@ public class CumulativeOffsetSPInstruction extends BinarySPInstruction {
sec.addLineage(output.getName(), input2.getName(), broadcast);
}
+ public double getInitValue() {
+ return _initValue;
+ }
+
+ public boolean getBroadcast() {
+ return _broadcast;
+ }
+
private static class RDDCumSplitFunction implements PairFlatMapFunction<Tuple2<MatrixIndexes, MatrixBlock>, MatrixIndexes, MatrixBlock>
{
private static final long serialVersionUID = -8407407527406576965L;
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/spark/IndexingSPInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/spark/IndexingSPInstruction.java
index 76949fe679..a023fa220a 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/spark/IndexingSPInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/spark/IndexingSPInstruction.java
@@ -53,6 +53,22 @@ public abstract class IndexingSPInstruction extends UnarySPInstruction {
colLower = cl;
colUpper = cu;
}
+
+ public CPOperand getRowLower() {
+ return rowLower;
+ }
+
+ public CPOperand getRowUpper() {
+ return rowUpper;
+ }
+
+ public CPOperand getColLower() {
+ return colLower;
+ }
+
+ public CPOperand getColUpper() {
+ return colUpper;
+ }
public static IndexingSPInstruction parseInstruction ( String str ) {
String[] parts = InstructionUtils.getInstructionPartsWithValueType(str);
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/spark/MapmmSPInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/spark/MapmmSPInstruction.java
index bd8d2ad9bb..f50c46aceb 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/spark/MapmmSPInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/spark/MapmmSPInstruction.java
@@ -56,7 +56,7 @@ import org.apache.sysds.runtime.meta.DataCharacteristics;
import scala.Tuple2;
-public class MapmmSPInstruction extends BinarySPInstruction {
+public class MapmmSPInstruction extends AggregateBinarySPInstruction {
private static final Log LOG = LogFactory.getLog(MapmmSPInstruction.class.getName());
private CacheType _type = null;
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/spark/MatrixReshapeSPInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/spark/MatrixReshapeSPInstruction.java
index 507acc9d08..d8b8e4e384 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/spark/MatrixReshapeSPInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/spark/MatrixReshapeSPInstruction.java
@@ -127,7 +127,19 @@ public class MatrixReshapeSPInstruction extends UnarySPInstruction
sec.addLineageRDD(output.getName(), input1.getName());
}
}
-
+
+ public CPOperand getOpRows() {
+ return _opRows;
+ }
+
+ public CPOperand getOpCols() {
+ return _opCols;
+ }
+
+ public CPOperand getOpByRow() {
+ return _opByRow;
+ }
+
private static class RDDReshapeFunction implements PairFlatMapFunction<Tuple2<MatrixIndexes, MatrixBlock>, MatrixIndexes, MatrixBlock>
{
private static final long serialVersionUID = 2819309412002224478L;
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/spark/MultiReturnParameterizedBuiltinSPInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/spark/MultiReturnParameterizedBuiltinSPInstruction.java
index bca0825935..241c692846 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/spark/MultiReturnParameterizedBuiltinSPInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/spark/MultiReturnParameterizedBuiltinSPInstruction.java
@@ -180,6 +180,10 @@ public class MultiReturnParameterizedBuiltinSPInstruction extends ComputationSPI
return acc;
}
+ public List<CPOperand> getOutputs() {
+ return _outputs;
+ }
+
private static class MaxLongAccumulator extends AccumulatorV2<Long, Long> {
private static final long serialVersionUID = -3739727823287550826L;
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/spark/ParameterizedBuiltinSPInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/spark/ParameterizedBuiltinSPInstruction.java
index f513898aa7..7135f141c6 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/spark/ParameterizedBuiltinSPInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/spark/ParameterizedBuiltinSPInstruction.java
@@ -547,6 +547,10 @@ public class ParameterizedBuiltinSPInstruction extends ComputationSPInstruction
}
}
+ public HashMap<String, String> getParameterMap() {
+ return params;
+ }
+
public static class RDDReplaceFunction implements Function<MatrixBlock, MatrixBlock> {
private static final long serialVersionUID = 6576713401901671659L;
private final double _pattern;
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/spark/QuantilePickSPInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/spark/QuantilePickSPInstruction.java
index 3f9daefa91..845c94aba2 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/spark/QuantilePickSPInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/spark/QuantilePickSPInstruction.java
@@ -235,7 +235,11 @@ public class QuantilePickSPInstruction extends BinarySPInstruction {
pos + " in block of size " + tmp.getNumRows()+"x"+tmp.getNumColumns());
return val.get(0).quickGetValue((int)pos, 0);
}
-
+
+ public OperationTypes getOperationType() {
+ return _type;
+ }
+
private static class FilterFunction implements Function<Tuple2<MatrixIndexes,MatrixBlock>, Boolean>
{
private static final long serialVersionUID = -8249102381116157388L;
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/spark/QuaternarySPInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/spark/QuaternarySPInstruction.java
index 7f2210895c..9c9a063d31 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/spark/QuaternarySPInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/spark/QuaternarySPInstruction.java
@@ -323,6 +323,10 @@ public class QuaternarySPInstruction extends ComputationSPInstruction {
}
}
+ public CPOperand getInput4() {
+ return _input4;
+ }
+
private abstract static class RDDQuaternaryBaseFunction implements Serializable
{
private static final long serialVersionUID = -3175397651350954930L;
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/spark/ReblockSPInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/spark/ReblockSPInstruction.java
index b8192a9d1d..38928ee3c3 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/spark/ReblockSPInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/spark/ReblockSPInstruction.java
@@ -272,4 +272,12 @@ public class ReblockSPInstruction extends UnarySPInstruction {
//default reblock w/ active lineage tracing
return super.getLineageItem(ec);
}
+
+ public int getBlockLength() {
+ return blen;
+ }
+
+ public boolean getOutputEmptyBlocks() {
+ return outputEmptyBlocks;
+ }
}
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/spark/RmmSPInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/spark/RmmSPInstruction.java
index d6eda445ea..70d4bef1dd 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/spark/RmmSPInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/spark/RmmSPInstruction.java
@@ -47,7 +47,7 @@ import scala.Tuple2;
import java.util.Iterator;
import java.util.LinkedList;
-public class RmmSPInstruction extends BinarySPInstruction {
+public class RmmSPInstruction extends AggregateBinarySPInstruction {
private RmmSPInstruction(Operator op, CPOperand in1, CPOperand in2, CPOperand out, String opcode, String istr) {
super(SPType.RMM, op, in1, in2, out, opcode, istr);
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/spark/SpoofSPInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/spark/SpoofSPInstruction.java
index c0ccf9fc69..09d41e6866 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/spark/SpoofSPInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/spark/SpoofSPInstruction.java
@@ -354,6 +354,10 @@ public class SpoofSPInstruction extends SPInstruction {
}
}
+ public CPOperand getOutput() {
+ return _out;
+ }
+
private static class SpoofFunction implements Serializable
{
private static final long serialVersionUID = 2953479427746463003L;