This is an automated email from the ASF dual-hosted git repository.
mboehm7 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 540d68e [SYSTEMDS-3234] Fix cov/cm instruction parsing
540d68e is described below
commit 540d68e0f10eabe2374a0f8a32ac1642ed00b78d
Author: Matthias Boehm <[email protected]>
AuthorDate: Sat Dec 18 17:42:09 2021 +0100
[SYSTEMDS-3234] Fix cov/cm instruction parsing
The recent change on multi-threaded cov/cm operations lacked
consistent parsing for spark and federated cov/cm instructions. We
fixed this by now relying on the CP parsing logic to guarantee
consistency for cp/fed instruction while also avoiding code
duplication.
---
.../java/org/apache/sysds/lops/CoVariance.java | 6 +-
.../fed/CentralMomentFEDInstruction.java | 64 ++++------------------
.../instructions/fed/CovarianceFEDInstruction.java | 47 ++++------------
.../instructions/fed/FEDInstructionUtils.java | 6 +-
4 files changed, 31 insertions(+), 92 deletions(-)
diff --git a/src/main/java/org/apache/sysds/lops/CoVariance.java
b/src/main/java/org/apache/sysds/lops/CoVariance.java
index dc5427d..2357380 100644
--- a/src/main/java/org/apache/sysds/lops/CoVariance.java
+++ b/src/main/java/org/apache/sysds/lops/CoVariance.java
@@ -97,8 +97,10 @@ public class CoVariance extends Lop
}
sb.append( prepOutputOperand(output));
- sb.append( OPERAND_DELIMITOR );
- sb.append(_numThreads);
+ if( getExecType() == ExecType.CP ) {
+ sb.append( OPERAND_DELIMITOR );
+ sb.append(_numThreads);
+ }
return sb.toString();
}
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 4bd522f..ab9c9ed 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
@@ -24,7 +24,6 @@ import java.util.List;
import java.util.Optional;
import org.apache.commons.lang3.tuple.Pair;
-import org.apache.sysds.common.Types;
import org.apache.sysds.runtime.DMLRuntimeException;
import org.apache.sysds.runtime.controlprogram.caching.MatrixObject;
import org.apache.sysds.runtime.controlprogram.context.ExecutionContext;
@@ -33,74 +32,33 @@ 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.functionobjects.CM;
-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;
+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.lineage.LineageItem;
import org.apache.sysds.runtime.matrix.data.MatrixBlock;
import org.apache.sysds.runtime.matrix.operators.CMOperator;
+import org.apache.sysds.runtime.matrix.operators.Operator;
public class CentralMomentFEDInstruction extends AggregateUnaryFEDInstruction {
- private CentralMomentFEDInstruction(CMOperator cm, CPOperand in1,
CPOperand in2, CPOperand in3, CPOperand out,
- String opcode, String str) {
+ private CentralMomentFEDInstruction(Operator cm, CPOperand in1,
+ CPOperand in2, CPOperand in3, CPOperand out, String opcode,
String str)
+ {
super(cm, in1, in2, in3, out, opcode, str);
}
public static CentralMomentFEDInstruction parseInstruction(String str) {
- CPOperand in1 = new CPOperand("", Types.ValueType.UNKNOWN,
Types.DataType.UNKNOWN);
- CPOperand in2 = null;
- CPOperand in3 = null;
- CPOperand out = new CPOperand("", Types.ValueType.UNKNOWN,
Types.DataType.UNKNOWN);
-
- String[] parts =
InstructionUtils.getInstructionPartsWithValueType(str);
- String opcode = parts[0];
-
- // check supported opcode
- if (!opcode.equalsIgnoreCase("cm")) {
- throw new DMLRuntimeException("Unsupported opcode " +
opcode);
- }
-
- if (parts.length == 4) {
- // Example: CP.cm.mVar0.Var1.mVar2; (without weights)
- in2 = new CPOperand("", Types.ValueType.UNKNOWN,
Types.DataType.UNKNOWN);
- parseUnaryInstruction(str, in1, in2, out);
- }
- else if (parts.length == 5) {
- // CP.cm.mVar0.mVar1.Var2.mVar3; (with weights)
- in2 = new CPOperand("", Types.ValueType.UNKNOWN,
Types.DataType.UNKNOWN);
- in3 = new CPOperand("", Types.ValueType.UNKNOWN,
Types.DataType.UNKNOWN);
- parseUnaryInstruction(str, in1, in2, in3, out);
- }
-
- /*
- * Exact order of the central moment MAY NOT be known at
compilation time. We
- * first try to parse the second argument as an integer, and if
we fail, we
- * simply pass -1 so that getCMAggOpType() picks up
- * AggregateOperationTypes.INVALID. It must be updated at run
time in
- * processInstruction() method.
- */
-
- int cmOrder;
- try {
- if (in3 == null) {
- cmOrder = Integer.parseInt(in2.getName());
- }
- else {
- cmOrder = Integer.parseInt(in3.getName());
- }
- }
- catch (NumberFormatException e) {
- cmOrder = -1; // unknown at compilation time
- }
+ return
parseInstruction(CentralMomentCPInstruction.parseInstruction(str));
+ }
- CMOperator.AggregateOperationTypes opType =
CMOperator.getCMAggOpType(cmOrder);
- CMOperator cm = new CMOperator(CM.getCMFnObject(opType),
opType);
- return new CentralMomentFEDInstruction(cm, in1, in2, in3, out,
opcode, str);
+ public static CentralMomentFEDInstruction
parseInstruction(CentralMomentCPInstruction inst) {
+ return new CentralMomentFEDInstruction(inst.getOperator(),
+ inst.input1, inst.input2, inst.input3, inst.output,
+ inst.getOpcode(), inst.getInstructionString());
}
@Override
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 cc5974f..16cb5ff 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
@@ -27,7 +27,6 @@ import java.util.stream.IntStream;
import org.apache.commons.lang3.tuple.ImmutableTriple;
import org.apache.commons.lang3.tuple.Pair;
-import org.apache.sysds.common.Types;
import org.apache.sysds.runtime.DMLRuntimeException;
import org.apache.sysds.runtime.controlprogram.caching.MatrixObject;
import org.apache.sysds.runtime.controlprogram.context.ExecutionContext;
@@ -37,10 +36,9 @@ 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.functionobjects.COV;
-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;
+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;
@@ -50,44 +48,23 @@ import
org.apache.sysds.runtime.matrix.operators.COVOperator;
import org.apache.sysds.runtime.matrix.operators.Operator;
public class CovarianceFEDInstruction extends BinaryFEDInstruction {
- private CovarianceFEDInstruction(Operator op, CPOperand in1, CPOperand
in2, CPOperand out, String opcode,
- String istr) {
- super(FEDInstruction.FEDType.AggregateBinary, op, in1, in2,
out, opcode, istr);
- }
-
- private CovarianceFEDInstruction(Operator op, CPOperand in1, CPOperand
in2, CPOperand in3, CPOperand out,
- String opcode, String istr) {
+
+ private CovarianceFEDInstruction(Operator op, CPOperand in1,
+ CPOperand in2, CPOperand in3, CPOperand out, String opcode,
String istr)
+ {
super(FEDInstruction.FEDType.AggregateBinary, op, in1, in2,
in3, out, opcode, istr);
}
-
public static CovarianceFEDInstruction parseInstruction(String str) {
- CPOperand in1 = new CPOperand("", Types.ValueType.UNKNOWN,
Types.DataType.UNKNOWN);
- CPOperand in2 = new CPOperand("", Types.ValueType.UNKNOWN,
Types.DataType.UNKNOWN);
- CPOperand in3 = null;
- CPOperand out = new CPOperand("", Types.ValueType.UNKNOWN,
Types.DataType.UNKNOWN);
-
- String[] parts =
InstructionUtils.getInstructionPartsWithValueType(str);
- String opcode = parts[0];
-
- if( !opcode.equalsIgnoreCase("cov") ) {
- throw new
DMLRuntimeException("CovarianceCPInstruction.parseInstruction():: Unknown
opcode " + opcode);
- }
-
- COVOperator cov = new COVOperator(COV.getCOMFnObject());
- if ( parts.length == 4 ) {
- parseBinaryInstruction(str, in1, in2, out);
- return new CovarianceFEDInstruction(cov, in1, in2, out,
opcode, str);
- } else if ( parts.length == 5 ) {
- in3 = new CPOperand("", Types.ValueType.UNKNOWN,
Types.DataType.UNKNOWN);
- parseBinaryInstruction(str, in1, in2, in3, out);
- return new CovarianceFEDInstruction(cov, in1, in2, in3,
out, opcode, str);
- }
- else {
- throw new DMLRuntimeException("Invalid number of
arguments in Instruction: " + str);
- }
+ return
parseInstruction(CovarianceCPInstruction.parseInstruction(str));
}
+ public static CovarianceFEDInstruction
parseInstruction(CovarianceCPInstruction 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/FEDInstructionUtils.java
b/src/main/java/org/apache/sysds/runtime/instructions/fed/FEDInstructionUtils.java
index d410042..12965f6 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
@@ -38,6 +38,8 @@ import
org.apache.sysds.runtime.instructions.cp.AggregateTernaryCPInstruction;
import org.apache.sysds.runtime.instructions.cp.AggregateUnaryCPInstruction;
import org.apache.sysds.runtime.instructions.cp.BinaryCPInstruction;
import org.apache.sysds.runtime.instructions.cp.BinaryFrameScalarCPInstruction;
+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;
@@ -181,7 +183,7 @@ public class FEDInstructionUtils {
fedinst =
QuantilePickFEDInstruction.parseInstruction(inst.getInstructionString());
else if("cov".equals(instruction.getOpcode())
&& (ec.getMatrixObject(instruction.input1).isFederated(FType.ROW) ||
ec.getMatrixObject(instruction.input2).isFederated(FType.ROW)))
- fedinst =
CovarianceFEDInstruction.parseInstruction(inst.getInstructionString());
+ fedinst =
CovarianceFEDInstruction.parseInstruction((CovarianceCPInstruction)inst);
else
fedinst =
BinaryFEDInstruction.parseInstruction(
InstructionUtils.concatOperands(inst.getInstructionString(),FederatedOutput.NONE.name()));
@@ -355,7 +357,7 @@ public class FEDInstructionUtils {
MatrixObject mo1 =
ec.getMatrixObject(instruction.input1);
if(mo1.isFederatedExcept(FType.BROADCAST)) {
if(instruction.getOpcode().equalsIgnoreCase("cm"))
- fedinst =
CentralMomentFEDInstruction.parseInstruction(inst.getInstructionString());
+ fedinst =
CentralMomentFEDInstruction.parseInstruction((CentralMomentCPInstruction)inst);
else
if(inst.getOpcode().equalsIgnoreCase("qsort")) {
if(mo1.getFedMapping().getFederatedRanges().length == 1)
fedinst =
QuantileSortFEDInstruction.parseInstruction(inst.getInstructionString());
