[SYSTEMDS-3762] Cumulative Row Aggregates and Rewrites

Closes #2279.

Co-authored-by: Vien Thanh Thai <thai@campus.tu-berlin.de>
Co-authored-by: Sandra Shi <sandra.shi@campus.tu-berlin.de>

Author: 3 people

docs/site/dml-language-reference.md

@@ -702,7 +702,8 @@ quantile () | The p-quantile for a random variable X is the value x such that Pr
 quantile () | Returns a column matrix with list of all quantiles requested in P. | Input: (X &lt;(n x 1) matrix&gt;, [W &lt;(n x 1) matrix&gt;),] P &lt;(q x 1) matrix&gt;) <br/> Output: matrix | quantile(X, P) <br/> quantile(X, W, P)
 median() | Computes the median in a given column matrix of values | Input: (X &lt;(n x 1) matrix&gt;, [W &lt;(n x 1) matrix&gt;),]) <br/> Output: &lt;scalar&gt; | median(X) <br/> median(X,W)
 rowSums() <br/> rowMeans() <br/> rowVars() <br/> rowSds() <br/> rowMaxs() <br/> rowMins() | Row-wise computations -- for each row, compute the sum/mean/variance/stdDev/max/min of cell value | Input: matrix <br/> Output: (n x 1) matrix | rowSums(X) <br/> rowMeans(X) <br/> rowVars(X) <br/> rowSds(X) <br/> rowMaxs(X) <br/> rowMins(X)
