You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@systemml.apache.org by lr...@apache.org on 2015/11/19 21:47:31 UTC
[49/50] [abbrv] incubator-systemml git commit: [SYSML-345] Add New
Unary Aggregate "SUM_SQ" Operator
[SYSML-345] Add New Unary Aggregate "SUM_SQ" Operator
This introduces a new fused SUM_SQ unary aggregate operation
into the existing unary aggregate framework, as well as a new
simplification rewrite that transforms a SUM(X^2) HOP expression
into the new SUM_SQ(X) expression. This operation fuses a summation
and square operation together, and applies to full (sum),
row (rowSums), and column (colSums) summation variants.
At a more detailed level, this contains the following components:
Creates a new SUM_SQ operation type for the existing AggUnaryOp HOP.
Rewrites SUM(POW(X, 2)) HOP expressions to the new SUM_SQ(X) HOP expression,
but only if X is not a column vector.
Creates a new KahanSumSq LOP operation type for the existing Aggregate
and PartialAggregate LOPs.
Compiles the SUM_SQ HOP type to the KahanSumSq LOP type.
Creates new runtime opcodes: "asqk+", "uasqk+", "uarsqk+", "uacsqk+".
Compiles the KahanSumSq LOP type to new opcodes.
Creates a new KahanPlusSq runtime function.
Parses the new opcodes into runtime aggregate operators that use the new
KahanPlusSq runtime function.
Creates a new KAHAN_SUM_SQ aggregate type in the core unary aggregate library.
Matches the runtime aggregate operators using the KahanPlusSq function to the
KAHAN_SUM_SQ case.
Extends the core unary aggregate library with the new sum squared functions
for the KAHAN_SUM_SQ case.
Project: http://git-wip-us.apache.org/repos/asf/incubator-systemml/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-systemml/commit/31bf3285
Tree: http://git-wip-us.apache.org/repos/asf/incubator-systemml/tree/31bf3285
Diff: http://git-wip-us.apache.org/repos/asf/incubator-systemml/diff/31bf3285
Branch: refs/heads/master
Commit: 31bf328559cb8d78c05d89cd473a1cd50379b026
Parents: 5be39df
Author: Mike Dusenberry <mw...@us.ibm.com>
Authored: Mon Nov 16 14:15:44 2015 -0800
Committer: Luciano Resende <lr...@apache.org>
Committed: Mon Nov 16 14:15:44 2015 -0800
----------------------------------------------------------------------
.../java/com/ibm/bi/dml/hops/AggUnaryOp.java | 7 +-
src/main/java/com/ibm/bi/dml/hops/Hop.java | 5 +-
.../hops/cost/CostEstimatorStaticRuntime.java | 19 +-
.../dml/hops/recompile/LiteralReplacement.java | 6 +-
.../RewriteAlgebraicSimplificationDynamic.java | 58 ++-
.../java/com/ibm/bi/dml/lops/Aggregate.java | 15 +-
.../com/ibm/bi/dml/lops/PartialAggregate.java | 13 +-
.../runtime/functionobjects/KahanFunction.java | 43 ++
.../dml/runtime/functionobjects/KahanPlus.java | 2 +-
.../runtime/functionobjects/KahanPlusSq.java | 101 +++++
.../instructions/CPInstructionParser.java | 3 +
.../runtime/instructions/InstructionUtils.java | 30 +-
.../instructions/MRInstructionParser.java | 4 +
.../instructions/SPInstructionParser.java | 3 +
.../instructions/mr/AggregateInstruction.java | 3 +-
.../dml/runtime/matrix/data/LibMatrixAgg.java | 392 ++++++++++++++++---
.../bi/dml/runtime/matrix/data/MatrixBlock.java | 43 +-
.../matrix/operators/AggregateOperator.java | 3 +-
.../operators/AggregateUnaryOperator.java | 6 +-
.../functions/aggregate/ColSumsSqTest.java | 268 +++++++++++++
.../functions/aggregate/RowSumsSqTest.java | 268 +++++++++++++
.../functions/aggregate/SumSqTest.java | 268 +++++++++++++
.../scripts/functions/aggregate/ColSumsSq.R | 26 ++
.../scripts/functions/aggregate/ColSumsSq.dml | 21 +
.../scripts/functions/aggregate/RowSumsSq.R | 26 ++
.../scripts/functions/aggregate/RowSumsSq.dml | 21 +
src/test/scripts/functions/aggregate/SumSq.R | 26 ++
src/test/scripts/functions/aggregate/SumSq.dml | 21 +
28 files changed, 1607 insertions(+), 94 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/31bf3285/src/main/java/com/ibm/bi/dml/hops/AggUnaryOp.java
----------------------------------------------------------------------
diff --git a/src/main/java/com/ibm/bi/dml/hops/AggUnaryOp.java b/src/main/java/com/ibm/bi/dml/hops/AggUnaryOp.java
index ab4e73f..a0be584 100644
--- a/src/main/java/com/ibm/bi/dml/hops/AggUnaryOp.java
+++ b/src/main/java/com/ibm/bi/dml/hops/AggUnaryOp.java
@@ -341,6 +341,7 @@ public class AggUnaryOp extends Hop implements MultiThreadedHop
val = dim2 * OptimizerUtils.INT_SIZE;
break;
case SUM:
+ case SUM_SQ:
//worst-case correction LASTROW / LASTCOLUMN
if( _direction == Direction.Col ) //(potentially sparse)
val = OptimizerUtils.estimateSizeExactSparsity(1, dim2, sparsity);
@@ -695,9 +696,9 @@ public class AggUnaryOp extends Hop implements MultiThreadedHop
public boolean isTransposeSafe()
{
boolean ret = (_direction == Direction.RowCol) && //full aggregate
- (_op == AggOp.SUM || _op == AggOp.MIN || //valid aggregration functions
- _op == AggOp.MAX || _op == AggOp.PROD ||
- _op == AggOp.MEAN);
+ (_op == AggOp.SUM || _op == AggOp.SUM_SQ || //valid aggregration functions
+ _op == AggOp.MIN || _op == AggOp.MAX ||
+ _op == AggOp.PROD || _op == AggOp.MEAN);
//note: trace and maxindex are not transpose-safe.
return ret;
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/31bf3285/src/main/java/com/ibm/bi/dml/hops/Hop.java
----------------------------------------------------------------------
diff --git a/src/main/java/com/ibm/bi/dml/hops/Hop.java b/src/main/java/com/ibm/bi/dml/hops/Hop.java
index 66c9f03..ab328a8 100644
--- a/src/main/java/com/ibm/bi/dml/hops/Hop.java
+++ b/src/main/java/com/ibm/bi/dml/hops/Hop.java
@@ -1059,7 +1059,7 @@ public abstract class Hop
public enum AggOp {
- SUM, MIN, MAX, TRACE, PROD, MEAN, MAXINDEX, MININDEX
+ SUM, SUM_SQ, MIN, MAX, TRACE, PROD, MEAN, MAXINDEX, MININDEX
};
public enum ReOrgOp {
@@ -1111,7 +1111,7 @@ public abstract class Hop
static {
HopsAgg2Lops = new HashMap<Hop.AggOp, com.ibm.bi.dml.lops.Aggregate.OperationTypes>();
HopsAgg2Lops.put(AggOp.SUM, com.ibm.bi.dml.lops.Aggregate.OperationTypes.KahanSum);
- // HopsAgg2Lops.put(AggOp.SUM, dml.lops.Aggregate.OperationTypes.Sum);
+ HopsAgg2Lops.put(AggOp.SUM_SQ, com.ibm.bi.dml.lops.Aggregate.OperationTypes.KahanSumSq);
HopsAgg2Lops.put(AggOp.TRACE, com.ibm.bi.dml.lops.Aggregate.OperationTypes.KahanTrace);
HopsAgg2Lops.put(AggOp.MIN, com.ibm.bi.dml.lops.Aggregate.OperationTypes.Min);
HopsAgg2Lops.put(AggOp.MAX, com.ibm.bi.dml.lops.Aggregate.OperationTypes.Max);
@@ -1385,6 +1385,7 @@ public abstract class Hop
static {
HopsAgg2String = new HashMap<Hop.AggOp, String>();
HopsAgg2String.put(AggOp.SUM, "+");
+ HopsAgg2String.put(AggOp.SUM_SQ, "sq+");
HopsAgg2String.put(AggOp.PROD, "*");
HopsAgg2String.put(AggOp.MIN, "min");
HopsAgg2String.put(AggOp.MAX, "max");
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/31bf3285/src/main/java/com/ibm/bi/dml/hops/cost/CostEstimatorStaticRuntime.java
----------------------------------------------------------------------
diff --git a/src/main/java/com/ibm/bi/dml/hops/cost/CostEstimatorStaticRuntime.java b/src/main/java/com/ibm/bi/dml/hops/cost/CostEstimatorStaticRuntime.java
index e14ac03..67d9cb2 100644
--- a/src/main/java/com/ibm/bi/dml/hops/cost/CostEstimatorStaticRuntime.java
+++ b/src/main/java/com/ibm/bi/dml/hops/cost/CostEstimatorStaticRuntime.java
@@ -919,10 +919,11 @@ public class CostEstimatorStaticRuntime extends CostEstimator
case AggregateTernary: //opcodes: tak+*
return 6 * d1m * d1n; //2*1(*) + 4 (k+)
- case AggregateUnary: //opcodes: uak+, uark+, uack+, uamean, uarmean, uacmean,
- // uamax, uarmax, uarimax, uacmax, uamin, uarmin, uacmin,
- // ua+, uar+, uac+, ua*, uatrace, uaktrace,
- // nrow, ncol, length, cm
+ case AggregateUnary: //opcodes: uak+, uark+, uack+, uasqk+, uarsqk+, uacsqk+,
+ // uamean, uarmean, uacmean,
+ // uamax, uarmax, uarimax, uacmax, uamin, uarmin, uacmin,
+ // ua+, uar+, uac+, ua*, uatrace, uaktrace,
+ // nrow, ncol, length, cm
if( optype.equals("nrow") || optype.equals("ncol") || optype.equals("length") )
return DEFAULT_NFLOP_NOOP;
@@ -949,6 +950,8 @@ public class CostEstimatorStaticRuntime extends CostEstimator
}
else if( optype.equals("uak+") || optype.equals("uark+") || optype.equals("uark+"))
return 4 * d1m * d1n; //1*k+
+ else if( optype.equals("uasqk+") || optype.equals("uarsqk+") || optype.equals("uacsqk+"))
+ return 5 * d1m * d1n; // +1 for multiplication to square term
else if( optype.equals("uamean") || optype.equals("uarmean") || optype.equals("uacmean"))
return 7 * d1m * d1n; //1*k+
else if( optype.equals("uamax") || optype.equals("uarmax") || optype.equals("uacmax")
@@ -1183,12 +1186,14 @@ public class CostEstimatorStaticRuntime extends CostEstimator
{
switch(mrtype)
{
- case Aggregate: //opcodes: a+, ak+, a*, amax, amin, amean
+ case Aggregate: //opcodes: a+, ak+, asqk+, a*, amax, amin, amean
//TODO should be aggregate unary
int numMap = Integer.parseInt(args[0]);
if( optype.equals("ak+") )
- return 4 * numMap * d1m * d1n * d1s;
- else
+ return 4 * numMap * d1m * d1n * d1s;
+ else if( optype.equals("asqk+") )
+ return 5 * numMap * d1m * d1n * d1s; // +1 for multiplication to square term
+ else
return numMap * d1m * d1n * d1s;
case AggregateBinary: //opcodes: cpmm, rmm, mapmult
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/31bf3285/src/main/java/com/ibm/bi/dml/hops/recompile/LiteralReplacement.java
----------------------------------------------------------------------
diff --git a/src/main/java/com/ibm/bi/dml/hops/recompile/LiteralReplacement.java b/src/main/java/com/ibm/bi/dml/hops/recompile/LiteralReplacement.java
index 666f598..4a645ff 100644
--- a/src/main/java/com/ibm/bi/dml/hops/recompile/LiteralReplacement.java
+++ b/src/main/java/com/ibm/bi/dml/hops/recompile/LiteralReplacement.java
@@ -486,6 +486,7 @@ public class LiteralReplacement
{
boolean cdir = (auop.getDirection() == Direction.RowCol);
boolean cop = ( auop.getOp() == AggOp.SUM
+ || auop.getOp() == AggOp.SUM_SQ
|| auop.getOp() == AggOp.MIN
|| auop.getOp() == AggOp.MAX );
@@ -512,7 +513,10 @@ public class LiteralReplacement
case SUM:
val = mb.sum();
break;
- case MIN:
+ case SUM_SQ:
+ val = mb.sumSq();
+ break;
+ case MIN:
val = mb.min();
break;
case MAX:
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/31bf3285/src/main/java/com/ibm/bi/dml/hops/rewrite/RewriteAlgebraicSimplificationDynamic.java
----------------------------------------------------------------------
diff --git a/src/main/java/com/ibm/bi/dml/hops/rewrite/RewriteAlgebraicSimplificationDynamic.java b/src/main/java/com/ibm/bi/dml/hops/rewrite/RewriteAlgebraicSimplificationDynamic.java
index a824de2..3c50a3a 100644
--- a/src/main/java/com/ibm/bi/dml/hops/rewrite/RewriteAlgebraicSimplificationDynamic.java
+++ b/src/main/java/com/ibm/bi/dml/hops/rewrite/RewriteAlgebraicSimplificationDynamic.java
@@ -61,11 +61,11 @@ public class RewriteAlgebraicSimplificationDynamic extends HopRewriteRule
//valid aggregation operation types for rowOp to Op conversions (not all operations apply)
- private static AggOp[] LOOKUP_VALID_ROW_COL_AGGREGATE = new AggOp[]{AggOp.SUM, AggOp.MIN, AggOp.MAX, AggOp.MEAN};
+ private static AggOp[] LOOKUP_VALID_ROW_COL_AGGREGATE = new AggOp[]{AggOp.SUM, AggOp.SUM_SQ, AggOp.MIN, AggOp.MAX, AggOp.MEAN};
//valid aggregation operation types for empty (sparse-safe) operations (not all operations apply)
//AggOp.MEAN currently not due to missing count/corrections
- private static AggOp[] LOOKUP_VALID_EMPTY_AGGREGATE = new AggOp[]{AggOp.SUM, AggOp.MIN, AggOp.MAX, AggOp.PROD, AggOp.TRACE};
+ private static AggOp[] LOOKUP_VALID_EMPTY_AGGREGATE = new AggOp[]{AggOp.SUM, AggOp.SUM_SQ, AggOp.MIN, AggOp.MAX, AggOp.PROD, AggOp.TRACE};
//valid unary operation types for empty (sparse-safe) operations (not all operations apply)
private static OpOp1[] LOOKUP_VALID_EMPTY_UNARY = new OpOp1[]{OpOp1.ABS, OpOp1.SIN, OpOp1.TAN, OpOp1.SQRT, OpOp1.ROUND, OpOp1.CUMSUM};
@@ -160,7 +160,8 @@ public class RewriteAlgebraicSimplificationDynamic extends HopRewriteRule
hi = simplifySumDiagToTrace(hi); //e.g., sum(diag(X)) -> trace(X); if col vector
hi = pushdownBinaryOperationOnDiag(hop, hi, i); //e.g., diag(X)*7 -> diag(X*7); if col vector
hi = simplifyDotProductSum(hop, hi, i); //e.g., sum(v^2) -> t(v)%*%v if ncol(v)==1
- hi = reorderMinusMatrixMult(hop, hi, i); //e.g., (-t(X))%*%y->-(t(X)%*%y), TODO size
+ hi = fuseSumSquared(hop, hi, i); //e.g., sum(X^2) -> sumSq(X), if ncol(X)>1
+ hi = reorderMinusMatrixMult(hop, hi, i); //e.g., (-t(X))%*%y->-(t(X)%*%y), TODO size
hi = simplifySumMatrixMult(hop, hi, i); //e.g., sum(A%*%B) -> sum(t(colSums(A))*rowSums(B)), if not dot product
hi = simplifyEmptyBinaryOperation(hop, hi, i); //e.g., X*Y -> matrix(0,nrow(X), ncol(X)) / X+Y->X / X-Y -> X
hi = simplifyScalarMVBinaryOperation(hi); //e.g., X*y -> X*as.scalar(y), if y is a 1-1 matrix
@@ -1411,6 +1412,57 @@ public class RewriteAlgebraicSimplificationDynamic extends HopRewriteRule
return hi;
}
+
+ /**
+ * Replace SUM(X^2) with a fused SUM_SQ(X) HOP.
+ *
+ * @param parent Parent HOP for which hi is an input.
+ * @param hi Current HOP for potential rewrite.
+ * @param pos Position of hi in parent's list of inputs.
+ *
+ * @return Either hi or the rewritten HOP replacing it.
+ *
+ * @throws HopsException
+ */
+ private Hop fuseSumSquared(Hop parent, Hop hi, int pos)
+ throws HopsException {
+ // if SUM
+ if (hi instanceof AggUnaryOp && ((AggUnaryOp) hi).getOp() == AggOp.SUM) {
+ Hop sumInput = hi.getInput().get(0);
+
+ // if input to SUM is POW(X,2), and no other consumers of the POW(X,2) HOP
+ if (sumInput instanceof BinaryOp && ((BinaryOp) sumInput).getOp() == OpOp2.POW
+ && sumInput.getInput().get(1) instanceof LiteralOp
+ && HopRewriteUtils.getDoubleValue((LiteralOp) sumInput.getInput().get(1)) == 2
+ && sumInput.getParent().size() == 1) {
+ Hop x = sumInput.getInput().get(0);
+
+ // if X is NOT a column vector
+ if (x.getDim2() > 1) {
+ // perform rewrite from SUM(POW(X,2)) to SUM_SQ(X)
+ DataType dt = hi.getDataType();
+ ValueType vt = hi.getValueType();
+ Direction dir = ((AggUnaryOp) hi).getDirection();
+ long brlen = hi.getRowsInBlock();
+ long bclen = hi.getColsInBlock();
+ AggUnaryOp sumSq = new AggUnaryOp("sumSq", dt, vt, AggOp.SUM_SQ, dir, x);
+ HopRewriteUtils.setOutputBlocksizes(sumSq, brlen, bclen);
+ HopRewriteUtils.removeChildReferenceByPos(parent, hi, pos);
+ HopRewriteUtils.addChildReference(parent, sumSq, pos);
+
+ // cleanup
+ if (hi.getParent().isEmpty())
+ HopRewriteUtils.removeAllChildReferences(hi);
+ if(sumInput.getParent().isEmpty())
+ HopRewriteUtils.removeAllChildReferences(sumInput);
+
+ // replace current HOP with new SUM_SQ HOP
+ hi = sumSq;
+ }
+ }
+ }
+ return hi;
+ }
/**
*
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/31bf3285/src/main/java/com/ibm/bi/dml/lops/Aggregate.java
----------------------------------------------------------------------
diff --git a/src/main/java/com/ibm/bi/dml/lops/Aggregate.java b/src/main/java/com/ibm/bi/dml/lops/Aggregate.java
index 3ee1d00..88d57e7 100644
--- a/src/main/java/com/ibm/bi/dml/lops/Aggregate.java
+++ b/src/main/java/com/ibm/bi/dml/lops/Aggregate.java
@@ -35,7 +35,9 @@ public class Aggregate extends Lop
/** Aggregate operation types **/
- public enum OperationTypes {Sum,Product,Min,Max,Trace,KahanSum,KahanTrace,Mean,MaxIndex, MinIndex};
+ public enum OperationTypes {
+ Sum, Product, Min, Max, Trace, KahanSum, KahanSumSq, KahanTrace, Mean,MaxIndex, MinIndex
+ }
OperationTypes operation;
private boolean isCorrectionUsed = false;
@@ -78,7 +80,8 @@ public class Aggregate extends Lop
// this function must be invoked during hop-to-lop translation
public void setupCorrectionLocation(CorrectionLocationType loc) {
- if ( operation == OperationTypes.KahanSum || operation == OperationTypes.KahanTrace || operation == OperationTypes.Mean ) {
+ if (operation == OperationTypes.KahanSum || operation == OperationTypes.KahanSumSq
+ || operation == OperationTypes.KahanTrace || operation == OperationTypes.Mean) {
isCorrectionUsed = true;
correctionLocation = loc;
}
@@ -121,10 +124,11 @@ public class Aggregate extends Lop
return "arimax";
case MinIndex:
return "arimin";
-
case KahanSum:
case KahanTrace:
- return "ak+";
+ return "ak+";
+ case KahanSumSq:
+ return "asqk+";
default:
throw new UnsupportedOperationException(this.printErrorLocation() + "Instruction is not defined for Aggregate operation: " + operation);
}
@@ -154,7 +158,8 @@ public class Aggregate extends Lop
boolean isCorrectionApplicable = false;
String opcode = getOpcode();
- if (operation == OperationTypes.Mean || operation == OperationTypes.KahanSum || operation == OperationTypes.KahanTrace )
+ if (operation == OperationTypes.Mean || operation == OperationTypes.KahanSum
+ || operation == OperationTypes.KahanSumSq || operation == OperationTypes.KahanTrace)
isCorrectionApplicable = true;
StringBuilder sb = new StringBuilder();
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/31bf3285/src/main/java/com/ibm/bi/dml/lops/PartialAggregate.java
----------------------------------------------------------------------
diff --git a/src/main/java/com/ibm/bi/dml/lops/PartialAggregate.java b/src/main/java/com/ibm/bi/dml/lops/PartialAggregate.java
index 3b31cf2..68f4b74 100644
--- a/src/main/java/com/ibm/bi/dml/lops/PartialAggregate.java
+++ b/src/main/java/com/ibm/bi/dml/lops/PartialAggregate.java
@@ -172,6 +172,7 @@ public class PartialAggregate extends Lop
switch (operation) {
case KahanSum:
+ case KahanSumSq:
case KahanTrace:
switch (direction) {
case Col:
@@ -344,7 +345,17 @@ public class PartialAggregate extends Lop
return "uack+";
break;
}
-
+
+ case KahanSumSq: {
+ if( dir == DirectionTypes.RowCol )
+ return "uasqk+";
+ else if( dir == DirectionTypes.Row )
+ return "uarsqk+";
+ else if( dir == DirectionTypes.Col )
+ return "uacsqk+";
+ break;
+ }
+
case Product: {
if( dir == DirectionTypes.RowCol )
return "ua*";
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/31bf3285/src/main/java/com/ibm/bi/dml/runtime/functionobjects/KahanFunction.java
----------------------------------------------------------------------
diff --git a/src/main/java/com/ibm/bi/dml/runtime/functionobjects/KahanFunction.java b/src/main/java/com/ibm/bi/dml/runtime/functionobjects/KahanFunction.java
new file mode 100644
index 0000000..6ac0a7c
--- /dev/null
+++ b/src/main/java/com/ibm/bi/dml/runtime/functionobjects/KahanFunction.java
@@ -0,0 +1,43 @@
+/**
+ * (C) Copyright IBM Corp. 2010, 2015
+ *
+ * Licensed 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 com.ibm.bi.dml.runtime.functionobjects;
+
+import com.ibm.bi.dml.runtime.instructions.cp.KahanObject;
+
+import java.io.Serializable;
+
+
+/**
+ * Runtime function type to perform the summation of values using
+ * the Kahan summation algorithm.
+ */
+public abstract class KahanFunction extends ValueFunction implements Serializable {
+
+ private static final long serialVersionUID = -8880016655817461398L;
+
+ /**
+ * Add the given term to the existing sum with a function applied
+ * using the Kahan summation algorithm.
+ *
+ * @param kObj A KahanObject containing the current sum and
+ * correction factor for the Kahan summation
+ * algorithm.
+ * @param in The current term to be added.
+ */
+ public abstract void execute2(KahanObject kObj, double in);
+}
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/31bf3285/src/main/java/com/ibm/bi/dml/runtime/functionobjects/KahanPlus.java
----------------------------------------------------------------------
diff --git a/src/main/java/com/ibm/bi/dml/runtime/functionobjects/KahanPlus.java b/src/main/java/com/ibm/bi/dml/runtime/functionobjects/KahanPlus.java
index 9046f8c..df5997b 100644
--- a/src/main/java/com/ibm/bi/dml/runtime/functionobjects/KahanPlus.java
+++ b/src/main/java/com/ibm/bi/dml/runtime/functionobjects/KahanPlus.java
@@ -24,7 +24,7 @@ import com.ibm.bi.dml.runtime.instructions.cp.Data;
import com.ibm.bi.dml.runtime.instructions.cp.KahanObject;
-public class KahanPlus extends ValueFunction implements Serializable
+public class KahanPlus extends KahanFunction implements Serializable
{
private static final long serialVersionUID = -8338160609569967791L;
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/31bf3285/src/main/java/com/ibm/bi/dml/runtime/functionobjects/KahanPlusSq.java
----------------------------------------------------------------------
diff --git a/src/main/java/com/ibm/bi/dml/runtime/functionobjects/KahanPlusSq.java b/src/main/java/com/ibm/bi/dml/runtime/functionobjects/KahanPlusSq.java
new file mode 100644
index 0000000..b02b8c2
--- /dev/null
+++ b/src/main/java/com/ibm/bi/dml/runtime/functionobjects/KahanPlusSq.java
@@ -0,0 +1,101 @@
+/**
+ * (C) Copyright IBM Corp. 2010, 2015
+ *
+ * Licensed 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 com.ibm.bi.dml.runtime.functionobjects;
+
+import com.ibm.bi.dml.runtime.DMLRuntimeException;
+import com.ibm.bi.dml.runtime.instructions.cp.Data;
+import com.ibm.bi.dml.runtime.instructions.cp.KahanObject;
+
+import java.io.Serializable;
+
+
+/**
+ * Runtime function to perform the summation of squared values using
+ * the Kahan summation algorithm.
+ */
+public class KahanPlusSq extends KahanFunction implements Serializable {
+
+ private static final long serialVersionUID = 5774388904472231717L;
+
+ private static KahanPlusSq singleObj = null;
+ private KahanPlus kplus = null;
+
+ private KahanPlusSq() {
+ kplus = KahanPlus.getKahanPlusFnObject();
+ }
+
+ /**
+ * Get the KahanPlusSq singleton object.
+ */
+ public static KahanPlusSq getKahanPlusSqFnObject() {
+ if (singleObj == null)
+ singleObj = new KahanPlusSq();
+ return singleObj;
+ }
+
+ public Object clone() throws CloneNotSupportedException {
+ // cloning is not supported for singleton classes
+ throw new CloneNotSupportedException();
+ }
+
+ /**
+ * Square the given term, then add to the existing sum using
+ * the Kahan summation algorithm.
+ *
+ * @param kObj A KahanObject supplied as a Data type containing
+ * the current sum and correction factor for the
+ * Kahan summation algorithm.
+ * @param in The current term to be squared and added.
+ */
+ @Override
+ public Data execute(Data kObj, double in)
+ throws DMLRuntimeException {
+ kplus.execute(kObj, in * in);
+ return kObj;
+ }
+
+ /**
+ * Add the given sum and correction factor to the existing
+ * sum in the KahanObject using the Kahan summation algorithm.
+ *
+ * @param kObj A KahanObject supplied as a Data type containing
+ * the current sum and correction factor for the
+ * Kahan summation algorithm.
+ * @param sum The current sum.
+ * @param corr The current correction factor.
+ */
+ @Override
+ public Data execute(Data kObj, double sum, double corr)
+ throws DMLRuntimeException {
+ kplus.execute(kObj, sum, corr);
+ return kObj;
+ }
+
+ /**
+ * Square the given term, then add to the existing sum using
+ * the Kahan summation algorithm.
+ *
+ * @param kObj A KahanObject containing the current sum and
+ * correction factor for the Kahan summation
+ * algorithm.
+ * @param in The current term to be squared and added.
+ */
+ public void execute2(KahanObject kObj, double in) {
+ kplus.execute2(kObj, in * in);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/31bf3285/src/main/java/com/ibm/bi/dml/runtime/instructions/CPInstructionParser.java
----------------------------------------------------------------------
diff --git a/src/main/java/com/ibm/bi/dml/runtime/instructions/CPInstructionParser.java b/src/main/java/com/ibm/bi/dml/runtime/instructions/CPInstructionParser.java
index df19c72..cf3dcaf 100644
--- a/src/main/java/com/ibm/bi/dml/runtime/instructions/CPInstructionParser.java
+++ b/src/main/java/com/ibm/bi/dml/runtime/instructions/CPInstructionParser.java
@@ -75,6 +75,9 @@ public class CPInstructionParser extends InstructionParser
String2CPInstructionType.put( "uak+" , CPINSTRUCTION_TYPE.AggregateUnary);
String2CPInstructionType.put( "uark+" , CPINSTRUCTION_TYPE.AggregateUnary);
String2CPInstructionType.put( "uack+" , CPINSTRUCTION_TYPE.AggregateUnary);
+ String2CPInstructionType.put( "uasqk+" , CPINSTRUCTION_TYPE.AggregateUnary);
+ String2CPInstructionType.put( "uarsqk+" , CPINSTRUCTION_TYPE.AggregateUnary);
+ String2CPInstructionType.put( "uacsqk+" , CPINSTRUCTION_TYPE.AggregateUnary);
String2CPInstructionType.put( "uamean" , CPINSTRUCTION_TYPE.AggregateUnary);
String2CPInstructionType.put( "uarmean" , CPINSTRUCTION_TYPE.AggregateUnary);
String2CPInstructionType.put( "uacmean" , CPINSTRUCTION_TYPE.AggregateUnary);
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/31bf3285/src/main/java/com/ibm/bi/dml/runtime/instructions/InstructionUtils.java
----------------------------------------------------------------------
diff --git a/src/main/java/com/ibm/bi/dml/runtime/instructions/InstructionUtils.java b/src/main/java/com/ibm/bi/dml/runtime/instructions/InstructionUtils.java
index 1529187..4f0fc1c 100644
--- a/src/main/java/com/ibm/bi/dml/runtime/instructions/InstructionUtils.java
+++ b/src/main/java/com/ibm/bi/dml/runtime/instructions/InstructionUtils.java
@@ -44,6 +44,7 @@ import com.ibm.bi.dml.runtime.functionobjects.GreaterThan;
import com.ibm.bi.dml.runtime.functionobjects.GreaterThanEquals;
import com.ibm.bi.dml.runtime.functionobjects.IntegerDivide;
import com.ibm.bi.dml.runtime.functionobjects.KahanPlus;
+import com.ibm.bi.dml.runtime.functionobjects.KahanPlusSq;
import com.ibm.bi.dml.runtime.functionobjects.LessThan;
import com.ibm.bi.dml.runtime.functionobjects.LessThanEquals;
import com.ibm.bi.dml.runtime.functionobjects.Mean;
@@ -326,6 +327,20 @@ public class InstructionUtils
AggregateOperator agg = new AggregateOperator(0, KahanPlus.getKahanPlusFnObject(), true, CorrectionLocationType.LASTROW);
aggun = new AggregateUnaryOperator(agg, ReduceRow.getReduceRowFnObject());
}
+ else if ( opcode.equalsIgnoreCase("uasqk+") ) {
+ AggregateOperator agg = new AggregateOperator(0, KahanPlusSq.getKahanPlusSqFnObject(), true, CorrectionLocationType.LASTCOLUMN);
+ aggun = new AggregateUnaryOperator(agg, ReduceAll.getReduceAllFnObject());
+ }
+ else if ( opcode.equalsIgnoreCase("uarsqk+") ) {
+ // RowSums
+ AggregateOperator agg = new AggregateOperator(0, KahanPlusSq.getKahanPlusSqFnObject(), true, CorrectionLocationType.LASTCOLUMN);
+ aggun = new AggregateUnaryOperator(agg, ReduceCol.getReduceColFnObject());
+ }
+ else if ( opcode.equalsIgnoreCase("uacsqk+") ) {
+ // ColSums
+ AggregateOperator agg = new AggregateOperator(0, KahanPlusSq.getKahanPlusSqFnObject(), true, CorrectionLocationType.LASTROW);
+ aggun = new AggregateUnaryOperator(agg, ReduceRow.getReduceRowFnObject());
+ }
else if ( opcode.equalsIgnoreCase("uamean") ) {
// Mean
AggregateOperator agg = new AggregateOperator(0, Mean.getMeanFnObject(), true, CorrectionLocationType.LASTTWOCOLUMNS);
@@ -418,7 +433,12 @@ public class InstructionUtils
boolean lcorrExists = (corrExists==null) ? true : Boolean.parseBoolean(corrExists);
CorrectionLocationType lcorrLoc = (corrLoc==null) ? CorrectionLocationType.LASTCOLUMN : CorrectionLocationType.valueOf(corrLoc);
agg = new AggregateOperator(0, KahanPlus.getKahanPlusFnObject(), lcorrExists, lcorrLoc);
- }
+ }
+ else if ( opcode.equalsIgnoreCase("asqk+") ) {
+ boolean lcorrExists = (corrExists==null) ? true : Boolean.parseBoolean(corrExists);
+ CorrectionLocationType lcorrLoc = (corrLoc==null) ? CorrectionLocationType.LASTCOLUMN : CorrectionLocationType.valueOf(corrLoc);
+ agg = new AggregateOperator(0, KahanPlusSq.getKahanPlusSqFnObject(), lcorrExists, lcorrLoc);
+ }
else if ( opcode.equalsIgnoreCase("a+") ) {
agg = new AggregateOperator(0, Plus.getPlusFnObject());
}
@@ -721,6 +741,8 @@ public class InstructionUtils
{
if ( opcode.equalsIgnoreCase("uak+") || opcode.equalsIgnoreCase("uark+") || opcode.equalsIgnoreCase("uack+"))
return "ak+";
+ else if ( opcode.equalsIgnoreCase("uasqk+") || opcode.equalsIgnoreCase("uarsqk+") || opcode.equalsIgnoreCase("uacsqk+") )
+ return "asqk+";
else if ( opcode.equalsIgnoreCase("uamean") || opcode.equalsIgnoreCase("uarmean") || opcode.equalsIgnoreCase("uacmean") )
return "amean";
else if ( opcode.equalsIgnoreCase("ua+") || opcode.equalsIgnoreCase("uar+") || opcode.equalsIgnoreCase("uac+") )
@@ -748,9 +770,11 @@ public class InstructionUtils
*/
public static CorrectionLocationType deriveAggregateOperatorCorrectionLocation(String opcode)
{
- if ( opcode.equalsIgnoreCase("uak+") || opcode.equalsIgnoreCase("uark+") || opcode.equalsIgnoreCase("uatrace") || opcode.equalsIgnoreCase("uaktrace") )
+ if ( opcode.equalsIgnoreCase("uak+") || opcode.equalsIgnoreCase("uark+") ||
+ opcode.equalsIgnoreCase("uasqk+") || opcode.equalsIgnoreCase("uarsqk+") ||
+ opcode.equalsIgnoreCase("uatrace") || opcode.equalsIgnoreCase("uaktrace") )
return CorrectionLocationType.LASTCOLUMN;
- else if ( opcode.equalsIgnoreCase("uack+") )
+ else if ( opcode.equalsIgnoreCase("uack+") || opcode.equalsIgnoreCase("uacsqk+") )
return CorrectionLocationType.LASTROW;
else if ( opcode.equalsIgnoreCase("uamean") || opcode.equalsIgnoreCase("uarmean") )
return CorrectionLocationType.LASTTWOCOLUMNS;
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/31bf3285/src/main/java/com/ibm/bi/dml/runtime/instructions/MRInstructionParser.java
----------------------------------------------------------------------
diff --git a/src/main/java/com/ibm/bi/dml/runtime/instructions/MRInstructionParser.java b/src/main/java/com/ibm/bi/dml/runtime/instructions/MRInstructionParser.java
index 43d92ee..51e5c84 100644
--- a/src/main/java/com/ibm/bi/dml/runtime/instructions/MRInstructionParser.java
+++ b/src/main/java/com/ibm/bi/dml/runtime/instructions/MRInstructionParser.java
@@ -86,6 +86,7 @@ public class MRInstructionParser extends InstructionParser
// AGG Instruction Opcodes
String2MRInstructionType.put( "a+" , MRINSTRUCTION_TYPE.Aggregate);
String2MRInstructionType.put( "ak+" , MRINSTRUCTION_TYPE.Aggregate);
+ String2MRInstructionType.put( "asqk+" , MRINSTRUCTION_TYPE.Aggregate);
String2MRInstructionType.put( "a*" , MRINSTRUCTION_TYPE.Aggregate);
String2MRInstructionType.put( "amax" , MRINSTRUCTION_TYPE.Aggregate);
String2MRInstructionType.put( "amin" , MRINSTRUCTION_TYPE.Aggregate);
@@ -105,6 +106,9 @@ public class MRInstructionParser extends InstructionParser
String2MRInstructionType.put( "uak+" , MRINSTRUCTION_TYPE.AggregateUnary);
String2MRInstructionType.put( "uark+" , MRINSTRUCTION_TYPE.AggregateUnary);
String2MRInstructionType.put( "uack+" , MRINSTRUCTION_TYPE.AggregateUnary);
+ String2MRInstructionType.put( "uasqk+" , MRINSTRUCTION_TYPE.AggregateUnary);
+ String2MRInstructionType.put( "uarsqk+", MRINSTRUCTION_TYPE.AggregateUnary);
+ String2MRInstructionType.put( "uacsqk+", MRINSTRUCTION_TYPE.AggregateUnary);
String2MRInstructionType.put( "uamean", MRINSTRUCTION_TYPE.AggregateUnary);
String2MRInstructionType.put( "uarmean",MRINSTRUCTION_TYPE.AggregateUnary);
String2MRInstructionType.put( "uacmean",MRINSTRUCTION_TYPE.AggregateUnary);
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/31bf3285/src/main/java/com/ibm/bi/dml/runtime/instructions/SPInstructionParser.java
----------------------------------------------------------------------
diff --git a/src/main/java/com/ibm/bi/dml/runtime/instructions/SPInstructionParser.java b/src/main/java/com/ibm/bi/dml/runtime/instructions/SPInstructionParser.java
index f2dd155..d6ba5ec 100644
--- a/src/main/java/com/ibm/bi/dml/runtime/instructions/SPInstructionParser.java
+++ b/src/main/java/com/ibm/bi/dml/runtime/instructions/SPInstructionParser.java
@@ -82,6 +82,9 @@ public class SPInstructionParser extends InstructionParser {
String2SPInstructionType.put( "uak+" , SPINSTRUCTION_TYPE.AggregateUnary);
String2SPInstructionType.put( "uark+" , SPINSTRUCTION_TYPE.AggregateUnary);
String2SPInstructionType.put( "uack+" , SPINSTRUCTION_TYPE.AggregateUnary);
+ String2SPInstructionType.put( "uasqk+" , SPINSTRUCTION_TYPE.AggregateUnary);
+ String2SPInstructionType.put( "uarsqk+" , SPINSTRUCTION_TYPE.AggregateUnary);
+ String2SPInstructionType.put( "uacsqk+" , SPINSTRUCTION_TYPE.AggregateUnary);
String2SPInstructionType.put( "uamean" , SPINSTRUCTION_TYPE.AggregateUnary);
String2SPInstructionType.put( "uarmean" , SPINSTRUCTION_TYPE.AggregateUnary);
String2SPInstructionType.put( "uacmean" , SPINSTRUCTION_TYPE.AggregateUnary);
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/31bf3285/src/main/java/com/ibm/bi/dml/runtime/instructions/mr/AggregateInstruction.java
----------------------------------------------------------------------
diff --git a/src/main/java/com/ibm/bi/dml/runtime/instructions/mr/AggregateInstruction.java b/src/main/java/com/ibm/bi/dml/runtime/instructions/mr/AggregateInstruction.java
index 97c3aee..c5c83a5 100644
--- a/src/main/java/com/ibm/bi/dml/runtime/instructions/mr/AggregateInstruction.java
+++ b/src/main/java/com/ibm/bi/dml/runtime/instructions/mr/AggregateInstruction.java
@@ -49,7 +49,8 @@ public class AggregateInstruction extends UnaryMRInstructionBase
out = Byte.parseByte(parts[2]);
AggregateOperator agg = null;
- if(opcode.equalsIgnoreCase("ak+") || opcode.equalsIgnoreCase("amean")) {
+ if(opcode.equalsIgnoreCase("ak+") || opcode.equalsIgnoreCase("asqk+")
+ || opcode.equalsIgnoreCase("amean")) {
InstructionUtils.checkNumFields ( str, 4 );
agg = InstructionUtils.parseAggregateOperator(opcode, parts[3], parts[4]);
}
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/31bf3285/src/main/java/com/ibm/bi/dml/runtime/matrix/data/LibMatrixAgg.java
----------------------------------------------------------------------
diff --git a/src/main/java/com/ibm/bi/dml/runtime/matrix/data/LibMatrixAgg.java b/src/main/java/com/ibm/bi/dml/runtime/matrix/data/LibMatrixAgg.java
index e715753..7881930 100644
--- a/src/main/java/com/ibm/bi/dml/runtime/matrix/data/LibMatrixAgg.java
+++ b/src/main/java/com/ibm/bi/dml/runtime/matrix/data/LibMatrixAgg.java
@@ -29,7 +29,9 @@ import com.ibm.bi.dml.runtime.DMLUnsupportedOperationException;
import com.ibm.bi.dml.runtime.functionobjects.Builtin;
import com.ibm.bi.dml.runtime.functionobjects.Builtin.BuiltinFunctionCode;
import com.ibm.bi.dml.runtime.functionobjects.IndexFunction;
+import com.ibm.bi.dml.runtime.functionobjects.KahanFunction;
import com.ibm.bi.dml.runtime.functionobjects.KahanPlus;
+import com.ibm.bi.dml.runtime.functionobjects.KahanPlusSq;
import com.ibm.bi.dml.runtime.functionobjects.Mean;
import com.ibm.bi.dml.runtime.functionobjects.Multiply;
import com.ibm.bi.dml.runtime.functionobjects.ReduceAll;
@@ -73,6 +75,7 @@ public class LibMatrixAgg
private enum AggType {
KAHAN_SUM,
+ KAHAN_SUM_SQ,
CUM_KAHAN_SUM,
CUM_MIN,
CUM_MAX,
@@ -218,7 +221,7 @@ public class LibMatrixAgg
return;
}
- //prepare meta data
+ //prepare meta data
AggType aggtype = getAggType(uaop);
final int m = in.rlen;
final int m2 = out.rlen;
@@ -430,17 +433,20 @@ public class LibMatrixAgg
ValueFunction vfn = op.aggOp.increOp.fn;
IndexFunction ifn = op.indexFn;
- //(kahan) sum / trace (for ReduceDiag)
- if( vfn instanceof KahanPlus
- && (op.aggOp.correctionLocation == CorrectionLocationType.LASTCOLUMN ||op.aggOp.correctionLocation == CorrectionLocationType.LASTROW )
+ //(kahan) sum / sum squared / trace (for ReduceDiag)
+ if( vfn instanceof KahanFunction
+ && (op.aggOp.correctionLocation == CorrectionLocationType.LASTCOLUMN || op.aggOp.correctionLocation == CorrectionLocationType.LASTROW)
&& (ifn instanceof ReduceAll || ifn instanceof ReduceCol || ifn instanceof ReduceRow || ifn instanceof ReduceDiag) )
{
- return AggType.KAHAN_SUM;
+ if (vfn instanceof KahanPlus)
+ return AggType.KAHAN_SUM;
+ else if (vfn instanceof KahanPlusSq)
+ return AggType.KAHAN_SUM_SQ;
}
-
- //mean
+
+ //mean
if( vfn instanceof Mean
- && (op.aggOp.correctionLocation == CorrectionLocationType.LASTTWOCOLUMNS || op.aggOp.correctionLocation == CorrectionLocationType.LASTTWOROWS )
+ && (op.aggOp.correctionLocation == CorrectionLocationType.LASTTWOCOLUMNS || op.aggOp.correctionLocation == CorrectionLocationType.LASTTWOROWS)
&& (ifn instanceof ReduceAll || ifn instanceof ReduceCol || ifn instanceof ReduceRow) )
{
return AggType.MEAN;
@@ -996,6 +1002,18 @@ public class LibMatrixAgg
d_uakptrace(a, c, m, n, kbuff, (KahanPlus)vFn, rl, ru);
break;
}
+ case KAHAN_SUM_SQ: //SUM_SQ via k+,
+ {
+ KahanObject kbuff = new KahanObject(0, 0);
+
+ if( ixFn instanceof ReduceAll ) //SUM_SQ
+ d_uasqkp(a, c, m, n, kbuff, (KahanPlusSq)vFn, rl, ru);
+ else if( ixFn instanceof ReduceCol ) //ROWSUM_SQ
+ d_uarsqkp(a, c, m, n, kbuff, (KahanPlusSq)vFn, rl, ru);
+ else if( ixFn instanceof ReduceRow ) //COLSUM_SQ
+ d_uacsqkp(a, c, m, n, kbuff, (KahanPlusSq)vFn, rl, ru);
+ break;
+ }
case CUM_KAHAN_SUM: //CUMSUM
{
KahanObject kbuff = new KahanObject(0, 0);
@@ -1101,7 +1119,18 @@ public class LibMatrixAgg
s_uackp(a, c, m, n, kbuff, (KahanPlus)vFn, rl, ru);
else if( ixFn instanceof ReduceDiag ) //TRACE
s_uakptrace(a, c, m, n, kbuff, (KahanPlus)vFn, rl, ru);
-
+ break;
+ }
+ case KAHAN_SUM_SQ: //SUM_SQ via k+
+ {
+ KahanObject kbuff = new KahanObject(0, 0);
+
+ if( ixFn instanceof ReduceAll ) //SUM_SQ
+ s_uasqkp(a, c, m, n, kbuff, (KahanPlusSq)vFn, rl, ru);
+ else if( ixFn instanceof ReduceCol ) //ROWSUM_SQ
+ s_uarsqkp(a, c, m, n, kbuff, (KahanPlusSq)vFn, rl, ru);
+ else if( ixFn instanceof ReduceRow ) //COLSUM_SQ
+ s_uacsqkp(a, c, m, n, kbuff, (KahanPlusSq)vFn, rl, ru);
break;
}
case CUM_KAHAN_SUM: //CUMSUM
@@ -1189,9 +1218,10 @@ public class LibMatrixAgg
throws DMLRuntimeException
{
//do nothing for pseudo sparse-safe operations
- if(optype==AggType.KAHAN_SUM || optype==AggType.MIN || optype==AggType.MAX || optype==AggType.PROD
- || optype == AggType.CUM_KAHAN_SUM || optype == AggType.CUM_PROD || optype == AggType.CUM_MIN
- || optype == AggType.CUM_MAX )
+ if(optype==AggType.KAHAN_SUM || optype==AggType.KAHAN_SUM_SQ
+ || optype==AggType.MIN || optype==AggType.MAX || optype==AggType.PROD
+ || optype == AggType.CUM_KAHAN_SUM || optype == AggType.CUM_PROD
+ || optype == AggType.CUM_MIN || optype == AggType.CUM_MAX)
{
return;
}
@@ -1293,7 +1323,81 @@ public class LibMatrixAgg
for( int i=rl, aix=rl*n; i<ru; i++, aix+=n )
sumAgg( a, c, aix, 0, n, kbuff, kplus );
}
-
+
+ /**
+ * SUM_SQ, opcode: uasqk+, dense input.
+ *
+ * @param a Array of values to square & sum.
+ * @param c Output array to store sum and correction factor.
+ * @param m Number of rows.
+ * @param n Number of values per row.
+ * @param kbuff A KahanObject to hold the current sum and
+ * correction factor for the Kahan summation
+ * algorithm.
+ * @param kplusSq A KahanPlusSq object to perform summation of
+ * squared values.
+ * @param rl Lower row limit.
+ * @param ru Upper row limit.
+ */
+ private static void d_uasqkp(double[] a, double[] c, int m, int n, KahanObject kbuff,
+ KahanPlusSq kplusSq, int rl, int ru)
+ {
+ int len = Math.min((ru-rl)*n, a.length);
+ sumSq(a, rl*n, len, kbuff, kplusSq);
+ c[0] = kbuff._sum;
+ c[1] = kbuff._correction;
+ }
+
+ /**
+ * ROWSUM_SQ, opcode: uarsqk+, dense input.
+ *
+ * @param a Array of values to square & sum row-wise.
+ * @param c Output array to store sum and correction factor
+ * for each row.
+ * @param m Number of rows.
+ * @param n Number of values per row.
+ * @param kbuff A KahanObject to hold the current sum and
+ * correction factor for the Kahan summation
+ * algorithm.
+ * @param kplusSq A KahanPlusSq object to perform summation of
+ * squared values.
+ * @param rl Lower row limit.
+ * @param ru Upper row limit.
+ */
+ private static void d_uarsqkp(double[] a, double[] c, int m, int n, KahanObject kbuff,
+ KahanPlusSq kplusSq, int rl, int ru)
+ {
+ for (int i=rl, aix=rl*n, cix=rl*2; i<ru; i++, aix+=n, cix+=2) {
+ kbuff.set(0, 0); //reset buffer
+ sumSq(a, aix, n, kbuff, kplusSq);
+ c[cix+0] = kbuff._sum;
+ c[cix+1] = kbuff._correction;
+ }
+ }
+
+ /**
+ * COLSUM_SQ, opcode: uacsqk+, dense input.
+ *
+ * @param a Array of values to square & sum column-wise.
+ * @param c Output array to store sum and correction factor
+ * for each column.
+ * @param m Number of rows.
+ * @param n Number of values per row.
+ * @param kbuff A KahanObject to hold the current sum and
+ * correction factor for the Kahan summation
+ * algorithm.
+ * @param kplusSq A KahanPlusSq object to perform summation of
+ * squared values.
+ * @param rl Lower row limit.
+ * @param ru Upper row limit.
+ */
+ private static void d_uacsqkp(double[] a, double[] c, int m, int n, KahanObject kbuff,
+ KahanPlusSq kplusSq, int rl, int ru)
+ {
+ for (int i=rl, aix=rl*n; i<ru; i++, aix+=n)
+ sumSqAgg(a, c, aix, 0, n, kbuff, kplusSq);
+ }
+
/**
* CUMSUM, opcode: ucumk+, dense input.
*
@@ -1314,7 +1418,7 @@ public class LibMatrixAgg
for( int i=0, aix=0; i<m; i++, aix+=n ) {
sumAgg( a, csums, aix, 0, n, kbuff, kplus );
System.arraycopy(csums, 0, c, aix, n);
- }
+ }
}
/**
@@ -1628,7 +1732,104 @@ public class LibMatrixAgg
}
}
}
-
+
+ /**
+ * SUM_SQ, opcode: uasqk+, sparse input.
+ *
+ * @param a Sparse array of values to square & sum.
+ * @param c Output array to store sum and correction factor.
+ * @param m Number of rows.
+ * @param n Number of values per row.
+ * @param kbuff A KahanObject to hold the current sum and
+ * correction factor for the Kahan summation
+ * algorithm.
+ * @param kplusSq A KahanPlusSq object to perform summation of
+ * squared values.
+ * @param rl Lower row limit.
+ * @param ru Upper row limit.
+ */
+ private static void s_uasqkp(SparseRow[] a, double[] c, int m, int n, KahanObject kbuff,
+ KahanPlusSq kplusSq, int rl, int ru )
+ {
+ for (int i=rl; i<ru; i++) {
+ SparseRow arow = a[i];
+ if (arow!=null && !arow.isEmpty()) {
+ int alen = arow.size();
+ double[] avals = arow.getValueContainer();
+ sumSq(avals, 0, alen, kbuff, kplusSq);
+ }
+ }
+ c[0] = kbuff._sum;
+ c[1] = kbuff._correction;
+ }
+
+ /**
+ * ROWSUM_SQ, opcode: uarsqk+, sparse input.
+ *
+ * @param a Sparse array of values to square & sum row-wise.
+ * @param c Output array to store sum and correction factor
+ * for each row.
+ * @param m Number of rows.
+ * @param n Number of values per row.
+ * @param kbuff A KahanObject to hold the current sum and
+ * correction factor for the Kahan summation
+ * algorithm.
+ * @param kplusSq A KahanPlusSq object to perform summation of
+ * squared values.
+ * @param rl Lower row limit.
+ * @param ru Upper row limit.
+ */
+ private static void s_uarsqkp(SparseRow[] a, double[] c, int m, int n, KahanObject kbuff,
+ KahanPlusSq kplusSq, int rl, int ru )
+ {
+ //compute row aggregates
+ for (int i=rl, cix=rl*2; i<ru; i++, cix+=2) {
+ SparseRow arow = a[i];
+ if (arow!=null && !arow.isEmpty()) {
+ int alen = arow.size();
+ double[] avals = arow.getValueContainer();
+ kbuff.set(0, 0); //reset buffer
+ sumSq(avals, 0, alen, kbuff, kplusSq);
+ c[cix+0] = kbuff._sum;
+ c[cix+1] = kbuff._correction;
+ }
+ }
+ }
+
+ /**
+ * COLSUM_SQ, opcode: uacsqk+, sparse input.
+ *
+ * @param a Sparse array of values to square & sum column-wise.
+ * @param c Output array to store sum and correction factor
+ * for each column.
+ * @param m Number of rows.
+ * @param n Number of values per row.
+ * @param kbuff A KahanObject to hold the current sum and
+ * correction factor for the Kahan summation
+ * algorithm.
+ * @param kplusSq A KahanPlusSq object to perform summation of
+ * squared values.
+ * @param rl Lower row limit.
+ * @param ru Upper row limit.
+ */
+ private static void s_uacsqkp(SparseRow[] a, double[] c, int m, int n, KahanObject kbuff,
+ KahanPlusSq kplusSq, int rl, int ru )
+ {
+ //init result (for empty columns)
+ Arrays.fill(c, 0);
+
+ //compute column aggregates
+ for (int i=rl; i<ru; i++) {
+ SparseRow arow = a[i];
+ if (arow!=null && !arow.isEmpty()) {
+ int alen = arow.size();
+ double[] avals = arow.getValueContainer();
+ int[] aix = arow.getIndexContainer();
+ sumSqAgg(avals, c, aix, alen, n, kbuff, kplusSq);
+ }
+ }
+ }
+
/**
* CUMSUM, opcode: ucumk+, sparse input.
*
@@ -2123,64 +2324,137 @@ public class LibMatrixAgg
////////////////////////////////////////////
/**
- *
- * @param a
- * @param ai
- * @param len
- * @param kbuff
- * @param kplus
+ * Summation using the Kahan summation algorithm with the
+ * KahanPlus function.
*/
- private static void sum( double[] a, int ai, final int len, KahanObject kbuff, KahanPlus kplus )
+ private static void sum(double[] a, int ai, final int len, KahanObject kbuff, KahanPlus kplus)
{
- for( int i=0; i<len; i++, ai++ )
- kplus.execute2(kbuff, a[ ai ]);
+ sumWithFn(a, ai, len, kbuff, kplus);
+ }
+
+ /**
+ * Aggregated summation using the Kahan summation algorithm with
+ * the KahanPlus function.
+ */
+ private static void sumAgg(double[] a, double[] c, int ai, int ci, final int len,
+ KahanObject kbuff, KahanPlus kplus)
+ {
+ sumAggWithFn(a, c, ai, ci, len, kbuff, kplus);
}
/**
- *
- * @param a
- * @param c
- * @param ai
- * @param ci
- * @param len
- * @param kbuff
- * @param kplus
+ * Aggregated summation using the Kahan summation algorithm with
+ * the KahanPlus function.
*/
- private static void sumAgg( double[] a, double[] c, int ai, int ci, final int len, KahanObject kbuff, KahanPlus kplus )
+ private static void sumAgg(double[] a, double[] c, int[] ai, final int len, final int n,
+ KahanObject kbuff, KahanPlus kplus)
{
- for( int i=0; i<len; i++, ai++, ci++ )
- {
- kbuff._sum = c[ci];
+ sumAggWithFn(a, c, ai, len, n, kbuff, kplus);
+ }
+
+ /**
+ * Summation of squared values using the Kahan summation algorithm
+ * with the KahanPlusSq function.
+ */
+ private static void sumSq(double[] a, int ai, final int len,
+ KahanObject kbuff, KahanPlusSq kplusSq)
+ {
+ sumWithFn(a, ai, len, kbuff, kplusSq);
+ }
+
+ /**
+ * Aggregated summation of squared values using the Kahan
+ * summation algorithm with the KahanPlusSq function.
+ */
+ private static void sumSqAgg(double[] a, double[] c, int ai, int ci, final int len,
+ KahanObject kbuff, KahanPlusSq kplusSq)
+ {
+ sumAggWithFn(a, c, ai, ci, len, kbuff, kplusSq);
+ }
+
+ /**
+ * Aggregated summation of squared values using the Kahan
+ * summation algorithm with the KahanPlusSq function.
+ */
+ private static void sumSqAgg(double[] a, double[] c, int[] ai, final int len, final int n,
+ KahanObject kbuff, KahanPlusSq kplusSq)
+ {
+ sumAggWithFn(a, c, ai, len, n, kbuff, kplusSq);
+ }
+
+ /**
+ * Summation using the Kahan summation algorithm with one of the
+ * Kahan functions.
+ *
+ * @param a Array of values to sum.
+ * @param ai Index at which to start processing.
+ * @param len Number of values to process, starting at index ai.
+ * @param kbuff A KahanObject to hold the current sum and
+ * correction factor for the Kahan summation
+ * algorithm.
+ * @param kfunc A KahanFunction object to perform the summation.
+ */
+ private static void sumWithFn(double[] a, int ai, final int len,
+ KahanObject kbuff, KahanFunction kfunc)
+ {
+ for (int i=0; i<len; i++, ai++)
+ kfunc.execute2(kbuff, a[ai]);
+ }
+
+ /**
+ * Aggregated summation using the Kahan summation algorithm
+ * with one of the Kahan functions.
+ *
+ * @param a Array of values to sum.
+ * @param c Output array to store aggregated sum and correction
+ * factors.
+ * @param ai Index at which to start processing array `a`.
+ * @param ci Index at which to start storing aggregated results
+ * into array `c`.
+ * @param len Number of values to process, starting at index ai.
+ * @param kbuff A KahanObject to hold the current sum and
+ * correction factor for the Kahan summation
+ * algorithm.
+ * @param kfunc A KahanFunction object to perform the summation.
+ */
+ private static void sumAggWithFn(double[] a, double[] c, int ai, int ci, final int len,
+ KahanObject kbuff, KahanFunction kfunc)
+ {
+ for (int i=0; i<len; i++, ai++, ci++) {
+ kbuff._sum = c[ci];
kbuff._correction = c[ci+len];
- kplus.execute2(kbuff, a[ai]);
- c[ci] = kbuff._sum;
+ kfunc.execute2(kbuff, a[ai]);
+ c[ci] = kbuff._sum;
c[ci+len] = kbuff._correction;
}
}
-
+
/**
- *
- * @param a
- * @param c
- * @param ai
- * @param len
- * @param clen
- * @param kbuff
- * @param kplus
- */
- private static void sumAgg( double[] a, double[] c, int[] ai, final int len, final int n, KahanObject kbuff, KahanPlus kplus )
- {
- for( int i=0; i<len; i++ )
- {
- kbuff._sum = c[ ai[i] ];
- kbuff._correction = c[ ai[i]+n ];
- kplus.execute2( kbuff, a[i] );
- c[ ai[i] ] = kbuff._sum;
- c[ ai[i]+n ] = kbuff._correction;
+ * Aggregated summation using the Kahan summation algorithm
+ * with one of the Kahan functions.
+ *
+ * @param a Array of values to sum.
+ * @param c Output array to store aggregated sum and correction
+ * factors.
+ * @param ai Array of indices to process for array `a`.
+ * @param len Number of indices in `ai` to process.
+ * @param n Number of values per row.
+ * @param kbuff A KahanObject to hold the current sum and
+ * correction factor for the Kahan summation
+ * algorithm.
+ * @param kfunc A KahanFunction object to perform the summation.
+ */
+ private static void sumAggWithFn(double[] a, double[] c, int[] ai, final int len, final int n,
+ KahanObject kbuff, KahanFunction kfunc)
+ {
+ for (int i=0; i<len; i++) {
+ kbuff._sum = c[ai[i]];
+ kbuff._correction = c[ai[i]+n];
+ kfunc.execute2(kbuff, a[i]);
+ c[ai[i]] = kbuff._sum;
+ c[ai[i]+n] = kbuff._correction;
}
}
-
-
/**
*
* @param a
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/31bf3285/src/main/java/com/ibm/bi/dml/runtime/matrix/data/MatrixBlock.java
----------------------------------------------------------------------
diff --git a/src/main/java/com/ibm/bi/dml/runtime/matrix/data/MatrixBlock.java b/src/main/java/com/ibm/bi/dml/runtime/matrix/data/MatrixBlock.java
index 594c75a..38ab456 100644
--- a/src/main/java/com/ibm/bi/dml/runtime/matrix/data/MatrixBlock.java
+++ b/src/main/java/com/ibm/bi/dml/runtime/matrix/data/MatrixBlock.java
@@ -49,7 +49,9 @@ import com.ibm.bi.dml.runtime.functionobjects.CM;
import com.ibm.bi.dml.runtime.functionobjects.CTable;
import com.ibm.bi.dml.runtime.functionobjects.DiagIndex;
import com.ibm.bi.dml.runtime.functionobjects.Divide;
+import com.ibm.bi.dml.runtime.functionobjects.KahanFunction;
import com.ibm.bi.dml.runtime.functionobjects.KahanPlus;
+import com.ibm.bi.dml.runtime.functionobjects.KahanPlusSq;
import com.ibm.bi.dml.runtime.functionobjects.Multiply;
import com.ibm.bi.dml.runtime.functionobjects.Plus;
import com.ibm.bi.dml.runtime.functionobjects.ReduceAll;
@@ -1105,20 +1107,51 @@ public class MatrixBlock extends MatrixValue implements Externalizable
/**
* Wrapper method for reduceall-sum of a matrix.
*
- * @return
+ * @return Sum of the values in the matrix.
* @throws DMLRuntimeException
*/
public double sum()
throws DMLRuntimeException
{
+ KahanPlus kplus = KahanPlus.getKahanPlusFnObject();
+ return sumWithFn(kplus);
+ }
+
+ /**
+ * Wrapper method for reduceall-sumSq of a matrix.
+ *
+ * @return Sum of the squared values in the matrix.
+ * @throws DMLRuntimeException
+ */
+ public double sumSq()
+ throws DMLRuntimeException
+ {
+ KahanPlusSq kplusSq = KahanPlusSq.getKahanPlusSqFnObject();
+ return sumWithFn(kplusSq);
+ }
+
+ /**
+ * Wrapper method for reduceall-sum of a matrix using the given
+ * Kahan function for summation.
+ *
+ * @param kfunc A Kahan function object to use for summation.
+ * @return Sum of the values in the matrix with the given
+ * function applied.
+ * @throws DMLRuntimeException
+ */
+ private double sumWithFn(KahanFunction kfunc)
+ throws DMLRuntimeException
+ {
//construct operator
- AggregateOperator aop = new AggregateOperator(0, KahanPlus.getKahanPlusFnObject(), true, CorrectionLocationType.LASTCOLUMN);
- AggregateUnaryOperator auop = new AggregateUnaryOperator( aop, ReduceAll.getReduceAllFnObject());
-
+ CorrectionLocationType corrLoc = CorrectionLocationType.LASTCOLUMN;
+ ReduceAll reduceAllObj = ReduceAll.getReduceAllFnObject();
+ AggregateOperator aop = new AggregateOperator(0, kfunc, true, corrLoc);
+ AggregateUnaryOperator auop = new AggregateUnaryOperator(aop, reduceAllObj);
+
//execute operation
MatrixBlock out = new MatrixBlock(1, 2, false);
LibMatrixAgg.aggregateUnaryMatrix(this, out, auop);
-
+
return out.quickGetValue(0, 0);
}
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/31bf3285/src/main/java/com/ibm/bi/dml/runtime/matrix/operators/AggregateOperator.java
----------------------------------------------------------------------
diff --git a/src/main/java/com/ibm/bi/dml/runtime/matrix/operators/AggregateOperator.java b/src/main/java/com/ibm/bi/dml/runtime/matrix/operators/AggregateOperator.java
index 62ca1f6..a87070d 100644
--- a/src/main/java/com/ibm/bi/dml/runtime/matrix/operators/AggregateOperator.java
+++ b/src/main/java/com/ibm/bi/dml/runtime/matrix/operators/AggregateOperator.java
@@ -22,6 +22,7 @@ import java.io.Serializable;
import com.ibm.bi.dml.lops.PartialAggregate.CorrectionLocationType;
import com.ibm.bi.dml.runtime.functionobjects.KahanPlus;
+import com.ibm.bi.dml.runtime.functionobjects.KahanPlusSq;
import com.ibm.bi.dml.runtime.functionobjects.Minus;
import com.ibm.bi.dml.runtime.functionobjects.Or;
import com.ibm.bi.dml.runtime.functionobjects.Plus;
@@ -45,7 +46,7 @@ public class AggregateOperator extends Operator implements Serializable
increOp=new BinaryOperator(op);
//increFn=op;
//as long as (v op 0)=v, then op is sparseSafe
- if(op instanceof Plus || op instanceof KahanPlus
+ if(op instanceof Plus || op instanceof KahanPlus || op instanceof KahanPlusSq
|| op instanceof Or || op instanceof Minus)
sparseSafe=true;
else
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/31bf3285/src/main/java/com/ibm/bi/dml/runtime/matrix/operators/AggregateUnaryOperator.java
----------------------------------------------------------------------
diff --git a/src/main/java/com/ibm/bi/dml/runtime/matrix/operators/AggregateUnaryOperator.java b/src/main/java/com/ibm/bi/dml/runtime/matrix/operators/AggregateUnaryOperator.java
index e7626d0..a249a42 100644
--- a/src/main/java/com/ibm/bi/dml/runtime/matrix/operators/AggregateUnaryOperator.java
+++ b/src/main/java/com/ibm/bi/dml/runtime/matrix/operators/AggregateUnaryOperator.java
@@ -20,6 +20,7 @@ package com.ibm.bi.dml.runtime.matrix.operators;
import com.ibm.bi.dml.runtime.functionobjects.IndexFunction;
import com.ibm.bi.dml.runtime.functionobjects.KahanPlus;
+import com.ibm.bi.dml.runtime.functionobjects.KahanPlusSq;
import com.ibm.bi.dml.runtime.functionobjects.Minus;
import com.ibm.bi.dml.runtime.functionobjects.Or;
import com.ibm.bi.dml.runtime.functionobjects.Plus;
@@ -48,8 +49,9 @@ public class AggregateUnaryOperator extends Operator
//decide on sparse safe
if( aggOp.increOp.fn instanceof Plus ||
- aggOp.increOp.fn instanceof KahanPlus ||
- aggOp.increOp.fn instanceof Or ||
+ aggOp.increOp.fn instanceof KahanPlus ||
+ aggOp.increOp.fn instanceof KahanPlusSq ||
+ aggOp.increOp.fn instanceof Or ||
aggOp.increOp.fn instanceof Minus )
{
sparseSafe=true;
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/31bf3285/src/test/java/com/ibm/bi/dml/test/integration/functions/aggregate/ColSumsSqTest.java
----------------------------------------------------------------------
diff --git a/src/test/java/com/ibm/bi/dml/test/integration/functions/aggregate/ColSumsSqTest.java b/src/test/java/com/ibm/bi/dml/test/integration/functions/aggregate/ColSumsSqTest.java
new file mode 100644
index 0000000..b9ea24d
--- /dev/null
+++ b/src/test/java/com/ibm/bi/dml/test/integration/functions/aggregate/ColSumsSqTest.java
@@ -0,0 +1,268 @@
+/**
+ * (C) Copyright IBM Corp. 2010, 2015
+ *
+ * Licensed 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 com.ibm.bi.dml.test.integration.functions.aggregate;
+
+import com.ibm.bi.dml.api.DMLScript;
+import com.ibm.bi.dml.api.DMLScript.RUNTIME_PLATFORM;
+import com.ibm.bi.dml.hops.OptimizerUtils;
+import com.ibm.bi.dml.lops.LopProperties.ExecType;
+import com.ibm.bi.dml.runtime.instructions.Instruction;
+import com.ibm.bi.dml.runtime.matrix.data.MatrixValue.CellIndex;
+import com.ibm.bi.dml.test.integration.AutomatedTestBase;
+import com.ibm.bi.dml.test.integration.TestConfiguration;
+import com.ibm.bi.dml.test.utils.TestUtils;
+import com.ibm.bi.dml.utils.Statistics;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.HashMap;
+
+/**
+ * Test for the column sums of squared values function, "colSums(X^2)".
+ */
+public class ColSumsSqTest extends AutomatedTestBase {
+
+ private static final String TEST_NAME = "ColSumsSq";
+ private static final String TEST_DIR = "functions/aggregate/";
+ private static final String INPUT_NAME = "X";
+ private static final String OUTPUT_NAME = "colSumsSq";
+
+ private static final String op = "uacsqk+";
+ private static final int rows = 1234;
+ private static final int cols = 567;
+ private static final double sparsity1 = 1;
+ private static final double sparsity2 = 0.2;
+ private static final double eps = Math.pow(10, -10);
+
+ @Override
+ public void setUp() {
+ TestUtils.clearAssertionInformation();
+ TestConfiguration config = new TestConfiguration(TEST_DIR, TEST_NAME);
+ addTestConfiguration(TEST_NAME, config);
+ }
+
+ // Dense matrix w/ rewrites
+ @Test
+ public void testColSumsSquaredDenseMatrixRewriteCP() {
+ testColSumsSquared(TEST_NAME, false, false, true, ExecType.CP);
+ }
+
+ @Test
+ public void testColSumsSquaredDenseMatrixRewriteSpark() {
+ testColSumsSquared(TEST_NAME, false, false, true, ExecType.SPARK);
+ }
+
+ @Test
+ public void testColSumsSquaredDenseMatrixRewriteMR() {
+ testColSumsSquared(TEST_NAME, false, false, true, ExecType.MR);
+ }
+
+ // Dense matrix w/o rewrites
+ @Test
+ public void testColSumsSquaredDenseMatrixNoRewriteCP() {
+ testColSumsSquared(TEST_NAME, false, false, false, ExecType.CP);
+ }
+
+ @Test
+ public void testColSumsSquaredDenseMatrixNoRewriteSpark() {
+ testColSumsSquared(TEST_NAME, false, false, false, ExecType.SPARK);
+ }
+
+ @Test
+ public void testColSumsSquaredDenseMatrixNoRewriteMR() {
+ testColSumsSquared(TEST_NAME, false, false, false, ExecType.MR);
+ }
+
+ // Dense vector w/ rewrites
+ @Test
+ public void testColSumsSquaredDenseVectorRewriteCP() {
+ testColSumsSquared(TEST_NAME, false, true, true, ExecType.CP);
+ }
+
+ @Test
+ public void testColSumsSquaredDenseVectorRewriteSpark() {
+ testColSumsSquared(TEST_NAME, false, true, true, ExecType.SPARK);
+ }
+
+ @Test
+ public void testColSumsSquaredDenseVectorRewriteMR() {
+ testColSumsSquared(TEST_NAME, false, true, true, ExecType.MR);
+ }
+
+ // Dense vector w/o rewrites
+ @Test
+ public void testColSumsSquaredDenseVectorNoRewriteCP() {
+ testColSumsSquared(TEST_NAME, false, true, false, ExecType.CP);
+ }
+
+ @Test
+ public void testColSumsSquaredDenseVectorNoRewriteSpark() {
+ testColSumsSquared(TEST_NAME, false, true, false, ExecType.SPARK);
+ }
+
+ @Test
+ public void testColSumsSquaredDenseVectorNoRewriteMR() {
+ testColSumsSquared(TEST_NAME, false, true, false, ExecType.MR);
+ }
+
+ // Sparse matrix w/ rewrites
+ @Test
+ public void testColSumsSquaredSparseMatrixRewriteCP() {
+ testColSumsSquared(TEST_NAME, true, false, true, ExecType.CP);
+ }
+
+ @Test
+ public void testColSumsSquaredSparseMatrixRewriteSpark() {
+ testColSumsSquared(TEST_NAME, true, false, true, ExecType.SPARK);
+ }
+
+ @Test
+ public void testColSumsSquaredSparseMatrixRewriteMR() {
+ testColSumsSquared(TEST_NAME, true, false, true, ExecType.MR);
+ }
+
+ // Sparse matrix w/o rewrites
+ @Test
+ public void testColSumsSquaredSparseMatrixNoRewriteCP() {
+ testColSumsSquared(TEST_NAME, true, false, false, ExecType.CP);
+ }
+
+ @Test
+ public void testColSumsSquaredSparseMatrixNoRewriteSpark() {
+ testColSumsSquared(TEST_NAME, true, false, false, ExecType.SPARK);
+ }
+
+ @Test
+ public void testColSumsSquaredSparseMatrixNoRewriteMR() {
+ testColSumsSquared(TEST_NAME, true, false, false, ExecType.MR);
+ }
+
+ // Sparse vector w/ rewrites
+ @Test
+ public void testColSumsSquaredSparseVectorRewriteCP() {
+ testColSumsSquared(TEST_NAME, true, true, true, ExecType.CP);
+ }
+
+ @Test
+ public void testColSumsSquaredSparseVectorRewriteSpark() {
+ testColSumsSquared(TEST_NAME, true, true, true, ExecType.SPARK);
+ }
+
+ @Test
+ public void testColSumsSquaredSparseVectorRewriteMR() {
+ testColSumsSquared(TEST_NAME, true, true, true, ExecType.MR);
+ }
+
+ // Sparse vector w/o rewrites
+ @Test
+ public void testColSumsSquaredSparseVectorNoRewriteCP() {
+ testColSumsSquared(TEST_NAME, true, true, false, ExecType.CP);
+ }
+
+ @Test
+ public void testColSumsSquaredSparseVectorNoRewriteSpark() {
+ testColSumsSquared(TEST_NAME, true, true, false, ExecType.SPARK);
+ }
+
+ @Test
+ public void testColSumsSquaredSparseVectorNoRewriteMR() {
+ testColSumsSquared(TEST_NAME, true, true, false, ExecType.MR);
+ }
+
+ /**
+ * Test the column sums of squared values function, "colSums(X^2)",
+ * on dense/sparse matrices/vectors with rewrites/no rewrites on
+ * the CP/Spark/MR platforms.
+ *
+ * @param testName The name of this test case.
+ * @param sparse Whether or not the matrix/vector should be sparse.
+ * @param vector Boolean value choosing between a vector and a matrix.
+ * @param rewrites Whether or not to employ algebraic rewrites.
+ * @param platform Selection between CP/Spark/MR platforms.
+ */
+ private void testColSumsSquared(String testName, boolean sparse, boolean vector,
+ boolean rewrites, ExecType platform) {
+ // Configure settings for this test case
+ boolean rewritesOld = OptimizerUtils.ALLOW_ALGEBRAIC_SIMPLIFICATION;
+ OptimizerUtils.ALLOW_ALGEBRAIC_SIMPLIFICATION = rewrites;
+
+ RUNTIME_PLATFORM platformOld = rtplatform;
+ switch (platform) {
+ case MR:
+ rtplatform = RUNTIME_PLATFORM.HADOOP;
+ break;
+ case SPARK:
+ rtplatform = RUNTIME_PLATFORM.SPARK;
+ break;
+ default:
+ rtplatform = RUNTIME_PLATFORM.SINGLE_NODE;
+ break;
+ }
+
+ boolean sparkConfigOld = DMLScript.USE_LOCAL_SPARK_CONFIG;
+ if (rtplatform == RUNTIME_PLATFORM.SPARK)
+ DMLScript.USE_LOCAL_SPARK_CONFIG = true;
+
+ try {
+ // Create and load test configuration
+ TestConfiguration config = getTestConfiguration(testName);
+ String HOME = SCRIPT_DIR + TEST_DIR;
+ fullDMLScriptName = HOME + testName + ".dml";
+ programArgs = new String[]{"-explain", "-stats", "-args",
+ HOME + INPUT_DIR + INPUT_NAME,
+ HOME + OUTPUT_DIR + OUTPUT_NAME};
+ fullRScriptName = HOME + testName + ".R";
+ rCmd = "Rscript" + " " + fullRScriptName + " " +
+ HOME + INPUT_DIR + " " + HOME + EXPECTED_DIR;
+ loadTestConfiguration(config);
+
+ // Generate data
+ double sparsity = sparse ? sparsity2 : sparsity1;
+ int columns = vector ? 1 : cols;
+ double[][] X = getRandomMatrix(rows, columns, -1, 1, sparsity, 7);
+ writeInputMatrixWithMTD(INPUT_NAME, X, true);
+
+ // Run DML and R scripts
+ runTest(true, false, null, -1);
+ runRScript(true);
+
+ // Compare output matrices
+ HashMap<CellIndex, Double> dmlfile = readDMLMatrixFromHDFS(OUTPUT_NAME);
+ HashMap<CellIndex, Double> rfile = readRMatrixFromFS(OUTPUT_NAME);
+ TestUtils.compareMatrices(dmlfile, rfile, eps, "Stat-DML", "Stat-R");
+
+ // On CP and Spark modes, check that the rewrite actually
+ // occurred for matrix cases and not for vector cases.
+ if (rewrites && (platform == ExecType.SPARK || platform == ExecType.CP)) {
+ String prefix = (platform == ExecType.SPARK) ? Instruction.SP_INST_PREFIX : "";
+ String opcode = prefix + op;
+ boolean rewriteApplied = Statistics.getCPHeavyHitterOpCodes().contains(opcode);
+ if (vector)
+ Assert.assertFalse("Rewrite applied to vector case.", rewriteApplied);
+ else
+ Assert.assertTrue("Rewrite not applied to matrix case.", rewriteApplied);
+ }
+ }
+ finally {
+ // Reset settings
+ OptimizerUtils.ALLOW_ALGEBRAIC_SIMPLIFICATION = rewritesOld;
+ rtplatform = platformOld;
+ DMLScript.USE_LOCAL_SPARK_CONFIG = sparkConfigOld;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/31bf3285/src/test/java/com/ibm/bi/dml/test/integration/functions/aggregate/RowSumsSqTest.java
----------------------------------------------------------------------
diff --git a/src/test/java/com/ibm/bi/dml/test/integration/functions/aggregate/RowSumsSqTest.java b/src/test/java/com/ibm/bi/dml/test/integration/functions/aggregate/RowSumsSqTest.java
new file mode 100644
index 0000000..9e8b4d3
--- /dev/null
+++ b/src/test/java/com/ibm/bi/dml/test/integration/functions/aggregate/RowSumsSqTest.java
@@ -0,0 +1,268 @@
+/**
+ * (C) Copyright IBM Corp. 2010, 2015
+ *
+ * Licensed 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 com.ibm.bi.dml.test.integration.functions.aggregate;
+
+import com.ibm.bi.dml.api.DMLScript;
+import com.ibm.bi.dml.api.DMLScript.RUNTIME_PLATFORM;
+import com.ibm.bi.dml.hops.OptimizerUtils;
+import com.ibm.bi.dml.lops.LopProperties.ExecType;
+import com.ibm.bi.dml.runtime.instructions.Instruction;
+import com.ibm.bi.dml.runtime.matrix.data.MatrixValue.CellIndex;
+import com.ibm.bi.dml.test.integration.AutomatedTestBase;
+import com.ibm.bi.dml.test.integration.TestConfiguration;
+import com.ibm.bi.dml.test.utils.TestUtils;
+import com.ibm.bi.dml.utils.Statistics;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.HashMap;
+
+/**
+ * Test for the row sums of squared values function, "rowSums(X^2)".
+ */
+public class RowSumsSqTest extends AutomatedTestBase {
+
+ private static final String TEST_NAME = "RowSumsSq";
+ private static final String TEST_DIR = "functions/aggregate/";
+ private static final String INPUT_NAME = "X";
+ private static final String OUTPUT_NAME = "rowSumsSq";
+
+ private static final String op = "uarsqk+";
+ private static final int rows = 1234;
+ private static final int cols = 567;
+ private static final double sparsity1 = 1;
+ private static final double sparsity2 = 0.2;
+ private static final double eps = Math.pow(10, -10);
+
+ @Override
+ public void setUp() {
+ TestUtils.clearAssertionInformation();
+ TestConfiguration config = new TestConfiguration(TEST_DIR, TEST_NAME);
+ addTestConfiguration(TEST_NAME, config);
+ }
+
+ // Dense matrix w/ rewrites
+ @Test
+ public void testRowSumsSquaredDenseMatrixRewriteCP() {
+ testRowSumsSquared(TEST_NAME, false, false, true, ExecType.CP);
+ }
+
+ @Test
+ public void testRowSumsSquaredDenseMatrixRewriteSpark() {
+ testRowSumsSquared(TEST_NAME, false, false, true, ExecType.SPARK);
+ }
+
+ @Test
+ public void testRowSumsSquaredDenseMatrixRewriteMR() {
+ testRowSumsSquared(TEST_NAME, false, false, true, ExecType.MR);
+ }
+
+ // Dense matrix w/o rewrites
+ @Test
+ public void testRowSumsSquaredDenseMatrixNoRewriteCP() {
+ testRowSumsSquared(TEST_NAME, false, false, false, ExecType.CP);
+ }
+
+ @Test
+ public void testRowSumsSquaredDenseMatrixNoRewriteSpark() {
+ testRowSumsSquared(TEST_NAME, false, false, false, ExecType.SPARK);
+ }
+
+ @Test
+ public void testRowSumsSquaredDenseMatrixNoRewriteMR() {
+ testRowSumsSquared(TEST_NAME, false, false, false, ExecType.MR);
+ }
+
+ // Dense vector w/ rewrites
+ @Test
+ public void testRowSumsSquaredDenseVectorRewriteCP() {
+ testRowSumsSquared(TEST_NAME, false, true, true, ExecType.CP);
+ }
+
+ @Test
+ public void testRowSumsSquaredDenseVectorRewriteSpark() {
+ testRowSumsSquared(TEST_NAME, false, true, true, ExecType.SPARK);
+ }
+
+ @Test
+ public void testRowSumsSquaredDenseVectorRewriteMR() {
+ testRowSumsSquared(TEST_NAME, false, true, true, ExecType.MR);
+ }
+
+ // Dense vector w/o rewrites
+ @Test
+ public void testRowSumsSquaredDenseVectorNoRewriteCP() {
+ testRowSumsSquared(TEST_NAME, false, true, false, ExecType.CP);
+ }
+
+ @Test
+ public void testRowSumsSquaredDenseVectorNoRewriteSpark() {
+ testRowSumsSquared(TEST_NAME, false, true, false, ExecType.SPARK);
+ }
+
+ @Test
+ public void testRowSumsSquaredDenseVectorNoRewriteMR() {
+ testRowSumsSquared(TEST_NAME, false, true, false, ExecType.MR);
+ }
+
+ // Sparse matrix w/ rewrites
+ @Test
+ public void testRowSumsSquaredSparseMatrixRewriteCP() {
+ testRowSumsSquared(TEST_NAME, true, false, true, ExecType.CP);
+ }
+
+ @Test
+ public void testRowSumsSquaredSparseMatrixRewriteSpark() {
+ testRowSumsSquared(TEST_NAME, true, false, true, ExecType.SPARK);
+ }
+
+ @Test
+ public void testRowSumsSquaredSparseMatrixRewriteMR() {
+ testRowSumsSquared(TEST_NAME, true, false, true, ExecType.MR);
+ }
+
+ // Sparse matrix w/o rewrites
+ @Test
+ public void testRowSumsSquaredSparseMatrixNoRewriteCP() {
+ testRowSumsSquared(TEST_NAME, true, false, false, ExecType.CP);
+ }
+
+ @Test
+ public void testRowSumsSquaredSparseMatrixNoRewriteSpark() {
+ testRowSumsSquared(TEST_NAME, true, false, false, ExecType.SPARK);
+ }
+
+ @Test
+ public void testRowSumsSquaredSparseMatrixNoRewriteMR() {
+ testRowSumsSquared(TEST_NAME, true, false, false, ExecType.MR);
+ }
+
+ // Sparse vector w/ rewrites
+ @Test
+ public void testRowSumsSquaredSparseVectorRewriteCP() {
+ testRowSumsSquared(TEST_NAME, true, true, true, ExecType.CP);
+ }
+
+ @Test
+ public void testRowSumsSquaredSparseVectorRewriteSpark() {
+ testRowSumsSquared(TEST_NAME, true, true, true, ExecType.SPARK);
+ }
+
+ @Test
+ public void testRowSumsSquaredSparseVectorRewriteMR() {
+ testRowSumsSquared(TEST_NAME, true, true, true, ExecType.MR);
+ }
+
+ // Sparse vector w/o rewrites
+ @Test
+ public void testRowSumsSquaredSparseVectorNoRewriteCP() {
+ testRowSumsSquared(TEST_NAME, true, true, false, ExecType.CP);
+ }
+
+ @Test
+ public void testRowSumsSquaredSparseVectorNoRewriteSpark() {
+ testRowSumsSquared(TEST_NAME, true, true, false, ExecType.SPARK);
+ }
+
+ @Test
+ public void testRowSumsSquaredSparseVectorNoRewriteMR() {
+ testRowSumsSquared(TEST_NAME, true, true, false, ExecType.MR);
+ }
+
+ /**
+ * Test the row sums of squared values function, "rowSums(X^2)",
+ * on dense/sparse matrices/vectors with rewrites/no rewrites on
+ * the CP/Spark/MR platforms.
+ *
+ * @param testName The name of this test case.
+ * @param sparse Whether or not the matrix/vector should be sparse.
+ * @param vector Boolean value choosing between a vector and a matrix.
+ * @param rewrites Whether or not to employ algebraic rewrites.
+ * @param platform Selection between CP/Spark/MR platforms.
+ */
+ private void testRowSumsSquared(String testName, boolean sparse, boolean vector,
+ boolean rewrites, ExecType platform) {
+ // Configure settings for this test case
+ boolean rewritesOld = OptimizerUtils.ALLOW_ALGEBRAIC_SIMPLIFICATION;
+ OptimizerUtils.ALLOW_ALGEBRAIC_SIMPLIFICATION = rewrites;
+
+ RUNTIME_PLATFORM platformOld = rtplatform;
+ switch (platform) {
+ case MR:
+ rtplatform = RUNTIME_PLATFORM.HADOOP;
+ break;
+ case SPARK:
+ rtplatform = RUNTIME_PLATFORM.SPARK;
+ break;
+ default:
+ rtplatform = RUNTIME_PLATFORM.SINGLE_NODE;
+ break;
+ }
+
+ boolean sparkConfigOld = DMLScript.USE_LOCAL_SPARK_CONFIG;
+ if (rtplatform == RUNTIME_PLATFORM.SPARK)
+ DMLScript.USE_LOCAL_SPARK_CONFIG = true;
+
+ try {
+ // Create and load test configuration
+ TestConfiguration config = getTestConfiguration(testName);
+ String HOME = SCRIPT_DIR + TEST_DIR;
+ fullDMLScriptName = HOME + testName + ".dml";
+ programArgs = new String[]{"-explain", "-stats", "-args",
+ HOME + INPUT_DIR + INPUT_NAME,
+ HOME + OUTPUT_DIR + OUTPUT_NAME};
+ fullRScriptName = HOME + testName + ".R";
+ rCmd = "Rscript" + " " + fullRScriptName + " " +
+ HOME + INPUT_DIR + " " + HOME + EXPECTED_DIR;
+ loadTestConfiguration(config);
+
+ // Generate data
+ double sparsity = sparse ? sparsity2 : sparsity1;
+ int columns = vector ? 1 : cols;
+ double[][] X = getRandomMatrix(rows, columns, -1, 1, sparsity, 7);
+ writeInputMatrixWithMTD(INPUT_NAME, X, true);
+
+ // Run DML and R scripts
+ runTest(true, false, null, -1);
+ runRScript(true);
+
+ // Compare output matrices
+ HashMap<CellIndex, Double> dmlfile = readDMLMatrixFromHDFS(OUTPUT_NAME);
+ HashMap<CellIndex, Double> rfile = readRMatrixFromFS(OUTPUT_NAME);
+ TestUtils.compareMatrices(dmlfile, rfile, eps, "Stat-DML", "Stat-R");
+
+ // On CP and Spark modes, check that the rewrite actually
+ // occurred for matrix cases and not for vector cases.
+ if (rewrites && (platform == ExecType.SPARK || platform == ExecType.CP)) {
+ String prefix = (platform == ExecType.SPARK) ? Instruction.SP_INST_PREFIX : "";
+ String opcode = prefix + op;
+ boolean rewriteApplied = Statistics.getCPHeavyHitterOpCodes().contains(opcode);
+ if (vector)
+ Assert.assertFalse("Rewrite applied to vector case.", rewriteApplied);
+ else
+ Assert.assertTrue("Rewrite not applied to matrix case.", rewriteApplied);
+ }
+ }
+ finally {
+ // Reset settings
+ OptimizerUtils.ALLOW_ALGEBRAIC_SIMPLIFICATION = rewritesOld;
+ rtplatform = platformOld;
+ DMLScript.USE_LOCAL_SPARK_CONFIG = sparkConfigOld;
+ }
+ }
+}