You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@systemds.apache.org by ba...@apache.org on 2021/11/21 13:40:04 UTC
[systemds] 02/02: [SYSTEMDS-3208] Local instruction to force block to local
This is an automated email from the ASF dual-hosted git repository.
baunsgaard pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/systemds.git
commit e7fd2c90acd7d6dcdb14f823fbc7d1559b2fe3f4
Author: baunsgaard <ba...@tugraz.at>
AuthorDate: Mon Nov 8 14:58:19 2021 +0100
[SYSTEMDS-3208] Local instruction to force block to local
This commit adds a new function that forces a block to local.
This makes it possible to test for pulling blocks back at arbitrary places
in code. The function is not exposed unless a boolean flag is set to true.
While adding this feature the SparkExecutionContext toMatrixBlock is modified.
To call a new matrix operation that enable putting a
MatrixBlock into another at an offset, called 'putInto'.
block.putInto(out, row_offset, col_offset, sparseCopyShallow);
This functionality is also added to CompressedMatrixBlock.
Closes #1447
---
.../java/org/apache/sysds/common/Builtins.java | 1 +
src/main/java/org/apache/sysds/common/Types.java | 3 +-
src/main/java/org/apache/sysds/hops/Hop.java | 8 +
.../java/org/apache/sysds/hops/OptimizerUtils.java | 7 +-
src/main/java/org/apache/sysds/hops/UnaryOp.java | 123 +++---
.../java/org/apache/sysds/lops/Compression.java | 2 +-
.../sysds/lops/{Compression.java => Local.java} | 25 +-
.../sysds/parser/BuiltinFunctionExpression.java | 14 +-
.../org/apache/sysds/parser/DMLTranslator.java | 3 +
.../runtime/compress/CompressedMatrixBlock.java | 25 +-
.../sysds/runtime/compress/colgroup/AColGroup.java | 45 ++-
.../runtime/compress/colgroup/AColGroupValue.java | 436 +++------------------
.../runtime/compress/colgroup/ColGroupConst.java | 16 +-
.../runtime/compress/colgroup/ColGroupDDC.java | 57 ++-
.../runtime/compress/colgroup/ColGroupEmpty.java | 9 +-
.../runtime/compress/colgroup/ColGroupOLE.java | 82 ++--
.../runtime/compress/colgroup/ColGroupRLE.java | 79 ++--
.../runtime/compress/colgroup/ColGroupSDC.java | 105 +++--
.../compress/colgroup/ColGroupSDCSingle.java | 47 ++-
.../compress/colgroup/ColGroupSDCSingleZeros.java | 73 ++--
.../compress/colgroup/ColGroupSDCZeros.java | 63 ++-
.../compress/colgroup/ColGroupUncompressed.java | 97 +++--
.../runtime/compress/lib/CLALibBinaryCellOp.java | 25 +-
.../sysds/runtime/compress/lib/CLALibCompAgg.java | 2 +-
.../runtime/compress/lib/CLALibDecompress.java | 72 +++-
.../sysds/runtime/compress/lib/CLALibUnary.java | 3 +-
.../compress/workload/WorkloadAnalyzer.java | 4 +-
.../context/SparkExecutionContext.java | 240 +++++++-----
.../runtime/instructions/CPInstructionParser.java | 16 +-
.../runtime/instructions/cp/CPInstruction.java | 2 +-
.../instructions/cp/LocalCPInstruction.java | 51 +++
.../sysds/runtime/matrix/data/MatrixBlock.java | 37 +-
.../sysds/utils/DMLCompressionStatistics.java | 55 ++-
.../java/org/apache/sysds/utils/Statistics.java | 9 +-
src/test/java/org/apache/sysds/test/TestUtils.java | 141 ++++++-
.../component/compress/CompressedMatrixTest.java | 93 ++++-
.../component/compress/CompressedTestBase.java | 3 +-
.../test/component/compress/InvalidInputTest.java | 5 -
.../functions/compress/CompressInstruction.java | 6 +-
.../compress/CompressInstructionRewrite.java | 2 +-
.../functions/compress/CompressRewriteSpark.java | 2 +
.../test/functions/compress/LocalInstruction.java | 84 ++++
.../compress/LocalInstructionWithCompression.java | 63 +++
.../LocalInstructionWithoutCompression.java | 64 +++
.../compress/configuration/CompressBase.java | 2 +
.../CompressedSparkInstructionsTest.java | 163 ++++----
.../CompressedSparkInstructionsTestDense.java | 2 -
.../CompressedSparkInstructionsTestSparse.java | 1 -
.../functions/compress/local/SystemDS-config.xml} | 25 +-
.../local/SystemDS-no-compress-config.xml} | 25 +-
.../scripts/functions/compress/local/local.dml | 29 ++
51 files changed, 1568 insertions(+), 978 deletions(-)
diff --git a/src/main/java/org/apache/sysds/common/Builtins.java b/src/main/java/org/apache/sysds/common/Builtins.java
index cfebfbc..affed9e 100644
--- a/src/main/java/org/apache/sysds/common/Builtins.java
+++ b/src/main/java/org/apache/sysds/common/Builtins.java
@@ -179,6 +179,7 @@ public enum Builtins {
LMCG("lmCG", true),
LMDS("lmDS", true),
LMPREDICT("lmPredict", true),
+ LOCAL("local", false),
LOG("log", false),
LOGSUMEXP("logSumExp", true),
LSTM("lstm", false, ReturnType.MULTI_RETURN),
diff --git a/src/main/java/org/apache/sysds/common/Types.java b/src/main/java/org/apache/sysds/common/Types.java
index f631f18..21a874e 100644
--- a/src/main/java/org/apache/sysds/common/Types.java
+++ b/src/main/java/org/apache/sysds/common/Types.java
@@ -241,7 +241,8 @@ public class Types
SIGMOID, //sigmoid function: 1 / (1 + exp(-X))
LOG_NZ, //sparse-safe log; ppred(X,0,"!=")*log(X)
- COMPRESS, DECOMPRESS,
+ COMPRESS, DECOMPRESS,
+ LOCAL, // instruction to pull data back from spark forcefully and return a CP matrix.
//low-level operators //TODO used?
MULT2, MINUS1_MULT, MINUS_RIGHT,
diff --git a/src/main/java/org/apache/sysds/hops/Hop.java b/src/main/java/org/apache/sysds/hops/Hop.java
index 9114c55..336beb0 100644
--- a/src/main/java/org/apache/sysds/hops/Hop.java
+++ b/src/main/java/org/apache/sysds/hops/Hop.java
@@ -1137,6 +1137,10 @@ public abstract class Hop implements ParseInfo {
setVisited();
}
+ /**
+ * Get the number of rows in the Hop.
+ * @return a long.
+ */
public long getDim1() {
return _dc.getRows();
}
@@ -1145,6 +1149,10 @@ public abstract class Hop implements ParseInfo {
_dc.setRows(dim1);
}
+ /**
+ * Get the number of columns in the Hop.
+ * @return a long.
+ */
public long getDim2() {
return _dc.getCols();
}
diff --git a/src/main/java/org/apache/sysds/hops/OptimizerUtils.java b/src/main/java/org/apache/sysds/hops/OptimizerUtils.java
index 6a41ea8..703b9f2 100644
--- a/src/main/java/org/apache/sysds/hops/OptimizerUtils.java
+++ b/src/main/java/org/apache/sysds/hops/OptimizerUtils.java
@@ -235,10 +235,15 @@ public class OptimizerUtils
public static final boolean ALLOW_COMBINE_FILE_INPUT_FORMAT = true;
/**
+ * This variable allows for use of explicit local command, that forces a spark block to be executed and returned as a local block.
+ */
+ public static boolean ALLOW_SCRIPT_LEVEL_LOCAL_COMMAND = false;
+
+ /**
* This variable allows for insertion of Compress and decompress in the dml script from the user.
* This is added because we want to have a way to test, and verify the correct placement of compress and decompress commands.
*/
- public static final boolean ALLOW_SCRIPT_LEVEL_COMPRESS_COMMAND = true;
+ public static boolean ALLOW_SCRIPT_LEVEL_COMPRESS_COMMAND = false;
/**
* Boolean specifying if compression rewrites is allowed. This is disabled at run time if the IPA for Workload aware compression
diff --git a/src/main/java/org/apache/sysds/hops/UnaryOp.java b/src/main/java/org/apache/sysds/hops/UnaryOp.java
index d4e8f34..009d6ff 100644
--- a/src/main/java/org/apache/sysds/hops/UnaryOp.java
+++ b/src/main/java/org/apache/sysds/hops/UnaryOp.java
@@ -19,18 +19,23 @@
package org.apache.sysds.hops;
+import java.util.ArrayList;
+
import org.apache.sysds.api.DMLScript;
import org.apache.sysds.common.Types.AggOp;
import org.apache.sysds.common.Types.DataType;
+import org.apache.sysds.common.Types.ExecType;
import org.apache.sysds.common.Types.OpOp1;
import org.apache.sysds.common.Types.ValueType;
import org.apache.sysds.hops.rewrite.HopRewriteUtils;
import org.apache.sysds.lops.Checkpoint;
+import org.apache.sysds.lops.Compression;
import org.apache.sysds.lops.CumulativeOffsetBinary;
import org.apache.sysds.lops.CumulativePartialAggregate;
import org.apache.sysds.lops.Data;
+import org.apache.sysds.lops.DeCompression;
+import org.apache.sysds.lops.Local;
import org.apache.sysds.lops.Lop;
-import org.apache.sysds.common.Types.ExecType;
import org.apache.sysds.lops.PickByCount;
import org.apache.sysds.lops.SortKeys;
import org.apache.sysds.lops.Unary;
@@ -40,8 +45,6 @@ import org.apache.sysds.runtime.meta.DataCharacteristics;
import org.apache.sysds.runtime.meta.MatrixCharacteristics;
import org.apache.sysds.runtime.util.UtilFunctions;
-import java.util.ArrayList;
-
/* Unary (cell operations): e.g, b_ij = round(a_ij)
@@ -127,67 +130,60 @@ public class UnaryOp extends MultiThreadedHop
//reuse existing lop
if( getLops() != null )
return getLops();
-
- try
- {
+ int k;
+ try {
Hop input = getInput().get(0);
-
- if( getDataType() == DataType.SCALAR //value type casts or matrix to scalar
- || (_op == OpOp1.CAST_AS_MATRIX && getInput().get(0).getDataType()==DataType.SCALAR)
- || (_op == OpOp1.CAST_AS_FRAME && getInput().get(0).getDataType()==DataType.SCALAR))
- {
- if (_op == OpOp1.IQM) //special handling IQM
- {
- Lop iqmLop = constructLopsIQM();
- setLops(iqmLop);
- }
- else if(_op == OpOp1.MEDIAN) {
- Lop medianLop = constructLopsMedian();
- setLops(medianLop);
- }
- else //general case SCALAR/CAST (always in CP)
- {
- UnaryCP unary1 = new UnaryCP(input.constructLops(), _op, getDataType(), getValueType());
- setOutputDimensions(unary1);
- setLineNumbers(unary1);
- setLops(unary1);
- }
- }
- else //general case MATRIX
- {
- ExecType et = optFindExecType();
-
- //special handling cumsum/cumprod/cummin/cumsum
- if( isCumulativeUnaryOperation() && !(et == ExecType.CP || et == ExecType.GPU) )
- {
- //TODO additional physical operation if offsets fit in memory
- Lop cumsumLop = constructLopsSparkCumulativeUnary();
- setLops(cumsumLop);
- }
- else //default unary
- {
- boolean inplace = false;
-
- //check in-place
- if (OptimizerUtils.ALLOW_UNARY_UPDATE_IN_PLACE
- && input.getParent().size() == 1)
- {
- inplace = !(input instanceof DataOp)
- || !((DataOp) input).isRead();
+ Lop ret = null;
+ switch(_op){
+ case COMPRESS:
+ k = OptimizerUtils.getConstrainedNumThreads(_maxNumThreads);
+ ret = new Compression(input.constructLops(), getDataType(), getValueType(), optFindExecType(), 0);
+ break;
+ case DECOMPRESS:
+ k = OptimizerUtils.getConstrainedNumThreads(_maxNumThreads);
+ ret = new DeCompression(input.constructLops(), getDataType(), getValueType(), optFindExecType());
+ break;
+ case LOCAL:
+ ret = new Local(input.constructLops(), getDataType(), getValueType());
+ break;
+ default:
+ if(getDataType() == DataType.SCALAR // value type casts or matrix to scalar
+ || (_op == OpOp1.CAST_AS_MATRIX && getInput().get(0).getDataType() == DataType.SCALAR) ||
+ (_op == OpOp1.CAST_AS_FRAME && getInput().get(0).getDataType() == DataType.SCALAR)) {
+ if(_op == OpOp1.IQM) { // special handling IQM
+ ret = constructLopsIQM();
+ }
+ else if(_op == OpOp1.MEDIAN) {
+ ret = constructLopsMedian();
+ }
+ else { // general case SCALAR/CAST (always in CP)
+ ret = new UnaryCP(input.constructLops(), _op, getDataType(), getValueType());
+ }
+ }
+ else { // general case MATRIX
+ ExecType et = optFindExecType();
+
+ // special handling cumsum/cumprod/cummin/cumsum
+ if(isCumulativeUnaryOperation() && !(et == ExecType.CP || et == ExecType.GPU)) {
+ // TODO additional physical operation if offsets fit in memory
+ ret = constructLopsSparkCumulativeUnary();
+ }
+ else // default unary
+ {
+ final boolean inplace = OptimizerUtils.ALLOW_UNARY_UPDATE_IN_PLACE &&
+ input.getParent().size() == 1 && (!(input instanceof DataOp) || !((DataOp) input).isRead());
+
+ k = isCumulativeUnaryOperation() || isExpensiveUnaryOperation() ? OptimizerUtils
+ .getConstrainedNumThreads(_maxNumThreads) : 1;
+ ret = new Unary(input.constructLops(), _op, getDataType(), getValueType(), et, k, inplace);
+ }
}
-
- int k = isCumulativeUnaryOperation() || isExpensiveUnaryOperation() ?
- OptimizerUtils.getConstrainedNumThreads( _maxNumThreads ) : 1;
- Unary unary1 = new Unary(input.constructLops(),
- _op, getDataType(), getValueType(), et, k, inplace);
- setOutputDimensions(unary1);
- setLineNumbers(unary1);
- setLops(unary1);
- }
}
- }
- catch (Exception e)
- {
+ setOutputDimensions(ret);
+ setLineNumbers(ret);
+ setLops(ret);
+ }
+ catch (Exception e) {
throw new HopsException(this.printErrorLocation() + "error constructing Lops for UnaryOp Hop -- \n " , e);
}
@@ -413,7 +409,8 @@ public class UnaryOp extends MultiThreadedHop
|| _op==OpOp1.ACOS || _op==OpOp1.ASIN || _op==OpOp1.ATAN
|| _op==OpOp1.COSH || _op==OpOp1.SINH || _op==OpOp1.TANH
|| _op==OpOp1.SQRT || _op==OpOp1.ROUND
- || _op==OpOp1.SPROP || _op== OpOp1.COMPRESS || _op== OpOp1.DECOMPRESS) //sparsity preserving
+ || _op==OpOp1.SPROP || _op== OpOp1.COMPRESS || _op== OpOp1.DECOMPRESS
+ || _op==OpOp1.LOCAL) //sparsity preserving
{
ret = new MatrixCharacteristics(dc.getRows(), dc.getCols(), -1, dc.getNonZeros());
}
@@ -571,7 +568,7 @@ public class UnaryOp extends MultiThreadedHop
|| _op==OpOp1.SINH || _op==OpOp1.TANH
|| _op==OpOp1.ASIN || _op==OpOp1.ATAN
|| _op==OpOp1.SQRT || _op==OpOp1.ROUND || _op==OpOp1.SPROP
- || _op==OpOp1.COMPRESS || _op==OpOp1.DECOMPRESS) //sparsity preserving
+ || _op==OpOp1.COMPRESS || _op==OpOp1.DECOMPRESS || _op==OpOp1.LOCAL) //sparsity preserving
{
setNnz( input.getNnz() );
}
diff --git a/src/main/java/org/apache/sysds/lops/Compression.java b/src/main/java/org/apache/sysds/lops/Compression.java
index bd44341..b2999db 100644
--- a/src/main/java/org/apache/sysds/lops/Compression.java
+++ b/src/main/java/org/apache/sysds/lops/Compression.java
@@ -46,7 +46,7 @@ public class Compression extends Lop {
@Override
public String toString() {
- return "Compress";
+ return OPCODE;
}
@Override
diff --git a/src/main/java/org/apache/sysds/lops/Compression.java b/src/main/java/org/apache/sysds/lops/Local.java
similarity index 72%
copy from src/main/java/org/apache/sysds/lops/Compression.java
copy to src/main/java/org/apache/sysds/lops/Local.java
index bd44341..10ebc79 100644
--- a/src/main/java/org/apache/sysds/lops/Compression.java
+++ b/src/main/java/org/apache/sysds/lops/Local.java
@@ -23,30 +23,19 @@ import org.apache.sysds.common.Types.DataType;
import org.apache.sysds.common.Types.ExecType;
import org.apache.sysds.common.Types.ValueType;
-public class Compression extends Lop {
- public static final String OPCODE = "compress";
+public class Local extends Lop {
+ public static final String OPCODE = "local";
- private final int _singletonLookupKey;
-
- public enum CompressConfig {
- TRUE, FALSE, COST, AUTO, WORKLOAD;
-
- public boolean isEnabled() {
- return this != FALSE;
- }
- }
-
- public Compression(Lop input, DataType dt, ValueType vt, ExecType et, int singletonLookupKey) {
+ public Local(Lop input, DataType dt, ValueType vt) {
super(Lop.Type.Checkpoint, dt, vt);
addInput(input);
input.addOutput(this);
- lps.setProperties(inputs, et);
- _singletonLookupKey = singletonLookupKey;
+ lps.setProperties(inputs, ExecType.CP);
}
@Override
public String toString() {
- return "Compress";
+ return OPCODE;
}
@Override
@@ -59,10 +48,6 @@ public class Compression extends Lop {
sb.append(getInputs().get(0).prepInputOperand(input1));
sb.append(OPERAND_DELIMITOR);
sb.append(prepOutputOperand(output));
- if(_singletonLookupKey != 0){
- sb.append(OPERAND_DELIMITOR);
- sb.append(_singletonLookupKey);
- }
return sb.toString();
}
diff --git a/src/main/java/org/apache/sysds/parser/BuiltinFunctionExpression.java b/src/main/java/org/apache/sysds/parser/BuiltinFunctionExpression.java
index 5c3d364..cc70396 100644
--- a/src/main/java/org/apache/sysds/parser/BuiltinFunctionExpression.java
+++ b/src/main/java/org/apache/sysds/parser/BuiltinFunctionExpression.java
@@ -1580,6 +1580,17 @@ public class BuiltinFunctionExpression extends DataIdentifier
output.setBlocksize (id.getBlocksize());
break;
+ case LOCAL:
+ if(OptimizerUtils.ALLOW_SCRIPT_LEVEL_LOCAL_COMMAND){
+ checkNumParameters(1);
+ checkMatrixParam(getFirstExpr());
+ output.setDataType(DataType.MATRIX);
+ output.setDimensions(id.getDim1(), id.getDim2());
+ output.setBlocksize (id.getBlocksize());
+ output.setValueType(id.getValueType());
+ }
+ else
+ raiseValidateError("Local instruction not allowed in dml script");
case COMPRESS:
case DECOMPRESS:
if(OptimizerUtils.ALLOW_SCRIPT_LEVEL_COMPRESS_COMMAND){
@@ -1591,8 +1602,7 @@ public class BuiltinFunctionExpression extends DataIdentifier
output.setValueType(id.getValueType());
}
else
- raiseValidateError("Compress instruction not allowed in dml script");
-
+ raiseValidateError("Compress/DeCompress instruction not allowed in dml script");
break;
default:
if( isMathFunction() ) {
diff --git a/src/main/java/org/apache/sysds/parser/DMLTranslator.java b/src/main/java/org/apache/sysds/parser/DMLTranslator.java
index 50e2b09..dc51d46 100644
--- a/src/main/java/org/apache/sysds/parser/DMLTranslator.java
+++ b/src/main/java/org/apache/sysds/parser/DMLTranslator.java
@@ -2490,6 +2490,9 @@ public class DMLTranslator
case CAST_AS_BOOLEAN:
currBuiltinOp = new UnaryOp(target.getName(), target.getDataType(), ValueType.BOOLEAN, OpOp1.CAST_AS_BOOLEAN, expr);
break;
+ case LOCAL:
+ currBuiltinOp = new UnaryOp(target.getName(), target.getDataType(), ValueType.FP64, OpOp1.LOCAL, expr);
+ break;
case COMPRESS:
currBuiltinOp = new UnaryOp(target.getName(), target.getDataType(), ValueType.FP64, OpOp1.COMPRESS, expr);
break;
diff --git a/src/main/java/org/apache/sysds/runtime/compress/CompressedMatrixBlock.java b/src/main/java/org/apache/sysds/runtime/compress/CompressedMatrixBlock.java
index 654be34..c8bdd0a 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/CompressedMatrixBlock.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/CompressedMatrixBlock.java
@@ -35,11 +35,9 @@ import org.apache.commons.lang.NotImplementedException;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.commons.math3.random.Well1024a;
-import org.apache.sysds.api.DMLScript;
import org.apache.sysds.common.Types.CorrectionLocationType;
import org.apache.sysds.conf.ConfigurationManager;
import org.apache.sysds.conf.DMLConfig;
-import org.apache.sysds.hops.OptimizerUtils;
import org.apache.sysds.lops.MMTSJ.MMTSJType;
import org.apache.sysds.lops.MapMultChain.ChainType;
import org.apache.sysds.runtime.DMLRuntimeException;
@@ -224,7 +222,6 @@ public class CompressedMatrixBlock extends MatrixBlock {
* @return a new uncompressed matrix block containing the contents of this block
*/
public MatrixBlock decompress(int k) {
- Timing time = new Timing(true);
// Early out if empty.
if(isEmpty())
return new MatrixBlock(rlen, clen, true, 0);
@@ -239,16 +236,14 @@ public class CompressedMatrixBlock extends MatrixBlock {
// Set soft reference to the decompressed version
decompressedVersion = new SoftReference<>(ret);
- if(DMLScript.STATISTICS) {
- final double t = time.stop();
- DMLCompressionStatistics.addDecompressTime(t, k);
- if(LOG.isTraceEnabled())
- LOG.trace("decompressed block w/ k=" + k + " in " + t + "ms.");
- }
-
return ret;
}
+ @Override
+ public void putInto(MatrixBlock target, int rowOffset, int colOffset, boolean sparseCopyShallow) {
+ CLALibDecompress.decompressTo(this, target, rowOffset, colOffset, 1);
+ }
+
/**
* Get the cached decompressed matrix (if it exists otherwise null).
*
@@ -714,7 +709,7 @@ public class CompressedMatrixBlock extends MatrixBlock {
// and cu. It is not inclusive in decompression, and construction of MatrixBlock.
tmp = new MatrixBlock(ru + 1 - rl, getNumColumns(), false).allocateDenseBlock();
for(AColGroup g : getColGroups())
- g.decompressToBlock(tmp, rl, ru + 1, 0);
+ g.decompressToBlock(tmp, rl, ru + 1, -rl, 0);
tmp.recomputeNonZeros();
tmp.examSparsity();
return tmp;
@@ -795,6 +790,7 @@ public class CompressedMatrixBlock extends MatrixBlock {
res.allocateDenseBlock();
double[] resV = res.getDenseBlockValues();
AColGroup.colSum(_colGroups, resV, getNumRows());
+ res.recomputeNonZeros();
return res;
}
@@ -1148,7 +1144,7 @@ public class CompressedMatrixBlock extends MatrixBlock {
return d_compressed;
if(isEmpty())
return new MatrixBlock(getNumRows(), getNumColumns(), true);
- return this.decompress(OptimizerUtils.getConstrainedNumThreads(-1));
+ return this.decompress(InfrastructureAnalyzer.getLocalParallelism());
}
public MatrixBlock getUncompressed(String operation) {
@@ -1297,11 +1293,6 @@ public class CompressedMatrixBlock extends MatrixBlock {
}
@Override
- public void appendToSparse(MatrixBlock that, int rowoffset, int coloffset, boolean deep) {
- throw new DMLCompressionException("Can't append to compressed Matrix");
- }
-
- @Override
public void appendRowToSparse(SparseBlock dest, MatrixBlock src, int i, int rowoffset, int coloffset, boolean deep) {
throw new DMLCompressionException("Can't append row to compressed Matrix");
}
diff --git a/src/main/java/org/apache/sysds/runtime/compress/colgroup/AColGroup.java b/src/main/java/org/apache/sysds/runtime/compress/colgroup/AColGroup.java
index 7118fce..27a29cb 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/colgroup/AColGroup.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/colgroup/AColGroup.java
@@ -28,6 +28,8 @@ import java.util.List;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
+import org.apache.sysds.runtime.data.DenseBlock;
+import org.apache.sysds.runtime.data.SparseBlock;
import org.apache.sysds.runtime.matrix.data.MatrixBlock;
import org.apache.sysds.runtime.matrix.operators.AggregateUnaryOperator;
import org.apache.sysds.runtime.matrix.operators.BinaryOperator;
@@ -130,14 +132,14 @@ public abstract class AColGroup implements Serializable {
}
/**
- * Decompress the contents of the column group without modifying nonZeroCount.
+ * Decompress the contents of the column group into the target matrix,.
*
- * @param target a matrix block where the columns covered by this column group have not yet been filled in.
- * @param rl row lower
- * @param ru row upper
+ * @param target A matrix block where the columns covered by this column group have not yet been filled in.
+ * @param rl Row to start decompression from
+ * @param ru Row to end decompression at (not inclusive)
*/
public final void decompressToBlock(MatrixBlock target, int rl, int ru) {
- decompressToBlock(target, rl, ru, rl);
+ decompressToBlock(target, rl, ru, 0, 0);
}
/**
@@ -326,12 +328,31 @@ public abstract class AColGroup implements Serializable {
/**
* Decompress the contents of the column group without counting non zeros
*
+ * The offsets helps us decompress into specific target areas of the output matrix.
+ *
+ * If OffR and OffC is 0, then decompression output starts at row offset equal to rl,
+ *
+ * If for instance a MiniBatch of rows 10 to 15, then target would be 5 rows high and arguments would look like:
+ *
+ * cg.decompressToBlock(target, 10, 15, -10, 0)
+ *
* @param target a matrix block where the columns covered by this column group have not yet been filled in.
- * @param rl row lower
- * @param ru row upper
- * @param offT Offset into target to assign from, this allows us to decompress into smaller matrices.
- */
- public abstract void decompressToBlock(MatrixBlock target, int rl, int ru, int offT);
+ * @param rl Row to start decompression at.
+ * @param ru Row to end decompression at (not inclusive).
+ * @param offR RowOffset into target to assign from.
+ * @param offC ColumnOffset into the target matrix to assign from.
+ */
+ public final void decompressToBlock(MatrixBlock target, int rl, int ru, int offR, int offC){
+ if(target.isInSparseFormat())
+ decompressToSparseBlock(target.getSparseBlock(), rl, ru, offR, offC);
+ else
+ decompressToDenseBlock(target.getDenseBlock(), rl, ru, offR, offC);
+ }
+
+
+ protected abstract void decompressToDenseBlock(DenseBlock db, int rl, int ru,int offR, int offC);
+
+ protected abstract void decompressToSparseBlock(SparseBlock sb, int rl, int ru, int offR, int offC);
/**
* Right matrix multiplication with this column group.
@@ -405,8 +426,8 @@ public abstract class AColGroup implements Serializable {
/**
* Perform a binary row operation.
*
- * @param op The operation to execute
- * @param v The vector of values to apply, should be same length as dictionary length.
+ * @param op The operation to execute
+ * @param v The vector of values to apply, should be same length as dictionary length.
* @param isRowSafe True if the binary op is applied to an entire zero row and all results are zero
* @return A updated column group with the new values.
*/
diff --git a/src/main/java/org/apache/sysds/runtime/compress/colgroup/AColGroupValue.java b/src/main/java/org/apache/sysds/runtime/compress/colgroup/AColGroupValue.java
index 97d487e..067fa6f 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/colgroup/AColGroupValue.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/colgroup/AColGroupValue.java
@@ -33,6 +33,7 @@ import org.apache.sysds.runtime.compress.colgroup.dictionary.ADictionary;
import org.apache.sysds.runtime.compress.colgroup.dictionary.Dictionary;
import org.apache.sysds.runtime.compress.colgroup.dictionary.DictionaryFactory;
import org.apache.sysds.runtime.compress.colgroup.dictionary.MatrixBlockDictionary;
+import org.apache.sysds.runtime.data.DenseBlock;
import org.apache.sysds.runtime.data.SparseBlock;
import org.apache.sysds.runtime.functionobjects.Builtin;
import org.apache.sysds.runtime.matrix.data.LibMatrixMult;
@@ -86,43 +87,87 @@ public abstract class AColGroupValue extends AColGroupCompressed implements Clon
}
@Override
- public final void decompressToBlock(MatrixBlock target, int rl, int ru, int offT) {
+ public final void decompressToDenseBlock(DenseBlock db, int rl, int ru, int offR, int offC) {
if(_dict instanceof MatrixBlockDictionary) {
final MatrixBlockDictionary md = (MatrixBlockDictionary) _dict;
final MatrixBlock mb = md.getMatrixBlock();
- if(mb.isEmpty())
+ if(mb.isEmpty()) // Early abort if the dictionary is empty.
return;
else if(mb.isInSparseFormat())
- decompressToBlockSparseDictionary(target, rl, ru, offT, mb.getSparseBlock());
+ decompressToDenseBlockSparseDictionary(db, rl, ru, offR, offC, mb.getSparseBlock());
else
- decompressToBlockDenseDictionary(target, rl, ru, offT, mb.getDenseBlockValues());
+ decompressToDenseBlockDenseDictionary(db, rl, ru, offR, offC, mb.getDenseBlockValues());
}
else
- decompressToBlockDenseDictionary(target, rl, ru, offT, _dict.getValues());
+ decompressToDenseBlockDenseDictionary(db, rl, ru, offR, offC, _dict.getValues());
+ }
+
+ @Override
+ public final void decompressToSparseBlock(SparseBlock sb, int rl, int ru, int offR, int offC) {
+ if(_dict instanceof MatrixBlockDictionary) {
+ final MatrixBlockDictionary md = (MatrixBlockDictionary) _dict;
+ final MatrixBlock mb = md.getMatrixBlock();
+ if(mb.isEmpty()) // Early abort if the dictionary is empty.
+ return;
+ else if(mb.isInSparseFormat())
+ decompressToSparseBlockSparseDictionary(sb, rl, ru, offR, offC, mb.getSparseBlock());
+ else
+ decompressToSparseBlockDenseDictionary(sb, rl, ru, offR, offC, mb.getDenseBlockValues());
+ }
+ else
+ decompressToSparseBlockDenseDictionary(sb, rl, ru, offR, offC, _dict.getValues());
}
/**
- * Decompress to block using a sparse dictionary to lookup into.
+ * Decompress to DenseBlock using a sparse dictionary to lookup into.
*
- * @param target The dense target block to decompress into
+ * @param db The dense db block to decompress into
+ * @param rl The row to start decompression from
+ * @param ru The row to end decompression at
+ * @param offR The row offset to insert into
+ * @param offC The column offset to insert into
+ * @param sb The sparse dictionary block to take value tuples from
+ */
+ protected abstract void decompressToDenseBlockSparseDictionary(DenseBlock db, int rl, int ru, int offR, int offC,
+ SparseBlock sb);
+
+ /**
+ * Decompress to DenseBlock using a dense dictionary to lookup into.
+ *
+ * @param db The dense db block to decompress into
* @param rl The row to start decompression from
* @param ru The row to end decompression at
- * @param offT The offset into target block to decompress to (use full if the target it a multi block matrix)
- * @param sb the sparse dictionary block to take value tuples from
+ * @param offR The row offset to insert into
+ * @param offC The column offset to insert into
+ * @param values The dense dictionary values, linearized row major.
+ */
+ protected abstract void decompressToDenseBlockDenseDictionary(DenseBlock db, int rl, int ru, int offR, int offC,
+ double[] values);
+
+ /**
+ * Decompress to SparseBlock using a sparse dictionary to lookup into.
+ *
+ * @param ret The dense ret block to decompress into
+ * @param rl The row to start decompression from
+ * @param ru The row to end decompression at
+ * @param offR The row offset to insert into
+ * @param offC The column offset to insert into
+ * @param sb The sparse dictionary block to take value tuples from
*/
- protected abstract void decompressToBlockSparseDictionary(MatrixBlock target, int rl, int ru, int offT,
+ protected abstract void decompressToSparseBlockSparseDictionary(SparseBlock ret, int rl, int ru, int offR, int offC,
SparseBlock sb);
/**
- * Decompress to block using a dense dictionary to lookup into.
+ * Decompress to SparseBlock using a dense dictionary to lookup into.
*
- * @param target The dense target block to decompress into
+ * @param ret The dense ret block to decompress into
* @param rl The row to start decompression from
* @param ru The row to end decompression at
- * @param offT The offset into target block to decompress to (use full if the target it a multi block matrix)
+ * @param offR The row offset to insert into
+ * @param offC The column offset to insert into
* @param values The dense dictionary values, linearized row major.
*/
- protected abstract void decompressToBlockDenseDictionary(MatrixBlock target, int rl, int ru, int offT,
+ protected abstract void decompressToSparseBlockDenseDictionary(SparseBlock ret, int rl, int ru, int offR, int offC,
double[] values);
@Override
@@ -329,14 +374,17 @@ public abstract class AColGroupValue extends AColGroupCompressed implements Clon
_dict.colSum(c, getCounts(), _colIndexes, square);
}
+ @Override
protected void computeProduct(double[] c, int nRows) {
c[0] *= _dict.product(getCounts(), _colIndexes.length);
}
+ @Override
protected void computeRowProduct(double[] c, int rl, int ru) {
throw new NotImplementedException();
}
+ @Override
protected void computeColProduct(double[] c, int nRows) {
_dict.colProduct(c, getCounts(), _colIndexes);
}
@@ -396,126 +444,6 @@ public abstract class AColGroupValue extends AColGroupCompressed implements Clon
return ret;
}
- // private static final MatrixBlock allocatePreAggregate(MatrixBlock m, int numVals, int rl, int ru) {
- // final int lhsRows = ru - rl;
- // final double[] vals = allocDVector(lhsRows * numVals, true);
- // final DenseBlock retB = new DenseBlockFP64(new int[] {lhsRows, numVals}, vals);
- // return new MatrixBlock(lhsRows, numVals, retB);
- // }
-
- // /**
- // * Pre aggregate for left Multiplication.
- // *
- // * @param m Matrix to preAggregate
- // * @param preAgg Matrix to preAggregate into
- // * @param rl Start row
- // * @param ru End row
- // */
- // public abstract void preAggregate(MatrixBlock m, MatrixBlock preAgg, int rl, int ru);
-
- // public abstract void preAggregateDense(MatrixBlock m, MatrixBlock preAgg, int rl, int ru, int vl, int vu);
-
- // /**
- // * Pre aggregate into a dictionary. It is assumed that "that" have more distinct values than, "this".
- // *
- // * @param that the other column group whose indexes are used for aggregation.
- // * @return A aggregate dictionary
- // */
- // public final Dictionary preAggregateThatIndexStructure(AColGroupValue that) {
- // int outputLength = that._colIndexes.length * this.getNumValues();
- // Dictionary ret = new Dictionary(new double[outputLength]);
-
- // if(that instanceof ColGroupDDC)
- // return preAggregateThatDDCStructure((ColGroupDDC) that, ret);
- // else if(that instanceof ColGroupSDCSingleZeros)
- // return preAggregateThatSDCSingleZerosStructure((ColGroupSDCSingleZeros) that, ret);
- // else if(that instanceof ColGroupSDCZeros)
- // return preAggregateThatSDCZerosStructure((ColGroupSDCZeros) that, ret);
-
- // throw new NotImplementedException("Not supported pre aggregate using index structure of :"
- // + that.getClass().getSimpleName() + " in " + this.getClass().getSimpleName());
- // }
-
- // protected Dictionary preAggregateThatDDCStructure(ColGroupDDC that, Dictionary ret) {
- // throw new DMLCompressionException("Does not make sense to call this, implement function for sub class");
- // }
-
- // protected Dictionary preAggregateThatSDCZerosStructure(ColGroupSDCZeros that, Dictionary ret) {
- // throw new DMLCompressionException("Does not make sense to call this, implement function for sub class");
- // }
-
- // protected Dictionary preAggregateThatSDCSingleZerosStructure(ColGroupSDCSingleZeros that, Dictionary ret) {
- // throw new DMLCompressionException("Does not make sense to call this, implement function for sub class");
- // }
-
- // @Override
- // public final void leftMultByAColGroup(AColGroup lhs, MatrixBlock result) {
- // if(lhs instanceof ColGroupEmpty)
- // return;
- // else if(lhs instanceof AColGroupValue)
- // leftMultByColGroupValue((AColGroupValue) lhs, result);
- // else if(lhs instanceof ColGroupUncompressed)
- // leftMultByUncompressedColGroup((ColGroupUncompressed) lhs, result);
- // else
- // throw new DMLCompressionException(
- // "Not supported left multiplication with A ColGroup of type: " + lhs.getClass().getSimpleName());
- // }
-
- // @Override
- // public void tsmmAColGroup(AColGroup other, MatrixBlock result) {
- // if(other instanceof ColGroupEmpty)
- // return;
- // else if(other instanceof APreAgg)
- // tsmmAColGroupValue((APreAgg) other, result);
- // else if(other instanceof ColGroupUncompressed)
- // tsmmColGroupUncompressed((ColGroupUncompressed) other, result);
- // else
- // throw new DMLCompressionException("Unsupported column group type " + other.getClass().getSimpleName());
-
- // }
-
- // protected void tsmmColGroupUncompressed(ColGroupUncompressed other, MatrixBlock result) {
- // LOG.warn("Inefficient multiplication with uncompressed column group");
- // final int nCols = result.getNumColumns();
- // final MatrixBlock otherMBT = LibMatrixReorg.transpose(((ColGroupUncompressed) other).getData());
- // final int nRows = otherMBT.getNumRows();
- // final MatrixBlock tmp = new MatrixBlock(otherMBT.getNumRows(), nCols, false);
- // tmp.allocateDenseBlock();
- // leftMultByMatrix(otherMBT, tmp, 0, nRows);
-
- // final double[] r = tmp.getDenseBlockValues();
- // final double[] resV = result.getDenseBlockValues();
- // final int otLen = other._colIndexes.length;
- // final int thisLen = _colIndexes.length;
- // for(int i = 0; i < otLen; i++) {
- // final int oid = other._colIndexes[i];
- // final int offR = i * nCols;
- // for(int j = 0; j < thisLen; j++)
- // addToUpperTriangle(nCols, _colIndexes[j], oid, resV, r[offR + _colIndexes[j]]);
- // }
- // }
-
- // private void leftMultByColGroupValue(AColGroupValue lhs, MatrixBlock result) {
- // final int[] rightIdx = this._colIndexes;
- // final int[] leftIdx = lhs._colIndexes;
- // final ADictionary rDict = this._dict;
- // final ADictionary lDict = lhs._dict;
-
- // if(sameIndexStructure(lhs)) {
- // final int[] c = getCounts();
- // if(rDict == lDict)
- // tsmmDictionaryWithScaling(rDict, c, leftIdx, rightIdx, result);
- // else
- // MMDictsWithScaling(lDict, rDict, leftIdx, rightIdx, result, c);
- // }
- // else {
- // if(shouldPreAggregateLeft(lhs))
- // MMDicts(lDict, preAggLeft(lhs), leftIdx, rightIdx, result);
- // else
- // MMDicts(preAggRight(lhs), rDict, leftIdx, rightIdx, result);
- // }
- // }
-
@Override
protected final void tsmm(double[] result, int numColumns, int nRows) {
final int[] counts = getCounts();
@@ -535,201 +463,6 @@ public abstract class AColGroupValue extends AColGroupCompressed implements Clon
return _dict.getNumberNonZeros(counts, _colIndexes.length);
}
- // /**
- // * Matrix Multiply the two matrices, note that the left side is considered transposed,
- // *
- // * making the multiplication a: t(left) %*% right
- // *
- // * @param left The left side dictionary
- // * @param right The right side dictionary
- // * @param rowsLeft The row indexes on the left hand side
- // * @param colsRight The column indexes on the right hand side
- // * @param result The result matrix to put the results into.
- // */
- // private static void MMDicts(ADictionary left, ADictionary right, int[] rowsLeft, int[] colsRight,
- // MatrixBlock result) {
-
- // if(left instanceof MatrixBlockDictionary && right instanceof MatrixBlockDictionary) {
- // final MatrixBlock leftMB = left.getMBDict(rowsLeft.length).getMatrixBlock();
- // final MatrixBlock rightMB = right.getMBDict(colsRight.length).getMatrixBlock();
- // MMDicts(leftMB, rightMB, rowsLeft, colsRight, result);
- // return;
- // }
-
- // double[] leftV = null;
- // double[] rightV = null;
-
- // if(left instanceof MatrixBlockDictionary) {
- // final MatrixBlock leftMB = left.getMBDict(rowsLeft.length).getMatrixBlock();
- // if(leftMB.isEmpty())
- // return;
- // else if(leftMB.isInSparseFormat())
- // MMDictsSparseDense(leftMB.getSparseBlock(), right.getValues(), rowsLeft, colsRight, result);
- // else
- // leftV = leftMB.getDenseBlockValues();
- // }
- // else
- // leftV = left.getValues();
-
- // if(right instanceof MatrixBlockDictionary) {
- // final MatrixBlock rightMB = right.getMBDict(colsRight.length).getMatrixBlock();
- // if(rightMB.isEmpty())
- // return;
- // else if(rightMB.isInSparseFormat())
- // MMDictsDenseSparse(leftV, rightMB.getSparseBlock(), rowsLeft, colsRight, result);
- // else
- // rightV = rightMB.getDenseBlockValues();
- // }
- // else
- // rightV = right.getValues();
-
- // // if both sides were extracted.
- // if(leftV != null && rightV != null)
- // MMDictsDenseDense(leftV, rightV, rowsLeft, colsRight, result);
-
- // }
-
- // private static void MMDicts(MatrixBlock leftMB, MatrixBlock rightMB, int[] rowsLeft, int[] colsRight,
- // MatrixBlock result) {
- // if(leftMB.isEmpty() || rightMB.isEmpty())
- // return;
- // else if(rightMB.isInSparseFormat() && leftMB.isInSparseFormat())
- // throw new NotImplementedException("Not Supported sparse sparse dictionary multiplication");
- // else if(rightMB.isInSparseFormat())
- // MMDictsDenseSparse(leftMB.getDenseBlockValues(), rightMB.getSparseBlock(), rowsLeft, colsRight, result);
- // else if(leftMB.isInSparseFormat())
- // MMDictsSparseDense(leftMB.getSparseBlock(), rightMB.getDenseBlockValues(), rowsLeft, colsRight, result);
- // else
- // MMDictsDenseDense(leftMB.getDenseBlockValues(), rightMB.getDenseBlockValues(), rowsLeft, colsRight, result);
- // }
-
- // private static void MMDictsDenseDense(double[] left, double[] right, int[] rowsLeft, int[] colsRight,
- // MatrixBlock result) {
- // final int commonDim = Math.min(left.length / rowsLeft.length, right.length / colsRight.length);
- // final int resCols = result.getNumColumns();
- // final double[] resV = result.getDenseBlockValues();
- // for(int k = 0; k < commonDim; k++) {
- // final int offL = k * rowsLeft.length;
- // final int offR = k * colsRight.length;
- // for(int i = 0; i < rowsLeft.length; i++) {
- // final int offOut = rowsLeft[i] * resCols;
- // final double vl = left[offL + i];
- // if(vl != 0)
- // for(int j = 0; j < colsRight.length; j++) {
- // final double vr = right[offR + j];
- // resV[offOut + colsRight[j]] += vl * vr;
- // }
- // }
- // }
- // }
-
- // private static void MMDenseToUpperTriangle(double[] left, double[] right, int[] rowsLeft, int[] colsRight,
- // MatrixBlock result) {
- // final int commonDim = Math.min(left.length / rowsLeft.length, right.length / colsRight.length);
- // final int resCols = result.getNumColumns();
- // final double[] resV = result.getDenseBlockValues();
- // for(int k = 0; k < commonDim; k++) {
- // final int offL = k * rowsLeft.length;
- // final int offR = k * colsRight.length;
- // for(int i = 0; i < rowsLeft.length; i++) {
- // final int rowOut = rowsLeft[i];
- // final double vl = left[offL + i];
- // if(vl != 0)
- // for(int j = 0; j < colsRight.length; j++) {
- // final double vr = right[offR + j];
- // final int colOut = colsRight[j];
- // addToUpperTriangle(resCols, rowOut, colOut, resV, vl * vr);
- // }
- // }
- // }
- // }
-
- // private static void MMDenseToUpperTriangleScaling(double[] left, double[] right, int[] rowsLeft, int[] colsRight,
- // int[] scale, MatrixBlock result) {
- // final int commonDim = Math.min(left.length / rowsLeft.length, right.length / colsRight.length);
- // final int resCols = result.getNumColumns();
- // final double[] resV = result.getDenseBlockValues();
- // for(int k = 0; k < commonDim; k++) {
- // final int offL = k * rowsLeft.length;
- // final int offR = k * colsRight.length;
- // final int sv = scale[k];
- // for(int i = 0; i < rowsLeft.length; i++) {
- // final int rowOut = rowsLeft[i];
- // final double vl = left[offL + i] * sv;
- // if(vl != 0)
- // for(int j = 0; j < colsRight.length; j++) {
- // final double vr = right[offR + j];
- // final int colOut = colsRight[j];
- // addToUpperTriangle(resCols, rowOut, colOut, resV, vl * vr);
- // }
- // }
- // }
- // }
-
- // private static void MMDictsSparseDense(SparseBlock left, double[] right, int[] rowsLeft, int[] colsRight,
- // MatrixBlock result) {
- // final double[] resV = result.getDenseBlockValues();
- // final int commonDim = Math.min(left.numRows(), right.length / colsRight.length);
- // for(int i = 0; i < commonDim; i++) {
- // if(left.isEmpty(i))
- // continue;
- // final int apos = left.pos(i);
- // final int alen = left.size(i) + apos;
- // final int[] aix = left.indexes(i);
- // final double[] leftVals = left.values(i);
- // final int offRight = i * colsRight.length;
- // for(int k = apos; k < alen; k++) {
- // final int offOut = rowsLeft[aix[k]] * result.getNumColumns();
- // final double v = leftVals[k];
- // for(int j = 0; j < colsRight.length; j++)
- // resV[offOut + colsRight[j]] += v * right[offRight + j];
- // }
- // }
- // }
-
- // private static void MMDictsDenseSparse(double[] left, SparseBlock right, int[] rowsLeft, int[] colsRight,
- // MatrixBlock result) {
- // final double[] resV = result.getDenseBlockValues();
- // final int commonDim = Math.min(left.length / rowsLeft.length, right.numRows());
- // for(int i = 0; i < commonDim; i++) {
- // if(right.isEmpty(i))
- // continue;
- // final int apos = right.pos(i);
- // final int alen = right.size(i) + apos;
- // final int[] aix = right.indexes(i);
- // final double[] rightVals = right.values(i);
- // final int offLeft = i * rowsLeft.length;
- // for(int j = 0; j < rowsLeft.length; j++) {
- // final int offOut = rowsLeft[j] * result.getNumColumns();
- // final double v = left[offLeft + j];
- // if(v != 0)
- // for(int k = apos; k < alen; k++)
- // resV[offOut + colsRight[aix[k]]] += v * rightVals[k];
- // }
- // }
- // }
-
- // /**
- // * Multiply with a matrix on the left.
- // *
- // * @param matrix Matrix Block to left multiply with
- // * @param result Matrix Block result
- // * @param rl The row to start the matrix multiplication from
- // * @param ru The row to stop the matrix multiplication at.
- // */
- // @Override
- // public final void leftMultByMatrix(MatrixBlock matrix, MatrixBlock result, int rl, int ru) {
- // if(matrix.isEmpty())
- // return;
- // final int numVals = getNumValues();
- // // Pre aggregate the matrix into same size as dictionary
- // MatrixBlock preAgg = allocatePreAggregate(matrix, numVals, rl, ru);
- // preAggregate(matrix, preAgg, rl, ru);
- // preAgg.recomputeNonZeros();
- // MatrixBlock tmpRes = leftMultByPreAggregateMatrix(preAgg);
- // addMatrixToResult(tmpRes, result, rl, ru);
- // }
-
public final MatrixBlock leftMultByPreAggregateMatrix(MatrixBlock preAgg, MatrixBlock tmpRes) {
// Get dictionary.
MatrixBlock dictM = forceMatrixBlockDictionary().getMatrixBlock();
@@ -737,32 +470,19 @@ public abstract class AColGroupValue extends AColGroupCompressed implements Clon
return tmpRes;
}
- // protected void leftMultByUncompressedColGroup(ColGroupUncompressed lhs, MatrixBlock result) {
- // if(lhs.getData().isEmpty())
- // return;
- // LOG.warn("Transpose of uncompressed to fit to template need t(a) %*% b support");
- // MatrixBlock tmp = LibMatrixReorg.transpose(lhs.getData(), InfrastructureAnalyzer.getLocalParallelism());
- // final int numVals = getNumValues();
- // MatrixBlock preAgg = allocatePreAggregate(tmp, numVals, 0, tmp.getNumRows());
- // preAggregate(tmp, preAgg, 0, tmp.getNumRows());
- // preAgg.recomputeNonZeros();
- // MatrixBlock tmpRes = leftMultByPreAggregateMatrix(preAgg);
- // addMatrixToResult(tmpRes, result, lhs._colIndexes);
- // }
-
private MatrixBlockDictionary forceMatrixBlockDictionary() {
if(!(_dict instanceof MatrixBlockDictionary))
_dict = _dict.getMBDict(_colIndexes.length);
return((MatrixBlockDictionary) _dict);
}
- public void addMatrixToResult(MatrixBlock tmp, MatrixBlock result, int rl, int ru) {
+ public final void addMatrixToResult(MatrixBlock tmp, MatrixBlock result, int rl, int ru) {
if(tmp.isEmpty())
return;
final double[] retV = result.getDenseBlockValues();
final int nColRet = result.getNumColumns();
if(tmp.isInSparseFormat()) {
- SparseBlock sb = tmp.getSparseBlock();
+ final SparseBlock sb = tmp.getSparseBlock();
for(int row = rl, offT = 0; row < ru; row++, offT++) {
final int apos = sb.pos(offT);
final int alen = sb.size(offT);
@@ -784,34 +504,6 @@ public abstract class AColGroupValue extends AColGroupCompressed implements Clon
}
}
- // private void addMatrixToResult(MatrixBlock tmp, MatrixBlock result, int[] rowIndexes) {
- // if(tmp.isEmpty())
- // return;
- // final double[] retV = result.getDenseBlockValues();
- // final int nColRet = result.getNumColumns();
- // if(tmp.isInSparseFormat()) {
- // SparseBlock sb = tmp.getSparseBlock();
- // for(int row = 0; row < rowIndexes.length; row++) {
- // final int apos = sb.pos(row);
- // final int alen = sb.size(row);
- // final int[] aix = sb.indexes(row);
- // final double[] avals = sb.values(row);
- // final int offR = rowIndexes[row] * nColRet;
- // for(int i = apos; i < apos + alen; i++)
- // retV[offR + _colIndexes[aix[i]]] += avals[i];
- // }
- // }
- // else {
- // final double[] tmpV = tmp.getDenseBlockValues();
- // final int nCol = _colIndexes.length;
- // for(int row = 0, offT = 0; row < rowIndexes.length; row++, offT += nCol) {
- // final int offR = rowIndexes[row] * nColRet;
- // for(int col = 0; col < nCol; col++)
- // retV[offR + _colIndexes[col]] += tmpV[offT + col];
- // }
- // }
- // }
-
@Override
public final AColGroup rightMultByMatrix(MatrixBlock right) {
diff --git a/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupConst.java b/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupConst.java
index 2d77543..86335b9 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupConst.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupConst.java
@@ -30,6 +30,7 @@ import org.apache.sysds.runtime.compress.colgroup.dictionary.Dictionary;
import org.apache.sysds.runtime.compress.colgroup.dictionary.DictionaryFactory;
import org.apache.sysds.runtime.compress.colgroup.dictionary.MatrixBlockDictionary;
import org.apache.sysds.runtime.data.DenseBlock;
+import org.apache.sysds.runtime.data.SparseBlock;
import org.apache.sysds.runtime.functionobjects.Builtin;
import org.apache.sysds.runtime.matrix.data.LibMatrixMult;
import org.apache.sysds.runtime.matrix.data.MatrixBlock;
@@ -83,17 +84,24 @@ public class ColGroupConst extends AColGroupCompressed {
}
@Override
- public void decompressToBlock(MatrixBlock target, int rl, int ru, int offT) {
- final DenseBlock db = target.getDenseBlock();
- for(int i = rl; i < ru; i++, offT++) {
+ public void decompressToDenseBlock(DenseBlock db, int rl, int ru, int offR, int offC) {
+ for(int i = rl, offT = rl + offR; i < ru; i++, offT++) {
final double[] c = db.values(offT);
- final int off = db.pos(offT);
+ final int off = db.pos(offT) + offC;
for(int j = 0; j < _colIndexes.length; j++)
c[off + _colIndexes[j]] += _dict.getValue(j);
}
}
@Override
+ public void decompressToSparseBlock(SparseBlock ret, int rl, int ru, int offR, int offC) {
+ final int nCol = _colIndexes.length;
+ for(int i = rl, offT = rl + offR; i < ru; i++, offT++)
+ for(int j = 0; j < nCol; j++)
+ ret.append(offT, _colIndexes[j] + offC, _dict.getValue(j));
+ }
+
+ @Override
public double getIdx(int r, int colIdx) {
return _dict.getValue(colIdx);
}
diff --git a/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupDDC.java b/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupDDC.java
index e58a041..b6d4231 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupDDC.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupDDC.java
@@ -23,6 +23,7 @@ import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
+import org.apache.commons.lang.NotImplementedException;
import org.apache.sysds.runtime.compress.colgroup.dictionary.ADictionary;
import org.apache.sysds.runtime.compress.colgroup.dictionary.Dictionary;
import org.apache.sysds.runtime.compress.colgroup.mapping.AMapToData;
@@ -63,30 +64,31 @@ public class ColGroupDDC extends APreAgg {
}
@Override
- protected void decompressToBlockSparseDictionary(MatrixBlock target, int rl, int ru, int offT, SparseBlock sb) {
- final DenseBlock db = target.getDenseBlock();
- for(int i = rl; i < ru; i++, offT++) {
- final int rowIndex = _data.getIndex(i);
- if(sb.isEmpty(rowIndex))
- continue;
- final double[] c = db.values(offT);
- final int off = db.pos(offT);
- final int apos = sb.pos(rowIndex);
- final int alen = sb.size(rowIndex) + apos;
- final double[] avals = sb.values(rowIndex);
- final int[] aix = sb.indexes(rowIndex);
- for(int j = apos; j < alen; j++)
- c[off + _colIndexes[aix[j]]] += avals[j];
- }
+ protected void decompressToDenseBlockSparseDictionary(DenseBlock db, int rl, int ru, int offR, int offC,
+ SparseBlock sb) {
+ throw new NotImplementedException();
+ // for(int i = rl; i < ru; i++, offT++) {
+ // final int rowIndex = _data.getIndex(i);
+ // if(sb.isEmpty(rowIndex))
+ // continue;
+ // final double[] c = db.values(offT);
+ // final int off = db.pos(offT);
+ // final int apos = sb.pos(rowIndex);
+ // final int alen = sb.size(rowIndex) + apos;
+ // final double[] avals = sb.values(rowIndex);
+ // final int[] aix = sb.indexes(rowIndex);
+ // for(int j = apos; j < alen; j++)
+ // c[off + _colIndexes[aix[j]]] += avals[j];
+ // }
}
@Override
- protected void decompressToBlockDenseDictionary(MatrixBlock target, int rl, int ru, int offT, double[] values) {
+ protected void decompressToDenseBlockDenseDictionary(DenseBlock db, int rl, int ru, int offR, int offC,
+ double[] values) {
final int nCol = _colIndexes.length;
- final DenseBlock db = target.getDenseBlock();
- for(int i = rl; i < ru; i++, offT++) {
+ for(int i = rl,offT = rl + offR; i < ru; i++, offT++) {
final double[] c = db.values(offT);
- final int off = db.pos(offT);
+ final int off = db.pos(offT) + offC;
final int rowIndex = _data.getIndex(i) * nCol;
for(int j = 0; j < nCol; j++)
c[off + _colIndexes[j]] += values[rowIndex + j];
@@ -94,6 +96,23 @@ public class ColGroupDDC extends APreAgg {
}
@Override
+ protected void decompressToSparseBlockSparseDictionary(SparseBlock ret, int rl, int ru, int offR, int offC,
+ SparseBlock sb) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ protected void decompressToSparseBlockDenseDictionary(SparseBlock ret, int rl, int ru, int offR, int offC,
+ double[] values) {
+ final int nCol = _colIndexes.length;
+ for(int i = rl, offT = rl + offR; i < ru; i++, offT++) {
+ final int rowIndex = _data.getIndex(i) * nCol;
+ for(int j = 0; j < nCol; j++)
+ ret.append(offT, _colIndexes[j] + offC, values[rowIndex + j]);
+ }
+ }
+
+ @Override
public double getIdx(int r, int colIdx) {
return _dict.getValue(_data.getIndex(r) * _colIndexes.length + colIdx);
}
diff --git a/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupEmpty.java b/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupEmpty.java
index 7f0cd46..ec20674 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupEmpty.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupEmpty.java
@@ -20,6 +20,8 @@
package org.apache.sysds.runtime.compress.colgroup;
import org.apache.sysds.runtime.compress.colgroup.dictionary.Dictionary;
+import org.apache.sysds.runtime.data.DenseBlock;
+import org.apache.sysds.runtime.data.SparseBlock;
import org.apache.sysds.runtime.functionobjects.Builtin;
import org.apache.sysds.runtime.functionobjects.ValueFunction;
import org.apache.sysds.runtime.matrix.data.MatrixBlock;
@@ -62,7 +64,12 @@ public class ColGroupEmpty extends AColGroupCompressed {
}
@Override
- public void decompressToBlock(MatrixBlock target, int rl, int ru, int offT) {
+ public void decompressToDenseBlock(DenseBlock target, int rl, int ru, int offR, int offC) {
+ // do nothing.
+ }
+
+ @Override
+ public void decompressToSparseBlock(SparseBlock sb, int rl, int ru, int offR, int offC){
// do nothing.
}
diff --git a/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupOLE.java b/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupOLE.java
index 9920384..a303d98 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupOLE.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupOLE.java
@@ -25,6 +25,7 @@ import org.apache.commons.lang.NotImplementedException;
import org.apache.sysds.runtime.compress.CompressionSettings;
import org.apache.sysds.runtime.compress.DMLCompressionException;
import org.apache.sysds.runtime.compress.colgroup.dictionary.ADictionary;
+import org.apache.sysds.runtime.data.DenseBlock;
import org.apache.sysds.runtime.data.SparseBlock;
import org.apache.sysds.runtime.functionobjects.Builtin;
import org.apache.sysds.runtime.matrix.data.MatrixBlock;
@@ -65,45 +66,58 @@ public class ColGroupOLE extends AColGroupOffset {
}
@Override
- protected void decompressToBlockDenseDictionary(MatrixBlock target, int rl, int ru, int offT, double[] values) {
- final int blksz = CompressionSettings.BITMAP_BLOCK_SZ;
- final int numCols = getNumCols();
- final int numVals = getNumValues();
- final int offOut = (rl - offT);
- final int targetCols = target.getNumColumns();
+ protected void decompressToDenseBlockDenseDictionary(DenseBlock db, int rl, int ru, int offR, int offC, double[] values) {
+ throw new NotImplementedException();
+ // final int blksz = CompressionSettings.BITMAP_BLOCK_SZ;
+ // final int numCols = getNumCols();
+ // final int numVals = getNumValues();
+ // final int offOut = (rl - offT);
+ // final int targetCols = target.getNumColumns();
+
+ // // cache blocking config and position array
+ // int[] apos = skipScan(numVals, rl);
+ // double[] c = target.getDenseBlockValues();
+ // // cache conscious append via horizontal scans
+ // for(int bi = (rl / blksz) * blksz; bi < ru; bi += blksz) {
+ // for(int k = 0, off = 0; k < numVals; k++, off += numCols) {
+ // int boff = _ptr[k];
+ // int blen = len(k);
+ // int bix = apos[k];
+
+ // if(bix >= blen)
+ // continue;
+ // int pos = boff + bix;
+ // int len = _data[pos];
+ // int i = 1;
+ // int row = bi + _data[pos + 1];
+ // while(i <= len && row < rl)
+ // row = bi + _data[pos + i++];
+
+ // for(; i <= len && row < ru; i++) {
+ // row = bi + _data[pos + i];
+ // int rc = (row - offOut) * targetCols;
+ // for(int j = 0; j < numCols; j++)
+ // c[rc + _colIndexes[j]] += values[off + j];
+ // }
+ // apos[k] += len + 1;
+ // }
+ // }
+ }
- // cache blocking config and position array
- int[] apos = skipScan(numVals, rl);
- double[] c = target.getDenseBlockValues();
- // cache conscious append via horizontal scans
- for(int bi = (rl / blksz) * blksz; bi < ru; bi += blksz) {
- for(int k = 0, off = 0; k < numVals; k++, off += numCols) {
- int boff = _ptr[k];
- int blen = len(k);
- int bix = apos[k];
+ @Override
+ protected void decompressToDenseBlockSparseDictionary(DenseBlock db, int rl, int ru, int offR, int offC, SparseBlock values) {
+ throw new NotImplementedException();
+ }
- if(bix >= blen)
- continue;
- int pos = boff + bix;
- int len = _data[pos];
- int i = 1;
- int row = bi + _data[pos + 1];
- while(i <= len && row < rl)
- row = bi + _data[pos + i++];
-
- for(; i <= len && row < ru; i++) {
- row = bi + _data[pos + i];
- int rc = (row - offOut) * targetCols;
- for(int j = 0; j < numCols; j++)
- c[rc + _colIndexes[j]] += values[off + j];
- }
- apos[k] += len + 1;
- }
- }
+ @Override
+ protected void decompressToSparseBlockSparseDictionary(SparseBlock ret, int rl, int ru, int offR, int offC,
+ SparseBlock sb) {
+ throw new NotImplementedException();
}
@Override
- protected void decompressToBlockSparseDictionary(MatrixBlock target, int rl, int ru, int offT, SparseBlock values) {
+ protected void decompressToSparseBlockDenseDictionary(SparseBlock ret, int rl, int ru, int offR, int offC,
+ double[] values) {
throw new NotImplementedException();
}
diff --git a/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupRLE.java b/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupRLE.java
index bb84779..3d69b96 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupRLE.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupRLE.java
@@ -26,6 +26,7 @@ import java.util.List;
import org.apache.commons.lang.NotImplementedException;
import org.apache.sysds.runtime.compress.CompressionSettings;
import org.apache.sysds.runtime.compress.colgroup.dictionary.ADictionary;
+import org.apache.sysds.runtime.data.DenseBlock;
import org.apache.sysds.runtime.data.SparseBlock;
import org.apache.sysds.runtime.functionobjects.Builtin;
import org.apache.sysds.runtime.matrix.data.MatrixBlock;
@@ -64,44 +65,58 @@ public class ColGroupRLE extends AColGroupOffset {
}
@Override
- protected void decompressToBlockDenseDictionary(MatrixBlock target, int rl, int ru, int offT, double[] values) {
- final int blksz = CompressionSettings.BITMAP_BLOCK_SZ;
- final int numCols = getNumCols();
- final int numVals = getNumValues();
-
- // position and start offset arrays
- int[] astart = new int[numVals];
- int[] apos = skipScan(numVals, rl, astart);
+ protected void decompressToDenseBlockDenseDictionary(DenseBlock target, int rl, int ru, int offR, int offC, double[] values) {
+ throw new NotImplementedException();
+ // final int blksz = CompressionSettings.BITMAP_BLOCK_SZ;
+ // final int numCols = getNumCols();
+ // final int numVals = getNumValues();
+
+ // // position and start offset arrays
+ // int[] astart = new int[numVals];
+ // int[] apos = skipScan(numVals, rl, astart);
+
+ // double[] c = target.getDenseBlockValues();
+ // // cache conscious append via horizontal scans
+ // for(int bi = rl; bi < ru; bi += blksz) {
+ // int bimax = Math.min(bi + blksz, ru);
+ // for(int k = 0, off = 0; k < numVals; k++, off += numCols) {
+ // int boff = _ptr[k];
+ // int blen = len(k);
+ // int bix = apos[k];
+ // int start = astart[k];
+ // for(; bix < blen & start < bimax; bix += 2) {
+ // start += _data[boff + bix];
+ // int len = _data[boff + bix + 1];
+ // for(int i = Math.max(rl, start) - (rl - offT); i < Math.min(start + len, ru) - (rl - offT); i++) {
+
+ // int rc = i * target.getNumColumns();
+ // for(int j = 0; j < numCols; j++)
+ // c[rc + _colIndexes[j]] += values[off + j];
+
+ // }
+ // start += len;
+ // }
+ // apos[k] = bix;
+ // astart[k] = start;
+ // }
+ // }
+ }
- double[] c = target.getDenseBlockValues();
- // cache conscious append via horizontal scans
- for(int bi = rl; bi < ru; bi += blksz) {
- int bimax = Math.min(bi + blksz, ru);
- for(int k = 0, off = 0; k < numVals; k++, off += numCols) {
- int boff = _ptr[k];
- int blen = len(k);
- int bix = apos[k];
- int start = astart[k];
- for(; bix < blen & start < bimax; bix += 2) {
- start += _data[boff + bix];
- int len = _data[boff + bix + 1];
- for(int i = Math.max(rl, start) - (rl - offT); i < Math.min(start + len, ru) - (rl - offT); i++) {
+ @Override
+ protected void decompressToDenseBlockSparseDictionary(DenseBlock target, int rl, int ru, int offR, int offC, SparseBlock values) {
+ throw new NotImplementedException();
+ }
- int rc = i * target.getNumColumns();
- for(int j = 0; j < numCols; j++)
- c[rc + _colIndexes[j]] += values[off + j];
- }
- start += len;
- }
- apos[k] = bix;
- astart[k] = start;
- }
- }
+ @Override
+ protected void decompressToSparseBlockSparseDictionary(SparseBlock ret, int rl, int ru, int offR, int offC,
+ SparseBlock sb) {
+ throw new NotImplementedException();
}
@Override
- protected void decompressToBlockSparseDictionary(MatrixBlock target, int rl, int ru, int offT, SparseBlock values) {
+ protected void decompressToSparseBlockDenseDictionary(SparseBlock ret, int rl, int ru, int offR, int offC,
+ double[] values) {
throw new NotImplementedException();
}
diff --git a/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupSDC.java b/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupSDC.java
index 65bb8cb..fc011e0 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupSDC.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupSDC.java
@@ -23,6 +23,7 @@ import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
+import org.apache.commons.lang.NotImplementedException;
import org.apache.sysds.runtime.compress.DMLCompressionException;
import org.apache.sysds.runtime.compress.colgroup.dictionary.ADictionary;
import org.apache.sysds.runtime.compress.colgroup.mapping.AMapToData;
@@ -84,16 +85,17 @@ public class ColGroupSDC extends AColGroupValue {
}
@Override
- protected void decompressToBlockDenseDictionary(MatrixBlock target, int rl, int ru, int offT, double[] values) {
+ protected void decompressToDenseBlockDenseDictionary(DenseBlock db, int rl, int ru, int offR, int offC,
+ double[] values) {
final int nCol = _colIndexes.length;
final int offsetToDefault = values.length - nCol;
- final DenseBlock db = target.getDenseBlock();
+ final AIterator it = _indexes.getIterator(rl);
+ int offT = rl + offR;
int i = rl;
- AIterator it = _indexes.getIterator(rl);
for(; i < ru && it.hasNext(); i++, offT++) {
final double[] c = db.values(offT);
- final int off = db.pos(offT);
+ final int off = db.pos(offT) + offC;
if(it.value() == i) {
int offset = _data.getIndex(it.getDataIndexAndIncrement()) * nCol;
for(int j = 0; j < nCol; j++)
@@ -106,7 +108,7 @@ public class ColGroupSDC extends AColGroupValue {
for(; i < ru; i++, offT++) {
final double[] c = db.values(offT);
- final int off = db.pos(offT);
+ final int off = db.pos(offT) + offC;
for(int j = 0; j < nCol; j++)
c[off + _colIndexes[j]] += values[offsetToDefault + j];
}
@@ -115,40 +117,83 @@ public class ColGroupSDC extends AColGroupValue {
}
@Override
- protected void decompressToBlockSparseDictionary(MatrixBlock target, int rl, int ru, int offT, SparseBlock sb) {
- final int offsetToDefault = sb.numRows() - 1;
- final int defApos = sb.pos(offsetToDefault);
- final int defAlen = sb.size(offsetToDefault) + defApos;
- final double[] defAvals = sb.values(offsetToDefault);
- final int[] defAix = sb.indexes(offsetToDefault);
- final DenseBlock db = target.getDenseBlock();
+ protected void decompressToDenseBlockSparseDictionary(DenseBlock db, int rl, int ru, int offR, int offC,
+ SparseBlock sb) {
+ throw new NotImplementedException();
+ // final int offsetToDefault = sb.numRows() - 1;
+ // final int defApos = sb.pos(offsetToDefault);
+ // final int defAlen = sb.size(offsetToDefault) + defApos;
+ // final double[] defAvals = sb.values(offsetToDefault);
+ // final int[] defAix = sb.indexes(offsetToDefault);
+ // final DenseBlock db = target.getDenseBlock();
+
+ // int i = rl;
+ // AIterator it = _indexes.getIterator(rl);
+ // for(; i < ru && it.hasNext(); i++, offT++) {
+ // final double[] c = db.values(offT);
+ // final int off = db.pos(offT);
+ // if(it.value() == i) {
+ // int dictIndex = _data.getIndex(it.getDataIndexAndIncrement());
+ // if(sb.isEmpty(dictIndex))
+ // continue;
+ // final int apos = sb.pos(dictIndex);
+ // final int alen = sb.size(dictIndex) + apos;
+ // final double[] avals = sb.values(dictIndex);
+ // final int[] aix = sb.indexes(dictIndex);
+ // for(int j = apos; j < alen; j++)
+ // c[off + _colIndexes[aix[j]]] += avals[j];
+ // }
+ // else
+ // for(int j = defApos; j < defAlen; j++)
+ // c[off + _colIndexes[defAix[j]]] += defAvals[j];
+ // }
+
+ // for(; i < ru; i++, offT++) {
+ // final double[] c = db.values(offT);
+ // final int off = db.pos(offT);
+ // for(int j = defApos; j < defAlen; j++)
+ // c[off + _colIndexes[defAix[j]]] += defAvals[j];
+ // }
+
+ // _indexes.cacheIterator(it, ru);
+ }
+
+ @Override
+ protected void decompressToSparseBlockSparseDictionary(SparseBlock ret, int rl, int ru, int offR, int offC,
+ SparseBlock sb) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ protected void decompressToSparseBlockDenseDictionary(SparseBlock ret, int rl, int ru, int offR, int offC,
+ double[] values) {
+ final int nCol = _colIndexes.length;
+ final int offsetToDefault = values.length - nCol;
+ final AIterator it = _indexes.getIterator(rl);
+ int offT = rl + offR;
int i = rl;
- AIterator it = _indexes.getIterator(rl);
for(; i < ru && it.hasNext(); i++, offT++) {
- final double[] c = db.values(offT);
- final int off = db.pos(offT);
+ // final double[] c = db.values(offT);
+ // final int off = db.pos(offT) + offC;
if(it.value() == i) {
- int dictIndex = _data.getIndex(it.getDataIndexAndIncrement());
- if(sb.isEmpty(dictIndex))
- continue;
- final int apos = sb.pos(dictIndex);
- final int alen = sb.size(dictIndex) + apos;
- final double[] avals = sb.values(dictIndex);
- final int[] aix = sb.indexes(dictIndex);
- for(int j = apos; j < alen; j++)
- c[off + _colIndexes[aix[j]]] += avals[j];
+ int offset = _data.getIndex(it.getDataIndexAndIncrement()) * nCol;
+ for(int j = 0; j < nCol; j++)
+ ret.append(offT, _colIndexes[j] + offC, values[offset + j]);
+ // c[off + _colIndexes[j]] += values[offset + j];
}
else
- for(int j = defApos; j < defAlen; j++)
- c[off + _colIndexes[defAix[j]]] += defAvals[j];
+ for(int j = 0; j < nCol; j++)
+ ret.append(offT, _colIndexes[j] + offC, values[offsetToDefault + j]);
+ // c[off + _colIndexes[j]] += values[offsetToDefault + j];
}
for(; i < ru; i++, offT++) {
- final double[] c = db.values(offT);
- final int off = db.pos(offT);
- for(int j = defApos; j < defAlen; j++)
- c[off + _colIndexes[defAix[j]]] += defAvals[j];
+ // final double[] c = db.values(offT);
+ // final int off = db.pos(offT) + offC;
+ for(int j = 0; j < nCol; j++)
+ ret.append(offT, _colIndexes[j] + offC, values[offsetToDefault + j]);
+ // c[off + _colIndexes[j]] += values[offsetToDefault + j];
}
_indexes.cacheIterator(it, ru);
diff --git a/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupSDCSingle.java b/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupSDCSingle.java
index c319d2d..cb123ec 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupSDCSingle.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupSDCSingle.java
@@ -77,16 +77,17 @@ public class ColGroupSDCSingle extends AColGroupValue {
}
@Override
- protected void decompressToBlockDenseDictionary(MatrixBlock target, int rl, int ru, int offT, double[] values) {
+ protected void decompressToDenseBlockDenseDictionary(DenseBlock db, int rl, int ru, int offR, int offC,
+ double[] values) {
final int nCol = _colIndexes.length;
final int offsetToDefault = values.length - nCol;
- final DenseBlock db = target.getDenseBlock();
+ final AIterator it = _indexes.getIterator(rl);
+ int offT = rl + offR;
int i = rl;
- AIterator it = _indexes.getIterator(rl);
for(; i < ru && it.hasNext(); i++, offT++) {
final double[] c = db.values(offT);
- final int off = db.pos(offT);
+ final int off = db.pos(offT) + offC;
if(it.value() == i) {
for(int j = 0; j < nCol; j++)
c[off + _colIndexes[j]] += values[j];
@@ -99,7 +100,7 @@ public class ColGroupSDCSingle extends AColGroupValue {
for(; i < ru; i++, offT++) {
final double[] c = db.values(offT);
- final int off = db.pos(offT);
+ final int off = db.pos(offT) + offC;
for(int j = 0; j < nCol; j++)
c[off + _colIndexes[j]] += values[offsetToDefault + j];
}
@@ -108,11 +109,45 @@ public class ColGroupSDCSingle extends AColGroupValue {
}
@Override
- protected void decompressToBlockSparseDictionary(MatrixBlock target, int rl, int ru, int offT, SparseBlock values) {
+ protected void decompressToDenseBlockSparseDictionary(DenseBlock db, int rl, int ru, int offR, int offC,
+ SparseBlock values) {
throw new NotImplementedException();
}
@Override
+ protected void decompressToSparseBlockSparseDictionary(SparseBlock ret, int rl, int ru, int offR, int offC,
+ SparseBlock sb) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ protected void decompressToSparseBlockDenseDictionary(SparseBlock ret, int rl, int ru, int offR, int offC,
+ double[] values) {
+ final int nCol = _colIndexes.length;
+ final int offsetToDefault = values.length - nCol;
+ final AIterator it = _indexes.getIterator(rl);
+
+ int offT = rl + offR;
+ int i = rl;
+ for(; i < ru && it.hasNext(); i++, offT++) {
+ if(it.value() == i) {
+ for(int j = 0; j < nCol; j++)
+ ret.append(offT, _colIndexes[j] + offC, values[j]);
+ it.next();
+ }
+ else
+ for(int j = 0; j < nCol; j++)
+ ret.append(offT, _colIndexes[j] + offC, values[offsetToDefault + j]);
+ }
+
+ for(; i < ru; i++, offT++)
+ for(int j = 0; j < nCol; j++)
+ ret.append(offT, _colIndexes[j] + offC, values[offsetToDefault + j]);
+
+ _indexes.cacheIterator(it, ru);
+ }
+
+ @Override
public double getIdx(int r, int colIdx) {
AIterator it = _indexes.getIterator(r);
if(it.value() == r)
diff --git a/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupSDCSingleZeros.java b/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupSDCSingleZeros.java
index 28b5fdc..d8edd0d 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupSDCSingleZeros.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupSDCSingleZeros.java
@@ -23,6 +23,7 @@ import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
+import org.apache.commons.lang.NotImplementedException;
import org.apache.sysds.runtime.compress.colgroup.dictionary.ADictionary;
import org.apache.sysds.runtime.compress.colgroup.dictionary.Dictionary;
import org.apache.sysds.runtime.compress.colgroup.offset.AIterator;
@@ -77,47 +78,65 @@ public class ColGroupSDCSingleZeros extends APreAgg {
}
@Override
- protected void decompressToBlockDenseDictionary(MatrixBlock target, int rl, int ru, int offT, double[] values) {
+ protected void decompressToDenseBlockDenseDictionary(DenseBlock db, int rl, int ru, int offR, int offC,
+ double[] values) {
final int nCol = _colIndexes.length;
- final int offTCorr = offT - rl;
- final DenseBlock db = target.getDenseBlock();
-
- AIterator it = _indexes.getIterator(rl);
+ final AIterator it = _indexes.getIterator(rl);
while(it.hasNext() && it.value() < ru) {
- final int idx = offTCorr + it.value();
- final double[] c = db.values(idx);
- final int off = db.pos(idx);
+ final int row = offR + it.value();
+ final double[] c = db.values(row);
+ final int off = db.pos(row) + offC;
for(int j = 0; j < nCol; j++)
c[off + _colIndexes[j]] += values[j];
it.next();
}
-
_indexes.cacheIterator(it, ru);
}
@Override
- protected void decompressToBlockSparseDictionary(MatrixBlock target, int rl, int ru, int offT, SparseBlock values) {
+ protected void decompressToDenseBlockSparseDictionary(DenseBlock db, int rl, int ru, int offR, int offC,
+ SparseBlock values) {
+ throw new NotImplementedException();
+ // final int offTCorr = offT - rl;
+ // final DenseBlock db = target.getDenseBlock();
+ // final int apos = values.pos(0);
+ // final int alen = values.size(0) + apos;
+ // final int[] aix = values.indexes(0);
+ // final double[] avals = values.values(0);
+
+ // AIterator it = _indexes.getIterator(rl);
+ // while(it.hasNext() && it.value() < ru) {
+ // final int idx = offTCorr + it.value();
+ // final double[] c = db.values(idx);
+ // final int off = db.pos(idx);
+
+ // for(int j = apos; j < alen; j++)
+ // c[off + _colIndexes[aix[j]]] += avals[j];
+
+ // it.next();
+ // }
+
+ // _indexes.cacheIterator(it, ru);
+ }
- final int offTCorr = offT - rl;
- final DenseBlock db = target.getDenseBlock();
- final int apos = values.pos(0);
- final int alen = values.size(0) + apos;
- final int[] aix = values.indexes(0);
- final double[] avals = values.values(0);
+ @Override
+ protected void decompressToSparseBlockSparseDictionary(SparseBlock ret, int rl, int ru, int offR, int offC,
+ SparseBlock sb) {
+ throw new NotImplementedException();
+ }
- AIterator it = _indexes.getIterator(rl);
+ @Override
+ protected void decompressToSparseBlockDenseDictionary(SparseBlock ret, int rl, int ru, int offR, int offC,
+ double[] values) {
+ final int nCol = _colIndexes.length;
+ final AIterator it = _indexes.getIterator(rl);
while(it.hasNext() && it.value() < ru) {
- final int idx = offTCorr + it.value();
- final double[] c = db.values(idx);
- final int off = db.pos(idx);
-
- for(int j = apos; j < alen; j++)
- c[off + _colIndexes[aix[j]]] += avals[j];
-
+ final int row = offR + it.value();
+ for(int j = 0; j < nCol; j++)
+ ret.append(row, _colIndexes[j] + offC, values[j]);
it.next();
}
-
_indexes.cacheIterator(it, ru);
}
@@ -261,7 +280,7 @@ public class ColGroupSDCSingleZeros extends APreAgg {
}
@Override
- public AColGroup binaryRowOpLeft(BinaryOperator op, double[] v,boolean isRowSafe) {
+ public AColGroup binaryRowOpLeft(BinaryOperator op, double[] v, boolean isRowSafe) {
if(isRowSafe) {
ADictionary ret = _dict.binOpLeft(op, v, _colIndexes);
return new ColGroupSDCSingleZeros(_colIndexes, _numRows, ret, _indexes, getCachedCounts());
@@ -273,7 +292,7 @@ public class ColGroupSDCSingleZeros extends APreAgg {
}
@Override
- public AColGroup binaryRowOpRight(BinaryOperator op, double[] v,boolean isRowSafe) {
+ public AColGroup binaryRowOpRight(BinaryOperator op, double[] v, boolean isRowSafe) {
if(isRowSafe) {
ADictionary ret = _dict.binOpRight(op, v, _colIndexes);
return new ColGroupSDCSingleZeros(_colIndexes, _numRows, ret, _indexes, getCachedCounts());
diff --git a/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupSDCZeros.java b/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupSDCZeros.java
index ec93247..a7632dd 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupSDCZeros.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupSDCZeros.java
@@ -98,35 +98,51 @@ public class ColGroupSDCZeros extends APreAgg {
}
@Override
- protected void decompressToBlockDenseDictionary(MatrixBlock target, int rl, int ru, int offT, double[] values) {
+ protected void decompressToDenseBlockDenseDictionary(DenseBlock db, int rl, int ru, int offR, int offC,
+ double[] values) {
final int nCol = _colIndexes.length;
- final int offTCorrected = offT - rl;
- final DenseBlock db = target.getDenseBlock();
AIterator it = _indexes.getIterator(rl);
while(it.hasNext() && it.value() < ru) {
- final int idx = offTCorrected + it.value();
+ final int idx = offR + it.value();
final double[] c = db.values(idx);
- final int off = db.pos(idx);
- final int offC = getIndex(it.getDataIndexAndIncrement()) * nCol;
- for(int j = 0; j < nCol; j++) {
- c[off + _colIndexes[j]] += values[offC + j];
- }
+ final int off = db.pos(idx) + offC;
+ final int offDict = getIndex(it.getDataIndexAndIncrement()) * nCol;
+ for(int j = 0; j < nCol; j++)
+ c[off + _colIndexes[j]] += values[offDict + j];
+
}
_indexes.cacheIterator(it, ru);
}
@Override
- protected void decompressToBlockSparseDictionary(MatrixBlock target, int rl, int ru, int offT, SparseBlock sb) {
+ protected void decompressToDenseBlockSparseDictionary(DenseBlock db, int rl, int ru, int offR, int offC,
+ SparseBlock sb) {
+ AIterator it = _indexes.getIterator(rl);
+ while(it.hasNext() && it.value() < ru) {
+ final int idx = offR + it.value();
+ final int dictIndex = getIndex(it.getDataIndexAndIncrement());
+ if(sb.isEmpty(dictIndex))
+ continue;
- final int offTCorrected = offT - rl;
- final DenseBlock db = target.getDenseBlock();
+ final double[] c = db.values(idx);
+ final int off = db.pos(idx) + offC;
+ final int apos = sb.pos(dictIndex);
+ final int alen = sb.size(dictIndex) + apos;
+ final double[] avals = sb.values(dictIndex);
+ final int[] aix = sb.indexes(dictIndex);
+ for(int j = apos; j < alen; j++)
+ c[off + _colIndexes[aix[j]]] += avals[j];
+ }
+ _indexes.cacheIterator(it, ru);
+ }
+ @Override
+ protected void decompressToSparseBlockSparseDictionary(SparseBlock ret, int rl, int ru, int offR, int offC,
+ SparseBlock sb) {
AIterator it = _indexes.getIterator(rl);
while(it.hasNext() && it.value() < ru) {
- final int idx = offTCorrected + it.value();
- final double[] c = db.values(idx);
- final int off = db.pos(idx);
+ final int row = offR + it.value();
final int dictIndex = getIndex(it.getDataIndexAndIncrement());
if(sb.isEmpty(dictIndex))
continue;
@@ -136,7 +152,22 @@ public class ColGroupSDCZeros extends APreAgg {
final double[] avals = sb.values(dictIndex);
final int[] aix = sb.indexes(dictIndex);
for(int j = apos; j < alen; j++)
- c[off + _colIndexes[aix[j]]] += avals[j];
+ ret.append(row, _colIndexes[aix[j]] + offC, avals[j] );
+ }
+ _indexes.cacheIterator(it, ru);
+ }
+
+ @Override
+ protected void decompressToSparseBlockDenseDictionary(SparseBlock ret, int rl, int ru, int offR, int offC,
+ double[] values) {
+ final int nCol = _colIndexes.length;
+
+ AIterator it = _indexes.getIterator(rl);
+ while(it.hasNext() && it.value() < ru) {
+ final int row = offR + it.value();
+ final int offDict = getIndex(it.getDataIndexAndIncrement()) * nCol;
+ for(int j = 0; j < nCol; j++)
+ ret.append(row, _colIndexes[j] + offC, values[offDict + j]);
}
_indexes.cacheIterator(it, ru);
}
diff --git a/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupUncompressed.java b/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupUncompressed.java
index 1b001d6..07b3d88 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupUncompressed.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupUncompressed.java
@@ -174,35 +174,76 @@ public class ColGroupUncompressed extends AColGroup {
}
@Override
- public void decompressToBlock(MatrixBlock target, int rl, int ru, int offT) {
- final DenseBlock db = target.getDenseBlock();
+ public void decompressToDenseBlock(DenseBlock db, int rl, int ru, int offR, int offC) {
+ if(_data.isEmpty())
+ return;
+ else if(_data.isInSparseFormat())
+ decompressToDenseBlockSparseData(db, rl, ru, offR, offC);
+ else
+ decompressToDenseBlockDenseData(db, rl, ru, offR, offC);
+ }
+
+ private void decompressToDenseBlockDenseData(DenseBlock db, int rl, int ru, int offR, int offC) {
+ int offT = rl + offR;
final int nCol = _colIndexes.length;
- if(_data.isInSparseFormat()) {
- SparseBlock sb = _data.getSparseBlock();
- if(sb == null)
- return;
- for(int row = rl; row < ru; row++, offT++) {
- if(sb.isEmpty(row))
- continue;
- final double[] c = db.values(offT);
- final int off = db.pos(offT);
- final int apos = sb.pos(row);
- final int alen = sb.size(row) + apos;
- final int[] aix = sb.indexes(row);
- final double[] avals = sb.values(row);
- for(int col = apos; col < alen; col++)
- c[_colIndexes[aix[col]] + off] += avals[col];
- }
+ final double[] values = _data.getDenseBlockValues();
+ int offS = rl * nCol;
+ for(int row = rl; row < ru; row++, offT++, offS += nCol) {
+ final double[] c = db.values(offT);
+ final int off = db.pos(offT) + offC;
+ for(int j = 0; j < nCol; j++)
+ c[off + _colIndexes[j]] += values[offS + j];
}
- else {
- final double[] values = _data.getDenseBlockValues();
- int offS = rl * nCol;
- for(int row = rl; row < ru; row++, offT++, offS += nCol) {
- final double[] c = db.values(offT);
- final int off = db.pos(offT);
- for(int j = 0; j < nCol; j++)
- c[off + _colIndexes[j]] += values[offS + j];
- }
+ }
+
+ private void decompressToDenseBlockSparseData(DenseBlock db, int rl, int ru, int offR, int offC) {
+
+ final SparseBlock sb = _data.getSparseBlock();
+ for(int row = rl, offT = rl + offR; row < ru; row++, offT++) {
+ if(sb.isEmpty(row))
+ continue;
+ final double[] c = db.values(offT);
+ final int off = db.pos(offT) + offC;
+ final int apos = sb.pos(row);
+ final int alen = sb.size(row) + apos;
+ final int[] aix = sb.indexes(row);
+ final double[] avals = sb.values(row);
+ for(int col = apos; col < alen; col++)
+ c[_colIndexes[aix[col]] + off] += avals[col];
+ }
+ }
+
+ @Override
+ public void decompressToSparseBlock(SparseBlock ret, int rl, int ru, int offR, int offC) {
+ if(_data.isEmpty())
+ return;
+ else if(_data.isInSparseFormat())
+ decompressToSparseBlockSparseData(ret, rl, ru, offR, offC);
+ else
+ decompressToSparseBlockDenseData(ret, rl, ru, offR, offC);
+ }
+
+ private void decompressToSparseBlockDenseData(SparseBlock ret, int rl, int ru, int offR, int offC) {
+ final int nCol = _colIndexes.length;
+ final double[] values = _data.getDenseBlockValues();
+ int offS = rl * nCol;
+ for(int row = rl, offT = rl + offR; row < ru; row++, offT++, offS += nCol)
+ for(int j = 0; j < nCol; j++)
+ ret.append(offT, offC + _colIndexes[j], values[offS + j]);
+ }
+
+ private void decompressToSparseBlockSparseData(SparseBlock ret, int rl, int ru, int offR, int offC) {
+ int offT = rl + offR;
+ final SparseBlock sb = _data.getSparseBlock();
+ for(int row = rl; row < ru; row++, offT++) {
+ if(sb.isEmpty(row))
+ continue;
+ final int apos = sb.pos(row);
+ final int alen = sb.size(row) + apos;
+ final int[] aix = sb.indexes(row);
+ final double[] avals = sb.values(row);
+ for(int col = apos; col < alen; col++)
+ ret.append(offT, offC + _colIndexes[aix[col]], avals[col]);
}
}
@@ -579,6 +620,8 @@ public class ColGroupUncompressed extends AColGroup {
final SparseBlock subR = subBlockRight.getSparseBlock();
long nnz = 0;
for(int i = 0; i < _colIndexes.length; i++) {
+ if(sbR.isEmpty(_colIndexes[i]))
+ continue;
subR.set(i, sbR.get(_colIndexes[i]), false);
nnz += sbR.get(_colIndexes[i]).size();
}
diff --git a/src/main/java/org/apache/sysds/runtime/compress/lib/CLALibBinaryCellOp.java b/src/main/java/org/apache/sysds/runtime/compress/lib/CLALibBinaryCellOp.java
index 2e003d2..e4c3333 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/lib/CLALibBinaryCellOp.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/lib/CLALibBinaryCellOp.java
@@ -62,6 +62,7 @@ public class CLALibBinaryCellOp {
public static MatrixBlock binaryOperationsRight(BinaryOperator op, CompressedMatrixBlock m1, MatrixBlock that,
MatrixBlock result) {
+
if(that.getNumRows() == 1 && that.getNumColumns() == 1) {
ScalarOperator sop = new RightScalarOperator(op.fn, that.getValue(0, 0), op.getNumThreads());
return CLALibScalar.scalarOperations(sop, m1, result);
@@ -87,6 +88,7 @@ public class CLALibBinaryCellOp {
LibMatrixBincell.isValidDimensionsBinary(that, m1);
BinaryAccessType atype = LibMatrixBincell.getBinaryAccessType(that, m1);
return selectProcessingBasedOnAccessType(op, m1, that, result, atype, true);
+
}
private static MatrixBlock binaryOperationsEmpty(BinaryOperator op, CompressedMatrixBlock m1, MatrixBlock that,
@@ -106,28 +108,25 @@ public class CLALibBinaryCellOp {
}
else
throw new NotImplementedException("Function Type: " + fn);
-
return result;
-
}
private static MatrixBlock selectProcessingBasedOnAccessType(BinaryOperator op, CompressedMatrixBlock m1,
MatrixBlock that, MatrixBlock result, BinaryAccessType atype, boolean left) {
if(atype == BinaryAccessType.MATRIX_COL_VECTOR) {
-
+ // Column vector access
MatrixBlock d_compressed = m1.getCachedDecompressed();
-
if(d_compressed != null) {
if(left)
return that.binaryOperations(op, d_compressed);
else
return d_compressed.binaryOperations(op, that);
}
-
return binaryMVCol(m1, that, op, left);
}
else if(atype == BinaryAccessType.MATRIX_MATRIX) {
+ // Full matrix access.
MatrixBlock d_compressed = m1.getUncompressed("MatrixMatrix " + op);
if(left)
return that.binaryOperations(op, d_compressed);
@@ -135,8 +134,10 @@ public class CLALibBinaryCellOp {
return d_compressed.binaryOperations(op, that);
}
else if(isSupportedBinaryCellOp(op.fn))
- return bincellOp(m1, that, setupCompressedReturnMatrixBlock(m1, result), op, left);
+ // Row matrix access.
+ return rowBinCellOp(m1, that, result, op, left);
else
+ // All other, fallback to default execution.
return CompressedMatrixBlock.getUncompressed(m1, "BinaryOp: " + op.fn).binaryOperations(op, that, result);
}
@@ -157,14 +158,15 @@ public class CLALibBinaryCellOp {
return ret;
}
- private static MatrixBlock bincellOp(CompressedMatrixBlock m1, MatrixBlock m2, CompressedMatrixBlock ret,
+ private static MatrixBlock rowBinCellOp(CompressedMatrixBlock m1, MatrixBlock m2, MatrixBlock ret,
BinaryOperator op, boolean left) {
+ CompressedMatrixBlock cRet = setupCompressedReturnMatrixBlock(m1, ret);
if(isValidForOverlappingBinaryCellOperations(m1, op))
- overlappingBinaryCellOp(m1, m2, ret, op, left);
+ overlappingBinaryCellOp(m1, m2, cRet, op, left);
else
- nonOverlappingBinaryCellOp(m1, m2, ret, op, left);
- return ret;
-
+ nonOverlappingBinaryCellOp(m1, m2, cRet, op, left);
+ cRet.recomputeNonZeros();
+ return cRet;
}
private static void nonOverlappingBinaryCellOp(CompressedMatrixBlock m1, MatrixBlock m2, CompressedMatrixBlock ret,
@@ -184,7 +186,6 @@ public class CLALibBinaryCellOp {
default:
LOG.warn("Inefficient Decompression for " + op + " " + atype);
m1.decompress().binaryOperations(op, m2, ret);
-
}
}
diff --git a/src/main/java/org/apache/sysds/runtime/compress/lib/CLALibCompAgg.java b/src/main/java/org/apache/sysds/runtime/compress/lib/CLALibCompAgg.java
index 89a7c79..4a39eac 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/lib/CLALibCompAgg.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/lib/CLALibCompAgg.java
@@ -563,7 +563,7 @@ public class CLALibCompAgg {
private MatrixBlock decompressToTemp() {
MatrixBlock tmp = getTmp();
for(AColGroup g : _m1.getColGroups())
- g.decompressToBlock(tmp, _rl, _ru, 0);
+ g.decompressToBlock(tmp, _rl, _ru, -_rl, 0);
tmp.setNonZeros(_rl + _ru);
return tmp;
}
diff --git a/src/main/java/org/apache/sysds/runtime/compress/lib/CLALibDecompress.java b/src/main/java/org/apache/sysds/runtime/compress/lib/CLALibDecompress.java
index 31f381a..558ca7b 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/lib/CLALibDecompress.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/lib/CLALibDecompress.java
@@ -29,14 +29,18 @@ import java.util.concurrent.Future;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
+import org.apache.sysds.api.DMLScript;
import org.apache.sysds.runtime.compress.CompressedMatrixBlock;
import org.apache.sysds.runtime.compress.CompressionSettings;
import org.apache.sysds.runtime.compress.DMLCompressionException;
import org.apache.sysds.runtime.compress.colgroup.AColGroup;
+import org.apache.sysds.runtime.compress.colgroup.ColGroupFactory;
import org.apache.sysds.runtime.compress.colgroup.ColGroupUncompressed;
+import org.apache.sysds.runtime.controlprogram.parfor.stat.Timing;
import org.apache.sysds.runtime.data.DenseBlock;
import org.apache.sysds.runtime.matrix.data.MatrixBlock;
import org.apache.sysds.runtime.util.CommonThreadPool;
+import org.apache.sysds.utils.DMLCompressionStatistics;
/**
* Library to decompress a list of column groups into a matrix.
@@ -45,6 +49,65 @@ public class CLALibDecompress {
private static final Log LOG = LogFactory.getLog(CLALibDecompress.class.getName());
public static MatrixBlock decompress(CompressedMatrixBlock cmb, int k) {
+ Timing time = new Timing(true);
+ MatrixBlock ret = decompressExecute(cmb, k);
+ if(DMLScript.STATISTICS) {
+ final double t = time.stop();
+ DMLCompressionStatistics.addDecompressTime(t, k);
+ if(LOG.isTraceEnabled())
+ LOG.trace("decompressed block w/ k=" + k + " in " + t + "ms.");
+ }
+ return ret;
+ }
+
+ public static void decompressTo(CompressedMatrixBlock cmb, MatrixBlock ret, int rowOffset, int colOffset, int k) {
+ Timing time = new Timing(true);
+
+ final boolean outSparse = ret.isInSparseFormat();
+ if(outSparse && cmb.isOverlapping())
+ throw new DMLCompressionException("Not supported decompression into sparse block from overlapping state");
+ else if(outSparse)
+ decompressToSparseBlock(cmb, ret, rowOffset, colOffset);
+ else
+ decompressToDenseBlock(cmb, ret, rowOffset, colOffset);
+
+ if(DMLScript.STATISTICS) {
+ final double t = time.stop();
+ DMLCompressionStatistics.addDecompressToBlockTime(t, k);
+ if(LOG.isTraceEnabled())
+ LOG.trace("decompressed block w/ k=" + k + " in " + t + "ms.");
+ }
+ }
+
+ private static void decompressToSparseBlock(CompressedMatrixBlock cmb, MatrixBlock ret, int rowOffset,
+ int colOffset) {
+ final List<AColGroup> groups = new ArrayList<>(cmb.getColGroups());
+ final int nRows = cmb.getNumRows();
+
+ for(AColGroup g : groups)
+ g.decompressToBlock(ret, 0, nRows, rowOffset, colOffset);
+ }
+
+ private static void decompressToDenseBlock(CompressedMatrixBlock cmb, MatrixBlock ret, int rowOffset,
+ int colOffset) {
+ final List<AColGroup> groups = new ArrayList<>(cmb.getColGroups());
+ // final int nCols = cmb.getNumColumns();
+ final int nRows = cmb.getNumRows();
+
+ final boolean containsSDC = CLALibUtils.containsSDCOrConst(groups);
+ double[] constV = containsSDC ? new double[cmb.getNumColumns()] : null;
+ final List<AColGroup> filteredGroups = containsSDC ? CLALibUtils.filterGroups(groups, constV) : groups;
+
+ for(AColGroup g : filteredGroups)
+ g.decompressToBlock(ret, 0, nRows, rowOffset, colOffset);
+
+ if(constV != null) {
+ AColGroup cRet = ColGroupFactory.genColGroupConst(constV);
+ cRet.decompressToBlock(ret, 0, nRows, rowOffset, colOffset);
+ }
+ }
+
+ private static MatrixBlock decompressExecute(CompressedMatrixBlock cmb, int k) {
// Copy column groups to make sure we can modify the list if we want to.
final List<AColGroup> groups = new ArrayList<>(cmb.getColGroups());
@@ -55,11 +118,11 @@ public class CLALibDecompress {
MatrixBlock ret = getUncompressedColGroupAndRemoveFromListOfColGroups(groups, overlapping, nRows, nCols);
- if(ret != null && groups.size() == 0){
+ if(ret != null && groups.size() == 0) {
ret.setNonZeros(ret.recomputeNonZeros());
return ret; // if uncompressedColGroup is only colGroup.
}
- else if(ret == null){
+ else if(ret == null) {
ret = new MatrixBlock(nRows, nCols, false, -1);
ret.allocateDenseBlock();
}
@@ -97,7 +160,7 @@ public class CLALibDecompress {
// If we have a uncompressed column group that covers all of the matrix,
// it makes sense to use as the decompression target.
MatrixBlock ret = null;
-
+
// It is only relevant if we are in overlapping state, or we only have a Uncompressed ColumnGroup left.
if(overlapping || colGroups.size() == 1) {
for(int i = 0; i < colGroups.size(); i++) {
@@ -116,7 +179,6 @@ public class CLALibDecompress {
}
}
}
-
return ret;
}
@@ -171,7 +233,7 @@ public class CLALibDecompress {
* @return A Effect double value.
*/
private static double effect(AColGroup x) {
- return (x instanceof ColGroupUncompressed ) ? - Double.MAX_VALUE : -Math.max(x.getMax(), Math.abs(x.getMin()));
+ return (x instanceof ColGroupUncompressed) ? -Double.MAX_VALUE : -Math.max(x.getMax(), Math.abs(x.getMin()));
}
/**
diff --git a/src/main/java/org/apache/sysds/runtime/compress/lib/CLALibUnary.java b/src/main/java/org/apache/sysds/runtime/compress/lib/CLALibUnary.java
index 11182f5..a499198 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/lib/CLALibUnary.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/lib/CLALibUnary.java
@@ -52,7 +52,8 @@ public class CLALibUnary {
op = new UnaryOperator(op.fn, op.getNumThreads(), false);
}
- MatrixBlock ret = m.getUncompressed("unaryOperations " + op.fn.toString()).unaryOperations(op, result);
+ MatrixBlock uc = m.getUncompressed("unaryOperations " + op.fn.toString());
+ MatrixBlock ret = uc.unaryOperations(op, result);
return ret;
}
}
diff --git a/src/main/java/org/apache/sysds/runtime/compress/workload/WorkloadAnalyzer.java b/src/main/java/org/apache/sysds/runtime/compress/workload/WorkloadAnalyzer.java
index 6d767be..61b9aaa 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/workload/WorkloadAnalyzer.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/workload/WorkloadAnalyzer.java
@@ -138,9 +138,9 @@ public class WorkloadAnalyzer {
private static List<Hop> getCandidates(DMLProgram prog) {
List<Hop> candidates = new ArrayList<>();
- for(StatementBlock sb : prog.getStatementBlocks()) {
+ for(StatementBlock sb : prog.getStatementBlocks())
getCandidates(sb, prog, candidates, new HashSet<>());
- }
+
return candidates;
}
diff --git a/src/main/java/org/apache/sysds/runtime/controlprogram/context/SparkExecutionContext.java b/src/main/java/org/apache/sysds/runtime/controlprogram/context/SparkExecutionContext.java
index 4dfb320..af96a21 100644
--- a/src/main/java/org/apache/sysds/runtime/controlprogram/context/SparkExecutionContext.java
+++ b/src/main/java/org/apache/sysds/runtime/controlprogram/context/SparkExecutionContext.java
@@ -22,11 +22,16 @@ package org.apache.sysds.runtime.controlprogram.context;
import java.io.IOException;
import java.io.OutputStream;
import java.io.PrintStream;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
+import java.util.concurrent.atomic.LongAdder;
import java.util.stream.Collectors;
import java.util.stream.LongStream;
@@ -55,7 +60,6 @@ import org.apache.sysds.conf.DMLConfig;
import org.apache.sysds.hops.OptimizerUtils;
import org.apache.sysds.lops.Checkpoint;
import org.apache.sysds.runtime.DMLRuntimeException;
-import org.apache.sysds.runtime.compress.CompressedMatrixBlock;
import org.apache.sysds.runtime.controlprogram.Program;
import org.apache.sysds.runtime.controlprogram.caching.CacheBlock;
import org.apache.sysds.runtime.controlprogram.caching.CacheableData;
@@ -89,6 +93,7 @@ import org.apache.sysds.runtime.matrix.data.MatrixIndexes;
import org.apache.sysds.runtime.meta.DataCharacteristics;
import org.apache.sysds.runtime.meta.MatrixCharacteristics;
import org.apache.sysds.runtime.meta.TensorCharacteristics;
+import org.apache.sysds.runtime.util.CommonThreadPool;
import org.apache.sysds.runtime.util.HDFSTool;
import org.apache.sysds.runtime.util.UtilFunctions;
import org.apache.sysds.utils.MLContextProxy;
@@ -1035,107 +1040,156 @@ public class SparkExecutionContext extends ExecutionContext
}
/**
- * Utility method for creating a single matrix block out of a binary block RDD.
- * Note that this collect call might trigger execution of any pending transformations.
+ * Utility method for creating a single matrix block out of a binary block RDD. Note that this collect call might
+ * trigger execution of any pending transformations.
*
- * NOTE: This is an unguarded utility function, which requires memory for both the output matrix
- * and its collected, blocked representation.
+ * NOTE: This is an unguarded utility function, which requires memory for both the output matrix and its collected,
+ * blocked representation.
*
- * @param rdd JavaPairRDD for matrix block
+ * @param rdd JavaPairRDD for matrix block
* @param rlen number of rows
* @param clen number of columns
* @param blen block length
- * @param nnz number of non-zeros
- * @return matrix block
+ * @param nnz number of non-zeros
+ * @return Local matrix block
*/
public static MatrixBlock toMatrixBlock(JavaPairRDD<MatrixIndexes,MatrixBlock> rdd, int rlen, int clen, int blen, long nnz) {
long t0 = DMLScript.STATISTICS ? System.nanoTime() : 0;
+ final Boolean singleBlock = rlen <= blen && clen <= blen ;
+ final MatrixBlock out = singleBlock ? toMatrixBlockSingleBlock(rdd, rlen, clen) : toMatrixBlockMultiBlock(rdd, rlen, clen, blen, nnz);
- MatrixBlock out = null;
+ if (DMLScript.STATISTICS)
+ Statistics.accSparkCollectTime(System.nanoTime() - t0);
+
+ return out;
+ }
- if( rlen <= blen && clen <= blen ) //SINGLE BLOCK
- {
- //special case without copy and nnz maintenance
- List<Tuple2<MatrixIndexes,MatrixBlock>> list = rdd.collect();
-
- if( list.size()>1 )
- throw new DMLRuntimeException("Expecting no more than one result block but got: " + list.size());
- else if( list.size()==1 )
- out = list.get(0)._2();
- else //empty (e.g., after ops w/ outputEmpty=false)
- out = new MatrixBlock(rlen, clen, true);
- out.examSparsity();
- }
- else //MULTIPLE BLOCKS
- {
- //determine target sparse/dense representation
- long lnnz = (nnz >= 0) ? nnz : (long)rlen * clen;
- boolean sparse = MatrixBlock.evalSparseFormatInMemory(rlen, clen, lnnz);
+ private static MatrixBlock toMatrixBlockSingleBlock(JavaPairRDD<MatrixIndexes, MatrixBlock> rdd, int rlen,
+ int clen) {
+ // special case without copy and nnz maintenance
+ List<Tuple2<MatrixIndexes, MatrixBlock>> list = rdd.collect();
+ MatrixBlock out;
- //create output matrix block (w/ lazy allocation)
- out = new MatrixBlock(rlen, clen, sparse, lnnz);
-
- //kickoff asynchronous allocation
- Future<MatrixBlock> fout = out.allocateBlockAsync();
-
- //trigger pending RDD operations and collect blocks
- List<Tuple2<MatrixIndexes,MatrixBlock>> list = rdd.collect();
- out = IOUtilFunctions.get(fout); //wait for allocation
-
- //copy blocks one-at-a-time into output matrix block
- long aNnz = 0;
- boolean firstCompressed = true;
- for( Tuple2<MatrixIndexes,MatrixBlock> keyval : list )
- {
- //unpack index-block pair
- MatrixIndexes ix = keyval._1();
- MatrixBlock block = keyval._2();
-
- //compute row/column block offsets
- int row_offset = (int)(ix.getRowIndex()-1)*blen;
- int col_offset = (int)(ix.getColumnIndex()-1)*blen;
- int rows = block.getNumRows();
- int cols = block.getNumColumns();
-
- //handle compressed blocks (decompress for robustness)
- if( block instanceof CompressedMatrixBlock ){
- if(firstCompressed){
- // with warning.
- block =((CompressedMatrixBlock)block).getUncompressed("Spark RDD block to MatrixBlock Decompressing");
- firstCompressed = false;
- }else
- block = ((CompressedMatrixBlock)block).decompress(InfrastructureAnalyzer.getLocalParallelism());
- }
+ if(list.size() > 1)
+ throw new DMLRuntimeException("Expecting no more than one result block but got: " + list.size());
+ else if(list.size() == 1)
+ out = list.get(0)._2();
+ else // empty (e.g., after ops w/ outputEmpty=false)
+ out = new MatrixBlock(rlen, clen, true);
+ out.examSparsity();
+ return out;
+ }
- //append block
- if( sparse ) { //SPARSE OUTPUT
- //append block to sparse target in order to avoid shifting, where
- //we use a shallow row copy in case of MCSR and single column blocks
- //note: this append requires, for multiple column blocks, a final sort
- out.appendToSparse(block, row_offset, col_offset, clen>blen);
- }
- else { //DENSE OUTPUT
- out.copy( row_offset, row_offset+rows-1,
- col_offset, col_offset+cols-1, block, false );
- }
+ private static MatrixBlock toMatrixBlockMultiBlock(JavaPairRDD<MatrixIndexes, MatrixBlock> rdd, int rlen, int clen,
+ int blen, long nnz) {
+ // determine target sparse/dense representation
+ final long lnnz = (nnz >= 0) ? nnz : (long) rlen * clen;
+ final boolean sparse = MatrixBlock.evalSparseFormatInMemory(rlen, clen, lnnz);
- //incremental maintenance nnz
- aNnz += block.getNonZeros();
- }
+ // create output matrix block (w/ lazy allocation)
+ MatrixBlock out = new MatrixBlock(rlen, clen, sparse, lnnz);
+
+ // kickoff asynchronous allocation
+ Future<MatrixBlock> fout = out.allocateBlockAsync();
+
+ // trigger pending RDD operations and collect blocks
+ List<Tuple2<MatrixIndexes, MatrixBlock>> list = rdd.collect();
+ out = IOUtilFunctions.get(fout); // wait for allocation
+
+ LongAdder aNnz = new LongAdder();
+ // copy blocks one-at-a-time into output matrix block
+ blockPartitionsToMatrixBlock(list, out, aNnz, blen);
+
+ // post-processing output matrix
+ if(sparse)
+ out.sortSparseRows();
+ out.setNonZeros(aNnz.longValue());
+ out.examSparsity();
+ return out;
+ }
+
+ private static void blockPartitionsToMatrixBlock(List<Tuple2<MatrixIndexes, MatrixBlock>> tuples, MatrixBlock out,
+ LongAdder aNnz, int blen) {
+ if(out.isInSparseFormat())
+ blockPartitionsToMatrixBlockSingleThread(tuples, out, aNnz, blen);
+ else
+ blockPartitionsToMatrixBlockMultiThreaded(tuples, out, aNnz, blen);
+ }
+
+ private static void blockPartitionsToMatrixBlockSingleThread(List<Tuple2<MatrixIndexes, MatrixBlock>> tuples,
+ MatrixBlock out, LongAdder aNnz, int blen) {
+ final boolean sparse = out.isInSparseFormat();
+ final boolean sparseCopyShallow = sparse && out.getNumColumns() <= blen;
+ for(Tuple2<MatrixIndexes, MatrixBlock> tuple : tuples)
+ blockPartitionToMatrixBlock(tuple, out, aNnz, blen, sparseCopyShallow);
+ }
- //post-processing output matrix
- if( sparse && clen>blen )
- out.sortSparseRows();
- out.setNonZeros(aNnz);
- out.examSparsity();
+ private static void blockPartitionsToMatrixBlockMultiThreaded(List<Tuple2<MatrixIndexes, MatrixBlock>> tuples,
+ MatrixBlock out, LongAdder aNnz, int blen) {
+ try {
+ final int k = InfrastructureAnalyzer.getLocalParallelism();
+ final ExecutorService pool = CommonThreadPool.get(k);
+ final ArrayList<BlockPartitionToMatrixBlockTask> tasks = new ArrayList<>();
+ final int tSize = tuples.size();
+ final int blockSize = Math.max(tSize / k / 2, 1);
+ for(int i = 0; i < tSize; i += blockSize)
+ tasks.add(new BlockPartitionToMatrixBlockTask(tuples, i, Math.min(i + blockSize, tSize), out, aNnz, blen, false));
+ // for(Tuple2<MatrixIndexes, MatrixBlock> tuple : tuples)
+
+ for(Future<Object> f : pool.invokeAll(tasks))
+ f.get();
+
+ pool.shutdown();
+ }
+ catch(InterruptedException | ExecutionException ex) {
+ throw new DMLRuntimeException("Parallel block partitions to matrix block failed", ex);
}
+ }
- if (DMLScript.STATISTICS) {
- Statistics.accSparkCollectTime(System.nanoTime() - t0);
- Statistics.incSparkCollectCount(1);
+ private static void blockPartitionToMatrixBlock(Tuple2<MatrixIndexes, MatrixBlock> tuple, MatrixBlock out,
+ LongAdder aNnz, int blen, boolean sparseCopyShallow) {
+ // unpack index-block pair
+ final MatrixIndexes ix = tuple._1();
+ final MatrixBlock block = tuple._2();
+
+ // compute row/column block offsets
+ final int row_offset = (int) (ix.getRowIndex() - 1) * blen;
+ final int col_offset = (int) (ix.getColumnIndex() - 1) * blen;
+
+ // Add the values of this block into the output block.
+ block.putInto(out, row_offset, col_offset, sparseCopyShallow);
+
+ // incremental maintenance nnz
+ aNnz.add(block.getNonZeros());
+ }
+
+ private static class BlockPartitionToMatrixBlockTask implements Callable<Object> {
+ private final List<Tuple2<MatrixIndexes, MatrixBlock>> _tuples;
+ private final int _start;
+ private final int _end;
+ private final MatrixBlock _out;
+ private final LongAdder _aNnz;
+ private final int _blen;
+ private final boolean _sparseCopyShallow;
+
+ protected BlockPartitionToMatrixBlockTask(List<Tuple2<MatrixIndexes, MatrixBlock>> tuples, int start, int end, MatrixBlock out,
+ LongAdder aNnz, int blen, boolean sparseCopyShallow){
+ _tuples = tuples;
+ _start = start;
+ _end = end;
+ _out = out;
+ _aNnz = aNnz;
+ _blen = blen;
+ _sparseCopyShallow = sparseCopyShallow;
}
- return out;
+ @Override
+ public Object call(){
+ for(int i = _start; i < _end; i ++)
+ blockPartitionToMatrixBlock(_tuples.get(i), _out, _aNnz, _blen, _sparseCopyShallow);
+
+ return null;
+ }
}
@SuppressWarnings("unchecked")
@@ -1188,10 +1242,9 @@ public class SparkExecutionContext extends ExecutionContext
out.recomputeNonZeros();
out.examSparsity();
- if (DMLScript.STATISTICS) {
+ if (DMLScript.STATISTICS)
Statistics.accSparkCollectTime(System.nanoTime() - t0);
- Statistics.incSparkCollectCount(1);
- }
+
return out;
}
@@ -1235,10 +1288,9 @@ public class SparkExecutionContext extends ExecutionContext
// TODO post-processing output tensor (nnz, sparsity)
- if (DMLScript.STATISTICS) {
+ if (DMLScript.STATISTICS)
Statistics.accSparkCollectTime(System.nanoTime() - t0);
- Statistics.incSparkCollectCount(1);
- }
+
return out;
}
@@ -1257,10 +1309,8 @@ public class SparkExecutionContext extends ExecutionContext
out.setBlock((int)ix.getRowIndex(), (int)ix.getColumnIndex(), block);
}
- if (DMLScript.STATISTICS) {
+ if (DMLScript.STATISTICS)
Statistics.accSparkCollectTime(System.nanoTime() - t0);
- Statistics.incSparkCollectCount(1);
- }
return out;
}
@@ -1298,10 +1348,8 @@ public class SparkExecutionContext extends ExecutionContext
}
}
- if (DMLScript.STATISTICS) {
+ if (DMLScript.STATISTICS)
Statistics.accSparkCollectTime(System.nanoTime() - t0);
- Statistics.incSparkCollectCount(1);
- }
return out;
}
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/CPInstructionParser.java b/src/main/java/org/apache/sysds/runtime/instructions/CPInstructionParser.java
index 236f3e0..22cceea 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/CPInstructionParser.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/CPInstructionParser.java
@@ -21,11 +21,14 @@ package org.apache.sysds.runtime.instructions;
import java.util.HashMap;
+import org.apache.sysds.common.Types.ExecType;
import org.apache.sysds.hops.FunctionOp;
import org.apache.sysds.lops.Append;
+import org.apache.sysds.lops.Compression;
import org.apache.sysds.lops.DataGen;
+import org.apache.sysds.lops.DeCompression;
import org.apache.sysds.lops.LeftIndex;
-import org.apache.sysds.common.Types.ExecType;
+import org.apache.sysds.lops.Local;
import org.apache.sysds.lops.RightIndex;
import org.apache.sysds.lops.UnaryCP;
import org.apache.sysds.runtime.DMLRuntimeException;
@@ -47,9 +50,9 @@ import org.apache.sysds.runtime.instructions.cp.DeCompressionCPInstruction;
import org.apache.sysds.runtime.instructions.cp.DnnCPInstruction;
import org.apache.sysds.runtime.instructions.cp.FunctionCallCPInstruction;
import org.apache.sysds.runtime.instructions.cp.IndexingCPInstruction;
+import org.apache.sysds.runtime.instructions.cp.LocalCPInstruction;
import org.apache.sysds.runtime.instructions.cp.MMChainCPInstruction;
import org.apache.sysds.runtime.instructions.cp.MMTSJCPInstruction;
-import org.apache.sysds.runtime.instructions.cp.ReshapeCPInstruction;
import org.apache.sysds.runtime.instructions.cp.MultiReturnBuiltinCPInstruction;
import org.apache.sysds.runtime.instructions.cp.MultiReturnParameterizedBuiltinCPInstruction;
import org.apache.sysds.runtime.instructions.cp.PMMJCPInstruction;
@@ -59,6 +62,7 @@ import org.apache.sysds.runtime.instructions.cp.QuantilePickCPInstruction;
import org.apache.sysds.runtime.instructions.cp.QuantileSortCPInstruction;
import org.apache.sysds.runtime.instructions.cp.QuaternaryCPInstruction;
import org.apache.sysds.runtime.instructions.cp.ReorgCPInstruction;
+import org.apache.sysds.runtime.instructions.cp.ReshapeCPInstruction;
import org.apache.sysds.runtime.instructions.cp.SpoofCPInstruction;
import org.apache.sysds.runtime.instructions.cp.SqlCPInstruction;
import org.apache.sysds.runtime.instructions.cp.StringInitCPInstruction;
@@ -315,11 +319,12 @@ public class CPInstructionParser extends InstructionParser
String2CPInstructionType.put( "svd", CPType.MultiReturnBuiltin);
String2CPInstructionType.put( "partition", CPType.Partition);
- String2CPInstructionType.put( "compress", CPType.Compression);
- String2CPInstructionType.put( "decompress", CPType.DeCompression);
+ String2CPInstructionType.put( Compression.OPCODE, CPType.Compression);
+ String2CPInstructionType.put( DeCompression.OPCODE, CPType.DeCompression);
String2CPInstructionType.put( "spoof", CPType.SpoofFused);
String2CPInstructionType.put( "prefetch", CPType.Prefetch);
String2CPInstructionType.put( "broadcast", CPType.Broadcast);
+ String2CPInstructionType.put( Local.OPCODE, CPType.Local);
String2CPInstructionType.put( "sql", CPType.Sql);
}
@@ -454,6 +459,9 @@ public class CPInstructionParser extends InstructionParser
case DeCompression:
return DeCompressionCPInstruction.parseInstruction(str);
+ case Local:
+ return LocalCPInstruction.parseInstruction(str);
+
case SpoofFused:
return SpoofCPInstruction.parseInstruction(str);
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/cp/CPInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/cp/CPInstruction.java
index 47a14b1..95d224e 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/cp/CPInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/cp/CPInstruction.java
@@ -44,7 +44,7 @@ public abstract class CPInstruction extends Instruction
AggregateUnary, AggregateBinary, AggregateTernary,
Unary, Binary, Ternary, Quaternary, BuiltinNary, Ctable,
MultiReturnParameterizedBuiltin, ParameterizedBuiltin, MultiReturnBuiltin,
- Builtin, Reorg, Variable, FCall, Append, Rand, QSort, QPick,
+ Builtin, Reorg, Variable, FCall, Append, Rand, QSort, QPick, Local,
MatrixIndexing, MMTSJ, PMMJ, MMChain, Reshape, Partition, Compression, DeCompression, SpoofFused,
StringInit, CentralMoment, Covariance, UaggOuterChain, Dnn, Sql, Prefetch, Broadcast }
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/cp/LocalCPInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/cp/LocalCPInstruction.java
new file mode 100644
index 0000000..6fcdcb0
--- /dev/null
+++ b/src/main/java/org/apache/sysds/runtime/instructions/cp/LocalCPInstruction.java
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.sysds.runtime.instructions.cp;
+
+import org.apache.sysds.runtime.controlprogram.context.ExecutionContext;
+import org.apache.sysds.runtime.instructions.InstructionUtils;
+import org.apache.sysds.runtime.matrix.data.MatrixBlock;
+
+/**
+ * A simple instruction that take whatever input it is and return it as a local matrix. This forces any distributed
+ * matrices to become local.
+ */
+public class LocalCPInstruction extends ComputationCPInstruction {
+ // private static final Log LOG = LogFactory.getLog(LocalCPInstruction.class.getName());
+
+ private LocalCPInstruction(CPOperand in, CPOperand out, String instruction) {
+ super(CPType.Local, null, in, null, null, out, "LOCAL", instruction);
+ }
+
+ public static LocalCPInstruction parseInstruction(String str) {
+ InstructionUtils.checkNumFields(str, 2);
+ final String[] parts = InstructionUtils.getInstructionPartsWithValueType(str);
+ final CPOperand in = new CPOperand(parts[1]);
+ final CPOperand out = new CPOperand(parts[2]);
+ return new LocalCPInstruction(in, out, str);
+ }
+
+ @Override
+ public void processInstruction(ExecutionContext ec) {
+ final MatrixBlock in = ec.getMatrixInput(input1.getName());
+ ec.releaseMatrixInput(input1.getName());
+ ec.setMatrixOutput(output.getName(), in);
+ }
+}
diff --git a/src/main/java/org/apache/sysds/runtime/matrix/data/MatrixBlock.java b/src/main/java/org/apache/sysds/runtime/matrix/data/MatrixBlock.java
index d77e7af..922f7ab 100644
--- a/src/main/java/org/apache/sysds/runtime/matrix/data/MatrixBlock.java
+++ b/src/main/java/org/apache/sysds/runtime/matrix/data/MatrixBlock.java
@@ -50,8 +50,8 @@ import org.apache.sysds.runtime.compress.CompressedMatrixBlock;
import org.apache.sysds.runtime.compress.DMLCompressionException;
import org.apache.sysds.runtime.compress.lib.CLALibBinaryCellOp;
import org.apache.sysds.runtime.controlprogram.caching.CacheBlock;
-import org.apache.sysds.runtime.controlprogram.caching.LazyWriteBuffer;
import org.apache.sysds.runtime.controlprogram.caching.MatrixObject.UpdateType;
+import org.apache.sysds.runtime.controlprogram.parfor.stat.InfrastructureAnalyzer;
import org.apache.sysds.runtime.data.DenseBlock;
import org.apache.sysds.runtime.data.DenseBlockFactory;
import org.apache.sysds.runtime.data.SparseBlock;
@@ -106,6 +106,7 @@ import org.apache.sysds.runtime.matrix.operators.TernaryOperator;
import org.apache.sysds.runtime.matrix.operators.UnaryOperator;
import org.apache.sysds.runtime.meta.DataCharacteristics;
import org.apache.sysds.runtime.meta.MatrixCharacteristics;
+import org.apache.sysds.runtime.util.CommonThreadPool;
import org.apache.sysds.runtime.util.DataConverter;
import org.apache.sysds.runtime.util.FastBufferedDataInputStream;
import org.apache.sysds.runtime.util.FastBufferedDataOutputStream;
@@ -361,7 +362,7 @@ public class MatrixBlock extends MatrixValue implements CacheBlock, Externalizab
}
public Future<MatrixBlock> allocateBlockAsync() {
- ExecutorService pool = LazyWriteBuffer.getUtilThreadPool();
+ ExecutorService pool = CommonThreadPool.get(InfrastructureAnalyzer.getLocalParallelism());
return (pool != null) ? pool.submit(() -> allocateBlock()) : //async
ConcurrentUtils.constantFuture(allocateBlock()); //fallback sync
}
@@ -788,7 +789,7 @@ public class MatrixBlock extends MatrixValue implements CacheBlock, Externalizab
appendToSparse(that, rowoffset, coloffset, true);
}
- public void appendToSparse( MatrixBlock that, int rowoffset, int coloffset, boolean deep )
+ private void appendToSparse( MatrixBlock that, int rowoffset, int coloffset, boolean deep )
{
if( that==null || that.isEmptyBlock(false) )
return; //nothing to append
@@ -1522,7 +1523,35 @@ public class MatrixBlock extends MatrixValue implements CacheBlock, Externalizab
}
}
-
+ /**
+ * Method for copying this matrix into a target matrix.
+ *
+ * Note that this method does not maintain number of non zero values.
+ *
+ * The method should output into the allocated block type of the target, therefore before any calls an appropriate
+ * block must be allocated.
+ *
+ * CSR sparse format is not supported.
+ *
+ * If allocating into a sparse matrix MCSR block the rows have to be sorted afterwards with a call to
+ *
+ * target.sortSparseRows()
+ *
+ * @param target Target MatrixBlock, that can be allocated dense or sparse
+ * @param rowOffset The Row offset to allocate into.
+ * @param colOffset The column offset to allocate into.
+ * @param sparseCopyShallow If the output is sparse, and shallow copy of rows is allowed from this block
+ */
+ public void putInto(MatrixBlock target, int rowOffset, int colOffset, boolean sparseCopyShallow) {
+ if(target.isInSparseFormat())
+ // append block to sparse target in order to avoid shifting, where
+ // we use a shallow row copy in case of MCSR and single column blocks
+ // note: this append requires, for multiple column blocks, a final sort
+ target.appendToSparse(this, rowOffset, colOffset, !sparseCopyShallow);
+ else
+ target.copy(rowOffset, rowOffset + rlen - 1, colOffset, colOffset + clen - 1, this, false);
+ }
+
/**
* In-place copy of matrix src into the index range of the existing current matrix.
* Note that removal of existing nnz in the index range and nnz maintenance is
diff --git a/src/main/java/org/apache/sysds/utils/DMLCompressionStatistics.java b/src/main/java/org/apache/sysds/utils/DMLCompressionStatistics.java
index 92130e8..7436fa2 100644
--- a/src/main/java/org/apache/sysds/utils/DMLCompressionStatistics.java
+++ b/src/main/java/org/apache/sysds/utils/DMLCompressionStatistics.java
@@ -33,6 +33,11 @@ public class DMLCompressionStatistics {
private static int DecompressMTCount = 0;
private static double DecompressMT = 0.0;
+ private static int DecompressToSTCount = 0;
+ private static double DecompressToST = 0.0;
+ private static int DecompressToMTCount = 0;
+ private static double DecompressToMT = 0.0;
+
private static int DecompressSparkCount = 0;
private static int DecompressCacheCount = 0;
@@ -47,11 +52,15 @@ public class DMLCompressionStatistics {
DecompressST = 0.0;
DecompressMTCount = 0;
DecompressMT = 0.0;
+ DecompressToSTCount = 0;
+ DecompressToST = 0.0;
+ DecompressToMTCount = 0;
+ DecompressToMT = 0.0;
DecompressSparkCount = 0;
DecompressCacheCount = 0;
}
- public static boolean haveCompressed(){
+ public static boolean haveCompressed() {
return Phase0 > 0;
}
@@ -75,7 +84,6 @@ public class DMLCompressionStatistics {
case 5:
Phase5 += time;
break;
-
}
}
@@ -90,35 +98,42 @@ public class DMLCompressionStatistics {
}
}
- public static void addDecompressSparkCount(){
+ public static void addDecompressToBlockTime(double time, int threads) {
+ if(threads == 1) {
+ DecompressToSTCount++;
+ DecompressToST += time;
+ }
+ else {
+ DecompressToMTCount++;
+ DecompressToMT += time;
+ }
+ }
+
+ public static void addDecompressSparkCount() {
DecompressSTCount++;
}
- public static void addDecompressCacheCount(){
+ public static void addDecompressCacheCount() {
DecompressCacheCount++;
}
public static int getDecompressionCount() {
- return DecompressMTCount + DecompressSTCount + DecompressSparkCount + DecompressCacheCount;
+ return DecompressMTCount + DecompressSTCount + DecompressSparkCount + DecompressCacheCount + DecompressToSTCount +
+ DecompressToMTCount;
}
public static void display(StringBuilder sb) {
if(haveCompressed()) { // If compression have been used
- sb.append(String.format("CLA Compression Phases :\t%.3f/%.3f/%.3f/%.3f/%.3f/%.3f\n",
- Phase0 / 1000,
- Phase1 / 1000,
- Phase2 / 1000,
- Phase3 / 1000,
- Phase4 / 1000,
- Phase5 / 1000));
- sb.append(String.format("Decompression Counts (Single , Multi, Spark, Cache) thread :\t%d/%d/%d/%d\n",
- DecompressSTCount,
- DecompressMTCount,
- DecompressSparkCount,
- DecompressCacheCount));
- sb.append(String.format("Dedicated Decompression Time (Single , Multi) thread :\t%.3f/%.3f\n",
- DecompressST / 1000,
- DecompressMT / 1000));
+ sb.append(String.format("CLA Compression Phases :\t%.3f/%.3f/%.3f/%.3f/%.3f/%.3f\n", Phase0 / 1000,
+ Phase1 / 1000, Phase2 / 1000, Phase3 / 1000, Phase4 / 1000, Phase5 / 1000));
+ sb.append(String.format("Decompression with allocation (Single, Multi, Spark, Cache) : %d/%d/%d/%d\n",
+ DecompressSTCount, DecompressMTCount, DecompressSparkCount, DecompressCacheCount));
+ sb.append(String.format("Decompression with allocation Time (Single , Multi) : %.3f/%.3f sec.\n",
+ DecompressST / 1000, DecompressMT / 1000));
+ sb.append(String.format("Decompression to block (Single, Multi) : %d/%d\n",
+ DecompressToSTCount, DecompressToMTCount));
+ sb.append(String.format("Decompression to block Time (Single, Multi) : %.3f/%.3f sec.\n",
+ DecompressToST / 1000, DecompressToMT / 1000));
}
}
}
diff --git a/src/main/java/org/apache/sysds/utils/Statistics.java b/src/main/java/org/apache/sysds/utils/Statistics.java
index d6c63ff..3940921 100644
--- a/src/main/java/org/apache/sysds/utils/Statistics.java
+++ b/src/main/java/org/apache/sysds/utils/Statistics.java
@@ -515,6 +515,10 @@ public class Statistics
sparkBroadcastCount.reset();
sparkAsyncPrefetchCount.reset();
sparkAsyncBroadcastCount.reset();
+ sparkParallelize.reset();
+ sparkParallelizeCount.reset();
+ sparkCollect.reset();
+ sparkCollectCount.reset();
lTotalLix.reset();
lTotalLixUIP.reset();
@@ -580,10 +584,11 @@ public class Statistics
public static void accSparkCollectTime(long t) {
sparkCollect.add(t);
+ sparkCollectCount.add(1);
}
- public static void incSparkCollectCount(long c) {
- sparkCollectCount.add(c);
+ public static long getSparkCollectCount(){
+ return sparkCollectCount.longValue();
}
public static void accSparkBroadCastTime(long t) {
diff --git a/src/test/java/org/apache/sysds/test/TestUtils.java b/src/test/java/org/apache/sysds/test/TestUtils.java
index 1a647ad..125de36 100644
--- a/src/test/java/org/apache/sysds/test/TestUtils.java
+++ b/src/test/java/org/apache/sysds/test/TestUtils.java
@@ -64,6 +64,7 @@ import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.SequenceFile.Writer;
import org.apache.sysds.common.Types.FileFormat;
import org.apache.sysds.common.Types.ValueType;
+import org.apache.sysds.runtime.data.SparseBlock;
import org.apache.sysds.runtime.data.TensorBlock;
import org.apache.sysds.runtime.functionobjects.Builtin;
import org.apache.sysds.runtime.functionobjects.Builtin.BuiltinCode;
@@ -854,27 +855,36 @@ public class TestUtils
public static void compareMatricesBitAvgDistance(MatrixBlock expectedMatrix, MatrixBlock actualMatrix,
long maxUnitsOfLeastPrecision, long maxAvgDistance, String message) {
+ if(expectedMatrix.isEmpty() && actualMatrix.isEmpty())
+ return;
+ if(expectedMatrix.isEmpty())
+ fail(message + "\nThe expected output is empty while the actual matrix is not");
+ if(actualMatrix.isEmpty())
+ fail(message + "\nThe actual output is empty while the expected matrix is not");
+ if(expectedMatrix.isInSparseFormat() && actualMatrix.isInSparseFormat()){
+ compareMatricesBitAvgDistanceSparse(expectedMatrix.getSparseBlock(), actualMatrix.getSparseBlock(), maxUnitsOfLeastPrecision, maxAvgDistance, message, actualMatrix.getNumColumns());
+ return;
+ }
final int rows = expectedMatrix.getNumRows();
final int cols = actualMatrix.getNumColumns();
int countErrors = 0;
long sumDistance = 0;
- long distance;
for(int i = 0; i < rows && countErrors < 20; i++) {
for(int j = 0; j < cols && countErrors < 20; j++) {
double v1 = expectedMatrix.quickGetValue(i, j);
double v2 = actualMatrix.quickGetValue(i, j);
if(v1 == 0 && v2 == 0)
- continue;
+ continue;
else if(v1 == 0 || v2 == 0) {
if(Math.abs(v1 - v2) > 1E-16) {
message += ("\n Expected:" + v1 + " vs actual: " + v2 + " at " + i + " " + j
- + " Not using Bit distance since one value is 0");
+ + " Not using Bit distance since one value is 0");
countErrors++;
}
}
else {
- distance = compareScalarBits(expectedMatrix.quickGetValue(i, j), actualMatrix.quickGetValue(i, j));
+ final long distance = compareScalarBits(expectedMatrix.quickGetValue(i, j), actualMatrix.quickGetValue(i, j));
sumDistance += distance;
if(distance > maxUnitsOfLeastPrecision) {
message += ("\n Expected:" + v1 + " vs actual: " + v2 + " at " + i + " " + j + " Distance in bits: "
@@ -884,14 +894,57 @@ public class TestUtils
}
}
}
- if(countErrors == 20) {
- assertTrue(message + "\n At least 20 values are not in equal", countErrors == 0);
- }
+
+ if(countErrors == 20)
+ fail(message + "\n At least 20 values are not in equal");
else {
long avgDistance = sumDistance / (rows * cols);
- assertTrue(message + "\n" + countErrors + " values are not in equal", countErrors == 0);
- assertTrue(message + "\nThe avg distance in bits: " + avgDistance + " was higher than max: " + maxAvgDistance,
- avgDistance <= maxAvgDistance);
+ if(countErrors != 0)
+ fail(message + "\n" + countErrors + " values are not in equal");
+ if(avgDistance > maxAvgDistance)
+ fail(message + "\nThe avg distance in bits: " + avgDistance + " was higher than max: " + maxAvgDistance);
+ }
+ }
+
+ private static void compareMatricesBitAvgDistanceSparse(SparseBlock sbe, SparseBlock sba, long maxUnitsOfLeastPrecision, long maxAvgDistance, String message, int cols){
+ final int rows = sbe.numRows();
+ int countErrors = 0;
+ long sumDistance = 0;
+ for(int i = 0; i < rows && countErrors < 20; i++){
+ if( sbe.isEmpty(i) != sba.isEmpty(i))
+ fail(message +"\nBoth matrices are not equally empty on row : " + i);
+
+ if(sbe.isEmpty(i))
+ continue;
+
+ if(sba.size(i) != sbe.size(i))
+ fail(message+"\nNumber of values are not equal in row: " + i);
+
+ final double[] e = sbe.values(i);
+ final double[] a = sba.values(i);
+ final int epos = sbe.pos(i);
+ final int apos = sba.pos(i);
+ final int elen = sbe.size(i) + epos;
+ for(int j = apos, jj = epos; jj < elen; j++, jj++){
+ final long distance = compareScalarBits(e[jj], a[j]);
+ sumDistance += distance;
+
+ if(distance > maxUnitsOfLeastPrecision) {
+ message += ("\n Expected:" + e[jj] + " vs actual: " + a[j] + " at " + i + " " + j + " Distance in bits: " + distance);
+ countErrors++;
+ }
+ }
+ }
+
+ if(countErrors == 20){
+ fail(message + "\n At least 20 values are not in equal");
+ }
+ else{
+ long avgDistance = sumDistance / (rows * cols);
+ if(countErrors != 0)
+ fail(message + "\n" + countErrors + " values are not in equal");
+ if(avgDistance > maxAvgDistance)
+ fail(message + "\nThe avg distance in bits: " + avgDistance + " was higher than max: " + maxAvgDistance);
}
}
@@ -1008,15 +1061,21 @@ public class TestUtils
assertTrue("maxAveragePercentDistance should be between 1 and 0",
maxAveragePercentDistance >= 0.0 && maxAveragePercentDistance <= 1.0);
+ if(expectedMatrix.isEmpty() && actualMatrix.isEmpty())
+ return;
+ if(expectedMatrix.isInSparseFormat() && actualMatrix.isInSparseFormat()){
+ compareMatricesPercentageDistanceSparse(expectedMatrix.getSparseBlock(), actualMatrix.getSparseBlock(), percentDistanceAllowed, maxAveragePercentDistance, message, ignoreZero,actualMatrix.getNumColumns());
+ return;
+ }
+
final int rows = expectedMatrix.getNumRows();
final int cols = expectedMatrix.getNumColumns();
int countErrors = 0;
double sumPercentDistance = 0;
- double distance;
-
+
for(int i = 0; i < rows && countErrors < 20; i++) {
for(int j = 0; j < cols && countErrors < 20; j++) {
- distance = getPercentDistance(expectedMatrix.quickGetValue(i, j), actualMatrix.quickGetValue(i, j), ignoreZero);
+ final double distance = getPercentDistance(expectedMatrix.quickGetValue(i, j), actualMatrix.quickGetValue(i, j), ignoreZero);
sumPercentDistance += distance;
if(distance < percentDistanceAllowed) {
message += ("\nExpected: " + expectedMatrix.quickGetValue(i, j) + " vs actual: " + actualMatrix.quickGetValue(i, j) + " at " + i
@@ -1026,15 +1085,59 @@ public class TestUtils
}
}
if(countErrors == 20) {
- assertTrue(message + "\n At least 20 values are not in equal", countErrors == 0);
+ fail(message + "\n At least 20 values are not in equal");
}
else {
double avgDistance = sumPercentDistance / (rows * cols);
- assertTrue(message + "\n" + countErrors + " values are not in equal of total: " + (rows * cols),
- countErrors == 0);
- assertTrue(
- message + "\nThe avg distance: " + avgDistance + " was lower than threshold " + maxAveragePercentDistance,
- avgDistance > maxAveragePercentDistance);
+ if(countErrors != 0)
+ fail(message + "\n" + countErrors + " values are not in equal of total: " + (rows * cols));
+ if(avgDistance <= maxAveragePercentDistance)
+ fail(message + "\nThe avg distance: " + avgDistance + " was lower than threshold " + maxAveragePercentDistance);
+ }
+ }
+
+ private static void compareMatricesPercentageDistanceSparse(SparseBlock sbe, SparseBlock sba,
+ double percentDistanceAllowed, double maxAveragePercentDistance, String message, boolean ignoreZero, int cols) {
+ final int rows = sbe.numRows();
+ int countErrors = 0;
+ double sumPercentDistance = 0;
+
+ for(int i = 0; i < rows && countErrors < 20; i++){
+
+ if( sbe.isEmpty(i) != sba.isEmpty(i))
+ fail(message +"\nBoth matrices are not equally empty on row : " + i);
+
+ if(sbe.isEmpty(i))
+ continue;
+
+ if(sba.size(i) != sbe.size(i))
+ fail(message+"\nNumber of values are not equal in row: " + i);
+
+ final double[] e = sbe.values(i);
+ final double[] a = sba.values(i);
+ final int epos = sbe.pos(i);
+ final int apos = sba.pos(i);
+ final int elen = sbe.size(i) + epos;
+ for(int j = apos, jj = epos; jj < elen; j++, jj++){
+ final double distance = getPercentDistance(e[jj], a[j], ignoreZero);
+ sumPercentDistance += distance;
+ if(distance < percentDistanceAllowed) {
+ message += ("\nExpected: " + e[jj] + " vs actual: " + a[j] + " at " + i
+ + " " + j + " Distance in percent " + distance);
+ countErrors++;
+ }
+ }
+ }
+
+ if(countErrors == 20){
+ fail(message + "\n At least 20 values are not in equal");
+ }
+ else{
+ double avgDistance = sumPercentDistance / (rows * cols);
+ if(countErrors != 0)
+ fail(message + "\n" + countErrors + " values are not in equal");
+ if(avgDistance > maxAveragePercentDistance)
+ fail(message + "\nThe avg distance in bits: " + avgDistance + " was higher than max: " + maxAveragePercentDistance);
}
}
diff --git a/src/test/java/org/apache/sysds/test/component/compress/CompressedMatrixTest.java b/src/test/java/org/apache/sysds/test/component/compress/CompressedMatrixTest.java
index 0c9c164..fee58b9 100644
--- a/src/test/java/org/apache/sysds/test/component/compress/CompressedMatrixTest.java
+++ b/src/test/java/org/apache/sysds/test/component/compress/CompressedMatrixTest.java
@@ -597,12 +597,101 @@ public class CompressedMatrixTest extends AbstractCompressedUnaryTests {
MatrixBlock m2 = new MatrixBlock(1, cmb.getNumColumns(), 0);
MatrixBlock ret1 = cmb.binaryOperations(op, m2, new MatrixBlock());
ucRet = mb.binaryOperations(op, m2, ucRet);
-
compareResultMatrices(ucRet, ret1, 1);
}
catch(Exception e) {
e.printStackTrace();
- throw new DMLRuntimeException(e);
+ throw e;
+ }
+ }
+
+ @Test
+ public void testDecompressToNoOffsets() {
+ testDecompressToMatrixWithOffsets(0, 0, rows + 1, cols + 1);
+ }
+
+ @Test
+ public void testDecompressToOffset1Row() {
+ testDecompressToMatrixWithOffsets(1, 0, rows + 1, cols + 1);
+ }
+
+ @Test
+ public void testDecompressToOffset1Col() {
+ testDecompressToMatrixWithOffsets(0, 1, rows + 1, cols + 1);
+ }
+
+ @Test
+ public void testDecompressToOffsetBoth1() {
+ testDecompressToMatrixWithOffsets(1, 1, rows + 1, cols + 1);
+ }
+
+ @Test
+ public void testDecompressToMiddleOfMatrix() {
+ testDecompressToMatrixWithOffsets(10, 10, rows + 20, cols + 20);
+ }
+
+ public void testDecompressToMatrixWithOffsets(int rowOff, int colOff, int outRows, int outCols) {
+ try {
+ if(!(cmb instanceof CompressedMatrixBlock))
+ return;
+
+ MatrixBlock ret1 = new MatrixBlock(outRows, outCols, false);
+ ret1.allocateDenseBlock();
+ MatrixBlock ret2 = new MatrixBlock(outRows, outCols, false);
+ ret2.allocateDenseBlock();
+
+ cmb.putInto(ret1, rowOff, colOff, false);
+ ret1.setNonZeros(cmb.getNonZeros());
+
+ mb.putInto(ret2, rowOff, colOff, false);
+ ret2.setNonZeros(mb.getNonZeros());
+
+ compareResultMatrices(ret2, ret1, 1);
+ }
+ catch(Exception e) {
+ e.printStackTrace();
+ throw e;
+ }
+ }
+
+ @Test
+ public void testDecompressToOffsetSparseExtraColumns() {
+ if(cmb instanceof CompressedMatrixBlock && !((CompressedMatrixBlock) cmb).isOverlapping())
+ testsDecompressToSparseMatrixWithOffsets(10, 10, rows + 100, cols + 1000);
+ }
+
+ @Test
+ public void testDecompressToOffsetSparseExtraRows() {
+ if(cmb instanceof CompressedMatrixBlock && !((CompressedMatrixBlock) cmb).isOverlapping())
+ testsDecompressToSparseMatrixWithOffsets(10, 10, rows + 10000, cols + 10);
+ }
+
+ public void testsDecompressToSparseMatrixWithOffsets(int rowOff, int colOff, int outRows, int outCols) {
+ try {
+ if(!(cmb instanceof CompressedMatrixBlock))
+ return;
+
+ MatrixBlock ret1 = new MatrixBlock(outRows, outCols, true);
+ ret1.allocateSparseRowsBlock();
+ MatrixBlock ret2 = new MatrixBlock(outRows, outCols, true);
+ ret2.allocateSparseRowsBlock();
+
+ cmb.putInto(ret1, rowOff, colOff, false);
+ ret1.setNonZeros(cmb.getNonZeros());
+ if(ret1.isInSparseFormat())
+ ret1.sortSparseRows();
+
+ mb.putInto(ret2, rowOff, colOff, false);
+ ret2.setNonZeros(mb.getNonZeros());
+ if(ret2.isInSparseFormat())
+ ret2.sortSparseRows();
+
+ compareResultMatrices(ret2, ret1, 1);
+
+ }
+ catch(Exception e) {
+ e.printStackTrace();
+ throw e;
}
}
diff --git a/src/test/java/org/apache/sysds/test/component/compress/CompressedTestBase.java b/src/test/java/org/apache/sysds/test/component/compress/CompressedTestBase.java
index 4f8fe18..16ca8ad 100644
--- a/src/test/java/org/apache/sysds/test/component/compress/CompressedTestBase.java
+++ b/src/test/java/org/apache/sysds/test/component/compress/CompressedTestBase.java
@@ -527,7 +527,7 @@ public abstract class CompressedTestBase extends TestBase {
public void testRightMatrixMatrixMultSparse() {
if(!(cmb instanceof CompressedMatrixBlock))
return; // Input was not compressed then just pass test
- MatrixBlock matrix = TestUtils.generateTestMatrixBlock(cols, 25, 0.9, 1.5, 0.2, 3);
+ MatrixBlock matrix = TestUtils.generateTestMatrixBlock(cols, 25, 0.9, 1.5, 0.01, 3);
testRightMatrixMatrix(matrix);
}
@@ -979,7 +979,6 @@ public abstract class CompressedTestBase extends TestBase {
e.printStackTrace();
throw e;
}
-
}
@Test
diff --git a/src/test/java/org/apache/sysds/test/component/compress/InvalidInputTest.java b/src/test/java/org/apache/sysds/test/component/compress/InvalidInputTest.java
index 45f9853..f9f3be2 100644
--- a/src/test/java/org/apache/sysds/test/component/compress/InvalidInputTest.java
+++ b/src/test/java/org/apache/sysds/test/component/compress/InvalidInputTest.java
@@ -108,11 +108,6 @@ public class InvalidInputTest {
}
@Test(expected = DMLRuntimeException.class)
- public void invalid_call_appendToSparse() {
- cmb.appendToSparse(null, 1, 1, false);
- }
-
- @Test(expected = DMLRuntimeException.class)
public void invalid_call_appendToValue() {
cmb.appendValue(1, 1, 0.2);
}
diff --git a/src/test/java/org/apache/sysds/test/functions/compress/CompressInstruction.java b/src/test/java/org/apache/sysds/test/functions/compress/CompressInstruction.java
index 9498e77..b63c154 100644
--- a/src/test/java/org/apache/sysds/test/functions/compress/CompressInstruction.java
+++ b/src/test/java/org/apache/sysds/test/functions/compress/CompressInstruction.java
@@ -23,6 +23,7 @@ import static org.junit.Assert.assertTrue;
import org.apache.sysds.common.Types;
import org.apache.sysds.common.Types.ExecType;
+import org.apache.sysds.hops.OptimizerUtils;
import org.apache.sysds.runtime.controlprogram.parfor.stat.Timing;
import org.apache.sysds.test.AutomatedTestBase;
import org.apache.sysds.test.TestConfiguration;
@@ -85,6 +86,7 @@ public class CompressInstruction extends AutomatedTestBase {
public void compressTest(int cols, int rows, double sparsity, ExecType instType, int min, int max,
int decompressionCountExpected, int compressionCountsExpected, String name) {
+ OptimizerUtils.ALLOW_SCRIPT_LEVEL_COMPRESS_COMMAND = true;
Types.ExecMode platformOld = setExecMode(instType);
try {
@@ -92,8 +94,8 @@ public class CompressInstruction extends AutomatedTestBase {
fullDMLScriptName = SCRIPT_DIR + "/" + getTestDir() + "compress_" + name + ".dml";
- programArgs = new String[] {"-stats", "100", "-nvargs", "cols=" + cols, "rows=" + rows,
- "sparsity=" + sparsity, "min=" + min, "max= " + max};
+ programArgs = new String[] {"-stats", "100", "-nvargs", "cols=" + cols, "rows=" + rows, "sparsity=" + sparsity,
+ "min=" + min, "max= " + max};
runTest(null);
}
diff --git a/src/test/java/org/apache/sysds/test/functions/compress/CompressInstructionRewrite.java b/src/test/java/org/apache/sysds/test/functions/compress/CompressInstructionRewrite.java
index b1697e7..cfff020 100644
--- a/src/test/java/org/apache/sysds/test/functions/compress/CompressInstructionRewrite.java
+++ b/src/test/java/org/apache/sysds/test/functions/compress/CompressInstructionRewrite.java
@@ -63,7 +63,7 @@ public class CompressInstructionRewrite extends AutomatedTestBase {
@Test
public void testCompressInstruction_02_toSmallToCompress() {
- compressTest(1, 74, 0.2, ExecType.CP, 0, 5, 0, 0, "02");
+ compressTest(1, 20, 0.2, ExecType.CP, 0, 5, 0, 0, "02");
}
@Test
diff --git a/src/test/java/org/apache/sysds/test/functions/compress/CompressRewriteSpark.java b/src/test/java/org/apache/sysds/test/functions/compress/CompressRewriteSpark.java
index dc5d205..15d3fa2 100644
--- a/src/test/java/org/apache/sysds/test/functions/compress/CompressRewriteSpark.java
+++ b/src/test/java/org/apache/sysds/test/functions/compress/CompressRewriteSpark.java
@@ -25,6 +25,7 @@ import java.io.File;
import org.apache.sysds.common.Types;
import org.apache.sysds.common.Types.ExecMode;
+import org.apache.sysds.hops.OptimizerUtils;
import org.apache.sysds.test.AutomatedTestBase;
import org.apache.sysds.test.TestConfiguration;
import org.apache.sysds.test.TestUtils;
@@ -88,6 +89,7 @@ public class CompressRewriteSpark extends AutomatedTestBase {
public void compressTest(ExecMode instType, String name, String data) {
+ OptimizerUtils.ALLOW_SCRIPT_LEVEL_COMPRESS_COMMAND =true;
Types.ExecMode platformOld = setExecMode(instType);
try {
diff --git a/src/test/java/org/apache/sysds/test/functions/compress/LocalInstruction.java b/src/test/java/org/apache/sysds/test/functions/compress/LocalInstruction.java
new file mode 100644
index 0000000..001eee8
--- /dev/null
+++ b/src/test/java/org/apache/sysds/test/functions/compress/LocalInstruction.java
@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.sysds.test.functions.compress;
+
+import static org.junit.Assert.fail;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.sysds.common.Types.ExecMode;
+import org.apache.sysds.hops.OptimizerUtils;
+import org.apache.sysds.runtime.matrix.data.MatrixBlock;
+import org.apache.sysds.test.AutomatedTestBase;
+import org.apache.sysds.test.TestUtils;
+import org.apache.sysds.utils.Statistics;
+import org.junit.Assert;
+
+public abstract class LocalInstruction extends AutomatedTestBase {
+
+ private static final Log LOG = LogFactory.getLog(LocalInstruction.class.getName());
+
+ private final MatrixBlock X;
+
+ public LocalInstruction() {
+ X = TestUtils.round(TestUtils.generateTestMatrixBlock(10000, 1000, 0, 5, 1.0, 7));
+ }
+
+ protected abstract String getTestDir();
+
+ protected void run(String testName, int compressionCount) {
+ run(testName, compressionCount, 10);
+ }
+
+ protected void run(String testName, int compressionCount, int sparkCollectionCount) {
+ ExecMode mode = ExecMode.SPARK;
+ ExecMode oldPlatform = setExecMode(mode);
+
+ OptimizerUtils.ALLOW_SCRIPT_LEVEL_COMPRESS_COMMAND = true;
+ OptimizerUtils.ALLOW_SCRIPT_LEVEL_LOCAL_COMMAND = true;
+
+ try {
+ loadTestConfiguration(getTestConfiguration(testName));
+
+ String HOME = SCRIPT_DIR + getTestDir();
+ fullDMLScriptName = HOME + testName + ".dml";
+ programArgs = new String[] {"-explain", "-stats", "20", "-args", input("X"), sparkCollectionCount + ""};
+
+ writeInputMatrixWithMTD("X", X, false);
+
+ String ret = runTest(null).toString();
+ LOG.error(ret);
+ long actualCompressionCount = Statistics.getCPHeavyHitterCount("sp_compress");
+ long actualCollectCount = Statistics.getSparkCollectCount();
+ Assert.assertEquals(ret + "Compression count is incorrect", compressionCount, actualCompressionCount);
+ Assert.assertEquals(ret + "Collection count is incorrect", sparkCollectionCount, actualCollectCount);
+ Statistics.reset();
+
+ }
+ catch(Exception e) {
+ resetExecMode(oldPlatform);
+ e.printStackTrace();
+ fail("Failed workload test");
+ }
+ finally {
+ resetExecMode(oldPlatform);
+ }
+ }
+}
diff --git a/src/test/java/org/apache/sysds/test/functions/compress/LocalInstructionWithCompression.java b/src/test/java/org/apache/sysds/test/functions/compress/LocalInstructionWithCompression.java
new file mode 100644
index 0000000..87afea9
--- /dev/null
+++ b/src/test/java/org/apache/sysds/test/functions/compress/LocalInstructionWithCompression.java
@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.sysds.test.functions.compress;
+
+import java.io.File;
+
+import org.apache.sysds.test.TestConfiguration;
+import org.apache.sysds.test.TestUtils;
+import org.junit.Test;
+
+public class LocalInstructionWithCompression extends LocalInstruction {
+ private final static String TEST_NAME1 = "local";
+ private final static String TEST_DIR = "functions/compress/local/";
+ private final static String TEST_CLASS_DIR = TEST_DIR + LocalInstructionWithCompression.class.getSimpleName() + "/";
+
+ public LocalInstructionWithCompression() {
+ super();
+ }
+
+ @Override
+ public void setUp() {
+ final String dir = TEST_CLASS_DIR + "/local/";
+ TestUtils.clearAssertionInformation();
+ addTestConfiguration(TEST_NAME1, new TestConfiguration(dir, TEST_NAME1, new String[] {"B"}));
+ }
+
+ @Override
+ protected String getTestDir() {
+ return TEST_DIR;
+ }
+
+ @Test
+ public void tests_01() {
+ run(TEST_NAME1, 1, 1);
+ }
+
+ @Test
+ public void tests_02() {
+ run(TEST_NAME1, 1);
+ }
+
+ @Override
+ protected File getConfigTemplateFile() {
+ return new File(SCRIPT_DIR + TEST_DIR, "SystemDS-config.xml");
+ }
+}
diff --git a/src/test/java/org/apache/sysds/test/functions/compress/LocalInstructionWithoutCompression.java b/src/test/java/org/apache/sysds/test/functions/compress/LocalInstructionWithoutCompression.java
new file mode 100644
index 0000000..f22d285
--- /dev/null
+++ b/src/test/java/org/apache/sysds/test/functions/compress/LocalInstructionWithoutCompression.java
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.sysds.test.functions.compress;
+
+import java.io.File;
+
+import org.apache.sysds.test.TestConfiguration;
+import org.apache.sysds.test.TestUtils;
+import org.junit.Test;
+
+public class LocalInstructionWithoutCompression extends LocalInstruction {
+ private final static String TEST_NAME1 = "local";
+ private final static String TEST_DIR = "functions/compress/local/";
+ private final static String TEST_CLASS_DIR = TEST_DIR + LocalInstructionWithoutCompression.class.getSimpleName()
+ + "/";
+
+ public LocalInstructionWithoutCompression() {
+ super();
+ }
+
+ @Override
+ public void setUp() {
+ final String dir = TEST_CLASS_DIR + "/local/";
+ TestUtils.clearAssertionInformation();
+ addTestConfiguration(TEST_NAME1, new TestConfiguration(dir, TEST_NAME1, new String[] {"B"}));
+ }
+
+ @Override
+ protected String getTestDir() {
+ return TEST_DIR;
+ }
+
+ @Test
+ public void tests_01() {
+ run(TEST_NAME1, 0, 1);
+ }
+
+ @Test
+ public void tests_02() {
+ run(TEST_NAME1, 0);
+ }
+
+ @Override
+ protected File getConfigTemplateFile() {
+ return new File(SCRIPT_DIR + TEST_DIR, "SystemDS-no-compress-config.xml");
+ }
+}
diff --git a/src/test/java/org/apache/sysds/test/functions/compress/configuration/CompressBase.java b/src/test/java/org/apache/sysds/test/functions/compress/configuration/CompressBase.java
index d38bb32..2ae285c 100644
--- a/src/test/java/org/apache/sysds/test/functions/compress/configuration/CompressBase.java
+++ b/src/test/java/org/apache/sysds/test/functions/compress/configuration/CompressBase.java
@@ -23,6 +23,7 @@ import static org.junit.Assert.assertTrue;
import org.apache.sysds.common.Types;
import org.apache.sysds.common.Types.ExecType;
+import org.apache.sysds.hops.OptimizerUtils;
import org.apache.sysds.runtime.meta.MatrixCharacteristics;
import org.apache.sysds.test.AutomatedTestBase;
import org.apache.sysds.test.TestConfiguration;
@@ -53,6 +54,7 @@ public abstract class CompressBase extends AutomatedTestBase {
public void compressTest(int rows, int cols, double sparsity, ExecType instType, int min, int max, double delta,
int decompressionCountExpected, int compressionCountsExpected, String name) {
+ OptimizerUtils.ALLOW_SCRIPT_LEVEL_COMPRESS_COMMAND =true;
Types.ExecMode platformOld = setExecMode(instType);
try {
diff --git a/src/test/java/org/apache/sysds/test/functions/compress/instructionsSpark/CompressedSparkInstructionsTest.java b/src/test/java/org/apache/sysds/test/functions/compress/instructionsSpark/CompressedSparkInstructionsTest.java
index 8dc653c..b98e835 100644
--- a/src/test/java/org/apache/sysds/test/functions/compress/instructionsSpark/CompressedSparkInstructionsTest.java
+++ b/src/test/java/org/apache/sysds/test/functions/compress/instructionsSpark/CompressedSparkInstructionsTest.java
@@ -25,6 +25,7 @@ import static org.junit.Assert.fail;
import org.apache.sysds.common.Types;
import org.apache.sysds.common.Types.ExecType;
+import org.apache.sysds.hops.OptimizerUtils;
import org.apache.sysds.test.AutomatedTestBase;
import org.apache.sysds.test.TestConfiguration;
import org.apache.sysds.test.TestUtils;
@@ -32,96 +33,96 @@ import org.junit.Test;
public abstract class CompressedSparkInstructionsTest extends AutomatedTestBase {
- final double eps = 0.00000001;
+ final double eps = 0.00000001;
- protected String getTestClassDir() {
- return getTestDir() + this.getClass().getSimpleName() + "/";
- }
+ protected String getTestClassDir() {
+ return getTestDir() + this.getClass().getSimpleName() + "/";
+ }
- protected String getTestName() {
- return "compress";
- }
+ protected String getTestName() {
+ return "compress";
+ }
- protected String getTestDir() {
- return "functions/compress/instructionsSpark/";
- }
+ protected String getTestDir() {
+ return "functions/compress/instructionsSpark/";
+ }
- public abstract double getDensity();
+ public abstract double getDensity();
- @Test
- public void testSum() {
- run("sum");
- }
+ @Test
+ public void testSum() {
+ run("sum");
+ }
- @Test
- public void testMatrixMult(){
- run("mm");
- }
+ @Test
+ public void testMatrixMult() {
+ run("mm");
+ }
- private void run(String name) {
- run(10, 2500, getDensity(), 1, 5, name);
- }
+ private void run(String name) {
+ run(10, 2500, getDensity(), 1, 5, name);
+ }
-
- private void run(int cols, int rows, double sparsity, int min, int max, String name) {
+ private void run(int cols, int rows, double sparsity, int min, int max, String name) {
- Types.ExecMode platformOld = setExecMode(ExecType.SPARK);
- try {
- setOutputBuffering(true);
+ OptimizerUtils.ALLOW_SCRIPT_LEVEL_COMPRESS_COMMAND = true;
+ Types.ExecMode platformOld = setExecMode(ExecType.SPARK);
+ try {
+ setOutputBuffering(true);
loadTestConfiguration(getTestConfiguration(getTestName()));
fullDMLScriptName = SCRIPT_DIR + "/" + getTestDir() + "compress_" + name + ".dml";
- programArgs = new String[] {"-explain","-stats","-args", s(rows), s(cols), s(min), s(max), s(sparsity)};
- String stdout = runTest(null).toString();
- parseOutAndVerify(stdout);
- }
- catch(Exception e) {
- e.printStackTrace();
- assertTrue("Exception in execution: " + e.getMessage(), false);
- }
- finally {
- rtplatform = platformOld;
- }
- }
-
- private void parseOutAndVerify(String out){
- double C = 0;
- double uC = 0;
- try{
-
- String[] lines = out.split("\n");
- for(int i = 0; i < lines.length; i++)
- if(lines[i].equals("RESULTS:")){
- C =Double.parseDouble(lines[i+1].split("=")[1]);
- uC =Double.parseDouble(lines[i+2].split("=")[1]);
- break;
- }
- }
- catch(Exception e){
- fail("Error when passing results:\n" + out);
- }
- assertEquals(out, C, uC, eps);
- }
-
- private String s(int i){
- return String.valueOf(i);
- }
-
- private String s(double i){
- return String.valueOf(i);
- }
-
- @Override
- public void setUp() {
- TestUtils.clearAssertionInformation();
- addTestConfiguration(getTestName(), new TestConfiguration(getTestClassDir(), getTestName()));
- }
-
- // /**
- // * Override default configuration with custom test configuration to ensure scratch space and local temporary
- // * directory locations are also updated.
- // */
- // @Override
- // protected File getConfigTemplateFile() {
- // return TEST_CONF_FILE;
- // }
+ programArgs = new String[] {"-explain", "-stats", "-args", s(rows), s(cols), s(min), s(max), s(sparsity)};
+ String stdout = runTest(null).toString();
+ parseOutAndVerify(stdout);
+ }
+ catch(Exception e) {
+ e.printStackTrace();
+ assertTrue("Exception in execution: " + e.getMessage(), false);
+ }
+ finally {
+ rtplatform = platformOld;
+ }
+ }
+
+ private void parseOutAndVerify(String out) {
+ double C = 0;
+ double uC = 0;
+ try {
+
+ String[] lines = out.split("\n");
+ for(int i = 0; i < lines.length; i++)
+ if(lines[i].equals("RESULTS:")) {
+ C = Double.parseDouble(lines[i + 1].split("=")[1]);
+ uC = Double.parseDouble(lines[i + 2].split("=")[1]);
+ break;
+ }
+ }
+ catch(Exception e) {
+ fail("Error when passing results:\n" + out);
+ }
+ assertEquals(out, C, uC, eps);
+ }
+
+ private String s(int i) {
+ return String.valueOf(i);
+ }
+
+ private String s(double i) {
+ return String.valueOf(i);
+ }
+
+ @Override
+ public void setUp() {
+ TestUtils.clearAssertionInformation();
+ addTestConfiguration(getTestName(), new TestConfiguration(getTestClassDir(), getTestName()));
+ }
+
+ // /**
+ // * Override default configuration with custom test configuration to ensure scratch space and local temporary
+ // * directory locations are also updated.
+ // */
+ // @Override
+ // protected File getConfigTemplateFile() {
+ // return TEST_CONF_FILE;
+ // }
}
diff --git a/src/test/java/org/apache/sysds/test/functions/compress/instructionsSpark/CompressedSparkInstructionsTestDense.java b/src/test/java/org/apache/sysds/test/functions/compress/instructionsSpark/CompressedSparkInstructionsTestDense.java
index fc30bc5..2d1c906 100644
--- a/src/test/java/org/apache/sysds/test/functions/compress/instructionsSpark/CompressedSparkInstructionsTestDense.java
+++ b/src/test/java/org/apache/sysds/test/functions/compress/instructionsSpark/CompressedSparkInstructionsTestDense.java
@@ -25,6 +25,4 @@ public class CompressedSparkInstructionsTestDense extends CompressedSparkInstruc
public double getDensity() {
return 1.0;
}
-
-
}
diff --git a/src/test/java/org/apache/sysds/test/functions/compress/instructionsSpark/CompressedSparkInstructionsTestSparse.java b/src/test/java/org/apache/sysds/test/functions/compress/instructionsSpark/CompressedSparkInstructionsTestSparse.java
index 94f7be8..c2812b2 100644
--- a/src/test/java/org/apache/sysds/test/functions/compress/instructionsSpark/CompressedSparkInstructionsTestSparse.java
+++ b/src/test/java/org/apache/sysds/test/functions/compress/instructionsSpark/CompressedSparkInstructionsTestSparse.java
@@ -25,5 +25,4 @@ public class CompressedSparkInstructionsTestSparse extends CompressedSparkInstru
public double getDensity() {
return 0.1;
}
-
}
diff --git a/src/test/java/org/apache/sysds/test/functions/compress/instructionsSpark/CompressedSparkInstructionsTestDense.java b/src/test/scripts/functions/compress/local/SystemDS-config.xml
similarity index 76%
copy from src/test/java/org/apache/sysds/test/functions/compress/instructionsSpark/CompressedSparkInstructionsTestDense.java
copy to src/test/scripts/functions/compress/local/SystemDS-config.xml
index fc30bc5..8f2eaf4 100644
--- a/src/test/java/org/apache/sysds/test/functions/compress/instructionsSpark/CompressedSparkInstructionsTestDense.java
+++ b/src/test/scripts/functions/compress/local/SystemDS-config.xml
@@ -1,4 +1,4 @@
-/*
+<!--
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
@@ -6,25 +6,20 @@
* to you 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 org.apache.sysds.test.functions.compress.instructionsSpark;
-
-public class CompressedSparkInstructionsTestDense extends CompressedSparkInstructionsTest {
-
- @Override
- public double getDensity() {
- return 1.0;
- }
-
-
-}
+<root>
+ <sysds.compressed.linalg>true</sysds.compressed.linalg>
+ <sysds.defaultblocksize>600</sysds.defaultblocksize>
+ <sysds.cp.parallel.ops>true</sysds.cp.parallel.ops>
+ <sysds.scratch>target/local/</sysds.scratch>
+</root>
diff --git a/src/test/java/org/apache/sysds/test/functions/compress/instructionsSpark/CompressedSparkInstructionsTestDense.java b/src/test/scripts/functions/compress/local/SystemDS-no-compress-config.xml
similarity index 76%
copy from src/test/java/org/apache/sysds/test/functions/compress/instructionsSpark/CompressedSparkInstructionsTestDense.java
copy to src/test/scripts/functions/compress/local/SystemDS-no-compress-config.xml
index fc30bc5..3fa6b24 100644
--- a/src/test/java/org/apache/sysds/test/functions/compress/instructionsSpark/CompressedSparkInstructionsTestDense.java
+++ b/src/test/scripts/functions/compress/local/SystemDS-no-compress-config.xml
@@ -1,4 +1,4 @@
-/*
+<!--
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
@@ -6,25 +6,20 @@
* to you 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 org.apache.sysds.test.functions.compress.instructionsSpark;
-
-public class CompressedSparkInstructionsTestDense extends CompressedSparkInstructionsTest {
-
- @Override
- public double getDensity() {
- return 1.0;
- }
-
-
-}
+<root>
+ <sysds.compressed.linalg>false</sysds.compressed.linalg>
+ <sysds.defaultblocksize>600</sysds.defaultblocksize>
+ <sysds.cp.parallel.ops>true</sysds.cp.parallel.ops>
+ <sysds.scratch>target/local_noComp/</sysds.scratch>
+</root>
diff --git a/src/test/scripts/functions/compress/local/local.dml b/src/test/scripts/functions/compress/local/local.dml
new file mode 100644
index 0000000..a861c36
--- /dev/null
+++ b/src/test/scripts/functions/compress/local/local.dml
@@ -0,0 +1,29 @@
+#-------------------------------------------------------------
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you 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.
+#
+#-------------------------------------------------------------
+
+A = read($1)
+for(i in 1:$2){
+ AP = A + i
+ r = local(AP)
+}
+
+# Print something
+print(max(r))