-cumsum() | Column prefix-sum (For row-prefix sum, use cumsum(t(X)) | Input: matrix <br/> Output: matrix of the same dimensions | A = matrix("1 2 3 4 5 6", rows=3, cols=2) <br/> B = cumsum(A) <br/> The output matrix B = [[1, 2], [4, 6], [9, 12]]
+cumsum() | Column prefix-sum | Input: matrix <br/> Output: matrix of the same dimensions | A = matrix("1 2 3 4 5 6", rows=3, cols=2) <br/> B = cumsum(A) <br/> The output matrix B = [[1, 2], [4, 6], [9, 12]]
+rowcumsum() | Row prefix-sum | Input: matrix <br/> Output: matrix of the same dimensions | A = matrix("1 2 3 4 5 6", rows=2, cols=3) <br/> B = rowcumsum(A) <br/> The output matrix B = [[1, 3, 6], [4, 9, 15]]
 cumprod() | Column prefix-prod (For row-prefix prod, use cumprod(t(X)) | Input: matrix <br/> Output: matrix of the same dimensions | A = matrix("1 2 3 4 5 6", rows=3, cols=2) <br/> B = cumprod(A) <br/> The output matrix B = [[1, 2], [3, 8], [15, 48]]
 cummin() | Column prefix-min (For row-prefix min, use cummin(t(X)) | Input: matrix <br/> Output: matrix of the same dimensions | A = matrix("3 4 1 6 5 2", rows=3, cols=2) <br/> B = cummin(A) <br/> The output matrix B = [[3, 4], [1, 4], [1, 2]]
 cummax() | Column prefix-max (For row-prefix min, use cummax(t(X)) | Input: matrix <br/> Output: matrix of the same dimensions | A = matrix("3 4 1 6 5 2", rows=3, cols=2) <br/> B = cummax(A) <br/> The output matrix B = [[3, 4], [3, 6], [5, 6]]

src/main/java/org/apache/sysds/common/Builtins.java

@@ -291,6 +291,7 @@ public enum Builtins {
 	ROLL("roll", false),
 	ROUND("round", false),
 	ROW_COUNT_DISTINCT("rowCountDistinct",false),
+	ROWCUMSUM("rowcumsum", false),
 	ROWINDEXMAX("rowIndexMax", false),
 	ROWINDEXMIN("rowIndexMin", false),
 	ROWMAX("rowMaxs", false),

src/main/java/org/apache/sysds/common/Opcodes.java

@@ -34,6 +34,7 @@ public enum Opcodes {
 	UAKP("uak+", InstructionType.AggregateUnary),
 	UARKP("uark+", InstructionType.AggregateUnary),
 	UACKP("uack+", InstructionType.AggregateUnary),
+	UARCKP("uarck+", InstructionType.AggregateUnary),
 	UASQKP("uasqk+", InstructionType.AggregateUnary),
 	UARSQKP("uarsqk+", InstructionType.AggregateUnary),
 	UACSQKP("uacsqk+", InstructionType.AggregateUnary),
@@ -151,6 +152,7 @@ public enum Opcodes {
 	CEIL("ceil", InstructionType.Unary),
 	FLOOR("floor", InstructionType.Unary),
 	UCUMKP("ucumk+", InstructionType.Unary),
+	UROWCUMKP("urowcumk+", InstructionType.Unary),
 	UCUMM("ucum*", InstructionType.Unary),
 	UCUMKPM("ucumk+*", InstructionType.Unary),
 	UCUMMIN("ucummin", InstructionType.Unary),
@@ -383,6 +385,7 @@ public enum Opcodes {
 	UCUMACMIN("ucumacmin", InstructionType.CumsumAggregate),
 	UCUMACMAX("ucumacmax", InstructionType.CumsumAggregate),
 	BCUMOFFKP("bcumoffk+", InstructionType.CumsumOffset),
+	BROWCUMOFFKP("browcumoffk+", InstructionType.CumsumOffset),
 	BCUMOFFM("bcumoff*", InstructionType.CumsumOffset),
 	BCUMOFFPM("bcumoff+*", InstructionType.CumsumOffset),
 	BCUMOFFMIN("bcumoffmin", InstructionType.CumsumOffset),

src/main/java/org/apache/sysds/common/Types.java

@@ -547,7 +547,7 @@ public enum OpOp1 {
 		CEIL, CHOLESKY, COS, COSH, CUMMAX, CUMMIN, CUMPROD, CUMSUM,
 		CUMSUMPROD, DET, DETECTSCHEMA, COLNAMES, EIGEN, EXISTS, EXP, FLOOR, INVERSE,
 		IQM, ISNA, ISNAN, ISINF, LENGTH, LINEAGE, LOG, NCOL, NOT, NROW,
-		MEDIAN, PREFETCH, PRINT, ROUND, SIN, SINH, SIGN, SOFTMAX, SQRT, STOP, _EVICT,
+		MEDIAN, PREFETCH, PRINT, ROUND, ROWCUMSUM, SIN, SINH, SIGN, SOFTMAX, SQRT, STOP, _EVICT,
 		SVD, TAN, TANH, TYPEOF, TRIGREMOTE, SQRT_MATRIX_JAVA,
 		//fused ML-specific operators for performance 
 		SPROP, //sample proportion: P * (1 - P)
@@ -591,6 +591,7 @@ public String toString() {
 				case MULT2:           return Opcodes.MULT2.toString();
 				case NOT:             return Opcodes.NOT.toString();
 				case POW2:            return Opcodes.POW2.toString();
+				case ROWCUMSUM:		  return Opcodes.UROWCUMKP.toString();
 				case TYPEOF:          return Opcodes.TYPEOF.toString();
 				default:              return name().toLowerCase();
 			}
@@ -610,6 +611,7 @@ public static OpOp1 valueOfByOpcode(String opcode) {
 				case "ucummin": return CUMMIN;
 				case "ucum*":   return CUMPROD;
 				case "ucumk+":  return CUMSUM;
+				case "urowcumk+":  return ROWCUMSUM;
 				case "ucumk+*": return CUMSUMPROD;
 				case "detectSchema":    return DETECTSCHEMA;
 				case "*2":      return MULT2;

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

@@ -205,6 +205,7 @@ private void rule_AlgebraicSimplification(Hop hop, boolean descendFirst)
 			hi = simplifyNotOverComparisons(hop, hi, i);         //e.g., !(A>B) -> (A<=B)
 			hi = simplifyMatrixScalarPMOperation(hop, hi, i);    //e.g., a-A-b -> (a-b)-A; a+A-b -> (a-b)+A
 			//hi = removeUnecessaryPPred(hop, hi, i);            //e.g., ppred(X,X,"==")->matrix(1,rows=nrow(X),cols=ncol(X))
+			hi = simplifyTransposedCumsum(hop, hi, i);					 //e.g., t(cumsum(t(X))) -> rowcumsum(X)
 
 			//process childs recursively after rewrites (to investigate pattern newly created by rewrites)
 			if( !descendFirst )
@@ -214,6 +215,28 @@ private void rule_AlgebraicSimplification(Hop hop, boolean descendFirst)
 		hop.setVisited();
 	}
 
+	private static Hop simplifyTransposedCumsum( Hop parent, Hop hi, int pos )
+	{
+		//e.g., t(cumsum(t(X))) -> rowcumsum(X)
+		if(   HopRewriteUtils.isTransposeOperation(hi)
+				&& hi.getInput(0) instanceof UnaryOp
+				&& ((UnaryOp)hi.getInput(0)).getOp() == OpOp1.CUMSUM
+				&& hi.getInput(0).getParent().size() == 1
+				&& HopRewriteUtils.isTransposeOperation(hi.getInput(0).getInput(0), 1)) //inner transpose with single consumer
+		{
+			UnaryOp cumsum=(UnaryOp)hi.getInput(0);
+			Hop innerMatrix = cumsum.getInput(0).getInput(0);
+
+			UnaryOp rowcumsumOp = HopRewriteUtils.createUnary(innerMatrix, OpOp1.ROWCUMSUM);
+			HopRewriteUtils.replaceChildReference(parent,hi, rowcumsumOp, pos);
+
+			hi = rowcumsumOp;
+			LOG.debug("Applied simplifyTransposedCumsum (line "+hi.getBeginLine()+").");
+		}
+
+		return hi;
+	}
+
 	private Hop simplifyMatrixScalarPMOperation(Hop parent, Hop hi, int pos) {
 		if (!(hi instanceof BinaryOp))
 			return hi;

src/main/java/org/apache/sysds/parser/BuiltinFunctionExpression.java

@@ -1034,6 +1034,7 @@ else if( getAllExpr().length == 2 ) { //binary
 			break;
 
 		case CUMSUM:
+		case ROWCUMSUM:
 		case CUMPROD:
 		case CUMSUMPROD:
 		case CUMMIN:

src/main/java/org/apache/sysds/parser/DMLTranslator.java

@@ -2616,6 +2616,7 @@ else if ( sop.equalsIgnoreCase(Opcodes.NOTEQUAL.toString()) )
 		case CEIL:
 		case FLOOR:
 		case CUMSUM:
+		case ROWCUMSUM:
 		case CUMPROD:
 		case CUMSUMPROD:
 		case CUMMIN:

src/main/java/org/apache/sysds/runtime/functionobjects/Builtin.java

@@ -49,7 +49,7 @@ public class Builtin extends ValueFunction
 
 	public enum BuiltinCode { AUTODIFF, SIN, COS, TAN, SINH, COSH, TANH, ASIN, ACOS, ATAN, LOG, LOG_NZ, MIN,
 		MAX, ABS, SIGN, SQRT, EXP, PLOGP, PRINT, PRINTF, NROW, NCOL, LENGTH, LINEAGE, ROUND, MAXINDEX, MININDEX,
-		STOP, CEIL, FLOOR, CUMSUM, CUMPROD, CUMMIN, CUMMAX, CUMSUMPROD, INVERSE, SPROP, SIGMOID, EVAL, LIST,
+		STOP, CEIL, FLOOR, CUMSUM, ROWCUMSUM, CUMPROD, CUMMIN, CUMMAX, CUMSUMPROD, INVERSE, SPROP, SIGMOID, EVAL, LIST,
 		TYPEOF, APPLY_SCHEMA, DETECTSCHEMA, ISNA, ISNAN, ISINF, DROP_INVALID_TYPE, 
 		DROP_INVALID_LENGTH, VALUE_SWAP, FRAME_ROW_REPLICATE,
 		MAP, COUNT_DISTINCT, COUNT_DISTINCT_APPROX, UNIQUE}
@@ -95,6 +95,7 @@ public enum BuiltinCode { AUTODIFF, SIN, COS, TAN, SINH, COSH, TANH, ASIN, ACOS,
 		String2BuiltinCode.put( "ceil"   , BuiltinCode.CEIL);
 		String2BuiltinCode.put( "floor"  , BuiltinCode.FLOOR);
 		String2BuiltinCode.put( "ucumk+" , BuiltinCode.CUMSUM);
+		String2BuiltinCode.put( "urowcumk+" , BuiltinCode.ROWCUMSUM);
 		String2BuiltinCode.put( "ucum*"  , BuiltinCode.CUMPROD);
 		String2BuiltinCode.put( "ucumk+*", BuiltinCode.CUMSUMPROD);
 		String2BuiltinCode.put( "ucummin", BuiltinCode.CUMMIN);

src/main/java/org/apache/sysds/runtime/instructions/InstructionUtils.java

@@ -557,6 +557,8 @@ public static AggregateUnaryOperator parseBasicCumulativeAggregateUnaryOperator(
 		Builtin f = (Builtin)uop.fn;
 		if( f.getBuiltinCode()==BuiltinCode.CUMSUM ) 
 			return parseBasicAggregateUnaryOperator(Opcodes.UACKP.toString()) ;
+		else if( f.getBuiltinCode()==BuiltinCode.ROWCUMSUM )
+			return parseBasicAggregateUnaryOperator(Opcodes.UARCKP.toString()) ;
 		else if( f.getBuiltinCode()==BuiltinCode.CUMPROD ) 
 			return parseBasicAggregateUnaryOperator(Opcodes.UACM.toString()) ;
 		else if( f.getBuiltinCode()==BuiltinCode.CUMMIN ) 

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

@@ -56,6 +56,8 @@ private CumulativeOffsetSPInstruction(Operator op, CPOperand in1, CPOperand in2,
 
 		if (Opcodes.BCUMOFFKP.toString().equals(opcode))
 			_uop = new UnaryOperator(Builtin.getBuiltinFnObject("ucumk+"));
+		else if (Opcodes.BROWCUMOFFKP.toString().equals(opcode))
+			_uop = new UnaryOperator(Builtin.getBuiltinFnObject("urowcumk+"));
 		else if (Opcodes.BCUMOFFM.toString().equals(opcode))
 			_uop = new UnaryOperator(Builtin.getBuiltinFnObject("ucum*"));
 		else if (Opcodes.BCUMOFFPM.toString().equals(opcode)) {