spark and cp support

Author: Baunsgaard

src/main/java/org/apache/sysds/hops/TernaryOp.java

@@ -480,6 +480,10 @@ protected DataCharacteristics inferOutputCharacteristics( MemoTable memo )
 	}
 
 
+	public ExecType findExecTypeTernaryOp(){
+		return _etype == null ? optFindExecType(OptimizerUtils.ALLOW_TRANSITIVE_SPARK_EXEC_TYPE) : _etype;
+	}
+
 	@Override
 	protected ExecType optFindExecType(boolean transitive) 
 	{

src/main/java/org/apache/sysds/hops/rewrite/HopRewriteUtils.java

@@ -25,7 +25,6 @@
 import org.apache.sysds.api.DMLScript;
 import org.apache.sysds.common.Types.DataType;
 import org.apache.sysds.common.Types.ExecMode;
-import org.apache.sysds.common.Types.ExecType;
 import org.apache.sysds.common.Types.FileFormat;
 import org.apache.sysds.common.Types.OpOp1;
 import org.apache.sysds.common.Types.OpOp2;
@@ -1397,7 +1396,7 @@ public static boolean isSequenceSizeOfA(Hop hop, Hop A)
 	{
 		boolean ret = false;
 
-		if((hop instanceof DataGenOp)) {
+		if((hop instanceof DataGenOp) ) {
 			DataGenOp dgop = (DataGenOp) hop;
 			if(dgop.getOp() == OpOpDG.SEQ) {
 				Hop from = dgop.getInput().get(dgop.getParamIndex(Statement.SEQ_FROM));

src/main/java/org/apache/sysds/hops/rewrite/RewriteAlgebraicSimplificationStatic.java

@@ -27,6 +27,19 @@
 import java.util.List;
 import java.util.Set;
 
+import org.apache.sysds.common.Types.AggOp;
+import org.apache.sysds.common.Types.DataType;
+import org.apache.sysds.common.Types.Direction;
+import org.apache.sysds.common.Types.ExecType;
+import org.apache.sysds.common.Types.OpOp1;
+import org.apache.sysds.common.Types.OpOp2;
+import org.apache.sysds.common.Types.OpOp3;
+import org.apache.sysds.common.Types.OpOpDG;
+import org.apache.sysds.common.Types.OpOpData;
+import org.apache.sysds.common.Types.OpOpN;
+import org.apache.sysds.common.Types.ParamBuiltinOp;
+import org.apache.sysds.common.Types.ReOrgOp;
+import org.apache.sysds.common.Types.ValueType;
 import org.apache.sysds.hops.AggBinaryOp;
 import org.apache.sysds.hops.AggUnaryOp;
 import org.apache.sysds.hops.BinaryOp;
@@ -40,20 +53,8 @@
 import org.apache.sysds.hops.ReorgOp;
 import org.apache.sysds.hops.TernaryOp;
 import org.apache.sysds.hops.UnaryOp;
-import org.apache.sysds.common.Types.AggOp;
-import org.apache.sysds.common.Types.Direction;
-import org.apache.sysds.common.Types.OpOp1;
-import org.apache.sysds.common.Types.OpOp2;
-import org.apache.sysds.common.Types.OpOp3;
-import org.apache.sysds.common.Types.OpOpDG;
-import org.apache.sysds.common.Types.OpOpData;
-import org.apache.sysds.common.Types.OpOpN;
-import org.apache.sysds.common.Types.ParamBuiltinOp;
-import org.apache.sysds.common.Types.ReOrgOp;
 import org.apache.sysds.parser.DataExpression;
 import org.apache.sysds.parser.Statement;
-import org.apache.sysds.common.Types.DataType;
-import org.apache.sysds.common.Types.ValueType;
 
 /**
  * Rule: Algebraic Simplifications. Simplifies binary expressions
@@ -2199,10 +2200,10 @@ private static void removeTWriteTReadPairs(ArrayList<Hop> roots) {
 
 	private static Hop fuseSeqAndTableExpand(Hop hi) {
 
-		if(hi instanceof TernaryOp) {
+		if(hi instanceof TernaryOp ) {
 			TernaryOp thop = (TernaryOp) hi;
 			thop.getOp();
-			if(thop.getOp() == OpOp3.CTABLE) {
+			if(thop.getOp() == OpOp3.CTABLE && thop.findExecTypeTernaryOp() == ExecType.CP) {
 				Hop input1 = thop.getInput(0);
 				Hop input2 = thop.getInput(1);
 				// Hop input3 = thop.getInput().size() == 3 ? thop.getInput(2) : null;

src/test/java/org/apache/sysds/test/functions/compress/wordembedding/WordEmbeddingUseCase.java

@@ -21,6 +21,7 @@
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
 
 import java.io.File;
 
@@ -80,6 +81,31 @@ public void testWordEmb_moreUniqueWordsThanSentences() {
 		wordEmb(100, 200, 5, 2, ExecType.CP, "01");
 	}
 
+	@Test
+	public void testWordEmbSP() {
+		wordEmb(10, 2, 2, 2, ExecType.SPARK, "01");
+	}
+
+	@Test
+	public void testWordEmb_mediumSP() {
+		wordEmb(100, 30, 4, 3, ExecType.SPARK, "01");
+	}
+
+	@Test
+	public void testWordEmb_bigWordsSP() {
+		wordEmb(10, 2, 2, 10, ExecType.SPARK, "01");
+	}
+
+	@Test
+	public void testWordEmb_longSentencesSP() {
+		wordEmb(100, 30, 5, 2, ExecType.SPARK, "01");
+	}
+
+	@Test
+	public void testWordEmb_moreUniqueWordsThanSentencesSP() {
+		wordEmb(100, 200, 5, 2, ExecType.SPARK, "01");
+	}
+
 	public void wordEmb(int rows, int unique, int l, int embeddingSize, ExecType instType, String name) {
 
 		OptimizerUtils.ALLOW_SCRIPT_LEVEL_COMPRESS_COMMAND = true;
@@ -101,12 +127,16 @@ public void wordEmb(int rows, int unique, int l, int embeddingSize, ExecType ins
 			MatrixBlock W = TestUtils.generateTestMatrixBlock(unique, embeddingSize, 1.0, -1, 1, 32);
 			writeBinaryWithMTD("W", W);
 
-			runTest(null);
+			String r = runTest(null).toString();
 
 			MatrixBlock R = TestUtils.readBinary(output("R"));
 
 			analyzeResult(X, W, R, l);
 
+			if( instType == ExecType.CP && heavyHittersContainsString("seq")){
+				fail("cp should not have seq instruction\n" + r);
+			}
+
 		}
 		catch(Exception e) {
 			e.printStackTrace();