rewrite introduced ... maybe not working in spark and federated

Author: Baunsgaard

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

@@ -651,7 +651,9 @@ private boolean isSequenceRewriteApplicable( boolean left )
 			{
 				Hop input1 = getInput().get(0);
 				Hop input2 = getInput().get(1);
-				if( input1.getDataType() == DataType.MATRIX && input2.getDataType() == DataType.MATRIX )
+				if( (input1.getDataType() == DataType.MATRIX 
+					|| input1.getDataType() == DataType.SCALAR )
+					 && input2.getDataType() == DataType.MATRIX )
 				{
 					//probe rewrite on left input
 					if( left && input1 instanceof DataGenOp )
@@ -663,6 +665,9 @@ private boolean isSequenceRewriteApplicable( boolean left )
 								  || dgop.getIncrementValue()==1.0; //set by recompiler
 						}
 					}
+					if( left && input1 instanceof LiteralOp){
+						ret = true;
+					}
 					//probe rewrite on right input
 					if( !left && input2 instanceof DataGenOp )
 					{

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

@@ -1397,7 +1397,7 @@ public static boolean isSequenceSizeOfA(Hop hop, Hop A)
 	{
 		boolean ret = false;
 
-		if((hop instanceof DataGenOp) && hop.getExecType() == ExecType.CP && A.getExecType() != ExecType.CP ) {
+		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

@@ -198,6 +198,7 @@ private void rule_AlgebraicSimplification(Hop hop, boolean descendFirst)
 			//hi = removeUnecessaryPPred(hop, hi, i);            //e.g., ppred(X,X,"==")->matrix(1,rows=nrow(X),cols=ncol(X))
 
 			hi = fixNonScalarPrint(hop, hi, i);                  //e.g., print(m) -> print(toString(m))
+			hi = fuseSeqAndTableExpand(hi);
 
 			//process childs recursively after rewrites (to investigate pattern newly created by rewrites)
 			if( !descendFirst )
@@ -2195,4 +2196,24 @@ private static void removeTWriteTReadPairs(ArrayList<Hop> roots) {
 			}
 		}
 	}
+
+	private static Hop fuseSeqAndTableExpand(Hop hi) {
+
+		if(hi instanceof TernaryOp) {
+			TernaryOp thop = (TernaryOp) hi;
+			thop.getOp();
+			if(thop.getOp() == OpOp3.CTABLE) {
+				Hop input1 = thop.getInput(0);
+				Hop input2 = thop.getInput(1);
+				// Hop input3 = thop.getInput().size() == 3 ? thop.getInput(2) : null;
+
+				if(HopRewriteUtils.isSequenceSizeOfA(input1, input2)) {
+					Hop literal = new LiteralOp(input1.getDim1());
+					HopRewriteUtils.replaceChildReference(hi, input1, literal);
+				}
+			}
+
+		}
+		return hi;
+	}
 }

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

@@ -66,8 +66,9 @@ private CtableFEDInstruction(CPOperand in1, CPOperand in2, CPOperand in3, CPOper
 	}
 
 	public static CtableFEDInstruction parseInstruction(CtableCPInstruction inst, ExecutionContext ec) {
-		if((inst.getOpcode().equalsIgnoreCase("ctable") || inst.getOpcode().equalsIgnoreCase("ctableexpand")) &&
-			(ec.getCacheableData(inst.input1).isFederated(FType.ROW) ||
+		if((inst.getOpcode().equalsIgnoreCase("ctable") 
+		|| inst.getOpcode().equalsIgnoreCase("ctableexpand")) &&
+				(inst.input1.isMatrix() && ec.getCacheableData(inst.input1).isFederated(FType.ROW) ||
 				(inst.input2.isMatrix() && ec.getCacheableData(inst.input2).isFederated(FType.ROW)) ||
 				(inst.input3.isMatrix() && ec.getCacheableData(inst.input3).isFederated(FType.ROW))))
 			return CtableFEDInstruction.parseInstruction(inst);

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

@@ -102,7 +102,7 @@ public void wordEmb(int rows, int unique, int l, int embeddingSize, ExecType ins
 			writeBinaryWithMTD("W", W);
 
 			runTest(null);
-
+			
 			MatrixBlock R = TestUtils.readBinary(output("R"));
 
 			analyzeResult(X, W, R, l);