[SYSTEMDS-2686] Compressed overlapping column groups
This commit change the compressed right multiplication to perform in
compressed space, resulting in significantly faster execution.
The technique employed results in overlapping column groups, that contain
partial results of the matrix multiplication.
The downside is that some operations that previously was possible on
compressed space no longer works, for the overlapping column groups.
To still support all the operations, then decompression is used for
cases where it is impossible to execute on the compressed matrices.
Another addition is that statistics now contain compression and
decompression times if compression is enabled.
Furthermore the tests have been extended to include tests for the
overlapping resulting in + 10000 tests in compression now.
diff --git a/src/main/java/org/apache/sysds/runtime/compress/AbstractCompressedMatrixBlock.java b/src/main/java/org/apache/sysds/runtime/compress/AbstractCompressedMatrixBlock.java
index 9eeaf8f..7055f0d 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/AbstractCompressedMatrixBlock.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/AbstractCompressedMatrixBlock.java
@@ -22,6 +22,7 @@
import java.util.ArrayList;
import java.util.List;
+import org.apache.commons.lang3.tuple.Pair;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.commons.math3.random.Well1024a;
@@ -35,8 +36,6 @@
import org.apache.sysds.runtime.instructions.cp.ScalarObject;
import org.apache.sysds.runtime.instructions.spark.data.IndexedMatrixValue;
import org.apache.sysds.runtime.matrix.data.CTableMap;
-import org.apache.sysds.runtime.matrix.data.LibMatrixBincell;
-import org.apache.sysds.runtime.matrix.data.LibMatrixBincell.BinaryAccessType;
import org.apache.sysds.runtime.matrix.data.MatrixBlock;
import org.apache.sysds.runtime.matrix.data.MatrixIndexes;
import org.apache.sysds.runtime.matrix.data.MatrixValue;
@@ -63,6 +62,17 @@
protected List<ColGroup> _colGroups;
/**
+ * list of lengths of dictionaries, including a longest length in left variable. Note Should not be called directly
+ * since it is constructed on first use, on calls to : getMaxNumValues()
+ */
+ protected Pair<Integer, int[]> v = null;
+
+ /**
+ * Boolean specifying if the colGroups are overlapping each other. This happens after a right matrix multiplication.
+ */
+ protected boolean overlappingColGroups = false;
+
+ /**
* Constructor for building an empty Compressed Matrix block object.
*/
public AbstractCompressedMatrixBlock() {
@@ -70,6 +80,15 @@
}
/**
+ * Create a potentially overlapping Compressed Matrix Block.
+ * @param overLapping boolean specifying if the matrix blocks columns are overlapping.
+ */
+ public AbstractCompressedMatrixBlock(boolean overLapping) {
+ super();
+ overlappingColGroups = overLapping;
+ }
+
+ /**
* Main constructor for building a block from scratch.
*
* @param rl number of rows in the block
@@ -136,61 +155,6 @@
}
@Override
- public MatrixBlock binaryOperations(BinaryOperator op, MatrixValue thatValue, MatrixValue result) {
-
- MatrixBlock that = getUncompressed(thatValue);
-
- if(!LibMatrixBincell.isValidDimensionsBinary(this, that)) {
- throw new RuntimeException("Block sizes are not matched for binary " + "cell operations: " + this.rlen + "x"
- + this.clen + " vs " + that.getNumRows() + "x" + that.getNumColumns());
- }
-
- MatrixBlock right = getUncompressed(thatValue);
-
- CompressedMatrixBlock ret = null;
- if(result == null || !(result instanceof CompressedMatrixBlock))
- ret = new CompressedMatrixBlock(getNumRows(), getNumColumns(), sparse);
- else {
- ret = (CompressedMatrixBlock) result;
- ret.reset(rlen, clen);
- }
-
- // MatrixBlock ret = (MatrixBlock) result;
- bincellOp(right, ret, op);
- return ret;
- }
-
- /**
- * matrix-matrix binary operations, MM, MV
- *
- * @param m2 input matrix 2
- * @param ret result matrix
- * @param op binary operator
- */
- private void bincellOp(MatrixBlock m2, CompressedMatrixBlock ret, BinaryOperator op) {
-
-
- BinaryAccessType atype = LibMatrixBincell.getBinaryAccessType((MatrixBlock) this, m2);
- if(atype == BinaryAccessType.MATRIX_COL_VECTOR // MATRIX - VECTOR
- || atype == BinaryAccessType.MATRIX_ROW_VECTOR) {
- binaryMV(m2, ret, op, atype);
- }
- else if(atype == BinaryAccessType.OUTER_VECTOR_VECTOR) // VECTOR - VECTOR
- {
- binaryVV(m2, ret, op, atype);
- }
- else {
- binaryMM(m2, ret, op);
- }
- }
-
- protected abstract void binaryMV(MatrixBlock m2, CompressedMatrixBlock ret, BinaryOperator op, BinaryAccessType atype );
-
- protected abstract void binaryVV(MatrixBlock m2, CompressedMatrixBlock ret, BinaryOperator op, BinaryAccessType atype );
-
- protected abstract void binaryMM(MatrixBlock m2, CompressedMatrixBlock ret, BinaryOperator op);
-
- @Override
public MatrixBlock binaryOperationsInPlace(BinaryOperator op, MatrixValue thatValue) {
printDecompressWarning("binaryOperationsInPlace", (MatrixBlock) thatValue);
MatrixBlock left = decompress();
@@ -542,7 +506,7 @@
return(mb instanceof CompressedMatrixBlock);
}
- private static MatrixBlock getUncompressed(MatrixValue mVal) {
+ protected static MatrixBlock getUncompressed(MatrixValue mVal) {
return isCompressed((MatrixBlock) mVal) ? ((CompressedMatrixBlock) mVal).decompress() : (MatrixBlock) mVal;
}
@@ -575,4 +539,4 @@
public void toShallowSerializeBlock() {
// do nothing
}
-}
\ No newline at end of file
+}
diff --git a/src/main/java/org/apache/sysds/runtime/compress/BitmapEncoder.java b/src/main/java/org/apache/sysds/runtime/compress/BitmapEncoder.java
index 2420f0d..d3f15fd 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/BitmapEncoder.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/BitmapEncoder.java
@@ -155,24 +155,32 @@
*/
private static Bitmap extractBitmap(int[] colIndices, MatrixBlock rawBlock, ReaderColumnSelection rowReader) {
// probe map for distinct items (for value or value groups)
- DblArrayIntListHashMap distinctVals = new DblArrayIntListHashMap();
+ DblArrayIntListHashMap distinctVals;
+ if(colIndices.length > 10) {
+ distinctVals = new DblArrayIntListHashMap(2048);
+ }
+ else {
+ distinctVals = new DblArrayIntListHashMap();
+ }
// scan rows and probe/build distinct items
DblArray cellVals = null;
int zero = 0;
while((cellVals = rowReader.nextRow()) != null) {
- IntArrayList lstPtr = distinctVals.get(cellVals);
- if(lstPtr == null) {
- // create new objects only on demand
- lstPtr = new IntArrayList();
- distinctVals.appendValue(new DblArray(cellVals), lstPtr);
+ if(cellVals.getData() == null) {
+ zero += 1;
}
- zero += DblArray.isZero(cellVals) ? 1 : 0;
-
- lstPtr.appendValue(rowReader.getCurrentRowIndex());
+ else {
+ IntArrayList lstPtr = distinctVals.get(cellVals);
+ if(lstPtr == null) {
+ // create new objects only on demand
+ lstPtr = new IntArrayList();
+ distinctVals.appendValue(new DblArray(cellVals), lstPtr);
+ }
+ lstPtr.appendValue(rowReader.getCurrentRowIndex());
+ }
}
-
return makeBitmap(distinctVals, colIndices.length, zero);
}
@@ -488,7 +496,8 @@
max = fp[fp.length - 1];
if(fp[fp.length - 1] < min)
min = fp[fp.length - 1];
- }else{
+ }
+ else {
max = fp[0];
min = fp[0];
maxDelta = 0;
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 8a8e27b..d69ebad 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/CompressedMatrixBlock.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/CompressedMatrixBlock.java
@@ -38,26 +38,25 @@
import org.apache.commons.lang3.tuple.Pair;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
-import org.apache.sysds.hops.OptimizerUtils;
+import org.apache.sysds.api.DMLScript;
import org.apache.sysds.lops.MMTSJ.MMTSJType;
import org.apache.sysds.lops.MapMultChain.ChainType;
-import org.apache.sysds.runtime.DMLCompressionException;
import org.apache.sysds.runtime.DMLRuntimeException;
import org.apache.sysds.runtime.compress.colgroup.ColGroup;
import org.apache.sysds.runtime.compress.colgroup.ColGroup.CompressionType;
import org.apache.sysds.runtime.compress.colgroup.ColGroupConverter;
-import org.apache.sysds.runtime.compress.colgroup.ColGroupDDC;
import org.apache.sysds.runtime.compress.colgroup.ColGroupIO;
-import org.apache.sysds.runtime.compress.colgroup.ColGroupOLE;
-import org.apache.sysds.runtime.compress.colgroup.ColGroupRLE;
import org.apache.sysds.runtime.compress.colgroup.ColGroupUncompressed;
import org.apache.sysds.runtime.compress.colgroup.ColGroupValue;
import org.apache.sysds.runtime.compress.colgroup.DenseRowIterator;
import org.apache.sysds.runtime.compress.colgroup.SparseRowIterator;
+import org.apache.sysds.runtime.compress.lib.LibBinaryCellOp;
+import org.apache.sysds.runtime.compress.lib.LibLeftMultBy;
+import org.apache.sysds.runtime.compress.lib.LibRightMultBy;
+import org.apache.sysds.runtime.compress.lib.LibScalar;
import org.apache.sysds.runtime.compress.utils.ColumnGroupIterator;
import org.apache.sysds.runtime.compress.utils.LinearAlgebraUtils;
import org.apache.sysds.runtime.controlprogram.parfor.stat.Timing;
-import org.apache.sysds.runtime.data.DenseBlock;
import org.apache.sysds.runtime.data.SparseBlock;
import org.apache.sysds.runtime.data.SparseRow;
import org.apache.sysds.runtime.functionobjects.Builtin;
@@ -66,7 +65,12 @@
import org.apache.sysds.runtime.functionobjects.KahanPlus;
import org.apache.sysds.runtime.functionobjects.KahanPlusSq;
import org.apache.sysds.runtime.functionobjects.Mean;
+import org.apache.sysds.runtime.functionobjects.Minus;
+import org.apache.sysds.runtime.functionobjects.MinusMultiply;
import org.apache.sysds.runtime.functionobjects.Multiply;
+import org.apache.sysds.runtime.functionobjects.Plus;
+import org.apache.sysds.runtime.functionobjects.PlusMultiply;
+import org.apache.sysds.runtime.functionobjects.Power2;
import org.apache.sysds.runtime.functionobjects.ReduceAll;
import org.apache.sysds.runtime.functionobjects.ReduceCol;
import org.apache.sysds.runtime.functionobjects.ReduceRow;
@@ -74,7 +78,6 @@
import org.apache.sysds.runtime.matrix.data.IJV;
import org.apache.sysds.runtime.matrix.data.LibMatrixBincell;
import org.apache.sysds.runtime.matrix.data.LibMatrixBincell.BinaryAccessType;
-import org.apache.sysds.runtime.matrix.data.LibMatrixReorg;
import org.apache.sysds.runtime.matrix.data.MatrixBlock;
import org.apache.sysds.runtime.matrix.data.MatrixIndexes;
import org.apache.sysds.runtime.matrix.data.MatrixValue;
@@ -84,6 +87,7 @@
import org.apache.sysds.runtime.matrix.operators.LeftScalarOperator;
import org.apache.sysds.runtime.matrix.operators.ScalarOperator;
import org.apache.sysds.runtime.util.CommonThreadPool;
+import org.apache.sysds.utils.DMLCompressionStatistics;
public class CompressedMatrixBlock extends AbstractCompressedMatrixBlock {
private static final Log LOG = LogFactory.getLog(CompressedMatrixBlock.class.getName());
@@ -102,6 +106,15 @@
}
/**
+ * Create a base Compressed matrix block with overlapping column groups.
+ *
+ * @param overLapping boolean specifier of the if the groups are overlapping.
+ */
+ public CompressedMatrixBlock(boolean overLapping) {
+ super(overLapping);
+ }
+
+ /**
* Main constructor for building a block from scratch.
*
* @param rl number of rows in the block
@@ -152,7 +165,8 @@
Timing time = new Timing(true);
// preallocation sparse rows to avoid repeated reallocations
- MatrixBlock ret = new MatrixBlock(getNumRows(), getNumColumns(), isInSparseFormat(), getNonZeros());
+ MatrixBlock ret = (nonZeros == -1) ? new MatrixBlock(rlen, clen, false, -1)
+ .allocateBlock() : new MatrixBlock(rlen, clen, sparse, nonZeros).allocateBlock();
if(ret.isInSparseFormat()) {
int[] rnnz = new int[rlen];
for(ColGroup grp : _colGroups)
@@ -168,13 +182,20 @@
grp.decompressToBlock(ret, 0, rlen);
// post-processing (for append in decompress)
- ret.setNonZeros(nonZeros);
+ if(ret.getNonZeros() == -1 || nonZeros == -1) {
+ ret.recomputeNonZeros();
+ }
+ else {
+ ret.setNonZeros(nonZeros);
+ }
if(ret.isInSparseFormat())
ret.sortSparseRows();
- if(LOG.isDebugEnabled())
- LOG.debug("decompressed block in " + time.stop() + "ms.");
-
+ if(DMLScript.STATISTICS || LOG.isDebugEnabled()) {
+ double t = time.stop();
+ LOG.debug("decompressed block w/ k=" + 1 + " in " + t + "ms.");
+ DMLCompressionStatistics.addDecompressTime(t,1);
+ }
return ret;
}
@@ -191,30 +212,38 @@
Timing time = new Timing(true);
- MatrixBlock ret = new MatrixBlock(rlen, clen, sparse, nonZeros).allocateBlock();
-
+ MatrixBlock ret = (nonZeros == -1) ? new MatrixBlock(rlen, clen, false, -1)
+ .allocateBlock() : new MatrixBlock(rlen, clen, sparse, nonZeros).allocateBlock();
// multi-threaded decompression
+ nonZeros = 0;
try {
ExecutorService pool = CommonThreadPool.get(k);
int rlen = getNumRows();
- int blklen = getAlignedBlockSize((int) (Math.ceil((double) rlen / k)));
+ final int blkz = CompressionSettings.BITMAP_BLOCK_SZ;
+ int blklen = (int) Math.ceil((double) rlen / k);
+ blklen += (blklen % blkz != 0) ? blkz - blklen % blkz : 0;
ArrayList<DecompressTask> tasks = new ArrayList<>();
for(int i = 0; i < k & i * blklen < getNumRows(); i++)
tasks.add(new DecompressTask(_colGroups, ret, i * blklen, Math.min((i + 1) * blklen, rlen)));
- List<Future<Object>> rtasks = pool.invokeAll(tasks);
+ List<Future<Long>> rtasks = pool.invokeAll(tasks);
pool.shutdown();
- for(Future<Object> rt : rtasks)
- rt.get(); // error handling
+ for(Future<Long> rt : rtasks)
+ nonZeros += rt.get(); // error handling
}
catch(InterruptedException | ExecutionException ex) {
LOG.error("Parallel decompression failed defaulting to non parallel implementation " + ex.getMessage());
+ nonZeros = -1;
+ ex.printStackTrace();
return decompress();
}
- // post-processing
ret.setNonZeros(nonZeros);
- LOG.debug("decompressed block w/ k=" + k + " in " + time.stop() + "ms.");
+ if(DMLScript.STATISTICS || LOG.isDebugEnabled()) {
+ double t = time.stop();
+ LOG.debug("decompressed block w/ k=" + k + " in " + time.stop() + "ms.");
+ DMLCompressionStatistics.addDecompressTime(t, k);
+ }
return ret;
}
@@ -247,18 +276,17 @@
public double quickGetValue(int r, int c) {
// TODO Optimize Quick Get Value, to located the correct column group without having to search for it
- ColGroup grp = null;
+ double v = 0.0;
for(ColGroup group : _colGroups) {
if(Arrays.binarySearch(group.getColIndices(), c) >= 0) {
- grp = group;
- break;
+ v += group.get(r, c);
+ if(!isOverlapping())
+ break;
}
}
- if(grp == null) {
- throw new DMLCompressionException("ColGroup for column index not found");
- }
+
// find row value
- return grp.get(r, c);
+ return v;
}
//////////////////////////////////////////
@@ -281,6 +309,7 @@
rlen = in.readInt();
clen = in.readInt();
nonZeros = in.readLong();
+ overlappingColGroups = in.readBoolean();
_colGroups = ColGroupIO.readGroups(in);
}
@@ -290,6 +319,7 @@
out.writeInt(rlen);
out.writeInt(clen);
out.writeLong(nonZeros);
+ out.writeBoolean(overlappingColGroups);
ColGroupIO.writeGroups(out, _colGroups);
}
@@ -342,119 +372,55 @@
@Override
public MatrixBlock scalarOperations(ScalarOperator sop, MatrixValue result) {
+ if(overlappingColGroups && !(sop.fn instanceof Multiply || sop.fn instanceof Plus || sop.fn instanceof Minus ||
+ (sop instanceof LeftScalarOperator && sop.fn instanceof Power2))) {
+ MatrixBlock m1d = decompress(sop.getNumThreads());
+ result = m1d.scalarOperations(sop, result);
+ return (MatrixBlock) result;
+ }
- // allocate the output matrix block
CompressedMatrixBlock ret = null;
if(result == null || !(result instanceof CompressedMatrixBlock))
ret = new CompressedMatrixBlock(getNumRows(), getNumColumns(), sparse);
- else {
- ret = (CompressedMatrixBlock) result;
- ret.reset(rlen, clen);
+ result = LibScalar.scalarOperations(sop, this, ret, overlappingColGroups);
+ return (MatrixBlock) result;
+ }
+
+ @Override
+ public MatrixBlock binaryOperations(BinaryOperator op, MatrixValue thatValue, MatrixValue result) {
+
+ MatrixBlock that = getUncompressed(thatValue);
+ if(!LibMatrixBincell.isValidDimensionsBinary(this, that)) {
+ throw new DMLRuntimeException("Block sizes are not matched for binary " + "cell operations: " + this.rlen
+ + "x" + this.clen + " vs " + that.getNumRows() + "x" + that.getNumColumns());
}
- int threads = OptimizerUtils.getConstrainedNumThreads(_colGroups.size());
-
- if(threads > 1) {
- ExecutorService pool = CommonThreadPool.get(sop.getNumThreads());
- ArrayList<ScalarTask> tasks = new ArrayList<>();
-
- ArrayList<ColGroup> small = new ArrayList<>();
-
- for(ColGroup grp : _colGroups) {
- if(grp instanceof ColGroupUncompressed) {
- ArrayList<ColGroup> uc = new ArrayList<>();
- uc.add(grp);
- tasks.add(new ScalarTask(uc, sop));
- }
- else {
- int nv = ((ColGroupValue) grp).getNumValues();
- if(nv < 256) {
- small.add(grp);
- }
- else {
- ArrayList<ColGroup> large = new ArrayList<>();
- large.add(grp);
- tasks.add(new ScalarTask(large, sop));
-
- }
- }
- if(small.size() > 10) {
- tasks.add(new ScalarTask(small, sop));
- small = new ArrayList<>();
- }
- }
- if(small.size() > 0) {
- tasks.add(new ScalarTask(small, sop));
- }
- try {
- List<Future<List<ColGroup>>> rtasks = pool.invokeAll(tasks);
- pool.shutdown();
-
- ArrayList<ColGroup> newColGroups = new ArrayList<>();
- for(Future<List<ColGroup>> f : rtasks) {
- for(ColGroup x : f.get()) {
- newColGroups.add(x);
- }
- }
- ret._colGroups = newColGroups;
- ret.setNonZeros(rlen * clen);
- }
- catch(InterruptedException | ExecutionException e) {
- LOG.fatal("UnaryAggregate Exception: " + e.getMessage(), e);
- throw new DMLRuntimeException(e);
- }
+ if(LibMatrixBincell.getBinaryAccessType(this, that) == BinaryAccessType.MATRIX_COL_VECTOR ||
+ (this.getNumColumns() == 1 && that.getNumColumns() == 1 && that.getNumRows() != 1) ||
+ !(op.fn instanceof Multiply || op.fn instanceof Plus || op.fn instanceof Minus ||
+ op.fn instanceof MinusMultiply || op.fn instanceof PlusMultiply)) {
+ // case MATRIX_COL_VECTOR:
+ // TODO make partial decompress and do operation.
+ // TODO support more of the operations... since it is possible.
+ MatrixBlock m2 = getUncompressed(this);
+ MatrixBlock ret = m2.binaryOperations(op, thatValue, result);
+ result = ret;
+ return ret;
}
else {
- // Apply the operation to each of the column groups.
- // Most implementations will only modify metadata.
- ArrayList<ColGroup> newColGroups = new ArrayList<>();
- for(ColGroup grp : _colGroups) {
- newColGroups.add(grp.scalarOperation(sop));
+ CompressedMatrixBlock ret = null;
+ if(result == null || !(result instanceof CompressedMatrixBlock))
+ ret = new CompressedMatrixBlock(getNumRows(), getNumColumns(), sparse);
+ else {
+ ret = (CompressedMatrixBlock) result;
+ ret.reset(rlen, clen);
}
- ret._colGroups = newColGroups;
- ret.setNonZeros(rlen * clen);
+ result = LibBinaryCellOp.bincellOp(this, that, ret, op);
+ result = ret;
+ return ret;
}
- return ret;
-
- }
-
- protected void binaryMV(MatrixBlock m2, CompressedMatrixBlock ret, BinaryOperator op, BinaryAccessType aType) {
- if(aType == BinaryAccessType.MATRIX_COL_VECTOR) {
- throw new NotImplementedException("Binary Matrix Col Vector operations are not implemented CLA");
- }
- else if(aType == BinaryAccessType.MATRIX_ROW_VECTOR) {
- // Apply the operation to each of the column groups.
- // Most implementations will only modify metadata.
- ArrayList<ColGroup> newColGroups = new ArrayList<>();
-
- for(ColGroup grp : _colGroups) {
- if(grp instanceof ColGroupUncompressed) {
- throw new DMLCompressionException("Not supported Binary MV");
- }
- else {
-
- if(grp.getNumCols() == 1) {
- ScalarOperator sop = new LeftScalarOperator(op.fn, m2.getValue(0, grp.getColIndices()[0]), 1);
- newColGroups.add(grp.scalarOperation(sop));
- }
- else {
- throw new NotImplementedException("Cocoded columns (nr cols:" + grp.getNumCols()
- + ") groupType: not implemented for Binary Matrix Row Vector operations");
- }
- }
- }
- ret._colGroups = newColGroups;
- }
- }
-
- protected void binaryVV(MatrixBlock m2, CompressedMatrixBlock ret, BinaryOperator op, BinaryAccessType aType) {
- throw new NotImplementedException("Binary Vector Vector operations are not implemented");
- }
-
- protected void binaryMM(MatrixBlock m2, CompressedMatrixBlock ret, BinaryOperator op) {
- throw new NotImplementedException("Binary Matrix Matrix operations are not implemented");
}
@Override
@@ -497,42 +463,7 @@
@Override
public MatrixBlock chainMatrixMultOperations(MatrixBlock v, MatrixBlock w, MatrixBlock out, ChainType ctype) {
-
- if(this.getNumColumns() != v.getNumRows())
- throw new DMLRuntimeException(
- "Dimensions mismatch on mmchain operation (" + this.getNumColumns() + " != " + v.getNumRows() + ")");
- if(v.getNumColumns() != 1)
- throw new DMLRuntimeException(
- "Invalid input vector (column vector expected, but ncol=" + v.getNumColumns() + ")");
- if(w != null && w.getNumColumns() != 1)
- throw new DMLRuntimeException(
- "Invalid weight vector (column vector expected, but ncol=" + w.getNumColumns() + ")");
-
- // single-threaded MMChain of single uncompressed ColGroup
- if(isSingleUncompressedGroup()) {
- return ((ColGroupUncompressed) _colGroups.get(0)).getData().chainMatrixMultOperations(v, w, out, ctype);
- }
-
- // prepare result
- if(out != null)
- out.reset(clen, 1, false);
- else
- out = new MatrixBlock(clen, 1, false);
-
- // empty block handling
- if(isEmptyBlock(false))
- return out;
-
- // compute matrix mult
- MatrixBlock tmp = new MatrixBlock(rlen, 1, false);
- rightMultByVector(v, tmp);
- if(ctype == ChainType.XtwXv) {
- BinaryOperator bop = new BinaryOperator(Multiply.getMultiplyFnObject());
- LibMatrixBincell.bincellOpInPlace(tmp, w, bop);
- }
- leftMultByVectorTranspose(_colGroups, tmp, out, true, true);
-
- return out;
+ return chainMatrixMultOperations(v, w, out, ctype, 1);
}
@Override
@@ -568,12 +499,12 @@
// compute matrix mult
MatrixBlock tmp = new MatrixBlock(rlen, 1, false);
- rightMultByVector(v, tmp, k);
+ tmp = LibRightMultBy.rightMultByMatrix(_colGroups, v, tmp, k, getMaxNumValues(), false);
if(ctype == ChainType.XtwXv) {
BinaryOperator bop = new BinaryOperator(Multiply.getMultiplyFnObject());
LibMatrixBincell.bincellOpInPlace(tmp, w, bop);
}
- leftMultByVectorTranspose(_colGroups, tmp, out, true, k);
+ LibLeftMultBy.leftMultByVectorTranspose(_colGroups, tmp, out, true, k, getMaxNumValues(), isOverlapping());
return out;
}
@@ -595,38 +526,24 @@
int cl = m2.getNumColumns();
// create output matrix block
- if(ret == null)
- ret = new MatrixBlock(rl, cl, false, rl * cl);
- else if(!(ret.getNumColumns() == cl && ret.getNumRows() == rl && ret.isAllocated()))
- ret.reset(rl, cl, false, rl * cl);
-
if(right) {
- if(that.getNumColumns() == 1) {
- // Right Matrix Vector Multiplication
- if(op.getNumThreads() > 1)
- rightMultByVector(that, ret, op.getNumThreads());
- else
- rightMultByVector(that, ret);
- }
- else {
- that = that instanceof CompressedMatrixBlock ? ((CompressedMatrixBlock) that).decompress() : that;
- ret = rightMultByMatrix(_colGroups, that, ret, op.getNumThreads(), that.getNumColumns());
- }
- }
- else { // Left
that = that instanceof CompressedMatrixBlock ? ((CompressedMatrixBlock) that).decompress() : that;
- if(that.getNumRows() == 1) {
- if(op.getNumThreads() > 1)
- return leftMultByVectorTranspose(_colGroups, that, ret, false, op.getNumThreads());
- else
- return leftMultByVectorTranspose(_colGroups, that, ret, false, true);
- }
- else {
- return leftMultByMatrix(_colGroups, that, ret, op.getNumThreads(), this.getNumColumns());
- }
+ return ret = LibRightMultBy
+ .rightMultByMatrix(_colGroups, that, ret, op.getNumThreads(), getMaxNumValues(), true);
+ }
+ else {
+ return LibLeftMultBy.leftMultByMatrix(_colGroups,
+ that,
+ ret,
+ false,
+ true,
+ rl,
+ cl,
+ isOverlapping(),
+ op.getNumThreads(),
+ getMaxNumValues());
}
- return ret;
}
@Override
@@ -642,6 +559,14 @@
throw new NotImplementedException("Unary aggregate " + op.aggOp.increOp.fn + " not supported yet.");
}
+ if(overlappingColGroups &&
+ (op.aggOp.increOp.fn instanceof KahanPlusSq || (op.aggOp.increOp.fn instanceof Builtin &&
+ (((Builtin) op.aggOp.increOp.fn).getBuiltinCode() == BuiltinCode.MIN ||
+ ((Builtin) op.aggOp.increOp.fn).getBuiltinCode() == BuiltinCode.MAX)))) {
+ MatrixBlock m1d = decompress(op.getNumThreads());
+ return m1d.aggregateUnaryOperations(op, result, blen, indexesIn, inCP);
+ }
+
// prepare output dimensions
CellIndex tempCellIndex = new CellIndex(-1, -1);
op.indexFn.computeDimension(rlen, clen, tempCellIndex);
@@ -697,6 +622,7 @@
(op.indexFn instanceof ReduceCol) ? 1 : op.getNumThreads(),
false);
ColGroupUncompressed uc = getUncompressedColGroup();
+
try {
// compute uncompressed column group in parallel (otherwise bottleneck)
if(uc != null)
@@ -705,14 +631,20 @@
ExecutorService pool = CommonThreadPool.get(op.getNumThreads());
ArrayList<UnaryAggregateTask> tasks = new ArrayList<>();
if(op.indexFn instanceof ReduceCol && grpParts.length > 0) {
- int blklen = getAlignedBlockSize((int) (Math.ceil((double) rlen / op.getNumThreads())));
- for(int i = 0; i < op.getNumThreads() & i * blklen < rlen; i++)
+ final int blkz = CompressionSettings.BITMAP_BLOCK_SZ;
+ int blklen = (int) Math.ceil((double) rlen / op.getNumThreads());
+ blklen += (blklen % blkz != 0) ? blkz - blklen % blkz : 0;
+ for(int i = 0; i < op.getNumThreads() & i * blklen < rlen; i++) {
tasks.add(
new UnaryAggregateTask(grpParts[0], ret, i * blklen, Math.min((i + 1) * blklen, rlen), op));
+
+ }
}
else
- for(ArrayList<ColGroup> grp : grpParts)
- tasks.add(new UnaryAggregateTask(grp, ret, 0, rlen, op));
+ for(ArrayList<ColGroup> grp : grpParts) {
+ if(grp != null)
+ tasks.add(new UnaryAggregateTask(grp, ret, 0, rlen, op));
+ }
List<Future<MatrixBlock>> rtasks = pool.invokeAll(tasks);
pool.shutdown();
@@ -750,11 +682,13 @@
}
}
else {
- for(ColGroup grp : _colGroups)
- if(grp instanceof ColGroupUncompressed)
- ((ColGroupUncompressed) grp).unaryAggregateOperations(op, ret);
+ if(_colGroups != null) {
- aggregateUnaryOperations(op, _colGroups, ret, 0, rlen);
+ for(ColGroup grp : _colGroups)
+ if(grp instanceof ColGroupUncompressed)
+ ((ColGroupUncompressed) grp).unaryAggregateOperations(op, ret);
+ aggregateUnaryOperations(op, _colGroups, ret, 0, rlen);
+ }
}
// special handling zeros for rowmins/rowmax
@@ -806,50 +740,18 @@
// note: UC group never passed into this function
double[] c = ret.getDenseBlockValues();
for(ColGroup grp : groups)
- if(!(grp instanceof ColGroupUncompressed))
+ if(grp != null && !(grp instanceof ColGroupUncompressed))
grp.unaryAggregateOperations(op, c, rl, ru);
}
@Override
public MatrixBlock transposeSelfMatrixMultOperations(MatrixBlock out, MMTSJType tstype) {
-
- Timing time = LOG.isDebugEnabled() ? new Timing(true) : null;
-
- // check for transpose type
- if(tstype != MMTSJType.LEFT) // right not supported yet
- throw new DMLRuntimeException("Invalid MMTSJ type '" + tstype.toString() + "'.");
-
- // create output matrix block
- if(out == null)
- out = new MatrixBlock(clen, clen, false);
- else
- out.reset(clen, clen, false);
- out.allocateDenseBlock();
-
- if(!isEmptyBlock(false)) {
- // compute matrix mult
- leftMultByTransposeSelf(_colGroups, out, 0, _colGroups.size());
-
- // post-processing
- out.setNonZeros(LinearAlgebraUtils.copyUpperToLowerTriangle(out));
- }
-
- if(LOG.isDebugEnabled())
- LOG.debug("Compressed TSMM in " + time.stop());
-
- return out;
+ return transposeSelfMatrixMultOperations(out, tstype, 1);
}
@Override
public MatrixBlock transposeSelfMatrixMultOperations(MatrixBlock out, MMTSJType tstype, int k) {
-
- if(k <= 1) {
- return transposeSelfMatrixMultOperations(out, tstype);
- }
-
- Timing time = LOG.isDebugEnabled() ? new Timing(true) : null;
-
// check for transpose type
if(tstype != MMTSJType.LEFT) // right not supported yet
throw new DMLRuntimeException("Invalid MMTSJ type '" + tstype.toString() + "'.");
@@ -863,658 +765,20 @@
if(!isEmptyBlock(false)) {
// compute matrix mult
- try {
- ExecutorService pool = CommonThreadPool.get(k);
- ArrayList<MatrixMultTransposeTask> tasks = new ArrayList<>();
- int numgrp = _colGroups.size();
- int blklen = (int) (Math.ceil((double) numgrp / (2 * k)));
- for(int i = 0; i < 2 * k & i * blklen < clen; i++)
- tasks.add(
- new MatrixMultTransposeTask(_colGroups, out, i * blklen, Math.min((i + 1) * blklen, numgrp)));
- List<Future<Object>> ret = pool.invokeAll(tasks);
- for(Future<Object> tret : ret)
- tret.get(); // check for errors
- pool.shutdown();
- }
- catch(InterruptedException | ExecutionException e) {
- throw new DMLRuntimeException(e);
- }
-
+ LibLeftMultBy.leftMultByTransposeSelf(_colGroups,
+ out,
+ 0,
+ _colGroups.size(),
+ k,
+ getNumColumns(),
+ getMaxNumValues(),
+ isOverlapping());
// post-processing
out.setNonZeros(LinearAlgebraUtils.copyUpperToLowerTriangle(out));
}
-
- if(LOG.isDebugEnabled())
- LOG.debug("Compressed TSMM k=" + k + " in " + time.stop());
-
return out;
}
- /**
- * Multiply this matrix block by a column vector on the right.
- *
- * @param vector right-hand operand of the multiplication
- * @param result buffer to hold the result; must have the appropriate size already
- */
- private void rightMultByVector(MatrixBlock vector, MatrixBlock result) {
- // initialize and allocate the result
- result.allocateDenseBlock();
-
- // delegate matrix-vector operation to each column group
- rightMultByVector(_colGroups, vector, result, 0, result.getNumRows());
-
- // post-processing
- result.recomputeNonZeros();
- }
-
- /**
- * Multi-threaded version of rightMultByVector.
- *
- * @param vector matrix block vector
- * @param result matrix block result
- * @param k number of threads
- */
- private void rightMultByVector(MatrixBlock vector, MatrixBlock result, int k) {
- // initialize and allocate the result
- result.allocateDenseBlock();
-
- // multi-threaded execution of all groups
- try {
- // ColGroupUncompressed uc = getUncompressedColGroup();
-
- // compute uncompressed column group in parallel
- // if(uc != null)
- // uc.rightMultByVector(vector, result, k);
-
- // compute remaining compressed column groups in parallel
- // note: OLE needs alignment to segment size, otherwise wrong entry
- ExecutorService pool = CommonThreadPool.get(k);
- int rlen = getNumRows();
- int seqsz = CompressionSettings.BITMAP_BLOCK_SZ;
- int blklen = (int) (Math.ceil((double) rlen / k));
- blklen += (blklen % seqsz != 0) ? seqsz - blklen % seqsz : 0;
-
- ArrayList<RightMatrixVectorMultTask> tasks = new ArrayList<>();
- for(int i = 0; i < k & i * blklen < getNumRows(); i++) {
- tasks.add(new RightMatrixVectorMultTask(_colGroups, vector, result, i * blklen,
- Math.min((i + 1) * blklen, rlen)));
- }
-
- List<Future<Long>> ret = pool.invokeAll(tasks);
- pool.shutdown();
-
- // error handling and nnz aggregation
- long lnnz = 0;
- for(Future<Long> tmp : ret)
- lnnz += tmp.get();
- result.setNonZeros(lnnz);
- }
- catch(InterruptedException | ExecutionException e) {
- throw new DMLRuntimeException(e);
- }
-
- }
-
- private static void rightMultByVector(List<ColGroup> groups, MatrixBlock vect, MatrixBlock ret, int rl, int ru) {
- // + 1 to enable containing a single 0 value in the dictionary that was not materialized.
- // This is to handle the case of a DDC dictionary not materializing the zero values.
- // A fine tradeoff!
- ColGroupValue.setupThreadLocalMemory(getMaxNumValues(groups).getLeft() + 1);
-
- // boolean cacheDDC1 = ru - rl > CompressionSettings.BITMAP_BLOCK_SZ * 2;
-
- // process uncompressed column group (overwrites output)
- // if(inclUC) {
- for(ColGroup grp : groups) {
- if(grp instanceof ColGroupUncompressed)
- ((ColGroupUncompressed) grp).rightMultByVector(vect, ret, rl, ru);
- }
-
- // process cache-conscious DDC1 groups (adds to output)
-
- // if(cacheDDC1) {
- // ArrayList<ColGroupDDC1> tmp = new ArrayList<>();
- // for(ColGroup grp : groups)
- // if(grp instanceof ColGroupDDC1)
- // tmp.add((ColGroupDDC1) grp);
- // if(!tmp.isEmpty())
- // ColGroupDDC1.rightMultByVector(tmp.toArray(new ColGroupDDC1[0]), vect, ret, rl, ru);
- // }
- // process remaining groups (adds to output)
- double[] values = ret.getDenseBlockValues();
- for(ColGroup grp : groups) {
- if(!(grp instanceof ColGroupUncompressed)) {
- grp.rightMultByVector(vect.getDenseBlockValues(), values, rl, ru, grp.getValues());
- }
- }
-
- ColGroupValue.cleanupThreadLocalMemory();
-
- }
-
- /**
- * Multiply this matrix block by the transpose of a column vector (i.e. t(v)%*%X)
- *
- * @param colGroups list of column groups
- * @param vector left-hand operand of the multiplication
- * @param result buffer to hold the result; must have the appropriate size already
- * @param doTranspose if true, transpose vector
- */
- private static MatrixBlock leftMultByVectorTranspose(List<ColGroup> colGroups, MatrixBlock vector,
- MatrixBlock result, boolean doTranspose, boolean allocTmp) {
-
- MatrixBlock rowVector = vector;
- // Note that transpose here is a metadata operation since the input is a vector.
- if(doTranspose) {
- rowVector = new MatrixBlock(1, vector.getNumRows(), false);
- LibMatrixReorg.transpose(vector, rowVector);
- }
-
- // initialize and allocate the result
- result.reset();
- result.allocateDenseBlock();
-
- // setup memory pool for reuse
- if(allocTmp) {
- Pair<Integer, int[]> v = getMaxNumValues(colGroups);
- ColGroupValue.setupThreadLocalMemory(v.getLeft() + 1); // +1 for efficiency in DDC groups.
- for(int i = 0; i < colGroups.size(); i++) {
- colGroups.get(i).leftMultByRowVector(rowVector.getDenseBlockValues(),
- result.getDenseBlockValues(),
- v.getRight()[i]);
- }
- }
- else {
- for(ColGroup grp : colGroups) {
- grp.leftMultByRowVector(rowVector.getDenseBlockValues(), result.getDenseBlockValues(), -1);
- }
- }
-
- // delegate matrix-vector operation to each column group
-
- // post-processing
- if(allocTmp)
- ColGroupValue.cleanupThreadLocalMemory();
- result.recomputeNonZeros();
-
- return result;
- }
-
- // private static void leftMultByVectorTranspose(List<ColGroup> colGroups, ColGroupDDC vector, MatrixBlock result) {
- // // initialize and allocate the result
- // result.reset();
- // // delegate matrix-vector operation to each column group
- // for(ColGroup grp : colGroups)
- // grp.leftMultByRowVector(vector, result);
- // // post-processing
- // result.recomputeNonZeros();
- // }
-
- /**
- * Multi-thread version of leftMultByVectorTranspose.
- *
- * @param colGroups list of column groups
- * @param vector left-hand operand of the multiplication
- * @param result buffer to hold the result; must have the appropriate size already
- * @param doTranspose if true, transpose vector
- * @param k number of threads
- */
- private MatrixBlock leftMultByVectorTranspose(List<ColGroup> colGroups, MatrixBlock vector, MatrixBlock result,
- boolean doTranspose, int k) {
- // transpose vector if required
- MatrixBlock rowVector = vector;
- if(doTranspose) {
- rowVector = new MatrixBlock(1, vector.getNumRows(), false);
- LibMatrixReorg.transpose(vector, rowVector);
- }
-
- // initialize and allocate the result
- result.reset();
- result.allocateDenseBlock();
-
- // multi-threaded execution
- try {
- // compute uncompressed column group in parallel
- // ColGroupUncompressed uc = getUncompressedColGroup();
- // if(uc != null)
- // uc.leftMultByRowVector(rowVector, result, k);
-
- // compute remaining compressed column groups in parallel
- ExecutorService pool = CommonThreadPool.get(Math.min(colGroups.size(), k));
- ArrayList<ColGroup>[] grpParts = createStaticTaskPartitioning(_colGroups, 4 * k, true);
- ArrayList<LeftMatrixVectorMultTask> tasks = new ArrayList<>();
- for(ArrayList<ColGroup> groups : grpParts)
- tasks.add(new LeftMatrixVectorMultTask(groups, rowVector, result));
- List<Future<Object>> ret;
-
- ret = pool.invokeAll(tasks);
-
- pool.shutdown();
- for(Future<Object> tmp : ret)
- tmp.get();
-
- }
- catch(InterruptedException | ExecutionException e) {
- LOG.error(e);
- throw new DMLRuntimeException(e);
- }
-
- // post-processing
- result.recomputeNonZeros();
- return result;
- }
-
- /**
- * Multiply this matrix block by a matrix (i.e. v%*%X)
- *
- * @param colGroups List of column groups
- * @param that Left-hand operand of the multiplication
- * @param ret The result matrix to insert the results
- * @param tmp buffer to hold the result; must have the appropriate size already
- * @param tmpIn buffer to hold a since row of input.
- * @param k The number of threads used
- * @param numColumns The number of columns in this colGroup
- */
- private static MatrixBlock leftMultByMatrix(List<ColGroup> colGroups, MatrixBlock that, MatrixBlock ret, int k,
- int numColumns) {
- ret.allocateDenseBlock();
- if(that.isInSparseFormat()) {
- ret = leftMultBySparseMatrix(colGroups, that, ret, k, numColumns);
- }
- else {
- ret = leftMultByDenseMatrix(colGroups, that, ret, k, numColumns);
- }
-
- ret.setNonZeros(ret.getNumColumns() * ret.getNumRows());
- return ret;
- }
-
- private static MatrixBlock rightMultByMatrix(List<ColGroup> colGroups, MatrixBlock that, MatrixBlock ret, int k,
- int numColumns) {
- ret.allocateDenseBlock();
-
- if(that.isInSparseFormat()) {
- ret = rightMultBySparseMatrix(colGroups, that, ret, k, numColumns);
- }
- else {
- ret = rightMultByDenseMatrix(colGroups, that, ret, k, numColumns);
-
- }
- ret.setNonZeros(ret.getNumColumns() * ret.getNumRows());
- return ret;
-
- }
-
- private static MatrixBlock leftMultByDenseMatrix(List<ColGroup> colGroups, MatrixBlock that, MatrixBlock ret, int k,
- int numColumns) {
- DenseBlock db = that.getDenseBlock();
- if(db == null)
- throw new DMLRuntimeException("Invalid LeftMult By Dense matrix, input matrix was sparse");
-
- double[] retV = ret.getDenseBlockValues();
- double[] thatV;
- int blockU;
- int blockL = 0;
- for(ColGroup grp : colGroups)
- if(grp instanceof ColGroupUncompressed)
- ((ColGroupUncompressed) grp).leftMultByMatrix(that, ret);
-
- for(int b = 0; b < db.numBlocks(); b++) {
- int blockSize = db.blockSize(b);
- blockU = Math.min(blockL + blockSize, ret.getNumRows());
- thatV = db.valuesAt(b);
-
- if(k == 1) {
- Pair<Integer, int[]> v = getMaxNumValues(colGroups);
- for(int j = 0; j < colGroups.size(); j++) {
- colGroups.get(j).leftMultByMatrix(thatV,
- retV,
- v.getRight()[j],
- colGroups.get(j).getValues(),
- that.getNumRows(),
- ret.getNumColumns(),
- 0,
- ret.getNumRows(),
- 0);
- }
- }
- else {
- try {
- ExecutorService pool = CommonThreadPool.get(k);
- // compute remaining compressed column groups in parallel
- ArrayList<LeftMatrixMatrixMultTask> tasks = new ArrayList<>();
- List<ColGroup>[] parts = createStaticTaskPartitioningForMatrixMult(colGroups, k, false);
- int rowBlockSize = 10;
- for(List<ColGroup> part : parts) {
- for(int blo = blockL; blo < blockU; blo += rowBlockSize) {
- tasks.add(new LeftMatrixMatrixMultTask(part, thatV, retV, that.getNumRows(), numColumns,
- blo, Math.min(blo + rowBlockSize, blockU), blo - blockL));
- }
- }
-
- List<Future<Object>> futures = pool.invokeAll(tasks);
-
- pool.shutdown();
- for(Future<Object> future : futures)
- future.get();
- }
- catch(InterruptedException | ExecutionException e) {
- throw new DMLRuntimeException(e);
- }
- }
- blockL += blockSize;
- }
- return ret;
- }
-
- private static MatrixBlock leftMultBySparseMatrix(List<ColGroup> colGroups, MatrixBlock that, MatrixBlock ret,
- int k, int numColumns) {
-
- SparseBlock sb = that.getSparseBlock();
- if(sb == null)
- throw new DMLRuntimeException("Invalid Left Mult by Sparse matrix, input matrix was dense");
-
- for(ColGroup grp : colGroups) {
- if(grp instanceof ColGroupUncompressed)
- ((ColGroupUncompressed) grp).leftMultByMatrix(that, ret);
- }
-
- if(k == 1) {
- double[][] materialized = new double[colGroups.size()][];
- boolean containsOLE = false;
- for(int i = 0; i < colGroups.size(); i++) {
- materialized[i] = colGroups.get(i).getValues();
- if(colGroups.get(i) instanceof ColGroupOLE) {
- containsOLE = true;
- }
- }
- double[] materializedRow = containsOLE ? new double[CompressionSettings.BITMAP_BLOCK_SZ * 2] : null;
-
- Pair<Integer, int[]> v = getMaxNumValues(colGroups);
- for(int r = 0; r < that.getNumRows(); r++) {
- SparseRow row = sb.get(r);
- if(row != null) {
-
- for(int j = 0; j < colGroups.size(); j++) {
- colGroups.get(j).leftMultBySparseMatrix(row.size(),
- row.indexes(),
- row.values(),
- ret.getDenseBlockValues(),
- v.getRight()[j],
- materialized[j],
- that.getNumRows(),
- ret.getNumColumns(),
- r,
- materializedRow);
- }
- }
- }
- }
- else {
- ExecutorService pool = CommonThreadPool.get(k);
- ArrayList<LeftMatrixSparseMatrixMultTask> tasks = new ArrayList<>();
- try {
- // compute remaining compressed column groups in parallel
- List<ColGroup>[] parts = createStaticTaskPartitioningForSparseMatrixMult(colGroups, k, false);
- for(List<ColGroup> part : parts) {
- tasks.add(new LeftMatrixSparseMatrixMultTask(part, sb, ret.getDenseBlockValues(), that.getNumRows(),
- numColumns));
- }
-
- List<Future<Object>> futures = pool.invokeAll(tasks);
- pool.shutdown();
- for(Future<Object> future : futures)
- future.get();
- }
- catch(InterruptedException | ExecutionException e) {
- throw new DMLRuntimeException(e);
- }
- }
-
- return ret;
-
- }
-
- private static MatrixBlock rightMultByDenseMatrix(List<ColGroup> colGroups, MatrixBlock that, MatrixBlock ret,
- int k, int numColumns) {
-
- // long StartTime = System.currentTimeMillis();
- DenseBlock db = that.getDenseBlock();
- double[] retV = ret.getDenseBlockValues();
- double[] thatV;
-
- for(ColGroup grp : colGroups) {
- if(grp instanceof ColGroupUncompressed) {
- ((ColGroupUncompressed) grp).rightMultByMatrix(that, ret, 0, ret.getNumRows());
- }
- }
-
- if(k == 1) {
- Pair<Integer, int[]> v = getMaxNumValues(colGroups);
- ColGroupValue.setupThreadLocalMemory((v.getLeft()) * that.getNumColumns());
- for(int b = 0; b < db.numBlocks(); b++) {
- // int blockSize = db.blockSize(b);
- thatV = db.valuesAt(b);
- for(int j = 0; j < colGroups.size(); j++) {
- int colBlockSize = 128;
- for(int i = 0; i < that.getNumColumns(); i += colBlockSize) {
- if(colGroups.get(j) instanceof ColGroupValue) {
- double[] preAggregatedB = ((ColGroupValue) colGroups.get(j)).preaggValues(v.getRight()[j],
- thatV,
- colGroups.get(j).getValues(),
- i,
- Math.min(i + colBlockSize, that.getNumColumns()),
- that.getNumColumns());
- int blklenRows = CompressionSettings.BITMAP_BLOCK_SZ;
- for(int n = 0; n * blklenRows < ret.getNumRows(); n++) {
- colGroups.get(j).rightMultByMatrix(preAggregatedB,
- retV,
- numColumns,
- n * blklenRows,
- Math.min((n + 1) * blklenRows, ret.getNumRows()),
- i,
- Math.min(i + colBlockSize, that.getNumColumns()));
- }
- }
- }
- }
- }
- ColGroupValue.cleanupThreadLocalMemory();
- }
- else {
- // for(int b = 0; b < db.numBlocks(); b++) {
- // compute remaining compressed column groups in parallel
- // int blockSize = db.blockSize(b);
- // int blockUCols = that.getNumColumns();
- thatV = db.valuesAt(0);
- ExecutorService pool = CommonThreadPool.get(k);
- ArrayList<RightMatrixMultTask> tasks = new ArrayList<>();
- ArrayList<RightMatrixPreAggregateTask> preTask = new ArrayList<>(colGroups.size());
- Pair<Integer, int[]> v;
- final int blkz = CompressionSettings.BITMAP_BLOCK_SZ;
- int blklenRows = (int) (Math.ceil((double) ret.getNumRows() / (4 * k)));
-
- List<ColGroup> ddcGroups = new ArrayList<>();
- List<ColGroup> oleGroups = new ArrayList<>();
- List<ColGroup> rleGroups = new ArrayList<>();
- for(ColGroup g : colGroups) {
- if(g instanceof ColGroupDDC) {
- ddcGroups.add(g);
- }
- else if(g instanceof ColGroupOLE) {
- oleGroups.add(g);
- }
- else if(g instanceof ColGroupRLE) {
- rleGroups.add(g);
- }
- }
-
- try {
- // Process DDC Groups!
- // int blklenRows = CompressionSettings.BITMAP_BLOCK_SZ;
- v = getMaxNumValues(ddcGroups);
- List<Future<double[]>> ag = pool.invokeAll(preAggregate(ddcGroups, thatV, that, preTask, v));
-
- for(int j = 0; j * blklenRows < ret.getNumRows(); j++) {
- RightMatrixMultTask rmmt = new RightMatrixMultTask(ddcGroups, retV, ag, v, numColumns,
- j * blklenRows, Math.min((j + 1) * blklenRows, ret.getNumRows()), 0, that.getNumColumns(),
- false);
- tasks.add(rmmt);
- }
- for(Future<Object> future : pool.invokeAll(tasks))
- future.get();
- tasks.clear();
-
- // Process RLE Groups
- blklenRows += (blklenRows % blkz != 0) ? blkz - blklenRows % blkz : 0;
- v = getMaxNumValues(rleGroups);
- preTask = preAggregate(rleGroups, thatV, that, preTask, v);
- for(int j = 0; j * blklenRows < ret.getNumRows(); j++) {
- RightMatrixMultTask rmmt = new RightMatrixMultTask(rleGroups, retV, pool.invokeAll(preTask), v,
- numColumns, j * blklenRows, Math.min((j + 1) * blklenRows, ret.getNumRows()), 0,
- that.getNumColumns(), true);
- tasks.add(rmmt);
- }
-
- for(Future<Object> future : pool.invokeAll(tasks))
- future.get();
- tasks.clear();
-
- // Process OLE Groups
- // blklenRows += (blklenRows % blkz != 0) ? blkz - blklenRows % blkz : 0;
-
- v = getMaxNumValues(oleGroups);
- preTask = preAggregate(oleGroups, thatV, that, preTask, v);
- for(int j = 0; j * blklenRows < ret.getNumRows(); j++) {
- RightMatrixMultTask rmmt = new RightMatrixMultTask(oleGroups, retV, pool.invokeAll(preTask), v,
- numColumns, j * blklenRows, Math.min((j + 1) * blklenRows, ret.getNumRows()), 0,
- that.getNumColumns(), true);
- tasks.add(rmmt);
- }
- for(Future<Object> future : pool.invokeAll(tasks))
- future.get();
- pool.shutdown();
- }
- catch(InterruptedException | ExecutionException e) {
- throw new DMLRuntimeException(e);
- }
- // }
- }
-
- return ret;
- }
-
- private static ArrayList<RightMatrixPreAggregateTask> preAggregate(List<ColGroup> colGroups, double[] thatV,
- MatrixBlock that, ArrayList<RightMatrixPreAggregateTask> preTask, Pair<Integer, int[]> v) {
- preTask.clear();
- for(int h = 0; h < colGroups.size(); h++) {
- RightMatrixPreAggregateTask pAggT = new RightMatrixPreAggregateTask((ColGroupValue) colGroups.get(h),
- v.getRight()[h], thatV, colGroups.get(h).getValues(), 0, that.getNumColumns(), that.getNumColumns());
- preTask.add(pAggT);
- }
- return preTask;
- }
-
- private static MatrixBlock rightMultBySparseMatrix(List<ColGroup> colGroups, MatrixBlock that, MatrixBlock ret,
- int k, int numColumns) {
- SparseBlock sb = that.getSparseBlock();
- double[] retV = ret.getDenseBlockValues();
-
- if(sb == null)
- throw new DMLRuntimeException("Invalid Right Mult by Sparse matrix, input matrix was dense");
-
- for(ColGroup grp : colGroups) {
- if(grp instanceof ColGroupUncompressed)
- ((ColGroupUncompressed) grp).rightMultByMatrix(that, ret, 0, ret.getNumColumns());
- }
-
- Pair<Integer, int[]> v = getMaxNumValues(colGroups);
- // if(k == 1) {
- for(int j = 0; j < colGroups.size(); j++) {
- double[] preAggregatedB = ((ColGroupValue) colGroups.get(j)).preaggValues(v.getRight()[j],
- sb,
- colGroups.get(j).getValues(),
- 0,
- that.getNumColumns(),
- that.getNumColumns());
- colGroups.get(j)
- .rightMultByMatrix(preAggregatedB, retV, numColumns, 0, ret.getNumRows(), 0, that.getNumColumns());
-
- }
- // }
- // else {
- // ExecutorService pool = CommonThreadPool.get(k);
- // ArrayList<RightMultBySparseMatrixTask> tasks = new ArrayList<>();
- // try {
-
- // for(int j = 0; j < ret.getNumColumns(); j += CompressionSettings.BITMAP_BLOCK_SZ) {
- // tasks.add(new RightMultBySparseMatrixTask(colGroups, retV, sb, materialized, v, numColumns, j,
- // Math.min(j + CompressionSettings.BITMAP_BLOCK_SZ, ret.getNumColumns())));
- // }
-
- // List<Future<Object>> futures = pool.invokeAll(tasks);
- // pool.shutdown();
- // for(Future<Object> future : futures)
- // future.get();
- // }
- // catch(InterruptedException | ExecutionException e) {
- // throw new DMLRuntimeException(e);
- // }
- // }
-
- return ret;
- }
-
- private static void leftMultByTransposeSelf(List<ColGroup> groups, MatrixBlock result, int gl, int gu) {
- final int numRows = groups.get(0).getNumRows();
- final int numGroups = groups.size();
- // final boolean containsUC = containsUncompressedColGroup(groups);
-
- // preallocated dense tmp matrix blocks
- MatrixBlock lhs = new MatrixBlock(1, numRows, false);
- MatrixBlock tmpret = new MatrixBlock(1, result.getNumColumns(), false);
- lhs.allocateDenseBlock();
- tmpret.allocateDenseBlock();
-
- // setup memory pool for reuse
- ColGroupValue.setupThreadLocalMemory(getMaxNumValues(groups).getLeft() + 1);
-
- // approach: for each colgroup, extract uncompressed columns one at-a-time
- // vector-matrix multiplies against remaining col groups
- for(int i = gl; i < gu; i++) {
- // get current group and relevant col groups
- ColGroup group = groups.get(i);
- int[] ixgroup = group.getColIndices();
- List<ColGroup> tmpList = groups.subList(i, numGroups);
-
- // if(group instanceof ColGroupDDC // single DDC group
- // && ixgroup.length == 1 && !containsUC && numRows < CompressionSettings.BITMAP_BLOCK_SZ) {
- // // compute vector-matrix partial result
- // leftMultByVectorTranspose(tmpList, (ColGroupDDC) group, tmpret);
-
- // // write partial results (disjoint non-zeros)
- // LinearAlgebraUtils.copyNonZerosToUpperTriangle(result, tmpret, ixgroup[0]);
- // }
- // else {
- // for all uncompressed lhs columns vectors
- for(int j = 0; j < ixgroup.length; j++) {
- group.decompressToBlock(lhs, j);
-
- if(!lhs.isEmptyBlock(false)) {
- // compute vector-matrix partial result
- leftMultByVectorTranspose(tmpList, lhs, tmpret, false, false);
-
- // write partial results (disjoint non-zeros)
- LinearAlgebraUtils.copyNonZerosToUpperTriangle(result, tmpret, ixgroup[j]);
- }
- }
- // }
- }
-
- // post processing
- ColGroupValue.cleanupThreadLocalMemory();
- }
-
@SuppressWarnings("unchecked")
private static ArrayList<ColGroup>[] createStaticTaskPartitioning(List<ColGroup> colGroups, int k,
boolean inclUncompressed) {
@@ -1540,76 +804,6 @@
return grpParts;
}
- @SuppressWarnings("unchecked")
- private static List<ColGroup>[] createStaticTaskPartitioningForMatrixMult(List<ColGroup> colGroups, int k,
- boolean inclUncompressed) {
- int numTasks = Math.min(k, colGroups.size());
- List<ColGroup>[] grpParts = new ArrayList[numTasks];
- int pos = 0;
- for(int i = 0; i < numTasks; i++) {
- grpParts[pos++] = new ArrayList<>();
- }
- pos = 0;
- for(ColGroup grp : colGroups) {
-
- if(grp instanceof ColGroupDDC) {
- grpParts[pos].add((ColGroupDDC) grp);
- pos = (pos == numTasks - 1) ? 0 : pos + 1;
- }
- }
- for(ColGroup grp : colGroups) {
- if(!(grp instanceof ColGroupDDC) && (inclUncompressed || !(grp instanceof ColGroupUncompressed))) {
- grpParts[pos].add(grp);
- pos = (pos == numTasks - 1) ? 0 : pos + 1;
- }
- }
-
- return grpParts;
- }
-
- @SuppressWarnings("unchecked")
- private static List<ColGroup>[] createStaticTaskPartitioningForSparseMatrixMult(List<ColGroup> colGroups, int k,
- boolean inclUncompressed) {
- int numTasks = Math.min(k, colGroups.size());
- List<ColGroup>[] grpParts = new ArrayList[numTasks];
- int pos = 0;
- for(int i = 0; i < numTasks; i++) {
- grpParts[pos++] = new ArrayList<>();
- }
- pos = 0;
- for(ColGroup grp : colGroups) {
-
- if(grp instanceof ColGroupOLE) {
- grpParts[pos].add((ColGroupOLE) grp);
- pos = (pos == numTasks - 1) ? 0 : pos + 1;
- }
- }
- for(ColGroup grp : colGroups) {
- if(!(grp instanceof ColGroupOLE) && (inclUncompressed || !(grp instanceof ColGroupUncompressed))) {
- grpParts[pos].add(grp);
- pos = (pos == numTasks - 1) ? 0 : pos + 1;
- }
- }
-
- return grpParts;
- }
-
- private static Pair<Integer, int[]> getMaxNumValues(List<ColGroup> groups) {
- int numVals = 1;
- int[] numValues = new int[groups.size()];
- int nr;
- for(int i = 0; i < groups.size(); i++)
- if(groups.get(i) instanceof ColGroupValue) {
- nr = ((ColGroupValue) groups.get(i)).getNumValues();
- numValues[i] = nr;
- numVals = Math.max(numVals, nr);
- }
- else {
- numValues[i] = -1;
- }
- return new ImmutablePair<>(numVals, numValues);
- }
-
public boolean hasUncompressedColGroup() {
return getUncompressedColGroup() != null;
}
@@ -1622,282 +816,26 @@
return null;
}
- private static class LeftMatrixVectorMultTask implements Callable<Object> {
- private final ArrayList<ColGroup> _groups;
- private final MatrixBlock _vect;
- private final MatrixBlock _ret;
+ public Pair<Integer, int[]> getMaxNumValues() {
+ if(v == null) {
- protected LeftMatrixVectorMultTask(ArrayList<ColGroup> groups, MatrixBlock vect, MatrixBlock ret) {
- _groups = groups;
- _vect = vect;
- _ret = ret;
- }
-
- @Override
- public Object call() {
- // setup memory pool for reuse
- try {
- Pair<Integer, int[]> v = getMaxNumValues(_groups);
- ColGroupValue.setupThreadLocalMemory(v.getLeft() + 1);
- for(int i = 0; i < _groups.size(); i++) {
- _groups.get(i)
- .leftMultByRowVector(_vect.getDenseBlockValues(), _ret.getDenseBlockValues(), v.getRight()[i]);
+ int numVals = 1;
+ int[] numValues = new int[_colGroups.size()];
+ int nr;
+ for(int i = 0; i < _colGroups.size(); i++)
+ if(_colGroups.get(i) instanceof ColGroupValue) {
+ nr = ((ColGroupValue) _colGroups.get(i)).getNumValues();
+ numValues[i] = nr;
+ numVals = Math.max(numVals, nr);
}
-
- ColGroupValue.cleanupThreadLocalMemory();
- }
- catch(Exception e) {
- throw new DMLRuntimeException(e);
- }
- return null;
- }
- }
-
- private static class LeftMatrixMatrixMultTask implements Callable<Object> {
- private final List<ColGroup> _group;
- private final double[] _that;
- private final double[] _ret;
- private final int _numRows;
- private final int _numCols;
- private final int _rl;
- private final int _ru;
- private final int _vOff;
-
- protected LeftMatrixMatrixMultTask(List<ColGroup> group, double[] that, double[] ret, int numRows, int numCols,
- int rl, int ru, int vOff) {
- _group = group;
- _that = that;
- _ret = ret;
- _numRows = numRows;
- _numCols = numCols;
- _rl = rl;
- _ru = ru;
- _vOff = vOff;
- }
-
- @Override
- public Object call() {
- // setup memory pool for reuse
-
- double[][] materialized = new double[_group.size()][];
- for(int i = 0; i < _group.size(); i++) {
- materialized[i] = _group.get(i).getValues();
- }
- Pair<Integer, int[]> v = getMaxNumValues(_group);
- try {
- ColGroupValue.setupThreadLocalMemory(v.getLeft() + 1);
- for(int j = 0; j < _group.size(); j++) {
- _group.get(j).leftMultByMatrix(_that,
- _ret,
- v.getRight()[j],
- materialized[j],
- _numRows,
- _numCols,
- _rl,
- _ru,
- _vOff);
+ else {
+ numValues[i] = -1;
}
- ColGroupValue.cleanupThreadLocalMemory();
-
- }
- catch(Exception e) {
- throw new DMLRuntimeException(e);
- }
- return null;
+ v = new ImmutablePair<>(numVals, numValues);
+ return v;
}
- }
-
- private static class LeftMatrixSparseMatrixMultTask implements Callable<Object> {
- private final List<ColGroup> _group;
- private final SparseBlock _that;
- private final double[] _ret;
- private final int _numRows;
- private final int _numCols;
-
- protected LeftMatrixSparseMatrixMultTask(List<ColGroup> group, SparseBlock that, double[] ret, int numRows,
- int numCols) {
- _group = group;
- _that = that;
- _ret = ret;
- _numRows = numRows;
- _numCols = numCols;
- }
-
- @Override
- public Object call() {
- // setup memory pool for reuse
-
- // double[][] materialized = new double[_group.size()][];
- // for(int i = 0; i < _group.size(); i++) {
- // materialized[i] = _group.get(i).getValues();
- // }
-
- boolean containsOLE = false;
- for(int j = 0; j < _group.size(); j++) {
- if(_group.get(j) instanceof ColGroupOLE) {
- containsOLE = true;
- }
- }
- // Temporary Array to store 2 * block size in
- double[] tmpA = containsOLE ? new double[CompressionSettings.BITMAP_BLOCK_SZ * 2] : null;
-
- Pair<Integer, int[]> v = getMaxNumValues(_group);
- ColGroupValue.setupThreadLocalMemory(v.getLeft());
- try {
- for(int j = 0; j < _group.size(); j++) {
- double[] materializedV = _group.get(j).getValues();
- for(int r = 0; r < _that.numRows(); r++) {
- if(_that.get(r) != null) {
- _group.get(j).leftMultBySparseMatrix(_that.get(r).size(),
- _that.get(r).indexes(),
- _that.get(r).values(),
- _ret,
- v.getRight()[j],
- materializedV,
- _numRows,
- _numCols,
- r,
- tmpA);
- }
- }
- }
- }
- catch(Exception e) {
- e.printStackTrace();
- throw new DMLRuntimeException(e);
- }
- ColGroupValue.cleanupThreadLocalMemory();
- return null;
- }
- }
-
- private static class RightMatrixVectorMultTask implements Callable<Long> {
- private final List<ColGroup> _groups;
- private final MatrixBlock _vect;
- private final MatrixBlock _ret;
- private final int _rl;
- private final int _ru;
-
- protected RightMatrixVectorMultTask(List<ColGroup> groups, MatrixBlock vect, MatrixBlock ret, int rl, int ru) {
- _groups = groups;
- _vect = vect;
- _ret = ret;
- _rl = rl;
- _ru = ru;
- }
-
- @Override
- public Long call() {
- try {
- rightMultByVector(_groups, _vect, _ret, _rl, _ru);
- return _ret.recomputeNonZeros(_rl, _ru - 1, 0, 0);
- }
- catch(Exception e) {
- LOG.error(e);
- throw new DMLRuntimeException(e);
- }
- }
- }
-
- private static class RightMatrixMultTask implements Callable<Object> {
- private final List<ColGroup> _colGroups;
- // private final double[] _thatV;
- private final double[] _retV;
- private final List<Future<double[]>> _aggB;
- private final Pair<Integer, int[]> _v;
- private final int _numColumns;
-
- private final int _rl;
- private final int _ru;
- private final int _cl;
- private final int _cu;
- private final boolean _mem;
-
- protected RightMatrixMultTask(List<ColGroup> groups, double[] retV, List<Future<double[]>> aggB,
- Pair<Integer, int[]> v, int numColumns, int rl, int ru, int cl, int cu, boolean mem) {
- _colGroups = groups;
- // _thatV = thatV;
- _retV = retV;
- _aggB = aggB;
- _v = v;
- _numColumns = numColumns;
- _rl = rl;
- _ru = ru;
- _cl = cl;
- _cu = cu;
- _mem = mem;
- }
-
- @Override
- public Object call() {
- try {
- if(_mem)
- ColGroupValue.setupThreadLocalMemory((_v.getLeft()));
- for(int j = 0; j < _colGroups.size(); j++) {
- // if (_colGroups.get(j) instanceof ColGroupRLE)
- _colGroups.get(j).rightMultByMatrix(_aggB.get(j).get(), _retV, _numColumns, _rl, _ru, _cl, _cu);
- }
- if(_mem)
- ColGroupValue.cleanupThreadLocalMemory();
- return null;
- }
- catch(Exception e) {
- LOG.error(e);
- throw new DMLRuntimeException(e);
- }
- }
- }
-
- private static class RightMatrixPreAggregateTask implements Callable<double[]> {
- private final ColGroupValue _colGroup;
- private final int _numVals;
- private final double[] _b;
- private final double[] _dict;
-
- private final int _cl;
- private final int _cu;
- private final int _cut;
-
- protected RightMatrixPreAggregateTask(ColGroupValue colGroup, int numVals, double[] b, double[] dict, int cl,
- int cu, int cut) {
- _colGroup = colGroup;
- _numVals = numVals;
- _b = b;
- _dict = dict;
- _cl = cl;
- _cu = cu;
- _cut = cut;
- }
-
- @Override
- public double[] call() {
- try {
- return _colGroup.preaggValues(_numVals, _b, _dict, _cl, _cu, _cut);
- }
- catch(Exception e) {
- LOG.error(e);
- throw new DMLRuntimeException(e);
- }
- }
- }
-
- private static class MatrixMultTransposeTask implements Callable<Object> {
- private final List<ColGroup> _groups;
- private final MatrixBlock _ret;
- private final int _gl;
- private final int _gu;
-
- protected MatrixMultTransposeTask(List<ColGroup> groups, MatrixBlock ret, int gl, int gu) {
- _groups = groups;
- _ret = ret;
- _gl = gl;
- _gu = gu;
- }
-
- @Override
- public Object call() {
- leftMultByTransposeSelf(_groups, _ret, _gl, _gu);
- return null;
+ else {
+ return v;
}
}
@@ -1937,7 +875,7 @@
}
}
- private static class DecompressTask implements Callable<Object> {
+ private static class DecompressTask implements Callable<Long> {
private final List<ColGroup> _colGroups;
private final MatrixBlock _ret;
private final int _rl;
@@ -1951,9 +889,10 @@
}
@Override
- public Object call() {
+ public Long call() {
// preallocate sparse rows to avoid repeated alloc
+
if(_ret.isInSparseFormat()) {
int[] rnnz = new int[_ru - _rl];
for(ColGroup grp : _colGroups)
@@ -1971,40 +910,10 @@
if(_ret.isInSparseFormat())
_ret.sortSparseRows(_rl, _ru);
- return null;
+ return _ret.recomputeNonZeros(_rl, _ru - 1);
}
}
- private static class ScalarTask implements Callable<List<ColGroup>> {
- private final List<ColGroup> _colGroups;
- private final ScalarOperator _sop;
-
- protected ScalarTask(List<ColGroup> colGroups, ScalarOperator sop) {
- _colGroups = colGroups;
- _sop = sop;
- }
-
- @Override
- public List<ColGroup> call() {
- List<ColGroup> res = new ArrayList<>();
- for(ColGroup x : _colGroups) {
- res.add(x.scalarOperation(_sop));
- }
- return res;
- }
- }
-
- /**
- * Calculates the Aligned block size if the block is a certain length.
- *
- * @param blklen The Entered block length
- * @return The total size of aligned blocks rounded the entered value up to the next BITMAP_BLOCK_SZ
- */
- private static int getAlignedBlockSize(int blklen) {
- return blklen + ((blklen % CompressionSettings.BITMAP_BLOCK_SZ != 0) ? CompressionSettings.BITMAP_BLOCK_SZ -
- blklen % CompressionSettings.BITMAP_BLOCK_SZ : 0);
- }
-
@Override
public String toString() {
StringBuilder sb = new StringBuilder();
@@ -2015,4 +924,12 @@
}
return sb.toString();
}
-}
\ No newline at end of file
+
+ public boolean isOverlapping() {
+ return overlappingColGroups;
+ }
+
+ public void setOverlapping(boolean overlapping) {
+ overlappingColGroups = overlapping;
+ }
+}
diff --git a/src/main/java/org/apache/sysds/runtime/compress/CompressedMatrixBlockFactory.java b/src/main/java/org/apache/sysds/runtime/compress/CompressedMatrixBlockFactory.java
index ced8d63..ae29535 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/CompressedMatrixBlockFactory.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/CompressedMatrixBlockFactory.java
@@ -26,6 +26,7 @@
import org.apache.commons.lang3.tuple.Pair;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
+import org.apache.sysds.api.DMLScript;
import org.apache.sysds.runtime.DMLRuntimeException;
import org.apache.sysds.runtime.compress.cocode.PlanningCoCoder;
import org.apache.sysds.runtime.compress.colgroup.ColGroup;
@@ -33,9 +34,11 @@
import org.apache.sysds.runtime.compress.estim.CompressedSizeEstimator;
import org.apache.sysds.runtime.compress.estim.CompressedSizeEstimatorFactory;
import org.apache.sysds.runtime.compress.estim.CompressedSizeInfo;
+import org.apache.sysds.runtime.compress.utils.DblArrayIntListHashMap;
import org.apache.sysds.runtime.controlprogram.parfor.stat.Timing;
import org.apache.sysds.runtime.matrix.data.LibMatrixReorg;
import org.apache.sysds.runtime.matrix.data.MatrixBlock;
+import org.apache.sysds.utils.DMLCompressionStatistics;
/**
* Factory pattern to construct a CompressedMatrixBlock.
@@ -110,8 +113,10 @@
_stats.estimatedSizeCols = sizeInfos.memoryEstimate();
_stats.setNextTimePhase(time.stop());
- LOG.debug("Compression statistics:");
- LOG.debug("--compression phase 1: " + _stats.getLastTimePhase());
+ if(LOG.isDebugEnabled()){
+ LOG.debug("Compression statistics:");
+ LOG.debug("--compression phase 1: " + _stats.getLastTimePhase());
+ }
if(sizeInfos.colsC.isEmpty()) {
LOG.info("Abort block compression because all columns are incompressible.");
@@ -128,8 +133,10 @@
if(LOG.isDebugEnabled()) {
LOG.debug("--compression phase 2: " + _stats.getLastTimePhase());
+ StringBuilder sb = new StringBuilder();
for(int[] group : coCodeColGroups)
- LOG.debug(Arrays.toString(group));
+ sb.append(Arrays.toString(group));
+ LOG.debug(sb.toString());
}
// TODO: Make second estimate of memory usage if the ColGroups are as above?
@@ -151,6 +158,8 @@
res.allocateColGroupList(colGroupList);
_stats.setNextTimePhase(time.stop());
if(LOG.isDebugEnabled()) {
+ LOG.debug("Hash overlap count:" + DblArrayIntListHashMap.hashMissCount);
+ DblArrayIntListHashMap.hashMissCount = 0;
LOG.debug("--compression phase 3: " + _stats.getLastTimePhase());
}
// --------------------------------------------------
@@ -163,7 +172,7 @@
// applySharedDDC1Dictionary(colGroupList, dict);
// res._sharedDDC1Dict = true;
// }
- // _stats.setNextTimePhase(time.stop());
+ _stats.setNextTimePhase(time.stop());
if(LOG.isDebugEnabled()) {
LOG.debug("--compression phase 4: " + _stats.getLastTimePhase());
}
@@ -206,6 +215,9 @@
}
}
+ if (DMLScript.STATISTICS ){
+ DMLCompressionStatistics.addCompressionTimes(_stats.getTimeArrayList());
+ }
return new ImmutablePair<>(res, _stats);
// --------------------------------------------------
}
diff --git a/src/main/java/org/apache/sysds/runtime/compress/CompressionSettingsBuilder.java b/src/main/java/org/apache/sysds/runtime/compress/CompressionSettingsBuilder.java
index 2fe6642..43f45d1 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/CompressionSettingsBuilder.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/CompressionSettingsBuilder.java
@@ -40,6 +40,7 @@
private EnumSet<CompressionType> validCompressions;
private boolean sortValuesByLength = false;
private PartitionerType columnPartitioner = PartitionerType.COST;
+ // private PartitionerType columnPartitioner = PartitionerType.STATIC;
private int maxStaticColGroupCoCode = 1;
public CompressionSettingsBuilder() {
diff --git a/src/main/java/org/apache/sysds/runtime/compress/CompressionStatistics.java b/src/main/java/org/apache/sysds/runtime/compress/CompressionStatistics.java
index fc53dd1..140644e 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/CompressionStatistics.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/CompressionStatistics.java
@@ -29,7 +29,7 @@
public class CompressionStatistics {
- private ArrayList<Double> timePhases = new ArrayList<>();
+ private ArrayList<Double> timePhases = new ArrayList<>(5);
public double ratio;
public long originalSize;
public long estimatedSizeColGroups;
@@ -79,7 +79,7 @@
return colGroupCounts;
}
- public ArrayList<Double> getTimeArrayList() {
+ public List<Double> getTimeArrayList() {
return timePhases;
}
diff --git a/src/main/java/org/apache/sysds/runtime/compress/ReaderColumnSelection.java b/src/main/java/org/apache/sysds/runtime/compress/ReaderColumnSelection.java
index 79547b8..b8cdd03 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/ReaderColumnSelection.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/ReaderColumnSelection.java
@@ -19,10 +19,14 @@
package org.apache.sysds.runtime.compress;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
import org.apache.sysds.runtime.compress.utils.DblArray;
/** Base class for all column selection readers. */
public abstract class ReaderColumnSelection {
+
+ protected static final Log LOG = LogFactory.getLog(ReaderColumnSelection.class.getName());
protected int[] _colIndexes = null;
protected int _numRows = -1;
protected int _lastRow = -1;
diff --git a/src/main/java/org/apache/sysds/runtime/compress/ReaderColumnSelectionSparse.java b/src/main/java/org/apache/sysds/runtime/compress/ReaderColumnSelectionSparse.java
index 606c58a..d560a8d 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/ReaderColumnSelectionSparse.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/ReaderColumnSelectionSparse.java
@@ -37,18 +37,26 @@
private DblArray reusableReturn;
private double[] reusableArr;
+ // an empty array to return if the entire row was 0.
+ private DblArray empty = new DblArray();
+
// current sparse row positions
private SparseRow[] sparseCols = null;
private int[] sparsePos = null;
+ /**
+ * Reader of sparse matrix blocks for compression.
+ *
+ * This reader should not be used if the input data is not transposed and sparse
+ *
+ * @param data The transposed and sparse matrix
+ * @param colIndexes The column indexes to combine
+ * @param compSettings The compression settings.
+ */
public ReaderColumnSelectionSparse(MatrixBlock data, int[] colIndexes, CompressionSettings compSettings) {
super(colIndexes, compSettings.transposeInput ? data.getNumColumns() : data.getNumRows(), compSettings);
reusableArr = new double[colIndexes.length];
- reusableReturn = null;
-
- if(!_compSettings.transposeInput) {
- throw new RuntimeException("SparseColumnSelectionReader should not be used without transposed input.");
- }
+ reusableReturn = new DblArray(reusableArr);
sparseCols = new SparseRow[colIndexes.length];
sparsePos = new int[colIndexes.length];
@@ -58,13 +66,11 @@
}
protected DblArray getNextRow() {
- if(_lastRow == _numRows - 1)
- return null;
- _lastRow++;
+ if(_lastRow == _numRows - 1) {
- if(!_compSettings.transposeInput) {
- throw new RuntimeException("SparseColumnSelectionReader should not be used without transposed input.");
+ return null;
}
+ _lastRow++;
// move pos to current row if necessary (for all columns)
for(int i = 0; i < _colIndexes.length; i++)
@@ -72,7 +78,6 @@
(sparseCols[i].indexes().length <= sparsePos[i] || sparseCols[i].indexes()[sparsePos[i]] < _lastRow)) {
sparsePos[i]++;
}
-
// extract current values
Arrays.fill(reusableArr, 0);
boolean zeroResult = true;
@@ -82,7 +87,6 @@
reusableArr[i] = sparseCols[i].values()[sparsePos[i]];
zeroResult = false;
}
-
- return zeroResult ? null : reusableReturn;
+ return zeroResult ? empty : reusableReturn;
}
}
diff --git a/src/main/java/org/apache/sysds/runtime/compress/cocode/ColumnGroupPartitionerCost.java b/src/main/java/org/apache/sysds/runtime/compress/cocode/ColumnGroupPartitionerCost.java
index 24f33db..2f8221d 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/cocode/ColumnGroupPartitionerCost.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/cocode/ColumnGroupPartitionerCost.java
@@ -43,7 +43,7 @@
* of distinct rows not the total number of values. That value can be calculated by multiplying with the number of
* rows in the coCoded group.
*/
- private static final int largestDistinct = 512;
+ private static final int largestDistinct = 256;
@Override
public List<int[]> partitionColumns(List<Integer> groupCols, HashMap<Integer, GroupableColInfo> groupColsInfo,
diff --git a/src/main/java/org/apache/sysds/runtime/compress/colgroup/ADictionary.java b/src/main/java/org/apache/sysds/runtime/compress/colgroup/ADictionary.java
index efbb40e..67df821 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/colgroup/ADictionary.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/colgroup/ADictionary.java
@@ -25,6 +25,7 @@
import org.apache.sysds.runtime.functionobjects.Builtin;
import org.apache.sysds.runtime.functionobjects.KahanFunction;
+import org.apache.sysds.runtime.functionobjects.ValueFunction;
import org.apache.sysds.runtime.instructions.cp.KahanObject;
import org.apache.sysds.runtime.matrix.operators.ScalarOperator;
@@ -102,6 +103,8 @@
*/
public abstract ADictionary applyScalarOp(ScalarOperator op, double newVal, int numCols);
+ public abstract ADictionary applyBinaryRowOp(ValueFunction fn, double[] v, boolean sparseSafe, int[] colIndexes);
+
/**
* Returns a deep clone of the dictionary.
*/
@@ -196,4 +199,6 @@
protected abstract void colSum(double[] c, int[] counts, int[] colIndexes, KahanFunction kplus);
protected abstract double sum(int[] counts, int ncol, KahanFunction kplus);
+
+ public abstract StringBuilder getString(StringBuilder sb, int colIndexes);
}
diff --git a/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroup.java b/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroup.java
index 7a0e5ae..6e833b4 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroup.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroup.java
@@ -24,6 +24,7 @@
import java.io.IOException;
import java.io.Serializable;
import java.util.Iterator;
+import java.util.List;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -32,8 +33,11 @@
import org.apache.sysds.runtime.matrix.data.IJV;
import org.apache.sysds.runtime.matrix.data.MatrixBlock;
import org.apache.sysds.runtime.matrix.operators.AggregateUnaryOperator;
+import org.apache.sysds.runtime.matrix.operators.BinaryOperator;
import org.apache.sysds.runtime.matrix.operators.ScalarOperator;
+import edu.emory.mathcs.backport.java.util.Arrays;
+
/**
* Class that stores information about a column group within a compressed matrix block. There are subclasses specific to
* each compression type.
@@ -52,7 +56,7 @@
* DDC for Dense dictionary encoding
*/
public enum CompressionType {
- UNCOMPRESSED, RLE, OLE, DDC,
+ UNCOMPRESSED, RLE, OLE, DDC, CONST
}
/**
@@ -61,7 +65,7 @@
* Protected such that outside the ColGroup package it should be unknown which specific subtype is used.
*/
protected enum ColGroupType {
- UNCOMPRESSED, RLE, OLE, DDC1, DDC2,
+ UNCOMPRESSED, RLE, OLE, DDC1, DDC2, CONST
}
/** The ColGroup Indexes 0 offset, contained in the ColGroup */
@@ -125,6 +129,15 @@
}
/**
+ * Set the column indexes of the column group.
+ *
+ * @param colIndexes
+ */
+ protected void setColIndices(int[] colIndexes) {
+ _colIndexes = colIndexes;
+ }
+
+ /**
* Get number of rows contained in the ColGroup.
*
* @return An integer that is the number of rows.
@@ -187,6 +200,15 @@
*/
public abstract void decompressToBlock(MatrixBlock target, int[] colIndexTargets);
+ public static void decompressToBlock(MatrixBlock target, int colIndex, List<ColGroup> colGroups){
+ for(ColGroup g: colGroups){
+ int groupColIndex = Arrays.binarySearch(g._colIndexes, colIndex);
+ if( groupColIndex >= 0){
+ g.decompressToBlock(target, groupColIndex);
+ }
+ }
+ }
+
/**
* Decompress to block.
*
@@ -321,7 +343,6 @@
*
* @param matrix matrix to left multiply
* @param result matrix block result
- * @param numVals The Number of values contained in the Column.
* @param values The materialized list of values contained in the dictionary.
* @param numRows The number of rows in the matrix input
* @param numCols The number of columns in the colGroups parent matrix.
@@ -329,7 +350,7 @@
* @param ru The row to stop the matrix multiplication at.
* @param vOff The offset into the first argument matrix to start at.
*/
- public abstract void leftMultByMatrix(double[] matrix, double[] result, int numVals, double[] values, int numRows,
+ public abstract void leftMultByMatrix(double[] matrix, double[] result, double[] values, int numRows,
int numCols, int rl, int ru, int vOff);
/**
@@ -361,6 +382,16 @@
public abstract ColGroup scalarOperation(ScalarOperator op);
/**
+ * 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 sparseSafe True if the operation return 0 on all instances of values in v -- op(v[?], 0)
+ * @return A updated column group with the new values.
+ */
+ public abstract ColGroup binaryRowOp(BinaryOperator op, double[] v, boolean sparseSafe);
+
+ /**
* Unary Aggregate operator, since aggregate operators require new object output, the output becomes an uncompressed
* matrix.
*
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
new file mode 100644
index 0000000..2d28fb2
--- /dev/null
+++ b/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupConst.java
@@ -0,0 +1,264 @@
+/*
+ * 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.compress.colgroup;
+
+import java.util.Iterator;
+
+import org.apache.sysds.runtime.DMLCompressionException;
+import org.apache.sysds.runtime.data.SparseRow;
+import org.apache.sysds.runtime.functionobjects.Builtin;
+import org.apache.sysds.runtime.functionobjects.KahanFunction;
+import org.apache.sysds.runtime.functionobjects.KahanPlus;
+import org.apache.sysds.runtime.instructions.cp.KahanObject;
+import org.apache.sysds.runtime.matrix.data.IJV;
+import org.apache.sysds.runtime.matrix.data.MatrixBlock;
+import org.apache.sysds.runtime.matrix.operators.BinaryOperator;
+import org.apache.sysds.runtime.matrix.operators.ScalarOperator;
+
+import edu.emory.mathcs.backport.java.util.Arrays;
+
+public class ColGroupConst extends ColGroupValue {
+
+ private static final long serialVersionUID = 3204391661346504L;
+
+ /**
+ * Constructor for serialization
+ */
+ protected ColGroupConst() {
+ super();
+ }
+
+ /**
+ * Constructs an Constant Colum Group, that contains only one tuple, with the given value.
+ *
+ * @param colIndices The Colum indexes for the column group.
+ * @param numRows The number of rows contained in the group.
+ * @param dict The dictionary containing one tuple for the entire compression.
+ */
+ public ColGroupConst(int[] colIndices, int numRows, ADictionary dict) {
+ super(colIndices, numRows, dict);
+ }
+
+ @Override
+ public int[] getCounts(int[] out) {
+ out[0] = _numRows;
+ return out;
+ }
+
+ @Override
+ public int[] getCounts(int rl, int ru, int[] out) {
+ out[0] = ru - rl;
+ return out;
+ }
+
+ @Override
+ protected void computeSum(double[] c, KahanFunction kplus) {
+ c[0] += _dict.sum(getCounts(), _colIndexes.length, kplus);
+ }
+
+ @Override
+ protected void computeRowSums(double[] c, KahanFunction kplus, int rl, int ru, boolean mean) {
+ KahanObject kbuff = new KahanObject(0, 0);
+ KahanPlus kplus2 = KahanPlus.getKahanPlusFnObject();
+ double[] vals = _dict.sumAllRowsToDouble(kplus, kbuff, _colIndexes.length);
+ for(int rix = rl; rix < ru; rix++) {
+ setandExecute(c, kbuff, kplus2, vals[0], rix * (2 + (mean ? 1 : 0)));
+ }
+ }
+
+ @Override
+ protected void computeColSums(double[] c, KahanFunction kplus) {
+ _dict.colSum(c, getCounts(), _colIndexes, kplus);
+ }
+
+ @Override
+ protected void computeRowMxx(double[] c, Builtin builtin, int rl, int ru) {
+ throw new DMLCompressionException(
+ "Row max not supported for Const since Const is used for overlapping ColGroups, You have to materialize rows and then calculate row max");
+ }
+
+ @Override
+ public CompressionType getCompType() {
+ return CompressionType.CONST;
+ }
+
+ @Override
+ protected ColGroupType getColGroupType() {
+ return ColGroupType.CONST;
+ }
+
+ @Override
+ public long estimateInMemorySize() {
+ return ColGroupSizes.estimateInMemorySizeCONST(getNumCols(), getNumValues(), isLossy());
+ }
+
+ @Override
+ public void decompressToBlock(MatrixBlock target, int rl, int ru) {
+ final int ncol = getNumCols();
+ final double[] values = getValues();
+
+ for(int i = rl; i < ru; i++)
+ for(int j = 0; j < ncol; j++) {
+ double v = target.quickGetValue(i, _colIndexes[j]);
+ target.setValue(i, _colIndexes[j], values[j] + v);
+ }
+ }
+
+ @Override
+ public void decompressToBlock(MatrixBlock target, int[] colIndexTargets) {
+ int ncol = getNumCols();
+ double[] values = getValues();
+ for(int i = 0; i < _numRows; i++) {
+ for(int colIx = 0; colIx < ncol; colIx++) {
+ int origMatrixColIx = getColIndex(colIx);
+ int col = colIndexTargets[origMatrixColIx];
+ double cellVal = values[colIx];
+ target.quickSetValue(i, col, target.quickGetValue(i, col) + cellVal);
+ }
+ }
+ }
+
+ @Override
+ public void decompressToBlock(MatrixBlock target, int colpos) {
+ double[] c = target.getDenseBlockValues();
+
+ int nnz = 0;
+ double v = _dict.getValue(Arrays.binarySearch(_colIndexes, colpos));
+ if(v != 0) {
+ for(int i = 0; i < c.length; i++)
+ c[i] += v;
+ nnz = _numRows;
+ }
+ target.setNonZeros(nnz);
+ }
+
+ @Override
+ public double get(int r, int c) {
+ return _dict.getValue(Arrays.binarySearch(_colIndexes, c));
+ }
+
+ @Override
+ public void rightMultByVector(double[] b, double[] c, int rl, int ru, double[] dictVals) {
+ double[] vals = preaggValues(1, b, dictVals);
+ for(int i = 0; i < c.length; i++) {
+ c[i] += vals[0];
+ }
+ }
+
+ @Override
+ public void rightMultByMatrix(double[] preAggregatedB, double[] c, int thatNrColumns, int rl, int ru, int cl,
+ int cu) {
+
+ for(int i = rl * thatNrColumns; i < ru * thatNrColumns; i += thatNrColumns)
+ for(int j = i + cl; j < i + cu; j++)
+ c[j] += preAggregatedB[j % thatNrColumns];
+
+ }
+
+ @Override
+ public void rightMultBySparseMatrix(SparseRow[] rows, double[] c, int numVals, double[] dictVals, int nrColumns,
+ int rl, int ru) {
+ throw new DMLCompressionException(
+ "Depreciated and not supported right mult by sparse matrix Please preAggregate before calling");
+ }
+
+ private double preAggregate(double[] a, int aRows) {
+ double vals = 0;
+ for(int i = 0, off = _numRows * aRows; i < _numRows; i++, off++) {
+ vals += a[off];
+ }
+ return vals;
+ }
+
+ @Override
+ public void leftMultByRowVector(double[] a, double[] c, int numVals) {
+ double preAggVals = preAggregate(a, 0);
+ double[] dictVals = getValues();
+ for(int i = 0; i < _colIndexes.length; i++) {
+ c[i] += preAggVals * dictVals[i];
+ }
+ }
+
+ @Override
+ public void leftMultByRowVector(double[] a, double[] c, int numVals, double[] values) {
+ double preAggVals = preAggregate(a, 0);
+ for(int i = 0; i < _colIndexes.length; i++) {
+ c[i] += preAggVals * values[i];
+ }
+ }
+
+ @Override
+ public void leftMultByMatrix(double[] a, double[] c, double[] values, int numRows, int numCols, int rl, int ru,
+ int vOff) {
+ for(int i = rl; i < ru; i++) {
+ double preAggVals = preAggregate(a, i);
+ int offC = i * numCols;
+ for(int j = 0; j < _colIndexes.length; j++) {
+ c[offC + j] += preAggVals * values[j];
+ }
+ }
+ }
+
+ @Override
+ public void leftMultBySparseMatrix(int spNrVals, int[] indexes, double[] sparseV, double[] c, int numVals,
+ double[] values, int numRows, int numCols, int row, double[] MaterializedRow) {
+ double v = 0;
+ for(int i = 0; i < spNrVals; i++) {
+ v += sparseV[i];
+ }
+ int offC = row * numCols;
+ for(int j = 0; j < _colIndexes.length; j++) {
+ c[offC + j] += v * values[j];
+ }
+ }
+
+ @Override
+ public ColGroup scalarOperation(ScalarOperator op) {
+ return new ColGroupConst(_colIndexes, _numRows, applyScalarOp(op));
+ }
+
+ @Override
+ public ColGroup binaryRowOp(BinaryOperator op, double[] v, boolean sparseSafe) {
+ return new ColGroupConst(_colIndexes, _numRows, applyBinaryRowOp(op.fn, v, true));
+ }
+
+ @Override
+ public Iterator<IJV> getIterator(int rl, int ru, boolean inclZeros, boolean rowMajor) {
+ throw new DMLCompressionException("Unsupported Iterator of Const ColGroup");
+ }
+
+ @Override
+ public ColGroupRowIterator getRowIterator(int rl, int ru) {
+ throw new DMLCompressionException("Unsupported Row iterator of Const ColGroup");
+ }
+
+ @Override
+ public void countNonZerosPerRow(int[] rnnz, int rl, int ru) {
+
+ double[] values = _dict.getValues();
+ int base = 0;
+ for(int i = 0; i < values.length; i++) {
+ base += values[i] == 0 ? 0 : 1;
+ }
+ for(int i = 0; i < ru - rl; i++) {
+ rnnz[i] = base;
+ }
+ }
+}
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 1400bea..a3159ae 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
@@ -56,11 +56,13 @@
@Override
public void decompressToBlock(MatrixBlock target, int rl, int ru) {
- int ncol = getNumCols();
- double[] values = getValues();
+ final int nCol = getNumCols();
+ final double[] values = getValues();
for(int i = rl; i < ru; i++)
- for(int j = 0; j < ncol; j++)
- target.appendValue(i, _colIndexes[j], getData(i, j, values));
+ for(int j = 0; j < nCol; j++) {
+ double v = target.quickGetValue(i, _colIndexes[j]);
+ target.quickSetValue(i, _colIndexes[j], getData(i, j, values) + v);
+ }
}
@Override
@@ -72,7 +74,7 @@
int origMatrixColIx = getColIndex(colIx);
int col = colIndexTargets[origMatrixColIx];
double cellVal = getData(i, colIx, dictionary);
- target.quickSetValue(i, col, cellVal);
+ target.quickSetValue(i, col, target.quickGetValue(i, col) + cellVal);
}
}
}
@@ -86,10 +88,10 @@
for(int i = 0; i < _numRows; i++) {
int index = getIndex(i);
if(index < getNumValues()) {
- nnz += ((c[i] = values[(index) * ncol + colpos]) != 0) ? 1 : 0;
+ nnz += ((c[i] += values[(index) * ncol + colpos]) != 0) ? 1 : 0;
}
else {
- c[i] = 0.0;
+ nnz++;
}
}
target.setNonZeros(nnz);
@@ -204,10 +206,9 @@
}
@Override
- public void leftMultByMatrix(double[] a, double[] c, int numVals, double[] values, int numRows, int numCols, int rl,
- int ru, int voff) {
-
- numVals = (numVals == -1) ? getNumValues() : numVals;
+ public void leftMultByMatrix(double[] a, double[] c, double[] values, int numRows, int numCols, int rl, int ru,
+ int voff) {
+ int numVals = getNumValues();
for(int i = rl, j = voff; i < ru; i++, j++) {
int offC = i * numCols;
if(8 * numVals < _numRows) {
@@ -217,6 +218,9 @@
postScaling(values, vals, c, numVals, i, numCols);
}
else {
+ // Because we want to reduce the number of multiplies then we multiply the number of values with the
+ // number of columns before the for loop.
+ numVals = getNumValues() * _colIndexes.length;
for(int k = 0, aOff = j * _numRows; k < _numRows; k++, aOff++) {
double aval = a[aOff];
if(aval != 0) {
@@ -236,7 +240,7 @@
@Override
public void leftMultBySparseMatrix(int spNrVals, int[] indexes, double[] sparseV, double[] c, int numVals,
double[] values, int numRows, int numCols, int row, double[] MaterializedRow) {
- numVals = (numVals == -1) ? getNumValues() : numVals;
+ numVals = getNumValues();
for(int i = 0; i < spNrVals; i++) {
int k = indexes[i];
double aval = sparseV[i];
@@ -252,7 +256,7 @@
@Override
public void leftMultByRowVector(double[] a, double[] result, int numVals) {
- numVals = (numVals == -1) ? getNumValues() : numVals;
+ numVals = getNumValues();
double[] values = getValues();
leftMultByRowVector(a, result, numVals, values);
@@ -291,7 +295,7 @@
@Override
public void leftMultByRowVector(double[] a, double[] c, int numVals, double[] values) {
- numVals = (numVals == -1) ? getNumValues() : numVals;
+ numVals = getNumValues();
if(8 * numVals < _numRows) {
// iterative over codes and pre-aggregate inputs per code (guaranteed <=255)
@@ -302,6 +306,7 @@
postScaling(values, vals, c, numVals);
}
else {
+ numVals = numVals * _colIndexes.length;
// iterate over codes, compute all, and add to the result
for(int i = 0; i < _numRows; i++) {
double aval = a[i];
diff --git a/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupDDC1.java b/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupDDC1.java
index 25f45ed..84a7683 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupDDC1.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupDDC1.java
@@ -30,6 +30,7 @@
import org.apache.sysds.runtime.compress.utils.ABitmap;
import org.apache.sysds.runtime.compress.utils.LinearAlgebraUtils;
import org.apache.sysds.runtime.data.SparseRow;
+import org.apache.sysds.runtime.matrix.operators.BinaryOperator;
import org.apache.sysds.runtime.matrix.operators.ScalarOperator;
/**
@@ -130,13 +131,13 @@
@Override
protected double getData(int r, double[] values) {
int index = (_data[r] & 0xFF);
- return (index >= values.length) ? 0.0 : values[index];
+ return (index < values.length) ? values[index] : 0.0;
}
@Override
protected double getData(int r, int colIx, double[] values) {
int index = (_data[r] & 0xFF) * getNumCols() + colIx;
- return (index >= values.length) ? 0.0 : values[index];
+ return (index < values.length) ? values[index] : 0.0;
}
@Override
@@ -198,10 +199,17 @@
}
@Override
+ public ColGroup binaryRowOp(BinaryOperator op, double[] v, boolean sparseSafe) {
+ sparseSafe = sparseSafe || !_zeros;
+ return new ColGroupDDC1(_colIndexes, _numRows, applyBinaryRowOp(op.fn, v, sparseSafe), _data, !sparseSafe);
+ }
+
+ @Override
public String toString() {
StringBuilder sb = new StringBuilder();
sb.append(super.toString());
sb.append(" DataLength: " + this._data.length);
+ sb.append(Arrays.toString(this._data));
return sb.toString();
}
}
diff --git a/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupDDC2.java b/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupDDC2.java
index 0e8cf04..6236b68 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupDDC2.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupDDC2.java
@@ -28,6 +28,7 @@
import org.apache.sysds.runtime.compress.utils.ABitmap;
import org.apache.sysds.runtime.compress.utils.LinearAlgebraUtils;
import org.apache.sysds.runtime.data.SparseRow;
+import org.apache.sysds.runtime.matrix.operators.BinaryOperator;
import org.apache.sysds.runtime.matrix.operators.ScalarOperator;
/**
@@ -181,6 +182,12 @@
}
@Override
+ public ColGroup binaryRowOp(BinaryOperator op, double[] v, boolean sparseSafe) {
+ sparseSafe = sparseSafe || !_zeros;
+ return new ColGroupDDC2(_colIndexes, _numRows, applyBinaryRowOp(op.fn, v, sparseSafe), _data, !sparseSafe);
+ }
+
+ @Override
public String toString() {
StringBuilder sb = new StringBuilder();
sb.append(super.toString());
diff --git a/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupIO.java b/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupIO.java
index eb6e1a4..bf927fc 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupIO.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupIO.java
@@ -77,6 +77,9 @@
case DDC2:
grp = new ColGroupDDC2();
break;
+ case CONST:
+ grp = new ColGroupConst();
+ break;
default:
throw new DMLRuntimeException("Unsupported ColGroup Type used: " + ctype);
}
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 8845ce7..466c942 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
@@ -33,6 +33,7 @@
import org.apache.sysds.runtime.functionobjects.KahanPlus;
import org.apache.sysds.runtime.instructions.cp.KahanObject;
import org.apache.sysds.runtime.matrix.data.MatrixBlock;
+import org.apache.sysds.runtime.matrix.operators.BinaryOperator;
import org.apache.sysds.runtime.matrix.operators.ScalarOperator;
/**
@@ -109,8 +110,10 @@
int pos = boff + bix + 1;
for(int i = pos; i < pos + len; i++)
for(int j = 0, rix = bi + _data[i]; j < numCols; j++)
- if(values[off + j] != 0)
- target.appendValue(rix, _colIndexes[j], values[off + j]);
+ if(values[off + j] != 0) {
+ double v = target.quickGetValue(rix, _colIndexes[j]);
+ target.setValue(rix, _colIndexes[j], values[off + j] + v);
+ }
apos[k] += len + 1;
}
}
@@ -149,8 +152,10 @@
int pos = boff + bix + 1;
for(int i = pos; i < pos + len; i++)
for(int j = 0, rix = bi + _data[i]; j < numCols; j++)
- if(values[off + j] != 0)
- target.appendValue(rix, cix[j], values[off + j]);
+ if(values[off + j] != 0) {
+ double v = target.quickGetValue(rix, _colIndexes[j]);
+ target.setValue(rix, cix[j], values[off + j] + v);
+ }
apos[k] += len + 1;
}
}
@@ -175,7 +180,7 @@
// cache conscious append via horizontal scans
int nnz = 0;
for(int bi = 0; bi < _numRows; bi += blksz) {
- Arrays.fill(c, bi, Math.min(bi + blksz, _numRows), 0);
+ // Arrays.fill(c, bi, Math.min(bi + blksz, _numRows), 0);
for(int k = 0, off = 0; k < numVals; k++, off += numCols) {
int boff = _ptr[k];
int blen = len(k);
@@ -185,7 +190,7 @@
int len = _data[boff + bix];
int pos = boff + bix + 1;
for(int i = pos; i < pos + len; i++) {
- c[bi + _data[i]] = values[off + colpos];
+ c[bi + _data[i]] += values[off + colpos];
nnz++;
}
apos[k] += len + 1;
@@ -245,22 +250,48 @@
@Override
public ColGroup scalarOperation(ScalarOperator op) {
double val0 = op.executeScalar(0);
-
// fast path: sparse-safe operations
// Note that bitmaps don't change and are shallow-copied
if(op.sparseSafe || val0 == 0 || !_zeros) {
return new ColGroupOLE(_colIndexes, _numRows, _zeros, applyScalarOp(op), _data, _ptr);
}
+ // slow path: sparse-unsafe operations (potentially create new bitmap)
+ // note: for efficiency, we currently don't drop values that become 0
+ boolean[] lind = computeZeroIndicatorVector();
+ int[] loff = computeOffsets(lind);
+
+ if(loff.length == 0) { // empty offset list: go back to fast path
+ return new ColGroupOLE(_colIndexes, _numRows, false, applyScalarOp(op), _data, _ptr);
+ }
+
+ ADictionary rvalues = applyScalarOp(op, val0, getNumCols());
+ char[] lbitmap = genOffsetBitmap(loff, loff.length);
+ char[] rbitmaps = Arrays.copyOf(_data, _data.length + lbitmap.length);
+ System.arraycopy(lbitmap, 0, rbitmaps, _data.length, lbitmap.length);
+ int[] rbitmapOffs = Arrays.copyOf(_ptr, _ptr.length + 1);
+ rbitmapOffs[rbitmapOffs.length - 1] = rbitmaps.length;
+
+ return new ColGroupOLE(_colIndexes, _numRows, false, rvalues, rbitmaps, rbitmapOffs);
+ }
+
+ @Override
+ public ColGroup binaryRowOp(BinaryOperator op, double[] v, boolean sparseSafe) {
+
+ sparseSafe = sparseSafe || !_zeros;
+ // fast path: sparse-safe operations
+ // Note that bitmaps don't change and are shallow-copied
+ if(sparseSafe) {
+ return new ColGroupOLE(_colIndexes, _numRows, _zeros, applyBinaryRowOp(op.fn, v, sparseSafe), _data, _ptr);
+ }
// slow path: sparse-unsafe operations (potentially create new bitmap)
// note: for efficiency, we currently don't drop values that become 0
boolean[] lind = computeZeroIndicatorVector();
int[] loff = computeOffsets(lind);
if(loff.length == 0) { // empty offset list: go back to fast path
- return new ColGroupOLE(_colIndexes, _numRows, false, applyScalarOp(op), _data, _ptr);
+ return new ColGroupOLE(_colIndexes, _numRows, false, applyBinaryRowOp(op.fn, v, true), _data, _ptr);
}
-
- ADictionary rvalues = applyScalarOp(op, val0, getNumCols());
+ ADictionary rvalues = applyBinaryRowOp(op.fn, v, sparseSafe);
char[] lbitmap = genOffsetBitmap(loff, loff.length);
char[] rbitmaps = Arrays.copyOf(_data, _data.length + lbitmap.length);
System.arraycopy(lbitmap, 0, rbitmaps, _data.length, lbitmap.length);
@@ -510,10 +541,10 @@
}
@Override
- public void leftMultByMatrix(double[] a, double[] c, int numVals, double[] values, int numRows, int numCols, int rl,
- int ru, int voff) {
+ public void leftMultByMatrix(double[] a, double[] c, double[] values, int numRows, int numCols, int rl, int ru,
+ int voff) {
final int blksz = CompressionSettings.BITMAP_BLOCK_SZ;
-
+ final int numVals = getNumValues();
if(numVals >= 1 && _numRows > blksz) {
// cache blocking config (see matrix-vector mult for explanation)
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 1ce23d5..1c82862 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
@@ -35,6 +35,7 @@
import org.apache.sysds.runtime.instructions.cp.KahanObject;
import org.apache.sysds.runtime.matrix.data.MatrixBlock;
import org.apache.sysds.runtime.matrix.data.Pair;
+import org.apache.sysds.runtime.matrix.operators.BinaryOperator;
import org.apache.sysds.runtime.matrix.operators.ScalarOperator;
/** A group of columns compressed with a single run-length encoded bitmap. */
@@ -91,7 +92,7 @@
@Override
public void decompressToBlock(MatrixBlock target, int rl, int ru) {
if(getNumValues() > 1) {
- final int blksz = 128 * 1024;
+ final int blksz = CompressionSettings.BITMAP_BLOCK_SZ;
final int numCols = getNumCols();
final int numVals = getNumValues();
final double[] values = getValues();
@@ -112,9 +113,12 @@
start += _data[boff + bix];
int len = _data[boff + bix + 1];
for(int i = Math.max(rl, start); i < Math.min(start + len, ru); i++)
- for(int j = 0; j < numCols; j++)
- if(values[off + j] != 0)
- target.appendValue(i, _colIndexes[j], values[off + j]);
+ for(int j = 0; j < numCols; j++) {
+ if(values[off + j] != 0) {
+ double v = target.quickGetValue(i, _colIndexes[j]);
+ target.setValue(i, _colIndexes[j], values[off + j] + v);
+ }
+ }
start += len;
}
apos[k] = bix;
@@ -131,7 +135,7 @@
@Override
public void decompressToBlock(MatrixBlock target, int[] colixTargets) {
if(getNumValues() > 1) {
- final int blksz = 128 * 1024;
+ final int blksz = CompressionSettings.BITMAP_BLOCK_SZ;
final int numCols = getNumCols();
final int numVals = getNumValues();
final double[] values = getValues();
@@ -160,8 +164,11 @@
int len = _data[boff + bix + 1];
for(int i = start; i < start + len; i++)
for(int j = 0; j < numCols; j++)
- if(values[off + j] != 0)
- target.appendValue(i, cix[j], values[off + j]);
+ if(values[off + j] != 0) {
+ double v = target.quickGetValue(i, _colIndexes[j]);
+ target.setValue(i, _colIndexes[j], values[off + j] + v);
+ }
+
start += len;
}
apos[k] = bix;
@@ -177,6 +184,7 @@
@Override
public void decompressToBlock(MatrixBlock target, int colpos) {
+ // LOG.error("Does not work");
final int blksz = 128 * 1024;
final int numCols = getNumCols();
final int numVals = getNumValues();
@@ -191,7 +199,7 @@
int nnz = 0;
for(int bi = 0; bi < _numRows; bi += blksz) {
int bimax = Math.min(bi + blksz, _numRows);
- Arrays.fill(c, bi, bimax, 0);
+ // Arrays.fill(c, bi, bimax, 0);
for(int k = 0, off = 0; k < numVals; k++, off += numCols) {
int boff = _ptr[k];
int blen = len(k);
@@ -202,7 +210,8 @@
for(; bix < blen & start < bimax; bix += 2) {
start += _data[boff + bix];
int len = _data[boff + bix + 1];
- Arrays.fill(c, start, start + len, values[off + colpos]);
+ for(int i = start; i< start + len; i++)
+ c[i] += values[off + colpos];
nnz += len;
start += len;
}
@@ -292,15 +301,13 @@
int start = astart[k];
// compute partial results, not aligned
- while(bix < blen) {
+ while(bix < blen & bix < bimax) {
int lstart = _data[boff + bix];
int llen = _data[boff + bix + 1];
int len = Math.min(start + lstart + llen, bimax) - Math.max(bi, start + lstart);
if(len > 0) {
LinearAlgebraUtils.vectAdd(val, c, Math.max(bi, start + lstart), len);
}
- if(start + lstart + llen >= bimax)
- break;
start += lstart + llen;
bix += 2;
}
@@ -508,9 +515,10 @@
}
@Override
- public void leftMultByMatrix(final double[] a, final double[] c, final int numVals, final double[] values,
- final int numRows, final int numCols, int rl, final int ru, final int vOff) {
+ public void leftMultByMatrix(final double[] a, final double[] c, final double[] values, final int numRows,
+ final int numCols, int rl, final int ru, final int vOff) {
+ final int numVals = getNumValues();
if(numVals >= 1 && _numRows > CompressionSettings.BITMAP_BLOCK_SZ) {
final int blksz = 2 * CompressionSettings.BITMAP_BLOCK_SZ;
@@ -650,6 +658,34 @@
}
@Override
+ public ColGroup binaryRowOp(BinaryOperator op, double[] v, boolean sparseSafe) {
+ sparseSafe = sparseSafe || !_zeros;
+
+ // fast path: sparse-safe operations
+ // Note that bitmaps don't change and are shallow-copied
+ if(sparseSafe) {
+ return new ColGroupRLE(_colIndexes, _numRows, _zeros, applyBinaryRowOp(op.fn, v, sparseSafe), _data, _ptr);
+ }
+
+ // slow path: sparse-unsafe operations (potentially create new bitmap)
+ // note: for efficiency, we currently don't drop values that become 0
+ boolean[] lind = computeZeroIndicatorVector();
+ int[] loff = computeOffsets(lind);
+ if(loff.length == 0) { // empty offset list: go back to fast path
+ return new ColGroupRLE(_colIndexes, _numRows, false, applyBinaryRowOp(op.fn, v, true), _data, _ptr);
+ }
+
+ ADictionary rvalues = applyBinaryRowOp(op.fn, v, sparseSafe);
+ char[] lbitmap = genRLEBitmap(loff, loff.length);
+ char[] rbitmaps = Arrays.copyOf(_data, _data.length + lbitmap.length);
+ System.arraycopy(lbitmap, 0, rbitmaps, _data.length, lbitmap.length);
+ int[] rbitmapOffs = Arrays.copyOf(_ptr, _ptr.length + 1);
+ rbitmapOffs[rbitmapOffs.length - 1] = rbitmaps.length;
+
+ return new ColGroupRLE(_colIndexes, _numRows, false, rvalues, rbitmaps, rbitmapOffs);
+ }
+
+ @Override
protected final void computeSum(double[] c, KahanFunction kplus) {
c[0] += _dict.sum(getCounts(), _colIndexes.length, kplus);
}
diff --git a/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupSizes.java b/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupSizes.java
index e18fb14..4688ecd 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupSizes.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupSizes.java
@@ -96,6 +96,11 @@
return size;
}
+ public static long estimateInMemorySizeCONST(int nrColumns, int nrValues, boolean lossy){
+ long size = estimateInMemorySizeGroupValue(nrColumns, nrValues, lossy);
+ return size;
+ }
+
public static long estimateInMemorySizeUncompressed(int nrRows, int nrColumns, double sparsity) {
long size = 0;
// Since the Object is a col group the overhead from the Memory Size group is added
@@ -104,5 +109,4 @@
size += MatrixBlock.estimateSizeInMemory(nrRows, nrColumns, sparsity);
return size;
}
-
-}
\ No newline at end of file
+}
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 47c1f0f..9898bb2 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
@@ -38,6 +38,7 @@
import org.apache.sysds.runtime.matrix.data.LibMatrixMult;
import org.apache.sysds.runtime.matrix.data.MatrixBlock;
import org.apache.sysds.runtime.matrix.operators.AggregateUnaryOperator;
+import org.apache.sysds.runtime.matrix.operators.BinaryOperator;
import org.apache.sysds.runtime.matrix.operators.ScalarOperator;
import org.apache.sysds.runtime.util.SortUtils;
@@ -276,13 +277,13 @@
public void rightMultByMatrix(MatrixBlock matrix, MatrixBlock result, int rl, int ru) {
// Pull out the relevant rows of the vector
-
+
int clen = _colIndexes.length;
MatrixBlock subMatrix = new MatrixBlock(clen, matrix.getNumColumns(), false);
subMatrix.allocateDenseBlock();
double[] b = subMatrix.getDenseBlockValues();
-
- for(int colIx = 0; colIx < clen; colIx++){
+
+ for(int colIx = 0; colIx < clen; colIx++) {
int row = _colIndexes[colIx];
for(int col = 0; col < matrix.getNumColumns(); col++)
b[colIx * matrix.getNumColumns() + col] = matrix.quickGetValue(row, col);
@@ -316,7 +317,7 @@
}
@Override
- public void leftMultByMatrix(double[] vector, double[] c, int numVals, double[] values, int numRows, int numCols,
+ public void leftMultByMatrix(double[] vector, double[] c, double[] values, int numRows, int numCols,
int rl, int ru, int vOff) {
throw new NotImplementedException("Should not be called use other matrix function for uncompressed columns");
}
@@ -348,6 +349,11 @@
return new ColGroupUncompressed(getColIndices(), _data.getNumRows(), retContent);
}
+ @Override
+ public ColGroup binaryRowOp(BinaryOperator op, double[] v, boolean sparseSafe) {
+ throw new NotImplementedException("Should not be called use other matrix function for uncompressed columns");
+ }
+
public void unaryAggregateOperations(AggregateUnaryOperator op, double[] ret) {
throw new NotImplementedException("Should not be called");
}
@@ -504,7 +510,11 @@
StringBuilder sb = new StringBuilder();
sb.append(super.toString());
sb.append("\n");
- sb.append(_data.toString());
+ sb.append(_data.getNumColumns() + " ");
+ sb.append(_data.getNumRows() + " ");
+ sb.append(_data.getNonZeros() + " ");
+ sb.append(_data.isInSparseFormat() + " ");
+ // sb.append(_data.toString());
return sb.toString();
}
diff --git a/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupValue.java b/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupValue.java
index 101308e..21dfa85 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupValue.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupValue.java
@@ -40,6 +40,7 @@
import org.apache.sysds.runtime.functionobjects.ReduceAll;
import org.apache.sysds.runtime.functionobjects.ReduceCol;
import org.apache.sysds.runtime.functionobjects.ReduceRow;
+import org.apache.sysds.runtime.functionobjects.ValueFunction;
import org.apache.sysds.runtime.instructions.cp.KahanObject;
import org.apache.sysds.runtime.matrix.data.MatrixBlock;
import org.apache.sysds.runtime.matrix.data.Pair;
@@ -50,7 +51,7 @@
* Base class for column groups encoded with value dictionary. This include column groups such as DDC OLE and RLE.
*
*/
-public abstract class ColGroupValue extends ColGroup {
+public abstract class ColGroupValue extends ColGroup implements Cloneable {
private static final long serialVersionUID = 3786247536054353658L;
/** thread-local pairs of reusable temporary vectors for positions and values */
@@ -63,6 +64,7 @@
/** Distinct value tuples associated with individual bitmaps. */
protected ADictionary _dict;
+ protected int[] counts;
protected ColGroupValue() {
super();
@@ -116,8 +118,12 @@
return _dict.getValues();
}
- public byte[] getByteValues() {
- return ((QDictionary) _dict).getValuesByte();
+ public ADictionary getDictionary() {
+ return _dict;
+ }
+
+ protected void setDictionary(ADictionary dict) {
+ _dict = dict;
}
@Override
@@ -133,8 +139,9 @@
}
/**
- * Returns the counts of values inside the MatrixBlock returned in getValuesAsBlock Throws an exception if the
- * getIfCountsType is false.
+ * Returns the counts of values inside the dictionary. If already calculated it will return the previous counts.
+ * This produce an overhead in cases where the count is calculated, but the overhead will be limited to number of
+ * distinct tuples in the dictionary.
*
* The returned counts always contains the number of zeros as well if there are some contained, even if they are not
* materialized.
@@ -142,14 +149,20 @@
* @return the count of each value in the MatrixBlock.
*/
public final int[] getCounts() {
- int[] tmp;
- if(_zeros) {
- tmp = allocIVector(getNumValues() + 1, true);
+ if(counts == null) {
+
+ counts = new int[getNumValues() + 1];
+ // if(_zeros) {
+ // tmp = allocIVector(getNumValues() + 1, true);
+ // }
+ // else {
+ // tmp = allocIVector(getNumValues(), true);
+ // }
+ return getCounts(counts);
}
else {
- tmp = allocIVector(getNumValues(), true);
+ return counts;
}
- return getCounts(tmp);
}
/**
@@ -245,14 +258,13 @@
* @param cl Lower column index to aggregate from
* @param cu Upper column index to aggregate to
* @param cut The total number of columns in b.
+ * @param ret The double list to return.
* @return The aggregated matrix output. Note this has to be mapped to the output matrix.
*/
public double[] preaggValues(final int numVals, final double[] b, double[] dictVals, final int cl, final int cu,
- final int cut) {
+ final int cut, double[] ret) {
final int retRows = (cu - cl);
- final int retCols = (numVals);
- final double[] ret = allocDVector(retCols * retRows, true);
for(int k = 0, off = 0; k < numVals * _colIndexes.length; k += _colIndexes.length, off += retRows) {
for(int h = 0; h < _colIndexes.length; h++) {
int idb = _colIndexes[h] * cut;
@@ -269,12 +281,18 @@
return ret;
}
- public double[] preaggValues(final int numVals, final SparseBlock b, double[] dictVals, final int cl, final int cu,
+ public double[] preaggValues(final int numVals, final double[] b, double[] dictVals, final int cl, final int cu,
final int cut) {
+ final double[] ret = allocDVector(numVals * (cu - cl), true);
+
+ return preaggValues(numVals, b, dictVals, cl, cu, cut, ret);
+ }
+
+ public double[] preaggValues(final int numVals, final SparseBlock b, double[] dictVals, final int cl, final int cu,
+ final int cut, final double[] ret) {
+
final int retRows = (cu - cl);
- final int retCols = (numVals);
- final double[] ret = allocDVector(retCols * retRows, true);
for(int h = 0; h < _colIndexes.length; h++) {
SparseRow row = b.get(_colIndexes[h]);
// SparseRow row = b[_colIndexes[h]];
@@ -290,6 +308,11 @@
return ret;
}
+ public double[] preaggValues(final int numVals, final SparseBlock b, double[] dictVals, final int cl, final int cu,
+ final int cut) {
+ return preaggValues(numVals, b, dictVals, cl, cu, cut, allocDVector(numVals * (cu - cl), true));
+ }
+
protected final double[] preaggValue(int k, double[] b, double[] dictVals, int cl, int cu, int cut) {
double[] ret = allocDVector(cu - cl, true);
for(int h = 0; h < _colIndexes.length; h++) {
@@ -369,6 +392,19 @@
return _dict.applyScalarOp(op, newVal, numCols);
}
+ /**
+ * Apply the binary row-wise operator to the dictionary, and copy it appropriately if needed.
+ *
+ * @param fn The function to apply.
+ * @param v The vector to apply on each tuple of the dictionary.
+ * @param sparseSafe Specify if the operation is sparseSafe. if false then allocate a new tuple.
+ * @return The new Dictionary with values.
+ */
+ protected ADictionary applyBinaryRowOp(ValueFunction fn, double[] v, boolean sparseSafe) {
+ return sparseSafe ? _dict.clone().applyBinaryRowOp(fn, v, sparseSafe, _colIndexes) : _dict
+ .applyBinaryRowOp(fn, v, sparseSafe, _colIndexes);
+ }
+
@Override
public void unaryAggregateOperations(AggregateUnaryOperator op, double[] c) {
unaryAggregateOperations(op, c, 0, _numRows);
@@ -462,7 +498,8 @@
sb.append(String.format("\n%15s%5d ", "Columns:", _colIndexes.length));
sb.append(Arrays.toString(_colIndexes));
sb.append(String.format("\n%15s%5d ", "Values:", _dict.getValues().length));
- sb.append(Arrays.toString(_dict.getValues()));
+ sb.append("\n");
+ _dict.getString(sb, _colIndexes.length);
return sb.toString();
}
@@ -529,4 +566,36 @@
protected abstract void computeRowMxx(double[] c, Builtin builtin, int rl, int ru);
+ protected Object clone() throws CloneNotSupportedException {
+ return super.clone();
+ }
+
+ public ColGroup copyAndSet(int[] colIndexes, double[] newDictionary) {
+ try {
+ ColGroupValue clone = (ColGroupValue) this.clone();
+ clone.setDictionary(new Dictionary(newDictionary));
+ clone.setColIndices(colIndexes);
+ return clone;
+ }
+ catch(CloneNotSupportedException e) {
+ e.printStackTrace();
+ }
+ return null;
+ }
+
+ /**
+ * shallow copy of the colGroup.
+ *
+ * @return a shallow copy of the colGroup.
+ */
+ public ColGroup copy() {
+ try {
+ ColGroupValue clone = (ColGroupValue) this.clone();
+ return clone;
+ }
+ catch(CloneNotSupportedException e) {
+ e.printStackTrace();
+ }
+ return null;
+ }
}
diff --git a/src/main/java/org/apache/sysds/runtime/compress/colgroup/Dictionary.java b/src/main/java/org/apache/sysds/runtime/compress/colgroup/Dictionary.java
index 230c121..ea9527e 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/colgroup/Dictionary.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/colgroup/Dictionary.java
@@ -29,6 +29,7 @@
import org.apache.sysds.runtime.functionobjects.Builtin;
import org.apache.sysds.runtime.functionobjects.KahanFunction;
import org.apache.sysds.runtime.functionobjects.KahanPlus;
+import org.apache.sysds.runtime.functionobjects.ValueFunction;
import org.apache.sysds.runtime.instructions.cp.KahanObject;
import org.apache.sysds.runtime.matrix.operators.ScalarOperator;
import org.apache.sysds.utils.MemoryEstimates;
@@ -112,6 +113,29 @@
}
@Override
+ public Dictionary applyBinaryRowOp(ValueFunction fn, double[] v, boolean sparseSafe, int[] colIndexes) {
+ final int len = _values.length;
+ final int lenV = colIndexes.length;
+ if(sparseSafe) {
+ for(int i = 0; i < len; i++) {
+ _values[i] = fn.execute(_values[i], v[colIndexes[i % lenV]]);
+ }
+ return this;
+ }
+ else {
+ double[] values = new double[len + lenV];
+ int i = 0;
+ for(; i < len; i++) {
+ values[i] = fn.execute(_values[i], v[colIndexes[i % lenV]]);
+ }
+ for(; i < len + lenV; i++) {
+ values[i] = fn.execute(0, v[colIndexes[i % lenV]]);
+ }
+ return new Dictionary(values);
+ }
+ }
+
+ @Override
public Dictionary clone() {
return new Dictionary(_values.clone());
}
@@ -175,7 +199,7 @@
protected void colSum(double[] c, int[] counts, int[] colIndexes, KahanFunction kplus) {
KahanObject kbuff = new KahanObject(0, 0);
int valOff = 0;
- final int rows = c.length/2;
+ final int rows = c.length / 2;
for(int k = 0; k < _values.length / colIndexes.length; k++) {
int cntk = counts[k];
for(int j = 0; j < colIndexes.length; j++) {
@@ -210,4 +234,12 @@
sb.append("\n " + Arrays.toString(_values));
return sb.toString();
}
+
+ public StringBuilder getString(StringBuilder sb, int colIndexes){
+ for(int i = 0; i< _values.length; i++){
+ sb.append(_values[i]);
+ sb.append((i) % (colIndexes ) == colIndexes - 1 ? "\n" : " ");
+ }
+ return sb;
+ }
}
diff --git a/src/main/java/org/apache/sysds/runtime/compress/colgroup/QDictionary.java b/src/main/java/org/apache/sysds/runtime/compress/colgroup/QDictionary.java
index 19c65fc..6d1906e 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/colgroup/QDictionary.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/colgroup/QDictionary.java
@@ -33,6 +33,7 @@
import org.apache.sysds.runtime.functionobjects.KahanPlusSq;
import org.apache.sysds.runtime.functionobjects.Multiply;
import org.apache.sysds.runtime.functionobjects.Plus;
+import org.apache.sysds.runtime.functionobjects.ValueFunction;
import org.apache.sysds.runtime.instructions.cp.KahanObject;
import org.apache.sysds.runtime.matrix.operators.ScalarOperator;
import org.apache.sysds.utils.MemoryEstimates;
@@ -60,6 +61,7 @@
@Override
public double[] getValues() {
+ // TODO: use a temporary double array for this.
double[] res = new double[_values.length];
for(int i = 0; i < _values.length; i++) {
res[i] = getValue(i);
@@ -139,9 +141,8 @@
}
else {
double[] temp = new double[_values.length];
- double max = op.executeScalar(getValue(0));
- temp[0] = max;
- for(int i = 1; i < _values.length; i++) {
+ double max = Math.abs(op.executeScalar(getValue(0)));
+ for(int i = 0; i < _values.length; i++) {
temp[i] = op.executeScalar(getValue(i));
double absTemp = Math.abs(temp[i]);
if(absTemp > max) {
@@ -153,14 +154,13 @@
_values[i] = (byte) Math.round(temp[i] / _scale);
}
}
-
return this;
}
@Override
public QDictionary applyScalarOp(ScalarOperator op, double newVal, int numCols) {
double[] temp = getValues();
- double max = newVal;
+ double max = Math.abs(newVal);
for(int i = 0; i < _values.length; i++) {
temp[i] = op.executeScalar(temp[i]);
double absTemp = Math.abs(temp[i]);
@@ -178,6 +178,38 @@
}
@Override
+ public QDictionary applyBinaryRowOp(ValueFunction fn, double[] v, boolean sparseSafe, int[] colIndexes) {
+ // TODO Use a temporary double array for this.
+ double[] temp = sparseSafe ? new double[_values.length] : new double[_values.length + colIndexes.length];
+ double max = Math.abs(fn.execute(0, v[0]));
+ final int colL = colIndexes.length;
+ int i = 0;
+ for(; i < _values.length; i++) {
+ temp[i] = fn.execute(_values[i] * _scale, v[colIndexes[i % colL]]);
+ double absTemp = Math.abs(temp[i]);
+ if(absTemp > max) {
+ max = absTemp;
+ }
+ }
+ if(!sparseSafe)
+ for(; i < _values.length + colL; i++) {
+ temp[i] = fn.execute(0, v[colIndexes[i % colL]]);
+ double absTemp = Math.abs(temp[i]);
+ if(absTemp > max) {
+ max = absTemp;
+ }
+ }
+
+ double scale = max / (double) (Byte.MAX_VALUE);
+ byte[] res = sparseSafe ? _values : new byte[_values.length + colIndexes.length];
+
+ for(i = 0; i < temp.length; i++) {
+ res[i] = (byte) Math.round(temp[i] / scale);
+ }
+ return new QDictionary(res, scale);
+ }
+
+ @Override
public int getValuesLength() {
return _values.length;
}
@@ -250,12 +282,11 @@
@Override
protected void colSum(double[] c, int[] counts, int[] colIndexes, KahanFunction kplus) {
-
- final int rows = c.length/2;
+ final int rows = c.length / 2;
if(!(kplus instanceof KahanPlusSq)) {
int[] sum = new int[colIndexes.length];
int valOff = 0;
- for(int k = 0; k < _values.length/ colIndexes.length; k++) {
+ for(int k = 0; k < _values.length / colIndexes.length; k++) {
int cntk = counts[k];
for(int j = 0; j < colIndexes.length; j++) {
sum[j] += cntk * getValueByte(valOff++);
@@ -268,7 +299,7 @@
else {
KahanObject kbuff = new KahanObject(0, 0);
int valOff = 0;
- for(int k = 0; k < _values.length/ colIndexes.length; k++) {
+ for(int k = 0; k < _values.length / colIndexes.length; k++) {
int cntk = counts[k];
for(int j = 0; j < colIndexes.length; j++) {
kbuff.set(c[colIndexes[j]], c[colIndexes[j] + rows]);
@@ -305,4 +336,12 @@
return kbuff._sum;
}
}
+
+ public StringBuilder getString(StringBuilder sb, int colIndexes) {
+ for(int i = 0; i < _values.length; i++) {
+ sb.append(_values[i]);
+ sb.append((i) % (colIndexes) == colIndexes - 1 ? "\n" : " ");
+ }
+ return sb;
+ }
}
diff --git a/src/main/java/org/apache/sysds/runtime/compress/estim/CompressedSizeEstimatorSample.java b/src/main/java/org/apache/sysds/runtime/compress/estim/CompressedSizeEstimatorSample.java
index 017abb1..e40035c 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/estim/CompressedSizeEstimatorSample.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/estim/CompressedSizeEstimatorSample.java
@@ -272,5 +272,4 @@
private static int[] getSortedUniformSample(int range, int smplSize, long seed) {
return UtilFunctions.getSortedSampleIndexes(range, smplSize, seed);
}
-
}
diff --git a/src/main/java/org/apache/sysds/runtime/compress/estim/EstimationFactors.java b/src/main/java/org/apache/sysds/runtime/compress/estim/EstimationFactors.java
index ef37d85..0c0d7f9 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/estim/EstimationFactors.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/estim/EstimationFactors.java
@@ -105,4 +105,4 @@
sb.append("\tcontains a 0: " + containsZero);
return sb.toString();
}
-}
\ No newline at end of file
+}
diff --git a/src/main/java/org/apache/sysds/runtime/compress/lib/LibBinaryCellOp.java b/src/main/java/org/apache/sysds/runtime/compress/lib/LibBinaryCellOp.java
new file mode 100644
index 0000000..01239f7
--- /dev/null
+++ b/src/main/java/org/apache/sysds/runtime/compress/lib/LibBinaryCellOp.java
@@ -0,0 +1,153 @@
+/*
+ * 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.compress.lib;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.commons.lang.NotImplementedException;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.sysds.runtime.DMLCompressionException;
+import org.apache.sysds.runtime.compress.CompressedMatrixBlock;
+import org.apache.sysds.runtime.compress.colgroup.ADictionary;
+import org.apache.sysds.runtime.compress.colgroup.ColGroup;
+import org.apache.sysds.runtime.compress.colgroup.ColGroupConst;
+import org.apache.sysds.runtime.compress.colgroup.ColGroupUncompressed;
+import org.apache.sysds.runtime.compress.colgroup.Dictionary;
+import org.apache.sysds.runtime.functionobjects.Minus;
+import org.apache.sysds.runtime.functionobjects.Multiply;
+import org.apache.sysds.runtime.functionobjects.Plus;
+import org.apache.sysds.runtime.matrix.data.LibMatrixBincell;
+import org.apache.sysds.runtime.matrix.data.LibMatrixBincell.BinaryAccessType;
+import org.apache.sysds.runtime.matrix.data.MatrixBlock;
+import org.apache.sysds.runtime.matrix.operators.BinaryOperator;
+import org.apache.sysds.runtime.matrix.operators.LeftScalarOperator;
+import org.apache.sysds.runtime.matrix.operators.RightScalarOperator;
+import org.apache.sysds.runtime.matrix.operators.ScalarOperator;
+
+public class LibBinaryCellOp {
+
+ private static final Log LOG = LogFactory.getLog(LibBinaryCellOp.class.getName());
+
+ /**
+ * matrix-matrix binary operations, MM, MV
+ *
+ * @param m1 Input matrix in compressed format to perform the operator on using m2
+ * @param m2 Input matrix 2 to use cell-wise on m1
+ * @param ret Result matrix to be returned in the operation.
+ * @param op Binary operator such as multiply, add, less than, etc.
+ * @return The ret matrix, modified appropriately.
+ */
+ public static MatrixBlock bincellOp(CompressedMatrixBlock m1, MatrixBlock m2, CompressedMatrixBlock ret,
+ BinaryOperator op) {
+ if(op.fn instanceof Minus) {
+ ScalarOperator sop = new RightScalarOperator(Multiply.getMultiplyFnObject(), -1);
+ m2 = m2.scalarOperations(sop, new MatrixBlock());
+ return LibBinaryCellOp.bincellOp(m1, m2, ret, new BinaryOperator(Plus.getPlusFnObject()));
+ }
+ if(m1.isOverlapping() && !(op.fn instanceof Multiply)) {
+ if(op.fn instanceof Plus || op.fn instanceof Minus) {
+ return binaryMVPlusStack(m1, m2, ret, op);
+ }
+ else {
+ throw new NotImplementedException(op + " not implemented for CLA");
+ }
+
+ }
+ else {
+ BinaryAccessType atype = LibMatrixBincell.getBinaryAccessType(m1, m2);
+ switch(atype) {
+ case MATRIX_ROW_VECTOR:
+ // Verify if it is okay to include all OuterVectorVector ops here.
+ return binaryMVRow(m1, m2, ret, op);
+
+ case OUTER_VECTOR_VECTOR:
+ if(m2.getNumRows() == 1 && m2.getNumColumns() == 1) {
+ return LibScalar.scalarOperations(new RightScalarOperator(op.fn, m2.quickGetValue(0, 0)),
+ m1,
+ ret,
+ m1.isOverlapping());
+ }
+ default:
+ LOG.warn("Inefficient Decompression for " + op + " " + atype);
+ MatrixBlock m1d = m1.decompress();
+ return m1d.binaryOperations(op, m2, ret);
+
+ }
+ }
+
+ }
+
+ protected static CompressedMatrixBlock binaryMVRow(CompressedMatrixBlock m1, MatrixBlock m2,
+ CompressedMatrixBlock ret, BinaryOperator op) {
+
+ // Apply the operation to each of the column groups.
+ // Most implementations will only modify metadata.
+ List<ColGroup> oldColGroups = m1.getColGroups();
+ List<ColGroup> newColGroups = new ArrayList<>(oldColGroups.size());
+ double[] v = m2.getDenseBlockValues();
+ boolean sparseSafe = true;
+ for(double x : v) {
+ if(op.fn.execute(x, 0.0) != 0.0) {
+ sparseSafe = false;
+ break;
+ }
+ }
+
+ for(ColGroup grp : oldColGroups) {
+ if(grp instanceof ColGroupUncompressed) {
+ throw new DMLCompressionException("Not supported Binary MV");
+ }
+ else {
+ if(grp.getNumCols() == 1) {
+ ScalarOperator sop = new LeftScalarOperator(op.fn, m2.getValue(0, grp.getColIndices()[0]), 1);
+ newColGroups.add(grp.scalarOperation(sop));
+ }
+ else {
+ ColGroup ncg = grp.binaryRowOp(op, v, sparseSafe);
+ newColGroups.add(ncg);
+ }
+ }
+ }
+ ret.allocateColGroupList(newColGroups);
+ ret.setNonZeros(m1.getNumColumns() * m1.getNumRows());
+ return ret;
+
+ }
+
+ protected static CompressedMatrixBlock binaryMVPlusStack(CompressedMatrixBlock m1, MatrixBlock m2,
+ CompressedMatrixBlock ret, BinaryOperator op) {
+ List<ColGroup> oldColGroups = m1.getColGroups();
+ List<ColGroup> newColGroups = new ArrayList<>(oldColGroups.size() + 1);
+ for(ColGroup grp : m1.getColGroups()) {
+ newColGroups.add(grp);
+ }
+ int[] colIndexes = oldColGroups.get(0).getColIndices();
+ double[] v = m2.getDenseBlockValues();
+ ADictionary newDict = new Dictionary(new double[colIndexes.length]);
+ newDict = newDict.applyBinaryRowOp(op.fn, v, true, colIndexes);
+ newColGroups.add(new ColGroupConst(colIndexes, m1.getNumRows(), newDict));
+ ret.allocateColGroupList(newColGroups);
+ ret.setOverlapping(true);
+ ret.setNonZeros(-1);
+ return ret;
+ }
+}
diff --git a/src/main/java/org/apache/sysds/runtime/compress/lib/LibLeftMultBy.java b/src/main/java/org/apache/sysds/runtime/compress/lib/LibLeftMultBy.java
new file mode 100644
index 0000000..004e601
--- /dev/null
+++ b/src/main/java/org/apache/sysds/runtime/compress/lib/LibLeftMultBy.java
@@ -0,0 +1,549 @@
+/*
+ * 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.compress.lib;
+
+import java.util.ArrayList;
+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 org.apache.commons.lang3.tuple.Pair;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.sysds.runtime.DMLRuntimeException;
+import org.apache.sysds.runtime.compress.CompressedMatrixBlock;
+import org.apache.sysds.runtime.compress.CompressionSettings;
+import org.apache.sysds.runtime.compress.colgroup.ColGroup;
+import org.apache.sysds.runtime.compress.colgroup.ColGroupOLE;
+import org.apache.sysds.runtime.compress.colgroup.ColGroupUncompressed;
+import org.apache.sysds.runtime.compress.colgroup.ColGroupValue;
+import org.apache.sysds.runtime.compress.utils.LinearAlgebraUtils;
+import org.apache.sysds.runtime.data.DenseBlock;
+import org.apache.sysds.runtime.data.SparseBlock;
+import org.apache.sysds.runtime.data.SparseRow;
+import org.apache.sysds.runtime.matrix.data.LibMatrixReorg;
+import org.apache.sysds.runtime.matrix.data.MatrixBlock;
+import org.apache.sysds.runtime.util.CommonThreadPool;
+
+public class LibLeftMultBy {
+ private static final Log LOG = LogFactory.getLog(LibLeftMultBy.class.getName());
+
+ public static MatrixBlock leftMultByMatrix(List<ColGroup> groups, MatrixBlock that, MatrixBlock ret,
+ boolean doTranspose, boolean allocTmp, int rl, int cl, boolean overlapping, int k, Pair<Integer, int[]> v) {
+
+ if(ret == null)
+ ret = new MatrixBlock(rl, cl, false, rl * cl);
+ else if(!(ret.getNumColumns() == cl && ret.getNumRows() == rl && ret.isAllocated()))
+ ret.reset(rl, cl, false, rl * cl);
+ that = that instanceof CompressedMatrixBlock ? ((CompressedMatrixBlock) that).decompress() : that;
+
+ // if(that.getNumRows() == 1) {
+ // if(k > 1) {
+ // return leftMultByVectorTranspose(groups, that, ret, doTranspose, k, v, overlapping);
+ // }
+ // else {
+ // return leftMultByVectorTranspose(groups, that, ret, doTranspose, true, v, overlapping);
+ // }
+ // }
+ // else {
+ return leftMultByMatrix(groups, that, ret, k, cl, v, overlapping);
+ // }
+ }
+
+ public static void leftMultByTransposeSelf(List<ColGroup> groups, MatrixBlock result, int gl, int gu, int k,
+ int numColumns, Pair<Integer, int[]> v, boolean overlapping) {
+ if(k <= 1 || overlapping) {
+ leftMultByTransposeSelf(groups, result, gl, gu, v, overlapping);
+ }
+ else {
+ try {
+ ExecutorService pool = CommonThreadPool.get(k);
+ ArrayList<MatrixMultTransposeTask> tasks = new ArrayList<>();
+ int numgrp = groups.size();
+ int blklen = (int) (Math.ceil((double) numgrp / (2 * k)));
+ for(int i = 0; i < 2 * k & i * blklen < numColumns; i++)
+ tasks.add(new MatrixMultTransposeTask(groups, result, i * blklen,
+ Math.min((i + 1) * blklen, numgrp), v, overlapping));
+ List<Future<Object>> ret = pool.invokeAll(tasks);
+ for(Future<Object> tret : ret)
+ tret.get(); // check for errors
+ pool.shutdown();
+ }
+ catch(InterruptedException | ExecutionException e) {
+ throw new DMLRuntimeException(e);
+ }
+ }
+ }
+
+ private static MatrixBlock leftMultByMatrix(List<ColGroup> colGroups, MatrixBlock that, MatrixBlock ret, int k,
+ int numColumns, Pair<Integer, int[]> v, boolean overlapping) {
+ ret.allocateDenseBlock();
+ if(that.isInSparseFormat()) {
+ ret = leftMultBySparseMatrix(colGroups, that, ret, k, numColumns, v);
+ }
+ else {
+ ret = leftMultByDenseMatrix(colGroups, that, ret, k, numColumns, v, overlapping);
+ }
+
+ ret.setNonZeros(ret.getNumColumns() * ret.getNumRows());
+ return ret;
+ }
+
+ private static MatrixBlock leftMultByDenseMatrix(List<ColGroup> colGroups, MatrixBlock that, MatrixBlock ret, int k,
+ int numColumns, Pair<Integer, int[]> v, boolean overlapping) {
+ DenseBlock db = that.getDenseBlock();
+ if(db == null)
+ throw new DMLRuntimeException("Invalid LeftMult By Dense matrix, input matrix was sparse");
+
+ double[] retV = ret.getDenseBlockValues();
+ double[] thatV;
+ int blockU;
+ int blockL = 0;
+ for(ColGroup grp : colGroups)
+ if(grp instanceof ColGroupUncompressed)
+ ((ColGroupUncompressed) grp).leftMultByMatrix(that, ret);
+
+ for(int b = 0; b < db.numBlocks(); b++) {
+ int blockSize = db.blockSize(b);
+ blockU = Math.min(blockL + blockSize, ret.getNumRows());
+ thatV = db.valuesAt(b);
+
+ if(k == 1 || overlapping) {
+ // Pair<Integer, int[]> v = getMaxNumValues(colGroups);
+ for(int j = 0; j < colGroups.size(); j++) {
+ colGroups.get(j).leftMultByMatrix(thatV,
+ retV,
+ colGroups.get(j).getValues(),
+ that.getNumRows(),
+ ret.getNumColumns(),
+ 0,
+ ret.getNumRows(),
+ 0);
+ }
+ }
+ else {
+ try {
+ ExecutorService pool = CommonThreadPool.get(k);
+ // compute remaining compressed column groups in parallel
+ ArrayList<LeftMatrixMatrixMultTask> tasks = new ArrayList<>();
+ int rowBlockSize = 1;
+ for(int blo = blockL; blo < blockU; blo += rowBlockSize) {
+ tasks.add(new LeftMatrixMatrixMultTask(colGroups, thatV, retV, that.getNumRows(), numColumns,
+ blo, Math.min(blo + rowBlockSize, blockU), blo - blockL, v));
+ }
+
+ List<Future<Object>> futures = pool.invokeAll(tasks);
+
+ pool.shutdown();
+ for(Future<Object> future : futures)
+ future.get();
+ }
+ catch(InterruptedException | ExecutionException e) {
+ throw new DMLRuntimeException(e);
+ }
+ }
+ blockL += blockSize;
+ }
+ return ret;
+ }
+
+ private static MatrixBlock leftMultByVectorTranspose(List<ColGroup> colGroups, MatrixBlock vector,
+ MatrixBlock result, boolean doTranspose, boolean allocTmp, Pair<Integer, int[]> v, boolean overlap) {
+
+ MatrixBlock rowVector = vector;
+ // Note that transpose here is a metadata operation since the input is a vector.
+ if(doTranspose) {
+ rowVector = new MatrixBlock(1, vector.getNumRows(), false);
+ LibMatrixReorg.transpose(vector, rowVector);
+ }
+
+ // initialize and allocate the result
+ result.reset();
+ result.allocateDenseBlock();
+
+ // setup memory pool for reuse
+ if(allocTmp) {
+ // Pair<Integer, int[]> v = getMaxNumValues(colGroups);
+ ColGroupValue.setupThreadLocalMemory(v.getLeft() + 1); // +1 for efficiency in DDC groups.
+ for(int i = 0; i < colGroups.size(); i++) {
+ colGroups.get(i).leftMultByRowVector(rowVector.getDenseBlockValues(),
+ result.getDenseBlockValues(),
+ v.getRight()[i]);
+ }
+ }
+ else {
+
+ for(ColGroup grp : colGroups) {
+ grp.leftMultByRowVector(rowVector.getDenseBlockValues(), result.getDenseBlockValues(), -1);
+ }
+ }
+
+ // delegate matrix-vector operation to each column group
+
+ // post-processing
+ if(allocTmp)
+ ColGroupValue.cleanupThreadLocalMemory();
+ result.recomputeNonZeros();
+
+ return result;
+ }
+
+ public static MatrixBlock leftMultByVectorTranspose(List<ColGroup> colGroups, MatrixBlock vector,
+ MatrixBlock result, boolean doTranspose, int k, Pair<Integer, int[]> v, boolean overlap) {
+ // transpose vector if required
+ MatrixBlock rowVector = vector;
+ if(doTranspose) {
+ rowVector = new MatrixBlock(1, vector.getNumRows(), false);
+ LibMatrixReorg.transpose(vector, rowVector);
+ }
+
+ // initialize and allocate the result
+ result.reset();
+ result.allocateDenseBlock();
+
+ // multi-threaded execution
+ try {
+ // compute uncompressed column group in parallel
+ // ColGroupUncompressed uc = getUncompressedColGroup();
+ // if(uc != null)
+ // uc.leftMultByRowVector(rowVector, result, k);
+
+ // compute remaining compressed column groups in parallel
+ ExecutorService pool = CommonThreadPool.get(Math.min(colGroups.size(), k));
+ ArrayList<LeftMatrixVectorMultTask> tasks = new ArrayList<>();
+
+ // if(overlap){
+ tasks.add(new LeftMatrixVectorMultTask(colGroups, rowVector, result, v));
+ // } else{
+ // ArrayList<ColGroup>[] grpParts = createStaticTaskPartitioning(colGroups, 4 * k, true);
+ // for(ArrayList<ColGroup> groups : grpParts)
+ // tasks.add(new LeftMatrixVectorMultTask(groups, rowVector, result, v));
+ // }
+
+ List<Future<Object>> ret = pool.invokeAll(tasks);
+ pool.shutdown();
+ for(Future<Object> tmp : ret)
+ tmp.get();
+
+ }
+ catch(InterruptedException | ExecutionException e) {
+ LOG.error(e);
+ throw new DMLRuntimeException(e);
+ }
+
+ // post-processing
+ result.recomputeNonZeros();
+
+ return result;
+ }
+
+ private static MatrixBlock leftMultBySparseMatrix(List<ColGroup> colGroups, MatrixBlock that, MatrixBlock ret,
+ int k, int numColumns, Pair<Integer, int[]> v) {
+
+ SparseBlock sb = that.getSparseBlock();
+ if(sb == null)
+ throw new DMLRuntimeException("Invalid Left Mult by Sparse matrix, input matrix was dense");
+
+ for(ColGroup grp : colGroups) {
+ if(grp instanceof ColGroupUncompressed)
+ ((ColGroupUncompressed) grp).leftMultByMatrix(that, ret);
+ }
+
+ if(k == 1) {
+ double[][] materialized = new double[colGroups.size()][];
+ boolean containsOLE = false;
+ for(int i = 0; i < colGroups.size(); i++) {
+ materialized[i] = colGroups.get(i).getValues();
+ if(colGroups.get(i) instanceof ColGroupOLE) {
+ containsOLE = true;
+ }
+ }
+ double[] materializedRow = containsOLE ? new double[CompressionSettings.BITMAP_BLOCK_SZ * 2] : null;
+
+ for(int r = 0; r < that.getNumRows(); r++) {
+ SparseRow row = sb.get(r);
+ if(row != null) {
+
+ for(int j = 0; j < colGroups.size(); j++) {
+ colGroups.get(j).leftMultBySparseMatrix(row.size(),
+ row.indexes(),
+ row.values(),
+ ret.getDenseBlockValues(),
+ v.getRight()[j],
+ materialized[j],
+ that.getNumRows(),
+ ret.getNumColumns(),
+ r,
+ materializedRow);
+ }
+ }
+ }
+ }
+ else {
+ ExecutorService pool = CommonThreadPool.get(k);
+ ArrayList<LeftMatrixSparseMatrixMultTask> tasks = new ArrayList<>();
+ try {
+ // compute remaining compressed column groups in parallel
+ // List<ColGroup>[] parts = createStaticTaskPartitioningForSparseMatrixMult(colGroups, k, false);
+ // for(List<ColGroup> part : parts) {
+ tasks.add(new LeftMatrixSparseMatrixMultTask(colGroups, sb, ret.getDenseBlockValues(),
+ that.getNumRows(), numColumns, v));
+ // }
+
+ List<Future<Object>> futures = pool.invokeAll(tasks);
+ pool.shutdown();
+ for(Future<Object> future : futures)
+ future.get();
+ }
+ catch(InterruptedException | ExecutionException e) {
+ throw new DMLRuntimeException(e);
+ }
+ }
+
+ return ret;
+
+ }
+
+ private static void leftMultByTransposeSelf(List<ColGroup> groups, MatrixBlock result, int gl, int gu,
+ Pair<Integer, int[]> v, boolean overlapping) {
+ final int numRows = groups.get(0).getNumRows();
+
+ // preallocated dense tmp matrix blocks
+ MatrixBlock lhs = new MatrixBlock(1, numRows, false);
+ MatrixBlock tmpret = new MatrixBlock(1, result.getNumColumns(), false);
+ lhs.allocateDenseBlock();
+ tmpret.allocateDenseBlock();
+
+ // setup memory pool for reuse
+ ColGroupValue.setupThreadLocalMemory(v.getLeft() + 1);
+
+ // approach: for each colgroup, extract uncompressed columns one at-a-time
+ // vector-matrix multiplies against remaining col groups
+ // for(int i = gl; i < gu; i++) {
+ // get current group and relevant col groups
+ // ColGroup group = groups.get(i);
+ // int[] ixgroup = group.getColIndices();
+ // List<ColGroup> tmpList = groups.subList(i, numGroups);
+
+ // if(group instanceof ColGroupDDC // single DDC group
+ // && ixgroup.length == 1 && !containsUC && numRows < CompressionSettings.BITMAP_BLOCK_SZ) {
+ // // compute vector-matrix partial result
+ // leftMultByVectorTranspose(tmpList, (ColGroupDDC) group, tmpret);
+
+ // // write partial results (disjoint non-zeros)
+ // LinearAlgebraUtils.copyNonZerosToUpperTriangle(result, tmpret, ixgroup[0]);
+ // }
+ // else {
+ // for all uncompressed lhs columns vectors
+ for(int j = 0; j < result.getNumColumns(); j++) {
+ ColGroup.decompressToBlock(lhs, j, groups);
+
+ if(!lhs.isEmptyBlock(false)) {
+ // tmpret.reset();
+ // compute vector-matrix partial result
+ // leftMultByMatrix(groups,lhs, tmpret, false, true, 0, 0, overlapping, 1, v );
+ leftMultByVectorTranspose(groups, lhs, tmpret, false, true, v, overlapping);
+ // LOG.error(tmpret);
+
+ // write partial results (disjoint non-zeros)
+ LinearAlgebraUtils.copyNonZerosToUpperTriangle(result, tmpret, j);
+ }
+ lhs.reset();
+ // }
+ // }
+ }
+
+ // post processing
+ ColGroupValue.cleanupThreadLocalMemory();
+ }
+
+ private static class LeftMatrixVectorMultTask implements Callable<Object> {
+ private final List<ColGroup> _groups;
+ private final MatrixBlock _vect;
+ private final MatrixBlock _ret;
+ private final Pair<Integer, int[]> _v;
+
+ protected LeftMatrixVectorMultTask(List<ColGroup> groups, MatrixBlock vect, MatrixBlock ret,
+ Pair<Integer, int[]> v) {
+ _groups = groups;
+ _vect = vect;
+ _ret = ret;
+ _v = v;
+ }
+
+ @Override
+ public Object call() {
+ // setup memory pool for reuse
+ try {
+ ColGroupValue.setupThreadLocalMemory(_v.getLeft() + 1);
+ for(int i = 0; i < _groups.size(); i++) {
+ _groups.get(i)
+ .leftMultByRowVector(_vect.getDenseBlockValues(), _ret.getDenseBlockValues(), _v.getRight()[i]);
+ }
+
+ ColGroupValue.cleanupThreadLocalMemory();
+ }
+ catch(Exception e) {
+ throw new DMLRuntimeException(e);
+ }
+ return null;
+ }
+ }
+
+ private static class LeftMatrixMatrixMultTask implements Callable<Object> {
+ private final List<ColGroup> _group;
+ private final double[] _that;
+ private final double[] _ret;
+ private final int _numRows;
+ private final int _numCols;
+ private final int _rl;
+ private final int _ru;
+ private final int _vOff;
+ private final Pair<Integer, int[]> _v;
+
+ protected LeftMatrixMatrixMultTask(List<ColGroup> group, double[] that, double[] ret, int numRows, int numCols,
+ int rl, int ru, int vOff, Pair<Integer, int[]> v) {
+ _group = group;
+ _that = that;
+ _ret = ret;
+ _numRows = numRows;
+ _numCols = numCols;
+ _rl = rl;
+ _ru = ru;
+ _vOff = vOff;
+ _v = v;
+ }
+
+ @Override
+ public Object call() {
+ // setup memory pool for reuse
+
+ double[][] materialized = new double[_group.size()][];
+ for(int i = 0; i < _group.size(); i++) {
+ materialized[i] = _group.get(i).getValues();
+ }
+ // Pair<Integer, int[]> v = getMaxNumValues(_group);
+ try {
+ ColGroupValue.setupThreadLocalMemory(_v.getLeft() + 1);
+ for(int j = 0; j < _group.size(); j++) {
+ _group.get(j).leftMultByMatrix(_that, _ret, materialized[j], _numRows, _numCols, _rl, _ru, _vOff);
+ }
+ ColGroupValue.cleanupThreadLocalMemory();
+
+ }
+ catch(Exception e) {
+ throw new DMLRuntimeException(e);
+ }
+ return null;
+ }
+ }
+
+ private static class LeftMatrixSparseMatrixMultTask implements Callable<Object> {
+ private final List<ColGroup> _group;
+ private final SparseBlock _that;
+ private final double[] _ret;
+ private final int _numRows;
+ private final int _numCols;
+ private final Pair<Integer, int[]> _v;
+
+ protected LeftMatrixSparseMatrixMultTask(List<ColGroup> group, SparseBlock that, double[] ret, int numRows,
+ int numCols, Pair<Integer, int[]> v) {
+ _group = group;
+ _that = that;
+ _ret = ret;
+ _numRows = numRows;
+ _numCols = numCols;
+ _v = v;
+ }
+
+ @Override
+ public Object call() {
+ // setup memory pool for reuse
+
+ // double[][] materialized = new double[_group.size()][];
+ // for(int i = 0; i < _group.size(); i++) {
+ // materialized[i] = _group.get(i).getValues();
+ // }
+
+ boolean containsOLE = false;
+ for(int j = 0; j < _group.size(); j++) {
+ if(_group.get(j) instanceof ColGroupOLE) {
+ containsOLE = true;
+ }
+ }
+ // Temporary Array to store 2 * block size in
+ double[] tmpA = containsOLE ? new double[CompressionSettings.BITMAP_BLOCK_SZ * 2] : null;
+
+ ColGroupValue.setupThreadLocalMemory(_v.getLeft());
+ try {
+ for(int j = 0; j < _group.size(); j++) {
+ double[] materializedV = _group.get(j).getValues();
+ for(int r = 0; r < _that.numRows(); r++) {
+ if(_that.get(r) != null) {
+ _group.get(j).leftMultBySparseMatrix(_that.get(r).size(),
+ _that.get(r).indexes(),
+ _that.get(r).values(),
+ _ret,
+ _v.getRight()[j],
+ materializedV,
+ _numRows,
+ _numCols,
+ r,
+ tmpA);
+ }
+ }
+ }
+ }
+ catch(Exception e) {
+ e.printStackTrace();
+ throw new DMLRuntimeException(e);
+ }
+ ColGroupValue.cleanupThreadLocalMemory();
+ return null;
+ }
+ }
+
+ private static class MatrixMultTransposeTask implements Callable<Object> {
+ private final List<ColGroup> _groups;
+ private final MatrixBlock _ret;
+ private final int _gl;
+ private final int _gu;
+ private final Pair<Integer, int[]> _v;
+ private final boolean _overlapping;
+
+ protected MatrixMultTransposeTask(List<ColGroup> groups, MatrixBlock ret, int gl, int gu,
+ Pair<Integer, int[]> v, boolean overlapping) {
+ _groups = groups;
+ _ret = ret;
+ _gl = gl;
+ _gu = gu;
+ _v = v;
+ _overlapping = overlapping;
+ }
+
+ @Override
+ public Object call() {
+ leftMultByTransposeSelf(_groups, _ret, _gl, _gu, _v, _overlapping);
+ return null;
+ }
+ }
+}
diff --git a/src/main/java/org/apache/sysds/runtime/compress/lib/LibRightMultBy.java b/src/main/java/org/apache/sysds/runtime/compress/lib/LibRightMultBy.java
new file mode 100644
index 0000000..df0f45e
--- /dev/null
+++ b/src/main/java/org/apache/sysds/runtime/compress/lib/LibRightMultBy.java
@@ -0,0 +1,645 @@
+/*
+ * 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.compress.lib;
+
+import java.util.ArrayList;
+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 org.apache.commons.lang3.tuple.Pair;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.sysds.runtime.DMLCompressionException;
+import org.apache.sysds.runtime.DMLRuntimeException;
+import org.apache.sysds.runtime.compress.CompressedMatrixBlock;
+import org.apache.sysds.runtime.compress.CompressionSettings;
+import org.apache.sysds.runtime.compress.colgroup.ColGroup;
+import org.apache.sysds.runtime.compress.colgroup.ColGroupOLE;
+import org.apache.sysds.runtime.compress.colgroup.ColGroupUncompressed;
+import org.apache.sysds.runtime.compress.colgroup.ColGroupValue;
+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.util.CommonThreadPool;
+
+public class LibRightMultBy {
+ private static final Log LOG = LogFactory.getLog(LibRightMultBy.class.getName());
+
+ /**
+ * Right multiply by matrix. Meaning a left hand side compressed matrix is multiplied with a right hand side
+ * uncompressed matrix.
+ *
+ * @param colGroups All Column groups in the compression
+ * @param that The right hand side matrix
+ * @param ret The MatrixBlock to return.
+ * @param k The parallelization degree to use.
+ * @param v The Precalculated counts and Maximum number of tuple entries in the column groups.
+ * @param allowOverlap Allow the multiplication to return an overlapped matrix.
+ * @return The Result Matrix, modified from the ret parameter.
+ */
+ public static MatrixBlock rightMultByMatrix(List<ColGroup> colGroups, MatrixBlock that, MatrixBlock ret, int k,
+ Pair<Integer, int[]> v, boolean allowOverlap) {
+
+ boolean containsUncompressable = false;
+ int distinctCount = 0;
+ for(ColGroup g : colGroups) {
+ if(g instanceof ColGroupValue) {
+ distinctCount += ((ColGroupValue) g).getNumValues();
+ }
+ else {
+ containsUncompressable = true;
+ }
+ }
+ int rl = colGroups.get(0).getNumRows();
+ int cl = that.getNumColumns();
+ if(!allowOverlap || (containsUncompressable || distinctCount >= rl / 2)) {
+ if(ret == null)
+ ret = new MatrixBlock(rl, cl, false, rl * cl);
+ else if(!(ret.getNumColumns() == cl && ret.getNumRows() == rl && ret.isAllocated()))
+ ret.reset(rl, cl, false, rl * cl);
+ ret.allocateDenseBlock();
+ if(that.isInSparseFormat()) {
+ ret = rightMultBySparseMatrix(colGroups, that, ret, k, v);
+ }
+ else {
+ ret = rightMultByDenseMatrix(colGroups, that, ret, k, v);
+
+ }
+ ret.setNonZeros(ret.getNumColumns() * ret.getNumRows());
+ }
+ else {
+ // Create an overlapping compressed Matrix Block.
+ ret = new CompressedMatrixBlock(true);
+
+ ret.setNumColumns(cl);
+ ret.setNumRows(rl);
+ CompressedMatrixBlock retC = (CompressedMatrixBlock) ret;
+ retC.setOverlapping(true);
+ if(that.isInSparseFormat()) {
+ ret = rightMultBySparseMatrixCompressed(colGroups, that, retC, k, v);
+ }
+ else {
+ ret = rightMultByDenseMatrixCompressed(colGroups, that, retC, k, v);
+ }
+ }
+
+ return ret;
+
+ }
+
+ /**
+ * Multi-threaded version of rightMultByVector.
+ *
+ * @param colGroups The Column groups used int the multiplication
+ * @param vector matrix block vector to multiply with
+ * @param result matrix block result to modify in the multiplication
+ * @param k number of threads to use
+ * @param v The Precalculated counts and Maximum number of tuple entries in the column groups
+ */
+ public static void rightMultByVector(List<ColGroup> colGroups, MatrixBlock vector, MatrixBlock result, int k,
+ Pair<Integer, int[]> v) {
+ // initialize and allocate the result
+ result.allocateDenseBlock();
+ if(k <= 1) {
+ rightMultByVector(colGroups, vector, result, v);
+ return;
+ }
+
+ // multi-threaded execution of all groups
+ try {
+ // ColGroupUncompressed uc = getUncompressedColGroup();
+
+ // compute uncompressed column group in parallel
+ // if(uc != null)
+ // uc.rightMultByVector(vector, result, k);
+
+ // compute remaining compressed column groups in parallel
+ // note: OLE needs alignment to segment size, otherwise wrong entry
+ ExecutorService pool = CommonThreadPool.get(k);
+ int rlen = colGroups.get(0).getNumRows();
+ int seqsz = CompressionSettings.BITMAP_BLOCK_SZ;
+ int blklen = (int) (Math.ceil((double) rlen / k));
+ blklen += (blklen % seqsz != 0) ? seqsz - blklen % seqsz : 0;
+
+ ArrayList<RightMatrixVectorMultTask> tasks = new ArrayList<>();
+ for(int i = 0; i < k & i * blklen < rlen; i++) {
+ tasks.add(new RightMatrixVectorMultTask(colGroups, vector, result, i * blklen,
+ Math.min((i + 1) * blklen, rlen), v));
+ }
+
+ List<Future<Long>> ret = pool.invokeAll(tasks);
+ pool.shutdown();
+
+ // error handling and nnz aggregation
+ long lnnz = 0;
+ for(Future<Long> tmp : ret)
+ lnnz += tmp.get();
+ result.setNonZeros(lnnz);
+ }
+ catch(InterruptedException | ExecutionException e) {
+ throw new DMLRuntimeException(e);
+ }
+ }
+
+ /**
+ * Multiply this matrix block by a column vector on the right.
+ *
+ * @param vector right-hand operand of the multiplication
+ * @param result buffer to hold the result; must have the appropriate size already
+ * @param v The Precalculated counts and Maximum number of tuple entries in the column groups.
+ */
+ private static void rightMultByVector(List<ColGroup> colGroups, MatrixBlock vector, MatrixBlock result,
+ Pair<Integer, int[]> v) {
+
+ // delegate matrix-vector operation to each column group
+ rightMultByVector(colGroups, vector, result, 0, result.getNumRows(), v);
+
+ // post-processing
+ result.recomputeNonZeros();
+ }
+
+ private static MatrixBlock rightMultBySparseMatrix(List<ColGroup> colGroups, MatrixBlock that, MatrixBlock ret,
+ int k, Pair<Integer, int[]> v) {
+ SparseBlock sb = that.getSparseBlock();
+ double[] retV = ret.getDenseBlockValues();
+
+ if(sb == null)
+ throw new DMLRuntimeException("Invalid Right Mult by Sparse matrix, input matrix was dense");
+
+ for(ColGroup grp : colGroups) {
+ if(grp instanceof ColGroupUncompressed)
+ ((ColGroupUncompressed) grp).rightMultByMatrix(that, ret, 0, ret.getNumColumns());
+ }
+
+ // Pair<Integer, int[]> v = Util.getMaxNumValues(colGroups);
+ // if(k == 1) {
+ for(int j = 0; j < colGroups.size(); j++) {
+ double[] preAggregatedB = ((ColGroupValue) colGroups.get(j)).preaggValues(v.getRight()[j],
+ sb,
+ colGroups.get(j).getValues(),
+ 0,
+ that.getNumColumns(),
+ that.getNumColumns());
+ colGroups.get(j).rightMultByMatrix(preAggregatedB,
+ retV,
+ that.getNumColumns(),
+ 0,
+ ret.getNumRows(),
+ 0,
+ that.getNumColumns());
+
+ }
+ // }
+ // else {
+ // ExecutorService pool = CommonThreadPool.get(k);
+ // ArrayList<RightMultBySparseMatrixTask> tasks = new ArrayList<>();
+ // try {
+
+ // for(int j = 0; j < ret.getNumColumns(); j += CompressionSettings.BITMAP_BLOCK_SZ) {
+ // tasks.add(new RightMultBySparseMatrixTask(colGroups, retV, sb, materialized, v, numColumns, j,
+ // Math.min(j + CompressionSettings.BITMAP_BLOCK_SZ, ret.getNumColumns())));
+ // }
+
+ // List<Future<Object>> futures = pool.invokeAll(tasks);
+ // pool.shutdown();
+ // for(Future<Object> future : futures)
+ // future.get();
+ // }
+ // catch(InterruptedException | ExecutionException e) {
+ // throw new DMLRuntimeException(e);
+ // }
+ // }
+
+ return ret;
+ }
+
+ private static MatrixBlock rightMultByDenseMatrix(List<ColGroup> colGroups, MatrixBlock that, MatrixBlock ret,
+ int k, Pair<Integer, int[]> v) {
+
+ // long StartTime = System.currentTimeMillis();
+ DenseBlock db = that.getDenseBlock();
+ double[] retV = ret.getDenseBlockValues();
+ double[] thatV;
+
+ for(ColGroup grp : colGroups) {
+ if(grp instanceof ColGroupUncompressed) {
+ ((ColGroupUncompressed) grp).rightMultByMatrix(that, ret, 0, ret.getNumRows());
+ }
+ }
+
+ if(k == 1) {
+ ColGroupValue.setupThreadLocalMemory((v.getLeft()));
+ for(int b = 0; b < db.numBlocks(); b++) {
+ // int blockSize = db.blockSize(b);
+ thatV = db.valuesAt(b);
+ for(int j = 0; j < colGroups.size(); j++) {
+ int colBlockSize = 128;
+ for(int i = 0; i < that.getNumColumns(); i += colBlockSize) {
+ if(colGroups.get(j) instanceof ColGroupValue) {
+ double[] preAggregatedB = ((ColGroupValue) colGroups.get(j)).preaggValues(v.getRight()[j],
+ thatV,
+ colGroups.get(j).getValues(),
+ i,
+ Math.min(i + colBlockSize, that.getNumColumns()),
+ that.getNumColumns());
+ int blklenRows = CompressionSettings.BITMAP_BLOCK_SZ;
+ for(int n = 0; n * blklenRows < ret.getNumRows(); n++) {
+ colGroups.get(j).rightMultByMatrix(preAggregatedB,
+ retV,
+ that.getNumColumns(),
+ n * blklenRows,
+ Math.min((n + 1) * blklenRows, ret.getNumRows()),
+ i,
+ Math.min(i + colBlockSize, that.getNumColumns()));
+ }
+ }
+ }
+ }
+ }
+ ColGroupValue.cleanupThreadLocalMemory();
+ }
+ else {
+
+ thatV = db.valuesAt(0);
+ ExecutorService pool = CommonThreadPool.get(k);
+ ArrayList<RightMatrixMultTask> tasks = new ArrayList<>();
+ ArrayList<RightMatrixPreAggregateTask> preTask = new ArrayList<>(colGroups.size());
+ // Pair<Integer, int[]> v;
+ final int blkz = CompressionSettings.BITMAP_BLOCK_SZ;
+ int blklenRows = (int) (Math.ceil((double) ret.getNumRows() / (2 * k)));
+
+ try {
+ List<Future<double[]>> ag = pool.invokeAll(preAggregate(colGroups, thatV, that, preTask, v));
+ // DDC and RLE
+ for(int j = 0; j * blklenRows < ret.getNumRows(); j++) {
+ RightMatrixMultTask rmmt = new RightMatrixMultTask(colGroups, retV, ag, v, that.getNumColumns(),
+ j * blklenRows, Math.min((j + 1) * blklenRows, ret.getNumRows()), 0, that.getNumColumns(),
+ false, false);
+ tasks.add(rmmt);
+ }
+ blklenRows += (blklenRows % blkz != 0) ? blkz - blklenRows % blkz : 0;
+ // OLE!
+ for(int j = 0; j * blklenRows < ret.getNumRows(); j++) {
+ RightMatrixMultTask rmmt = new RightMatrixMultTask(colGroups, retV, ag, v, that.getNumColumns(),
+ j * blklenRows, Math.min((j + 1) * blklenRows, ret.getNumRows()), 0, that.getNumColumns(),
+ false, true);
+ tasks.add(rmmt);
+ }
+ for(Future<Object> future : pool.invokeAll(tasks))
+ future.get();
+ tasks.clear();
+
+ }
+ catch(InterruptedException | ExecutionException e) {
+ throw new DMLRuntimeException(e);
+ }
+ }
+
+ return ret;
+ }
+
+ private static MatrixBlock rightMultByDenseMatrixCompressed(List<ColGroup> colGroups, MatrixBlock that,
+ CompressedMatrixBlock ret, int k, Pair<Integer, int[]> v) {
+
+ DenseBlock db = that.getDenseBlock();
+ double[] thatV;
+
+ for(ColGroup grp : colGroups) {
+ if(grp instanceof ColGroupUncompressed) {
+ throw new DMLCompressionException(
+ "Right Mult by dense with compressed output is not efficient to do with uncompressed Compressed ColGroups and therefore not supported.");
+ }
+ }
+
+ thatV = db.valuesAt(0);
+ List<ColGroup> retCg = new ArrayList<ColGroup>();
+ int[] newColIndexes = new int[that.getNumColumns()];
+ for(int i = 0; i < that.getNumColumns(); i++) {
+ newColIndexes[i] = i;
+ }
+ if(k == 1) {
+ for(int j = 0; j < colGroups.size(); j++) {
+ ColGroupValue g = (ColGroupValue) colGroups.get(j);
+ double[] preAggregatedB = g.preaggValues(v.getRight()[j],
+ thatV,
+ g.getValues(),
+ 0,
+ that.getNumColumns(),
+ that.getNumColumns(),
+ new double[v.getRight()[j] * that.getNumColumns()]);
+ retCg.add(g.copyAndSet(newColIndexes, preAggregatedB));
+ }
+ }
+ else {
+ thatV = db.valuesAt(0);
+ ExecutorService pool = CommonThreadPool.get(k);
+ ArrayList<RightMatrixPreAggregateTask> preTask = new ArrayList<>(colGroups.size());
+
+ try {
+ List<Future<double[]>> ag = pool.invokeAll(preAggregate(colGroups, thatV, that, preTask, v));
+ for(int j = 0; j < colGroups.size(); j++) {
+ retCg.add(((ColGroupValue) colGroups.get(j)).copyAndSet(newColIndexes, ag.get(j).get()));
+ }
+ }
+ catch(InterruptedException | ExecutionException e) {
+ throw new DMLRuntimeException(e);
+ }
+ }
+ ret.allocateColGroupList(retCg);
+ ret.setOverlapping(true);
+ ret.setNonZeros(-1);
+
+ return ret;
+ }
+
+ private static MatrixBlock rightMultBySparseMatrixCompressed(List<ColGroup> colGroups, MatrixBlock that,
+ CompressedMatrixBlock ret, int k, Pair<Integer, int[]> v) {
+
+ // long StartTime = System.currentTimeMillis();
+ SparseBlock sb = that.getSparseBlock();
+
+ for(ColGroup grp : colGroups) {
+ if(grp instanceof ColGroupUncompressed) {
+ throw new DMLCompressionException(
+ "Right Mult by dense with compressed output is not efficient to do with uncompressed Compressed ColGroups and therefore not supported.");
+ }
+ }
+
+ List<ColGroup> retCg = new ArrayList<ColGroup>();
+ int[] newColIndexes = new int[that.getNumColumns()];
+ for(int i = 0; i < that.getNumColumns(); i++) {
+ newColIndexes[i] = i;
+ }
+ if(k == 1) {
+ for(int j = 0; j < colGroups.size(); j++) {
+ ColGroupValue g = (ColGroupValue) colGroups.get(j);
+ double[] preAggregatedB = g.preaggValues(v.getRight()[j],
+ sb,
+ colGroups.get(j).getValues(),
+ 0,
+ that.getNumColumns(),
+ that.getNumColumns(),
+ new double[v.getRight()[j] * that.getNumColumns()]);
+ retCg.add(g.copyAndSet(newColIndexes, preAggregatedB));
+ }
+ }
+ else {
+ ExecutorService pool = CommonThreadPool.get(k);
+ ArrayList<RightMatrixPreAggregateSparseTask> preTask = new ArrayList<>(colGroups.size());
+
+ try {
+ List<Future<double[]>> ag = pool.invokeAll(preAggregate(colGroups, sb, that, preTask, v));
+ for(int j = 0; j < colGroups.size(); j++) {
+ retCg.add(((ColGroupValue) colGroups.get(j)).copyAndSet(newColIndexes, ag.get(j).get()));
+ }
+ }
+ catch(InterruptedException | ExecutionException e) {
+ throw new DMLRuntimeException(e);
+ }
+ }
+ ret.allocateColGroupList(retCg);
+ ret.setOverlapping(true);
+ ret.setNonZeros(-1);
+
+ return ret;
+ }
+
+ private static ArrayList<RightMatrixPreAggregateTask> preAggregate(List<ColGroup> colGroups, double[] thatV,
+ MatrixBlock that, ArrayList<RightMatrixPreAggregateTask> preTask, Pair<Integer, int[]> v) {
+ preTask.clear();
+ for(int h = 0; h < colGroups.size(); h++) {
+ RightMatrixPreAggregateTask pAggT = new RightMatrixPreAggregateTask((ColGroupValue) colGroups.get(h),
+ v.getRight()[h], thatV, colGroups.get(h).getValues(), 0, that.getNumColumns(), that.getNumColumns());
+ preTask.add(pAggT);
+ }
+ return preTask;
+ }
+
+ private static ArrayList<RightMatrixPreAggregateSparseTask> preAggregate(List<ColGroup> colGroups, SparseBlock sb,
+ MatrixBlock that, ArrayList<RightMatrixPreAggregateSparseTask> preTask, Pair<Integer, int[]> v) {
+ preTask.clear();
+ for(int h = 0; h < colGroups.size(); h++) {
+ RightMatrixPreAggregateSparseTask pAggT = new RightMatrixPreAggregateSparseTask(
+ (ColGroupValue) colGroups.get(h), v.getRight()[h], sb, colGroups.get(h).getValues(), 0,
+ that.getNumColumns(), that.getNumColumns());
+ preTask.add(pAggT);
+ }
+ return preTask;
+ }
+
+ private static void rightMultByVector(List<ColGroup> groups, MatrixBlock vect, MatrixBlock ret, int rl, int ru,
+ Pair<Integer, int[]> v) {
+ // + 1 to enable containing a single 0 value in the dictionary that was not materialized.
+ // This is to handle the case of a DDC dictionary not materializing the zero values.
+ // A fine tradeoff!
+ ColGroupValue.setupThreadLocalMemory(v.getLeft() + 1);
+
+ // boolean cacheDDC1 = ru - rl > CompressionSettings.BITMAP_BLOCK_SZ * 2;
+
+ // process uncompressed column group (overwrites output)
+ // if(inclUC) {
+ for(ColGroup grp : groups) {
+ if(grp instanceof ColGroupUncompressed)
+ ((ColGroupUncompressed) grp).rightMultByVector(vect, ret, rl, ru);
+ }
+
+ // process cache-conscious DDC1 groups (adds to output)
+
+ // if(cacheDDC1) {
+ // ArrayList<ColGroupDDC1> tmp = new ArrayList<>();
+ // for(ColGroup grp : groups)
+ // if(grp instanceof ColGroupDDC1)
+ // tmp.add((ColGroupDDC1) grp);
+ // if(!tmp.isEmpty())
+ // ColGroupDDC1.rightMultByVector(tmp.toArray(new ColGroupDDC1[0]), vect, ret, rl, ru);
+ // }
+ // process remaining groups (adds to output)
+ double[] values = ret.getDenseBlockValues();
+ for(ColGroup grp : groups) {
+ if(!(grp instanceof ColGroupUncompressed)) {
+ grp.rightMultByVector(vect.getDenseBlockValues(), values, rl, ru, grp.getValues());
+ }
+ }
+
+ ColGroupValue.cleanupThreadLocalMemory();
+
+ }
+
+ private static class RightMatrixMultTask implements Callable<Object> {
+ private final List<ColGroup> _colGroups;
+ // private final double[] _thatV;
+ private final double[] _retV;
+ private final List<Future<double[]>> _aggB;
+ private final Pair<Integer, int[]> _v;
+ private final int _numColumns;
+
+ private final int _rl;
+ private final int _ru;
+ private final int _cl;
+ private final int _cu;
+ private final boolean _mem;
+ private final boolean _skipOle;
+
+ protected RightMatrixMultTask(List<ColGroup> groups, double[] retV, List<Future<double[]>> aggB,
+ Pair<Integer, int[]> v, int numColumns, int rl, int ru, int cl, int cu, boolean mem, boolean skipOle) {
+ _colGroups = groups;
+ // _thatV = thatV;
+ _retV = retV;
+ _aggB = aggB;
+ _v = v;
+ _numColumns = numColumns;
+ _rl = rl;
+ _ru = ru;
+ _cl = cl;
+ _cu = cu;
+ _mem = mem;
+ _skipOle = skipOle;
+ }
+
+ @Override
+ public Object call() {
+ try {
+ if(_mem)
+ ColGroupValue.setupThreadLocalMemory((_v.getLeft()));
+ for(int j = 0; j < _colGroups.size(); j++) {
+ if(_colGroups.get(j) instanceof ColGroupOLE) {
+ if(_skipOle) {
+ _colGroups.get(j)
+ .rightMultByMatrix(_aggB.get(j).get(), _retV, _numColumns, _rl, _ru, _cl, _cu);
+ }
+ }
+ else {
+ if(!_skipOle) {
+ _colGroups.get(j)
+ .rightMultByMatrix(_aggB.get(j).get(), _retV, _numColumns, _rl, _ru, _cl, _cu);
+ }
+ }
+ }
+ if(_mem)
+ ColGroupValue.cleanupThreadLocalMemory();
+ return null;
+ }
+ catch(Exception e) {
+ LOG.error(e);
+ throw new DMLRuntimeException(e);
+ }
+ }
+ }
+
+ private static class RightMatrixPreAggregateTask implements Callable<double[]> {
+ private final ColGroupValue _colGroup;
+ private final int _numVals;
+ private final double[] _b;
+ private final double[] _dict;
+
+ private final int _cl;
+ private final int _cu;
+ private final int _cut;
+
+ protected RightMatrixPreAggregateTask(ColGroupValue colGroup, int numVals, double[] b, double[] dict, int cl,
+ int cu, int cut) {
+ _colGroup = colGroup;
+ _numVals = numVals;
+ _b = b;
+ _dict = dict;
+ _cl = cl;
+ _cu = cu;
+ _cut = cut;
+ }
+
+ @Override
+ public double[] call() {
+ try {
+ return _colGroup.preaggValues(_numVals, _b, _dict, _cl, _cu, _cut);
+ }
+ catch(Exception e) {
+ LOG.error(e);
+ throw new DMLRuntimeException(e);
+ }
+ }
+ }
+
+ private static class RightMatrixPreAggregateSparseTask implements Callable<double[]> {
+ private final ColGroupValue _colGroup;
+ private final int _numVals;
+ private final SparseBlock _b;
+ private final double[] _dict;
+
+ private final int _cl;
+ private final int _cu;
+ private final int _cut;
+
+ protected RightMatrixPreAggregateSparseTask(ColGroupValue colGroup, int numVals, SparseBlock b, double[] dict,
+ int cl, int cu, int cut) {
+ _colGroup = colGroup;
+ _numVals = numVals;
+ _b = b;
+ _dict = dict;
+ _cl = cl;
+ _cu = cu;
+ _cut = cut;
+ }
+
+ @Override
+ public double[] call() {
+ try {
+ return _colGroup.preaggValues(_numVals, _b, _dict, _cl, _cu, _cut);
+ }
+ catch(Exception e) {
+ LOG.error(e);
+ throw new DMLRuntimeException(e);
+ }
+ }
+ }
+
+ private static class RightMatrixVectorMultTask implements Callable<Long> {
+ private final List<ColGroup> _groups;
+ private final MatrixBlock _vect;
+ private final MatrixBlock _ret;
+ private final int _rl;
+ private final int _ru;
+ private final Pair<Integer, int[]> _v;
+
+ protected RightMatrixVectorMultTask(List<ColGroup> groups, MatrixBlock vect, MatrixBlock ret, int rl, int ru,
+ Pair<Integer, int[]> v) {
+ _groups = groups;
+ _vect = vect;
+ _ret = ret;
+ _rl = rl;
+ _ru = ru;
+ _v = v;
+ }
+
+ @Override
+ public Long call() {
+ try {
+ rightMultByVector(_groups, _vect, _ret, _rl, _ru, _v);
+ return _ret.recomputeNonZeros(_rl, _ru - 1, 0, 0);
+ }
+ catch(Exception e) {
+ LOG.error(e);
+ throw new DMLRuntimeException(e);
+ }
+ }
+ }
+}
diff --git a/src/main/java/org/apache/sysds/runtime/compress/lib/LibScalar.java b/src/main/java/org/apache/sysds/runtime/compress/lib/LibScalar.java
new file mode 100644
index 0000000..f555513
--- /dev/null
+++ b/src/main/java/org/apache/sysds/runtime/compress/lib/LibScalar.java
@@ -0,0 +1,194 @@
+/*
+ * 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.compress.lib;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+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 org.apache.sysds.runtime.DMLCompressionException;
+import org.apache.sysds.runtime.DMLRuntimeException;
+import org.apache.sysds.runtime.compress.CompressedMatrixBlock;
+import org.apache.sysds.runtime.compress.colgroup.ColGroup;
+import org.apache.sysds.runtime.compress.colgroup.ColGroupConst;
+import org.apache.sysds.runtime.compress.colgroup.ColGroupUncompressed;
+import org.apache.sysds.runtime.compress.colgroup.ColGroupValue;
+import org.apache.sysds.runtime.compress.colgroup.Dictionary;
+import org.apache.sysds.runtime.functionobjects.Minus;
+import org.apache.sysds.runtime.functionobjects.Multiply;
+import org.apache.sysds.runtime.functionobjects.Plus;
+import org.apache.sysds.runtime.functionobjects.Power2;
+import org.apache.sysds.runtime.matrix.data.MatrixBlock;
+import org.apache.sysds.runtime.matrix.operators.LeftScalarOperator;
+import org.apache.sysds.runtime.matrix.operators.RightScalarOperator;
+import org.apache.sysds.runtime.matrix.operators.ScalarOperator;
+import org.apache.sysds.runtime.util.CommonThreadPool;
+
+public class LibScalar {
+
+ // private static final Log LOG = LogFactory.getLog(LibScalar.class.getName());
+ private static final int MINIMUM_PARALLEL_SIZE = 8096;
+
+ public static MatrixBlock scalarOperations(ScalarOperator sop, CompressedMatrixBlock m1, CompressedMatrixBlock ret,
+ boolean overlapping) {
+ // LOG.error(sop);
+ if(sop instanceof LeftScalarOperator) {
+ if(sop.fn instanceof Minus) {
+ m1 = (CompressedMatrixBlock) scalarOperations(new RightScalarOperator(Multiply.getMultiplyFnObject(),
+ -1), m1, ret, overlapping);
+ return scalarOperations(new RightScalarOperator(Plus.getPlusFnObject(), sop.getConstant()),
+ m1,
+ ret,
+ overlapping);
+ }
+ else if(sop.fn instanceof Power2) {
+ throw new DMLCompressionException("Left Power does not make sense.");
+ // List<ColGroup> newColGroups = new ArrayList<>();
+ // double v = sop.executeScalar(0);
+
+ // double[] values = new double[m1.getNumColumns()];
+ // Arrays.fill(values, v);
+
+ // int[] colIndexes = new int[m1.getNumColumns()];
+ // for(int i = 0; i < colIndexes.length; i++) {
+ // colIndexes[i] = i;
+ // }
+ // newColGroups.add(new ColGroupConst(colIndexes, ret.getNumRows(), new Dictionary(values)));
+ // ret.allocateColGroupList(newColGroups);
+ // ret.setNonZeros(ret.getNumColumns() * ret.getNumRows());
+ // return ret;
+ }
+
+ }
+
+ List<ColGroup> colGroups = m1.getColGroups();
+ if(overlapping && !(sop.fn instanceof Multiply)) {
+ if(sop.fn instanceof Plus || sop.fn instanceof Minus) {
+
+ // If the colGroup is overlapping we know there are no incompressable colGroups.
+ List<ColGroup> newColGroups = new ArrayList<>();
+ for(ColGroup grp : colGroups) {
+ ColGroupValue g = (ColGroupValue) grp;
+ newColGroups.add(g.copy());
+ }
+ int[] colIndexes = newColGroups.get(0).getColIndices();
+ double v = sop.executeScalar(0);
+ double[] values = new double[colIndexes.length];
+ Arrays.fill(values, v);
+ newColGroups.add(new ColGroupConst(colIndexes, ret.getNumRows(), new Dictionary(values)));
+ ret.allocateColGroupList(newColGroups);
+ ret.setOverlapping(true);
+ ret.setNonZeros(-1);
+ }
+ }
+ else {
+
+ if(sop.getNumThreads() > 1) {
+ parallelScalarOperations(sop, colGroups, ret, sop.getNumThreads());
+ }
+ else {
+ // Apply the operation to each of the column groups.
+ // Most implementations will only modify metadata.
+ List<ColGroup> newColGroups = new ArrayList<>();
+ for(ColGroup grp : colGroups) {
+ newColGroups.add(grp.scalarOperation(sop));
+ }
+ ret.allocateColGroupList(newColGroups);
+ }
+ ret.setNonZeros(-1);
+ ret.setOverlapping(m1.isOverlapping());
+ }
+
+ return ret;
+
+ }
+
+ private static void parallelScalarOperations(ScalarOperator sop, List<ColGroup> colGroups,
+ CompressedMatrixBlock ret, int k) {
+ ExecutorService pool = CommonThreadPool.get(k);
+ List<ScalarTask> tasks = partition(sop, colGroups);
+ try {
+ List<Future<List<ColGroup>>> rtasks = pool.invokeAll(tasks);
+ pool.shutdown();
+ List<ColGroup> newColGroups = new ArrayList<>();
+ for(Future<List<ColGroup>> f : rtasks) {
+ newColGroups.addAll(f.get());
+ }
+ ret.allocateColGroupList(newColGroups);
+ }
+ catch(InterruptedException | ExecutionException e) {
+ throw new DMLRuntimeException(e);
+ }
+ }
+
+ private static List<ScalarTask> partition(ScalarOperator sop, List<ColGroup> colGroups) {
+ ArrayList<ScalarTask> tasks = new ArrayList<>();
+ ArrayList<ColGroup> small = new ArrayList<>();
+ for(ColGroup grp : colGroups) {
+ if(grp instanceof ColGroupUncompressed) {
+ ArrayList<ColGroup> uc = new ArrayList<>();
+ uc.add(grp);
+ tasks.add(new ScalarTask(uc, sop));
+ }
+ else {
+ int nv = ((ColGroupValue) grp).getNumValues() * grp.getColIndices().length;
+ if(nv < MINIMUM_PARALLEL_SIZE) {
+ small.add(grp);
+ }
+ else {
+ ArrayList<ColGroup> large = new ArrayList<>();
+ large.add(grp);
+ tasks.add(new ScalarTask(large, sop));
+ }
+ }
+ if(small.size() > 10) {
+ tasks.add(new ScalarTask(small, sop));
+ small = new ArrayList<>();
+ }
+ }
+ if(small.size() > 0) {
+ tasks.add(new ScalarTask(small, sop));
+ }
+ return tasks;
+ }
+
+ private static class ScalarTask implements Callable<List<ColGroup>> {
+ private final List<ColGroup> _colGroups;
+ private final ScalarOperator _sop;
+
+ protected ScalarTask(List<ColGroup> colGroups, ScalarOperator sop) {
+ _colGroups = colGroups;
+ _sop = sop;
+ }
+
+ @Override
+ public List<ColGroup> call() {
+ List<ColGroup> res = new ArrayList<>();
+ for(ColGroup x : _colGroups) {
+ res.add(x.scalarOperation(_sop));
+ }
+ return res;
+ }
+ }
+}
diff --git a/src/main/java/org/apache/sysds/runtime/compress/utils/ABitmap.java b/src/main/java/org/apache/sysds/runtime/compress/utils/ABitmap.java
index abc745d..aae9d41 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/utils/ABitmap.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/utils/ABitmap.java
@@ -97,4 +97,4 @@
sb.append("\nOffsets:" + Arrays.toString(_offsetsLists));
return sb.toString();
}
-}
\ No newline at end of file
+}
diff --git a/src/main/java/org/apache/sysds/runtime/compress/utils/BitmapLossy.java b/src/main/java/org/apache/sysds/runtime/compress/utils/BitmapLossy.java
index 7f86794..abd0f03 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/utils/BitmapLossy.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/utils/BitmapLossy.java
@@ -109,5 +109,4 @@
sb.append("\nOffsets:" + Arrays.toString(_offsetsLists));
return sb.toString();
}
-
}
diff --git a/src/main/java/org/apache/sysds/runtime/compress/utils/CustomHashMap.java b/src/main/java/org/apache/sysds/runtime/compress/utils/CustomHashMap.java
index 1362c6f..417aa81 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/utils/CustomHashMap.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/utils/CustomHashMap.java
@@ -22,9 +22,9 @@
* This class provides a memory-efficient base for Custom HashMaps for restricted use cases.
*/
public abstract class CustomHashMap {
- protected static final int INIT_CAPACITY = 8;
+ protected static final int INIT_CAPACITY = 32;
protected static final int RESIZE_FACTOR = 2;
- protected static final float LOAD_FACTOR = 0.75f;
+ protected static final float LOAD_FACTOR = 0.30f;
protected int _size = -1;
diff --git a/src/main/java/org/apache/sysds/runtime/compress/utils/DblArray.java b/src/main/java/org/apache/sysds/runtime/compress/utils/DblArray.java
index 49cb5d0..0810e8d 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/utils/DblArray.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/utils/DblArray.java
@@ -25,24 +25,18 @@
* Helper class used for bitmap extraction.
*/
public class DblArray {
- private double[] _arr = null;
- private boolean _zero = false;
+ private double[] _arr;
public DblArray() {
- this(null, false);
+ _arr = null;
}
public DblArray(double[] arr) {
- this(arr, false);
+ _arr = arr;
}
public DblArray(DblArray that) {
- this(Arrays.copyOf(that._arr, that._arr.length), that._zero);
- }
-
- public DblArray(double[] arr, boolean allZeros) {
- _arr = arr;
- _zero = allZeros;
+ this(Arrays.copyOf(that._arr, that._arr.length));
}
public double[] getData() {
@@ -51,12 +45,12 @@
@Override
public int hashCode() {
- return _zero ? 0 : Arrays.hashCode(_arr);
+ return _arr == null ? 0 : Arrays.hashCode(_arr);
}
@Override
public boolean equals(Object o) {
- return(o instanceof DblArray && _zero == ((DblArray) o)._zero && Arrays.equals(_arr, ((DblArray) o)._arr));
+ return(o instanceof DblArray && Arrays.equals(_arr, ((DblArray) o)._arr));
}
@Override
@@ -64,14 +58,7 @@
return Arrays.toString(_arr);
}
- public static boolean isZero(double[] ds) {
- for(int i = 0; i < ds.length; i++)
- if(ds[i] != 0.0)
- return false;
- return true;
- }
-
public static boolean isZero(DblArray val) {
- return val._zero || isZero(val._arr);
+ return val._arr == null;
}
}
diff --git a/src/main/java/org/apache/sysds/runtime/compress/utils/DblArrayIntListHashMap.java b/src/main/java/org/apache/sysds/runtime/compress/utils/DblArrayIntListHashMap.java
index 32d7ae7..55065b4 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/utils/DblArrayIntListHashMap.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/utils/DblArrayIntListHashMap.java
@@ -36,6 +36,7 @@
protected static final Log LOG = LogFactory.getLog(DblArrayIntListHashMap.class.getName());
private DArrayIListEntry[] _data = null;
+ public static int hashMissCount = 0;
public DblArrayIntListHashMap() {
_data = new DArrayIListEntry[INIT_CAPACITY];
@@ -51,7 +52,6 @@
// probe for early abort
if(_size == 0)
return null;
-
// compute entry index position
int hash = hash(key);
int ix = indexFor(hash, _data.length);
@@ -60,7 +60,10 @@
for(DArrayIListEntry e = _data[ix]; e != null; e = e.next) {
if(e.key.equals(key)) {
return e.value;
+ }else{
+ hashMissCount++;
}
+
}
return null;
@@ -98,8 +101,6 @@
}
}
Collections.sort(ret);
-
- LOG.info(ret);
return ret;
}
@@ -127,7 +128,7 @@
private static int hash(DblArray key) {
int h = key.hashCode();
-
+
// This function ensures that hashCodes that differ only by
// constant multiples at each bit position have a bounded
// number of collisions (approximately 8 at default load factor).
@@ -162,7 +163,10 @@
return -1;
}
}
- if(o1d.length > o2d.length) {
+ if(o1d.length == o2d.length){
+ return 0;
+ }
+ else if(o1d.length > o2d.length) {
return 1;
}
else {
@@ -177,11 +181,11 @@
@Override
public String toString(){
- StringBuilder sb = new StringBuilder();
- sb.append("[" + key + ", ");
- sb.append( value + ", ");
- sb.append( next + "]");
- return sb.toString();
+ if(next == null){
+ return key + ":" + value;
+ }else{
+ return key +":" + value + "," + next;
+ }
}
}
@@ -189,8 +193,18 @@
public String toString(){
StringBuilder sb = new StringBuilder();
sb.append(this.getClass().getSimpleName() + this.hashCode());
- for(DArrayIListEntry ent : _data)
- sb.append("\n" + ent);
+ sb.append(" "+ _size);
+ for(int i = 0 ; i < _data.length; i++){
+ DArrayIListEntry ent = _data[i];
+ if(ent != null){
+
+ sb.append("\n");
+ sb.append("id:" + i);
+ sb.append("[");
+ sb.append(ent);
+ sb.append("]");
+ }
+ }
return sb.toString();
}
}
diff --git a/src/main/java/org/apache/sysds/runtime/compress/utils/IntArrayList.java b/src/main/java/org/apache/sysds/runtime/compress/utils/IntArrayList.java
index e99ca2f..37d11dc 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/utils/IntArrayList.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/utils/IntArrayList.java
@@ -103,16 +103,16 @@
@Override
public String toString(){
StringBuilder sb = new StringBuilder();
- sb.append("IntArrayList ");
- sb.append("size: " + _size);
+
if(_size == 1){
- sb.append(" [" + _val0+ "] ");
+ sb.append(_val0);
} else{
- sb.append(" [");
- for(int i = 0; i < _size-1; i++){
- sb.append(_data[i] + ", ");
+ sb.append("[");
+ int i = 0;
+ for(; i < _size-1; i++){
+ sb.append(_data[i] + ",");
}
- sb.append(_data[_data.length-1]+"] ");
+ sb.append(_data[i]+"]");
}
return sb.toString();
}
diff --git a/src/main/java/org/apache/sysds/runtime/compress/utils/LinearAlgebraUtils.java b/src/main/java/org/apache/sysds/runtime/compress/utils/LinearAlgebraUtils.java
index 23e534e..1bce819 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/utils/LinearAlgebraUtils.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/utils/LinearAlgebraUtils.java
@@ -149,11 +149,14 @@
public static void vectListAddDDC(final double[] values, double[] c, byte[] bix, final int rl, final int ru,
final int cl, final int cu, final int cut, final int numVals) {
+
for(int j = rl, off = rl * cut; j < ru; j++, off += cut) {
int rowIdx = (bix[j] & 0xFF);
if(rowIdx < numVals)
- for(int k = cl, h = rowIdx * (cu - cl); k < cu; k++, h++)
+ for(int k = cl, h = rowIdx * (cu - cl); k < cu; k++, h++) {
+ // LOG.error((off + k) + " \t" + h);
c[off + k] += values[h];
+ }
}
}
@@ -268,8 +271,17 @@
double[] a = tmp.getDenseBlockValues();
DenseBlock c = ret.getDenseBlock();
for(int i = 0; i < tmp.getNumColumns(); i++)
- if(a[i] != 0)
- c.set((ix < i) ? ix : i, (ix < i) ? i : ix, a[i]);
+ if(a[i] != 0) {
+ int row = (ix < i) ? ix : i;
+ int col = (ix < i) ? i : ix;
+ // if(row == col) {
+ c.set(row, col, a[i]);
+ // }
+ // else {
+ // double v = c.get(row, col);
+ // c.set(row, col, a[i] + v);
+ // }
+ }
}
/**
diff --git a/src/main/java/org/apache/sysds/runtime/matrix/data/LibMatrixCountDistinct.java b/src/main/java/org/apache/sysds/runtime/matrix/data/LibMatrixCountDistinct.java
index e582ee9..3fce404 100644
--- a/src/main/java/org/apache/sysds/runtime/matrix/data/LibMatrixCountDistinct.java
+++ b/src/main/java/org/apache/sysds/runtime/matrix/data/LibMatrixCountDistinct.java
@@ -114,17 +114,27 @@
private static int countDistinctValuesNaive(MatrixBlock in) {
Set<Double> distinct = new HashSet<>();
double[] data;
- long nonZeros = in.getNonZeros();
- if(nonZeros < in.getNumColumns() * in.getNumRows()) {
- distinct.add(0d);
- }
- if(in.sparseBlock == null && in.denseBlock == null) {
- List<ColGroup> colGroups = ((CompressedMatrixBlock) in).getColGroups();
- for(ColGroup cg : colGroups) {
- countDistinctValuesNaive(cg.getValues(), distinct);
+ if(in instanceof CompressedMatrixBlock) {
+ CompressedMatrixBlock inC = (CompressedMatrixBlock) in;
+ if(inC.isOverlapping()) {
+ in = inC.decompress();
+ inC = null;
+ }
+ else {
+ List<ColGroup> colGroups = ((CompressedMatrixBlock) in).getColGroups();
+ for(ColGroup cg : colGroups) {
+ countDistinctValuesNaive(cg.getValues(), distinct);
+ }
}
}
- else if(in.sparseBlock != null) {
+
+ long nonZeros = in.getNonZeros();
+
+ if(nonZeros != -1 && nonZeros < in.getNumColumns() * in.getNumRows()) {
+ distinct.add(0d);
+ }
+
+ if(in.sparseBlock != null) {
SparseBlock sb = in.sparseBlock;
if(in.sparseBlock.isContiguous()) {
@@ -140,7 +150,7 @@
}
}
}
- else {
+ else if(in.denseBlock != null) {
DenseBlock db = in.denseBlock;
for(int i = 0; i <= db.numBlocks(); i++) {
data = db.valuesAt(i);
diff --git a/src/main/java/org/apache/sysds/runtime/util/DataConverter.java b/src/main/java/org/apache/sysds/runtime/util/DataConverter.java
index 25a09a0..c181b5c 100644
--- a/src/main/java/org/apache/sysds/runtime/util/DataConverter.java
+++ b/src/main/java/org/apache/sysds/runtime/util/DataConverter.java
@@ -38,6 +38,7 @@
import org.apache.sysds.common.Types.FileFormat;
import org.apache.sysds.common.Types.ValueType;
import org.apache.sysds.runtime.DMLRuntimeException;
+import org.apache.sysds.runtime.compress.CompressedMatrixBlock;
import org.apache.sysds.runtime.controlprogram.caching.FrameObject;
import org.apache.sysds.runtime.controlprogram.caching.MatrixObject;
import org.apache.sysds.runtime.controlprogram.caching.TensorObject;
@@ -258,7 +259,9 @@
int rows = mb.getNumRows();
int cols = mb.getNumColumns();
double[][] ret = new double[rows][cols]; //0-initialized
-
+ if(mb instanceof CompressedMatrixBlock){
+ mb = ((CompressedMatrixBlock)mb).decompress();
+ }
if( mb.getNonZeros() > 0 ) {
if( mb.isInSparseFormat() ) {
Iterator<IJV> iter = mb.getSparseBlockIterator();
diff --git a/src/main/java/org/apache/sysds/utils/DMLCompressionStatistics.java b/src/main/java/org/apache/sysds/utils/DMLCompressionStatistics.java
new file mode 100644
index 0000000..2b08a0a
--- /dev/null
+++ b/src/main/java/org/apache/sysds/utils/DMLCompressionStatistics.java
@@ -0,0 +1,76 @@
+/*
+ * 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.utils;
+
+import java.util.List;
+
+public class DMLCompressionStatistics {
+
+ // Compute compressed size info
+ private static double Phase1 = 0.0;
+ // Co-code columns
+ private static double Phase2 = 0.0;
+ // Compress the columns
+ private static double Phase3 = 0.0;
+ // Share resources
+ private static double Phase4 = 0.0;
+ // Cleanup
+ private static double Phase5 = 0.0;
+
+ private static int DecompressSTCount = 0;
+ private static double DecompressST = 0.0;
+ private static int DecompressMTCount = 0;
+ private static double DecompressMT = 0.0;
+
+ public static void addCompressionTimes(List<Double> times) {
+ Phase1 += times.get(0);
+ Phase2 += times.get(1);
+ Phase3 += times.get(2);
+ Phase4 += times.get(3);
+ Phase5 += times.get(4);
+ }
+
+ public static void addDecompressTime(double time, int threads) {
+ if(threads == 1) {
+ DecompressSTCount++;
+ DecompressST += time;
+ }
+ else {
+ DecompressMTCount++;
+ DecompressMT += time;
+ }
+ }
+
+ public static void display(StringBuilder sb) {
+ sb.append(String.format(
+ "CLA Compression Phases (classify, group, compress, share, clean) :\t%.3f/%.3f/%.3f/%.3f/%.3f\n",
+ Phase1 / 1000,
+ Phase2 / 1000,
+ Phase3 / 1000,
+ Phase4 / 1000,
+ Phase5 / 1000));
+ sb.append(String.format("Decompression Counts (Single , Multi) thread :\t%d/%d\n",
+ DecompressSTCount,
+ DecompressMTCount));
+ sb.append(String.format("Dedicated Decompression Time (Single , Multi) thread :\t%.3f/%.3f\n",
+ DecompressST / 1000,
+ DecompressMT / 1000));
+ }
+}
diff --git a/src/main/java/org/apache/sysds/utils/Statistics.java b/src/main/java/org/apache/sysds/utils/Statistics.java
index 7642397..6f6b26c 100644
--- a/src/main/java/org/apache/sysds/utils/Statistics.java
+++ b/src/main/java/org/apache/sysds/utils/Statistics.java
@@ -34,6 +34,7 @@
import org.apache.sysds.api.DMLScript;
import org.apache.sysds.conf.ConfigurationManager;
+import org.apache.sysds.conf.DMLConfig;
import org.apache.sysds.hops.OptimizerUtils;
import org.apache.sysds.runtime.controlprogram.caching.CacheStatistics;
import org.apache.sysds.runtime.controlprogram.context.SparkExecutionContext;
@@ -1036,6 +1037,10 @@
federatedExecuteUDFCount.longValue() + ".\n");
}
+ if( ConfigurationManager.getDMLConfig().getTextValue(DMLConfig.COMPRESSED_LINALG).contains("true")){
+ DMLCompressionStatistics.display(sb);
+ }
+
sb.append("Total JIT compile time:\t\t" + ((double)getJITCompileTime())/1000 + " sec.\n");
sb.append("Total JVM GC count:\t\t" + getJVMgcCount() + ".\n");
sb.append("Total JVM GC time:\t\t" + ((double)getJVMgcTime())/1000 + " sec.\n");
diff --git a/src/test/java/org/apache/sysds/test/component/compress/AbstractCompressedUnaryTests.java b/src/test/java/org/apache/sysds/test/component/compress/AbstractCompressedUnaryTests.java
index 90add0e..df15499 100644
--- a/src/test/java/org/apache/sysds/test/component/compress/AbstractCompressedUnaryTests.java
+++ b/src/test/java/org/apache/sysds/test/component/compress/AbstractCompressedUnaryTests.java
@@ -30,6 +30,7 @@
import org.apache.sysds.runtime.util.DataConverter;
import org.apache.sysds.test.TestUtils;
import org.apache.sysds.test.component.compress.TestConstants.MatrixTypology;
+import org.apache.sysds.test.component.compress.TestConstants.OverLapping;
import org.apache.sysds.test.component.compress.TestConstants.SparsityType;
import org.apache.sysds.test.component.compress.TestConstants.ValueRange;
import org.apache.sysds.test.component.compress.TestConstants.ValueType;
@@ -38,8 +39,8 @@
public abstract class AbstractCompressedUnaryTests extends CompressedTestBase {
public AbstractCompressedUnaryTests(SparsityType sparType, ValueType valType, ValueRange valRange,
- CompressionSettings compSettings, MatrixTypology matrixTypology, int parallelism) {
- super(sparType, valType, valRange, compSettings, matrixTypology, parallelism);
+ CompressionSettings compSettings, MatrixTypology matrixTypology, OverLapping ov, int parallelism) {
+ super(sparType, valType, valRange, compSettings, matrixTypology, ov, parallelism);
}
enum AggType {
@@ -183,7 +184,7 @@
assertTrue("dim 2 is equal in non compressed res", d1[0].length == dim2);
assertTrue("dim 2 is equal in compressed res", d2[0].length == dim2);
- String css = compressionSettings.toString();
+ String css = this.toString();
if(compressionSettings.lossy) {
if(aggType == AggType.COLSUMS) {
TestUtils.compareMatrices(d1, d2, lossyTolerance * 10 * rows, css);
@@ -208,12 +209,13 @@
}
}
else {
- if(aggType == AggType.ROWMEAN) {
+ if(aggType == AggType.ROWMEAN)
TestUtils.compareMatrices(d1, d2, 0.0001, css);
- }
- else {
- TestUtils.compareMatricesBitAvgDistance(d1, d2, 2048, 30, css);
- }
+ else if(overlappingType == OverLapping.MATRIX_MULT_NEGATIVE ||
+ overlappingType == OverLapping.MATRIX_PLUS || overlappingType == OverLapping.MATRIX)
+ TestUtils.compareMatricesBitAvgDistance(d1, d2, 8192, 128, css);
+ else
+ TestUtils.compareMatricesBitAvgDistance(d1, d2, 2048, 128, css);
}
}
catch(NotImplementedException e) {
@@ -224,4 +226,4 @@
throw new RuntimeException(this.toString() + "\n" + e.getMessage(), e);
}
}
-}
\ No newline at end of file
+}
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 76e8fb9..76fe85d 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
@@ -39,6 +39,7 @@
import org.apache.sysds.runtime.util.DataConverter;
import org.apache.sysds.test.TestUtils;
import org.apache.sysds.test.component.compress.TestConstants.MatrixTypology;
+import org.apache.sysds.test.component.compress.TestConstants.OverLapping;
import org.apache.sysds.test.component.compress.TestConstants.SparsityType;
import org.apache.sysds.test.component.compress.TestConstants.ValueRange;
import org.apache.sysds.test.component.compress.TestConstants.ValueType;
@@ -55,8 +56,8 @@
public class CompressedMatrixTest extends AbstractCompressedUnaryTests {
public CompressedMatrixTest(SparsityType sparType, ValueType valType, ValueRange valRange,
- CompressionSettings compSettings, MatrixTypology matrixTypology) {
- super(sparType, valType, valRange, compSettings, matrixTypology, 1);
+ CompressionSettings compSettings, MatrixTypology matrixTypology, OverLapping ov) {
+ super(sparType, valType, valRange, compSettings, matrixTypology, ov, 1);
}
@Test
@@ -67,14 +68,17 @@
for(int i = 0; i < rows; i++)
for(int j = 0; j < cols; j++) {
- double ulaVal = input[i][j];
+ double ulaVal = mb.quickGetValue(i, j);
double claVal = cmb.getValue(i, j); // calls quickGetValue internally
- if(compressionSettings.lossy) {
+ if(compressionSettings.lossy)
TestUtils.compareCellValue(ulaVal, claVal, lossyTolerance, false);
- }
- else {
+ else if(overlappingType == OverLapping.MATRIX_MULT_NEGATIVE ||
+ overlappingType == OverLapping.MATRIX_PLUS || overlappingType == OverLapping.MATRIX ||
+ overlappingType == OverLapping.COL)
+ TestUtils.compareScalarBitsJUnit(ulaVal, claVal, 8192);
+ else
TestUtils.compareScalarBitsJUnit(ulaVal, claVal, 0); // Should be exactly same value
- }
+
}
}
catch(Exception e) {
@@ -103,12 +107,15 @@
// compare result with input
double[][] d1 = DataConverter.convertToDoubleMatrix(ret1);
double[][] d2 = DataConverter.convertToDoubleMatrix(ret2);
- if(compressionSettings.lossy) {
+ if(compressionSettings.lossy)
TestUtils.compareMatrices(d1, d2, lossyTolerance);
- }
- else {
+
+ else if(overlappingType == OverLapping.MATRIX_MULT_NEGATIVE || overlappingType == OverLapping.MATRIX_PLUS ||
+ overlappingType == OverLapping.MATRIX || overlappingType == OverLapping.COL)
+ TestUtils.compareMatricesBitAvgDistance(d1, d2, 8192, 128, this.toString());
+ else
TestUtils.compareMatricesBitAvgDistance(d1, d2, 0, 1, "Test Append Matrix");
- }
+
}
catch(Exception e) {
e.printStackTrace();
@@ -119,13 +126,16 @@
@Test
public void testCountDistinct() {
try {
- if(!(cmb instanceof CompressedMatrixBlock))
+ // Counting distinct is potentially wrong in cases with overlapping, resulting in a few to many or few
+ // elements.
+ if(!(cmb instanceof CompressedMatrixBlock) || (overlappingType == OverLapping.MATRIX_MULT_NEGATIVE))
return; // Input was not compressed then just pass test
CountDistinctOperator op = new CountDistinctOperator(CountDistinctTypes.COUNT);
int ret1 = LibMatrixCountDistinct.estimateDistinctValues(mb, op);
int ret2 = LibMatrixCountDistinct.estimateDistinctValues(cmb, op);
- String base = compressionSettings.toString() + "\n";
+
+ String base = this.toString() + "\n";
if(compressionSettings.lossy) {
// The number of distinct values should be same or lower in lossy mode.
// assertTrue(base + "lossy distinct count " +ret2+ "is less than full " + ret1, ret1 >= ret2);
@@ -174,12 +184,14 @@
// compare result with input
double[][] d1 = DataConverter.convertToDoubleMatrix(mb);
double[][] d2 = DataConverter.convertToDoubleMatrix(tmp);
- if(compressionSettings.lossy) {
- TestUtils.compareMatrices(d1, d2, lossyTolerance, compressionSettings.toString());
- }
- else {
- TestUtils.compareMatricesBitAvgDistance(d1, d2, 0, 0, compressionSettings.toString());
- }
+ if(compressionSettings.lossy)
+ TestUtils.compareMatrices(d1, d2, lossyTolerance, this.toString());
+ else if(overlappingType == OverLapping.MATRIX_MULT_NEGATIVE || overlappingType == OverLapping.MATRIX_PLUS ||
+ overlappingType == OverLapping.MATRIX || overlappingType == OverLapping.COL)
+ TestUtils.compareMatricesBitAvgDistance(d1, d2, 8192, 128, this.toString());
+ else
+ TestUtils.compareMatricesBitAvgDistance(d1, d2, 0, 0, this.toString());
+
}
catch(Exception e) {
e.printStackTrace();
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 5d3c35f..e26ec03 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
@@ -36,46 +36,57 @@
import org.apache.sysds.runtime.compress.CompressionSettingsBuilder;
import org.apache.sysds.runtime.compress.CompressionStatistics;
import org.apache.sysds.runtime.compress.colgroup.ColGroup.CompressionType;
+import org.apache.sysds.runtime.functionobjects.GreaterThan;
+import org.apache.sysds.runtime.functionobjects.LessThanEquals;
+import org.apache.sysds.runtime.functionobjects.Minus;
import org.apache.sysds.runtime.functionobjects.Multiply;
import org.apache.sysds.runtime.functionobjects.Plus;
+import org.apache.sysds.runtime.functionobjects.Power2;
+import org.apache.sysds.runtime.functionobjects.ValueFunction;
+import org.apache.sysds.runtime.functionobjects.Xor;
import org.apache.sysds.runtime.instructions.InstructionUtils;
import org.apache.sysds.runtime.matrix.data.MatrixBlock;
import org.apache.sysds.runtime.matrix.operators.AggregateBinaryOperator;
-import org.apache.sysds.runtime.matrix.operators.AggregateOperator;
+import org.apache.sysds.runtime.matrix.operators.BinaryOperator;
+import org.apache.sysds.runtime.matrix.operators.LeftScalarOperator;
import org.apache.sysds.runtime.matrix.operators.RightScalarOperator;
import org.apache.sysds.runtime.matrix.operators.ScalarOperator;
import org.apache.sysds.runtime.util.DataConverter;
import org.apache.sysds.test.TestUtils;
import org.apache.sysds.test.component.compress.TestConstants.MatrixTypology;
+import org.apache.sysds.test.component.compress.TestConstants.OverLapping;
import org.apache.sysds.test.component.compress.TestConstants.SparsityType;
import org.apache.sysds.test.component.compress.TestConstants.ValueRange;
import org.apache.sysds.test.component.compress.TestConstants.ValueType;
-import org.junit.Ignore;
import org.junit.Test;
import org.junit.runners.Parameterized.Parameters;
-
public abstract class CompressedTestBase extends TestBase {
protected static final Log LOG = LogFactory.getLog(CompressedTestBase.class.getName());
protected static SparsityType[] usedSparsityTypes = new SparsityType[] { // Sparsity 0.9, 0.1, 0.01 and 0.0
// SparsityType.FULL,
- // SparsityType.DENSE,
- SparsityType.SPARSE,
- // SparsityType.ULTRA_SPARSE,
+ SparsityType.DENSE,
+ // SparsityType.SPARSE,
+ SparsityType.ULTRA_SPARSE,
// SparsityType.EMPTY
};
protected static ValueType[] usedValueTypes = new ValueType[] {
// ValueType.RAND,
// ValueType.CONST,
- ValueType.RAND_ROUND, ValueType.OLE_COMPRESSIBLE, ValueType.RLE_COMPRESSIBLE,};
+ ValueType.RAND_ROUND, ValueType.OLE_COMPRESSIBLE,
+ // ValueType.RLE_COMPRESSIBLE,
+ };
protected static ValueRange[] usedValueRanges = new ValueRange[] {ValueRange.SMALL,
// ValueRange.LARGE,
// ValueRange.BYTE
};
+ protected static OverLapping[] overLapping = new OverLapping[] {OverLapping.COL, OverLapping.MATRIX,
+ OverLapping.NONE, OverLapping.MATRIX_PLUS, OverLapping.MATRIX_MULT_NEGATIVE};
+
private static final int compressionSeed = 7;
protected static CompressionSettings[] usedCompressionSettings = new CompressionSettings[] {
@@ -94,16 +105,16 @@
// // // // LOSSY TESTS!
- new CompressionSettingsBuilder().setSamplingRatio(0.1).setSeed(compressionSeed)
- .setValidCompressions(EnumSet.of(CompressionType.DDC)).setInvestigateEstimate(true).setLossy(true).create(),
- new CompressionSettingsBuilder().setSamplingRatio(0.1).setSeed(compressionSeed)
- .setValidCompressions(EnumSet.of(CompressionType.OLE)).setInvestigateEstimate(true).setLossy(true).create(),
- new CompressionSettingsBuilder().setSamplingRatio(0.1).setSeed(compressionSeed)
- .setValidCompressions(EnumSet.of(CompressionType.RLE)).setInvestigateEstimate(true).setLossy(true).create(),
- new CompressionSettingsBuilder().setSamplingRatio(0.1).setSeed(compressionSeed).setInvestigateEstimate(true)
- .create(),
- new CompressionSettingsBuilder().setSamplingRatio(1.0).setSeed(compressionSeed).setInvestigateEstimate(true)
- .setAllowSharedDictionary(false).setmaxStaticColGroupCoCode(1).setLossy(true).create(),
+ // new CompressionSettingsBuilder().setSamplingRatio(0.1).setSeed(compressionSeed)
+ // .setValidCompressions(EnumSet.of(CompressionType.DDC)).setInvestigateEstimate(true).setLossy(true).create(),
+ // new CompressionSettingsBuilder().setSamplingRatio(0.1).setSeed(compressionSeed)
+ // .setValidCompressions(EnumSet.of(CompressionType.OLE)).setInvestigateEstimate(true).setLossy(true).create(),
+ // new CompressionSettingsBuilder().setSamplingRatio(0.1).setSeed(compressionSeed)
+ // .setValidCompressions(EnumSet.of(CompressionType.RLE)).setInvestigateEstimate(true).setLossy(true).create(),
+ // new CompressionSettingsBuilder().setSamplingRatio(0.1).setSeed(compressionSeed).setInvestigateEstimate(true)
+ // .create(),
+ // new CompressionSettingsBuilder().setSamplingRatio(1.0).setSeed(compressionSeed).setInvestigateEstimate(true)
+ // .setAllowSharedDictionary(false).setmaxStaticColGroupCoCode(1).setLossy(true).create(),
// COCODING TESTS!!
@@ -132,8 +143,8 @@
protected CompressionStatistics cmbStats;
// Decompressed Result
- protected MatrixBlock cmbDeCompressed;
- protected double[][] deCompressed;
+ // protected MatrixBlock cmbDeCompressed;
+ // protected double[][] deCompressed;
/** number of threads used for the operation */
protected final int _k;
@@ -143,8 +154,9 @@
protected double lossyTolerance;
public CompressedTestBase(SparsityType sparType, ValueType valType, ValueRange valueRange,
- CompressionSettings compSettings, MatrixTypology MatrixTypology, int parallelism) {
- super(sparType, valType, valueRange, compSettings, MatrixTypology);
+ CompressionSettings compSettings, MatrixTypology MatrixTypology, OverLapping ov, int parallelism) {
+ super(sparType, valType, valueRange, compSettings, MatrixTypology, ov);
+
_k = parallelism;
try {
@@ -154,16 +166,47 @@
.compress(mb, _k, compressionSettings);
cmb = pair.getLeft();
cmbStats = pair.getRight();
+ MatrixBlock tmp = null;
+ switch(ov) {
+ case COL:
+ tmp = DataConverter.convertToMatrixBlock(TestUtils.generateTestMatrix(cols, 1, 0.5, 1.5, 1.0, 6));
+ lossyTolerance = lossyTolerance * 80;
+ cols = 1;
+ break;
+ case MATRIX:
+ case MATRIX_MULT_NEGATIVE:
+ case MATRIX_PLUS:
+ tmp = DataConverter.convertToMatrixBlock(TestUtils.generateTestMatrix(cols, 2, 0.5, 1.5, 1.0, 2));
+ lossyTolerance = lossyTolerance * 160;
+ cols = 2;
+ break;
+ default:
+ break;
+ }
if(cmb instanceof CompressedMatrixBlock) {
- cmbDeCompressed = ((CompressedMatrixBlock) cmb).decompress(_k);
- if(cmbDeCompressed != null) {
- deCompressed = DataConverter.convertToDoubleMatrix(cmbDeCompressed);
+ if(tmp != null) {
+ // Make Operator
+ AggregateBinaryOperator abop = InstructionUtils.getMatMultOperator(_k);
+
+ // vector-matrix uncompressed
+ mb = mb.aggregateBinaryOperations(mb, tmp, new MatrixBlock(), abop);
+
+ // vector-matrix compressed
+ cmb = cmb.aggregateBinaryOperations(cmb, tmp, new MatrixBlock(), abop);
+ if(ov == OverLapping.MATRIX_PLUS) {
+
+ ScalarOperator sop = new LeftScalarOperator(Plus.getPlusFnObject(), 15);
+ mb = mb.scalarOperations(sop, new MatrixBlock());
+ cmb = cmb.scalarOperations(sop, new MatrixBlock());
+ }
+ else if(ov == OverLapping.MATRIX_MULT_NEGATIVE) {
+ ScalarOperator sop = new LeftScalarOperator(Multiply.getMultiplyFnObject(), -1.3);
+ mb = mb.scalarOperations(sop, new MatrixBlock());
+ cmb = cmb.scalarOperations(sop, new MatrixBlock());
+ }
}
}
- else {
- cmbDeCompressed = null;
- deCompressed = null;
- }
+
}
catch(Exception e) {
e.printStackTrace();
@@ -187,45 +230,16 @@
public static Collection<Object[]> data() {
ArrayList<Object[]> tests = new ArrayList<>();
- for(SparsityType st : usedSparsityTypes) {
- for(ValueType vt : usedValueTypes) {
- for(ValueRange vr : usedValueRanges) {
- for(CompressionSettings cs : usedCompressionSettings) {
- for(MatrixTypology mt : usedMatrixTypology) {
- tests.add(new Object[] {st, vt, vr, cs, mt});
- }
- }
- }
- }
- }
-
+ for(SparsityType st : usedSparsityTypes)
+ for(ValueType vt : usedValueTypes)
+ for(ValueRange vr : usedValueRanges)
+ for(CompressionSettings cs : usedCompressionSettings)
+ for(MatrixTypology mt : usedMatrixTypology)
+ for(OverLapping ov : overLapping)
+ tests.add(new Object[] {st, vt, vr, cs, mt, ov});
return tests;
}
- // %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
- // %%%%%%%%%%%%%%%%% TESTS START! %%%%%%%%%%%%%%%%%
- // %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
-
- @Test
- public void testConstruction() {
- try {
- if(!(cmb instanceof CompressedMatrixBlock)) {
- return; // Input was not compressed then just pass test
- // Assert.assertTrue("Compression Failed \n" + this.toString(), false);
- }
- if(compressionSettings.lossy) {
- TestUtils.compareMatrices(input, deCompressed, lossyTolerance, compressionSettings.toString() + "\n");
- }
- else {
- TestUtils.compareMatricesBitAvgDistance(input, deCompressed, 0, 0, compressionSettings.toString());
- }
- }
- catch(Exception e) {
- e.printStackTrace();
- throw new RuntimeException(this.toString() + "\n" + e.getMessage(), e);
- }
- }
-
@Test
public void testDecompress() {
try {
@@ -233,13 +247,16 @@
return; // Input was not compressed then just pass test
// Assert.assertTrue("Compression Failed \n" + this.toString(), false);
}
+ double[][] org = DataConverter.convertToDoubleMatrix(mb);
double[][] deCompressed = DataConverter.convertToDoubleMatrix(((CompressedMatrixBlock) cmb).decompress(_k));
- if(compressionSettings.lossy) {
- TestUtils.compareMatrices(input, deCompressed, lossyTolerance);
- }
- else {
- TestUtils.compareMatricesBitAvgDistance(input, deCompressed, 0, 0, compressionSettings.toString());
- }
+ if(compressionSettings.lossy)
+ TestUtils.compareMatrices(org, deCompressed, lossyTolerance, this.toString());
+ else if(overlappingType == OverLapping.MATRIX_MULT_NEGATIVE || overlappingType == OverLapping.MATRIX_PLUS ||
+ overlappingType == OverLapping.MATRIX || overlappingType == OverLapping.COL)
+ TestUtils.compareMatricesBitAvgDistance(org, deCompressed, 8192, 124, this.toString());
+ else
+ TestUtils.compareMatricesBitAvgDistance(org, deCompressed, 5, 1, this.toString());
+
}
catch(Exception e) {
e.printStackTrace();
@@ -282,16 +299,18 @@
// scaledTolerance *= d1.length * d1.length * 0.5;
// }
// TestUtils.compareMatrices(d1, d2, d1.length, d1[0].length, scaledTolerance );
- TestUtils.compareMatricesPercentageDistance(d1, d2, 0.95, 0.95, compressionSettings.toString());
+ TestUtils.compareMatricesPercentageDistance(d1, d2, 0.95, 0.95, this.toString());
}
else {
- if(rows > 50000) {
- TestUtils
- .compareMatricesPercentageDistance(d1, d2, 0.99, 0.999, compressionSettings.toString());
- }
- else {
- TestUtils.compareMatricesBitAvgDistance(d1, d2, 2048, 350, compressionSettings.toString());
- }
+ if(rows > 50000)
+ TestUtils.compareMatricesPercentageDistance(d1, d2, 0.99, 0.999, this.toString());
+ else if(overlappingType == OverLapping.MATRIX_MULT_NEGATIVE ||
+ overlappingType == OverLapping.MATRIX_PLUS || overlappingType == OverLapping.MATRIX ||
+ overlappingType == OverLapping.COL)
+ TestUtils.compareMatricesPercentageDistance(d1, d2, 0.98, 0.99, this.toString());
+ else
+ TestUtils.compareMatricesBitAvgDistance(d1, d2, 2048, 350, this.toString());
+
}
}
}
@@ -302,6 +321,89 @@
}
@Test
+ public void testVectorMatrixMult() {
+
+ if(!(cmb instanceof CompressedMatrixBlock))
+ return; // Input was not compressed then just pass test
+
+ MatrixBlock vector = DataConverter
+ .convertToMatrixBlock(TestUtils.generateTestMatrix(1, rows, 0.9, 1.5, 1.0, 3));
+
+ testLeftMatrixMatrix(vector);
+ }
+
+ @Test
+ public void testLeftMatrixMatrixMultSmall() {
+
+ if(!(cmb instanceof CompressedMatrixBlock))
+ return; // Input was not compressed then just pass test
+
+ MatrixBlock matrix = DataConverter
+ .convertToMatrixBlock(TestUtils.generateTestMatrix(3, rows, 0.9, 1.5, 1.0, 3));
+
+ testLeftMatrixMatrix(matrix);
+
+ }
+
+ @Test
+ public void testLeftMatrixMatrixMultMedium() {
+
+ if(!(cmb instanceof CompressedMatrixBlock))
+ return; // Input was not compressed then just pass test
+
+ MatrixBlock matrix = DataConverter
+ .convertToMatrixBlock(TestUtils.generateTestMatrix(50, rows, 0.9, 1.5, 1.0, 3));
+
+ testLeftMatrixMatrix(matrix);
+ }
+
+ @Test
+ public void testLeftMatrixMatrixMultSparse() {
+
+ if(!(cmb instanceof CompressedMatrixBlock))
+ return; // Input was not compressed then just pass test
+
+ MatrixBlock matrix = DataConverter.convertToMatrixBlock(TestUtils.generateTestMatrix(2, rows, 0.9, 1.5, .1, 3));
+
+ testLeftMatrixMatrix(matrix);
+ }
+
+ public void testLeftMatrixMatrix(MatrixBlock matrix) {
+ try {
+ // Make Operator
+ AggregateBinaryOperator abop = InstructionUtils.getMatMultOperator(_k);
+
+ // vector-matrix uncompressed
+ MatrixBlock ret1 = mb.aggregateBinaryOperations(matrix, mb, new MatrixBlock(), abop);
+
+ // vector-matrix compressed
+ MatrixBlock ret2 = cmb.aggregateBinaryOperations(matrix, cmb, new MatrixBlock(), abop);
+
+ // compare result with input
+ double[][] d1 = DataConverter.convertToDoubleMatrix(ret1);
+ double[][] d2 = DataConverter.convertToDoubleMatrix(ret2);
+ if(compressionSettings.lossy) {
+ TestUtils.compareMatricesPercentageDistance(d1, d2, 0.25, 0.83, compressionSettings.toString());
+ }
+ else {
+ if(rows > 65000)
+ TestUtils.compareMatricesPercentageDistance(d1, d2, 0.99, 0.99, compressionSettings.toString());
+ else if(overlappingType == OverLapping.MATRIX_MULT_NEGATIVE ||
+ overlappingType == OverLapping.MATRIX_PLUS || overlappingType == OverLapping.MATRIX ||
+ overlappingType == OverLapping.COL)
+ TestUtils.compareMatricesBitAvgDistance(d1, d2, 1500000, 1000, this.toString());
+ else
+ TestUtils.compareMatricesBitAvgDistance(d1, d2, 24000, 512, compressionSettings.toString());
+
+ }
+ }
+ catch(Exception e) {
+ e.printStackTrace();
+ throw new RuntimeException(this.toString() + "\n" + e.getMessage(), e);
+ }
+ }
+
+ @Test
public void testMatrixVectorMult01() {
testMatrixVectorMult(1.0, 1.1);
}
@@ -322,295 +424,52 @@
}
public void testMatrixVectorMult(double min, double max) {
- try {
- if(!(cmb instanceof CompressedMatrixBlock))
- return; // Input was not compressed then just pass test
- MatrixBlock vector = DataConverter
- .convertToMatrixBlock(TestUtils.generateTestMatrix(cols, 1, min, max, 1.0, 3));
+ if(!(cmb instanceof CompressedMatrixBlock))
+ return; // Input was not compressed then just pass test
- // Make Operator // matrix-vector uncompressed
- // AggregateBinaryOperator abop = InstructionUtils.getMatMultOperator(_k);
- AggregateOperator aop = new AggregateOperator(0, Plus.getPlusFnObject());
- AggregateBinaryOperator abop = new AggregateBinaryOperator(Multiply.getMultiplyFnObject(), aop, _k);
-
- // matrix-vector uncompressed
- MatrixBlock ret1 = mb.aggregateBinaryOperations(mb, vector, new MatrixBlock(), abop);
-
- // matrix-vector compressed
- MatrixBlock ret2 = cmb.aggregateBinaryOperations(cmb, vector, new MatrixBlock(), abop);
-
- // compare result with input
- double[][] d1 = DataConverter.convertToDoubleMatrix(ret1);
- double[][] d2 = DataConverter.convertToDoubleMatrix(ret2);
-
- if(compressionSettings.lossy) {
- // TODO Make actual calculation to know the actual tolerance
- double scaledTolerance = lossyTolerance * 30 * max;
- TestUtils.compareMatrices(d1, d2, scaledTolerance);
- }
- else {
- TestUtils.compareMatricesBitAvgDistance(d1, d2, 120000, 128, compressionSettings.toString());
- }
- }
- catch(Exception e) {
- e.printStackTrace();
- throw new RuntimeException(this.toString() + "\n" + e.getMessage(), e);
- }
- }
-
- @Test
- public void testVectorMatrixMult() {
- try {
- if(!(cmb instanceof CompressedMatrixBlock))
- return; // Input was not compressed then just pass test
-
- MatrixBlock vector = DataConverter
- .convertToMatrixBlock(TestUtils.generateTestMatrix(1, rows, 0.9, 1.5, 1.0, 3));
-
- // Make Operator
- AggregateBinaryOperator abop = InstructionUtils.getMatMultOperator(_k);
-
- // vector-matrix uncompressed
- MatrixBlock ret1 = mb.aggregateBinaryOperations(vector, mb, new MatrixBlock(), abop);
-
- // vector-matrix compressed
- MatrixBlock ret2 = cmb.aggregateBinaryOperations(vector, cmb, new MatrixBlock(), abop);
-
- // compare result with input
- double[][] d1 = DataConverter.convertToDoubleMatrix(ret1);
- double[][] d2 = DataConverter.convertToDoubleMatrix(ret2);
- if(compressionSettings.lossy) {
- TestUtils.compareMatricesPercentageDistance(d1, d2, 0.35, 0.92, compressionSettings.toString());
- }
- else {
- TestUtils.compareMatricesBitAvgDistance(d1, d2, 10000, 500, compressionSettings.toString());
- }
- }
- catch(Exception e) {
- e.printStackTrace();
- throw new RuntimeException(this.toString() + "\n" + e.getMessage(), e);
- }
- }
-
- @Test
- public void testLeftMatrixMatrixMultSmall() {
- try {
- if(!(cmb instanceof CompressedMatrixBlock))
- return; // Input was not compressed then just pass test
-
- MatrixBlock matrix = DataConverter
- .convertToMatrixBlock(TestUtils.generateTestMatrix(3, rows, 0.9, 1.5, 1.0, 3));
-
- // Make Operator
- AggregateBinaryOperator abop = InstructionUtils.getMatMultOperator(_k);
-
- // vector-matrix uncompressed
- MatrixBlock ret1 = mb.aggregateBinaryOperations(matrix, mb, new MatrixBlock(), abop);
-
- // vector-matrix compressed
- MatrixBlock ret2 = cmb.aggregateBinaryOperations(matrix, cmb, new MatrixBlock(), abop);
-
- // compare result with input
- double[][] d1 = DataConverter.convertToDoubleMatrix(ret1);
- double[][] d2 = DataConverter.convertToDoubleMatrix(ret2);
- if(compressionSettings.lossy) {
- TestUtils.compareMatricesPercentageDistance(d1, d2, 0.5, 0.92, compressionSettings.toString());
- }
- else {
- // rows
- if(rows > 65000) {
- TestUtils.compareMatricesPercentageDistance(d1, d2, 0.5, 0.99, compressionSettings.toString());
- }
- else {
-
- TestUtils.compareMatricesBitAvgDistance(d1, d2, 10000, 500, compressionSettings.toString());
- }
- }
- }
- catch(Exception e) {
- e.printStackTrace();
- throw new RuntimeException(this.toString() + "\n" + e.getMessage(), e);
- }
- }
-
- @Test
- public void testLeftMatrixMatrixMultMedium() {
- try {
- if(!(cmb instanceof CompressedMatrixBlock))
- return; // Input was not compressed then just pass test
-
- int cols = 50;
- MatrixBlock matrix = DataConverter
- .convertToMatrixBlock(TestUtils.generateTestMatrix(cols, rows, 0.9, 1.5, 1.0, 3));
-
- // Make Operator
- AggregateBinaryOperator abop = InstructionUtils.getMatMultOperator(_k);
-
- // vector-matrix uncompressed
- MatrixBlock ret1 = mb.aggregateBinaryOperations(matrix, mb, new MatrixBlock(), abop);
-
- // vector-matrix compressed
- MatrixBlock ret2 = cmb.aggregateBinaryOperations(matrix, cmb, new MatrixBlock(), abop);
-
- // compare result with input
- double[][] d1 = DataConverter.convertToDoubleMatrix(ret1);
- double[][] d2 = DataConverter.convertToDoubleMatrix(ret2);
- if(compressionSettings.lossy) {
- TestUtils.compareMatrices(d1, d2, lossyTolerance * cols * rows / 100, this.toString());
- }
- else {
- if(rows > 65000) {
- TestUtils.compareMatricesPercentageDistance(d1, d2, 0.90, 0.99, this.toString());
- }
- else {
- TestUtils.compareMatricesBitAvgDistance(d1, d2, 1000 * 1000, 5096, this.toString());
- }
- }
- }
- catch(Exception e) {
- e.printStackTrace();
- throw new RuntimeException(this.toString() + "\n" + e.getMessage(), e);
- }
- }
-
- @Test
- public void testLeftMatrixMatrixMultSparse() {
- try {
- if(!(cmb instanceof CompressedMatrixBlock))
- return; // Input was not compressed then just pass test
-
- MatrixBlock matrix = DataConverter
- .convertToMatrixBlock(TestUtils.generateTestMatrix(2, rows, 0.9, 1.5, .1, 3));
-
- // Make Operator
- AggregateBinaryOperator abop = InstructionUtils.getMatMultOperator(_k);
-
- // vector-matrix uncompressed
- MatrixBlock ret1 = mb.aggregateBinaryOperations(matrix, mb, new MatrixBlock(), abop);
-
- // vector-matrix compressed
- MatrixBlock ret2 = cmb.aggregateBinaryOperations(matrix, cmb, new MatrixBlock(), abop);
-
- // compare result with input
- double[][] d1 = DataConverter.convertToDoubleMatrix(ret1);
- double[][] d2 = DataConverter.convertToDoubleMatrix(ret2);
- if(compressionSettings.lossy) {
- TestUtils.compareMatricesPercentageDistance(d1, d2, 0.25, 0.83, compressionSettings.toString());
- }
- else {
- if(rows > 65000) {
- TestUtils.compareMatricesPercentageDistance(d1, d2, 0.99, 0.99, compressionSettings.toString());
- }
- else {
- TestUtils.compareMatricesBitAvgDistance(d1, d2, 1000, 500, compressionSettings.toString());
- }
- }
- }
- catch(Exception e) {
- e.printStackTrace();
- throw new RuntimeException(this.toString() + "\n" + e.getMessage(), e);
- }
+ MatrixBlock vector = DataConverter
+ .convertToMatrixBlock(TestUtils.generateTestMatrix(cols, 1, min, max, 1.0, 3));
+ testRightMatrixMatrix(vector);
}
@Test
public void testRightMatrixMatrixMultSmall() {
- try {
- if(!(cmb instanceof CompressedMatrixBlock))
- return; // Input was not compressed then just pass test
- MatrixBlock matrix = DataConverter
- .convertToMatrixBlock(TestUtils.generateTestMatrix(cols, 2, 0.9, 1.5, 1.0, 3));
+ if(!(cmb instanceof CompressedMatrixBlock))
+ return; // Input was not compressed then just pass test
- // Make Operator
- AggregateBinaryOperator abop = InstructionUtils.getMatMultOperator(_k);
+ MatrixBlock matrix = DataConverter
+ .convertToMatrixBlock(TestUtils.generateTestMatrix(cols, 2, 0.9, 1.5, 1.0, 3));
- // vector-matrix uncompressed
- MatrixBlock ret1 = mb.aggregateBinaryOperations(mb, matrix, new MatrixBlock(), abop);
-
- // vector-matrix compressed
- MatrixBlock ret2 = cmb.aggregateBinaryOperations(cmb, matrix, new MatrixBlock(), abop);
-
- // compare result with input
- double[][] d1 = DataConverter.convertToDoubleMatrix(ret1);
- double[][] d2 = DataConverter.convertToDoubleMatrix(ret2);
- if(compressionSettings.lossy) {
- if(rows > 65000) {
- TestUtils.compareMatrices(d1, d2, lossyTolerance * cols * rows / 50, this.toString());
- }
- else {
- TestUtils.compareMatrices(d1, d2, lossyTolerance * cols * rows / 100, this.toString());
- }
- }
- else {
- if(rows > 65000) {
- TestUtils.compareMatricesPercentageDistance(d1, d2, 0.5, 0.99, this.toString());
- }
- else {
- TestUtils.compareMatricesBitAvgDistance(d1, d2, 10000, 500, this.toString());
- }
- }
- }
- catch(Exception e) {
- e.printStackTrace();
- throw new RuntimeException(this.toString() + "\n" + e.getMessage(), e);
- }
+ testRightMatrixMatrix(matrix);
}
@Test
public void testRightMatrixMatrixMultMedium() {
- try {
- if(!(cmb instanceof CompressedMatrixBlock))
- return; // Input was not compressed then just pass test
- MatrixBlock matrix = DataConverter
- .convertToMatrixBlock(TestUtils.generateTestMatrix(cols, 16, 0.9, 1.5, 1.0, 3));
+ if(!(cmb instanceof CompressedMatrixBlock))
+ return; // Input was not compressed then just pass test
- // Make Operator
- AggregateBinaryOperator abop = InstructionUtils.getMatMultOperator(_k);
+ MatrixBlock matrix = DataConverter
+ .convertToMatrixBlock(TestUtils.generateTestMatrix(cols, 16, 0.9, 1.5, 1.0, 3));
+ testRightMatrixMatrix(matrix);
- // vector-matrix uncompressed
- MatrixBlock ret1 = mb.aggregateBinaryOperations(mb, matrix, new MatrixBlock(), abop);
-
- // vector-matrix compressed
- MatrixBlock ret2 = cmb.aggregateBinaryOperations(cmb, matrix, new MatrixBlock(), abop);
-
- // compare result with input
- double[][] d1 = DataConverter.convertToDoubleMatrix(ret1);
- double[][] d2 = DataConverter.convertToDoubleMatrix(ret2);
- if(compressionSettings.lossy) {
- if(rows > 65000) {
- TestUtils.compareMatrices(d1, d2, lossyTolerance * cols * rows / 50, this.toString());
- }
- else {
- TestUtils.compareMatrices(d1, d2, lossyTolerance * cols * rows / 100, this.toString());
-
- }
- }
- else {
- if(rows > 65000) {
- TestUtils.compareMatricesPercentageDistance(d1, d2, 0.5, 0.99, this.toString());
- }
- else {
- TestUtils.compareMatricesBitAvgDistance(d1, d2, 10000, 500, this.toString());
- }
- }
- }
- catch(Exception e) {
- e.printStackTrace();
- throw new RuntimeException(this.toString() + "\n" + e.getMessage(), e);
- }
}
@Test
public void testRightMatrixMatrixMultSparse() {
+
+ if(!(cmb instanceof CompressedMatrixBlock))
+ return; // Input was not compressed then just pass test
+
+ MatrixBlock matrix = DataConverter
+ .convertToMatrixBlock(TestUtils.generateTestMatrix(cols, 25, 0.9, 1.5, 0.2, 3));
+ testRightMatrixMatrix(matrix);
+ }
+
+ public void testRightMatrixMatrix(MatrixBlock matrix) {
try {
- if(!(cmb instanceof CompressedMatrixBlock))
- return; // Input was not compressed then just pass test
-
- MatrixBlock matrix = DataConverter
- .convertToMatrixBlock(TestUtils.generateTestMatrix(cols, 25, 0.9, 1.5, 0.2, 3));
-
matrix.quickSetValue(0, 0, 10);
// Make Operator
AggregateBinaryOperator abop = InstructionUtils.getMatMultOperator(_k);
@@ -634,12 +493,15 @@
}
}
else {
- if(rows > 65000) {
+ if(rows > 65000)
TestUtils.compareMatricesPercentageDistance(d1, d2, 0.5, 0.99, compressionSettings.toString());
- }
- else {
+ else if(overlappingType == OverLapping.MATRIX_MULT_NEGATIVE ||
+ overlappingType == OverLapping.MATRIX_PLUS || overlappingType == OverLapping.MATRIX ||
+ overlappingType == OverLapping.COL)
+ TestUtils.compareMatricesBitAvgDistance(d1, d2, 1600000, 1000, this.toString());
+ else
TestUtils.compareMatricesBitAvgDistance(d1, d2, 10000, 500, compressionSettings.toString());
- }
+
}
}
catch(Exception e) {
@@ -669,15 +531,19 @@
// High probability that The value is off by some amount
if(compressionSettings.lossy) {
// Probably the worst thing you can do to increase the amount the values are estimated wrong
- TestUtils.compareMatricesPercentageDistance(d1, d2, 0.5, 0.8, compressionSettings.toString());
+ TestUtils.compareMatricesPercentageDistance(d1, d2, 0.5, 0.8, this.toString());
}
else {
if(rows > 50000) {
TestUtils
- .compareMatricesPercentageDistance(d1, d2, 0.99, 0.999, compressionSettings.toString());
+ .compareMatricesPercentageDistance(d1, d2, 0.99, 0.999, this.toString());
}
+ else if(overlappingType == OverLapping.MATRIX_MULT_NEGATIVE ||
+ overlappingType == OverLapping.MATRIX_PLUS || overlappingType == OverLapping.MATRIX ||
+ overlappingType == OverLapping.COL)
+ TestUtils.compareMatricesPercentageDistance(d1, d2, 0.98, 0.98, this.toString());
else {
- TestUtils.compareMatricesBitAvgDistance(d1, d2, 2048, 512, compressionSettings.toString());
+ TestUtils.compareMatricesBitAvgDistance(d1, d2, 2048, 512, this.toString());
}
}
}
@@ -689,31 +555,127 @@
}
@Test
- public void testScalarOperations() {
+ public void testScalarOpRightMultiplyPositive() {
+ double mult = 7;
+ ScalarOperator sop = new RightScalarOperator(Multiply.getMultiplyFnObject(), mult, _k);
+ testScalarOperations(sop, lossyTolerance * 7);
+ }
+
+ @Test
+ public void testScalarOpRightMultiplyNegative() {
+ double mult = -7;
+ ScalarOperator sop = new RightScalarOperator(Multiply.getMultiplyFnObject(), mult, _k);
+ testScalarOperations(sop, lossyTolerance * 7);
+ }
+
+ @Test
+ public void testScalarRightOpAddition() {
+ double addValue = 4;
+ ScalarOperator sop = new RightScalarOperator(Plus.getPlusFnObject(), addValue);
+ testScalarOperations(sop, lossyTolerance + 0.05);
+ }
+
+ @Test
+ public void testScalarRightOpSubtract() {
+ double addValue = 15;
+ ScalarOperator sop = new RightScalarOperator(Minus.getMinusFnObject(), addValue);
+ testScalarOperations(sop, lossyTolerance + 0.1);
+ }
+
+ @Test
+ public void testScalarRightOpLess() {
+ double addValue = 0.11;
+ ScalarOperator sop = new RightScalarOperator(LessThanEquals.getLessThanEqualsFnObject(), addValue);
+ testScalarOperations(sop, lossyTolerance + 0.1);
+ }
+
+ @Test
+ public void testScalarRightOpGreater() {
+ double addValue = 0.11;
+ ScalarOperator sop = new RightScalarOperator(GreaterThan.getGreaterThanFnObject(), addValue);
+ testScalarOperations(sop, lossyTolerance + 0.1);
+ }
+
+ @Test
+ public void testScalarRightOpPower2() {
+ double addValue = 2;
+ ScalarOperator sop = new RightScalarOperator(Power2.getPower2FnObject(), addValue);
+ testScalarOperations(sop, lossyTolerance + 0.1);
+ }
+
+ @Test
+ public void testScalarOpLeftMultiplyPositive() {
+ double mult = 7;
+ ScalarOperator sop = new LeftScalarOperator(Multiply.getMultiplyFnObject(), mult, _k);
+ testScalarOperations(sop, lossyTolerance * 7);
+ }
+
+ @Test
+ public void testScalarOpLeftMultiplyNegative() {
+ double mult = -7;
+ ScalarOperator sop = new LeftScalarOperator(Multiply.getMultiplyFnObject(), mult, _k);
+ testScalarOperations(sop, lossyTolerance * 7);
+ }
+
+ @Test
+ public void testScalarLeftOpAddition() {
+ double addValue = 4;
+ ScalarOperator sop = new LeftScalarOperator(Plus.getPlusFnObject(), addValue);
+ testScalarOperations(sop, lossyTolerance + 0.05);
+ }
+
+ @Test
+ public void testScalarLeftOpSubtract() {
+ double addValue = 15;
+ ScalarOperator sop = new LeftScalarOperator(Minus.getMinusFnObject(), addValue);
+ testScalarOperations(sop, lossyTolerance + 0.1);
+ }
+
+ @Test
+ public void testScalarLeftOpLess() {
+ double addValue = 0.11;
+ ScalarOperator sop = new LeftScalarOperator(LessThanEquals.getLessThanEqualsFnObject(), addValue);
+ testScalarOperations(sop, lossyTolerance + 0.1);
+ }
+
+ @Test
+ public void testScalarLeftOpGreater() {
+ double addValue = 0.11;
+ ScalarOperator sop = new LeftScalarOperator(GreaterThan.getGreaterThanFnObject(), addValue);
+ testScalarOperations(sop, lossyTolerance + 0.1);
+ }
+
+ // @Test
+ // This test does not make sense to execute... since the result of left power always is 4.
+ // Furthermore it does not work consistently in our normal matrix blocks ... and should never be used.
+ // public void testScalarLeftOpPower2() {
+ // double addValue = 2;
+ // ScalarOperator sop = new LeftScalarOperator(Power2.getPower2FnObject(), addValue);
+ // testScalarOperations(sop, lossyTolerance + 0.1);
+ // }
+
+ public void testScalarOperations(ScalarOperator sop, double tolerance) {
try {
if(!(cmb instanceof CompressedMatrixBlock))
return; // Input was not compressed then just pass test
- double mult = 7;
- // matrix-scalar uncompressed
- ScalarOperator sop = new RightScalarOperator(Multiply.getMultiplyFnObject(), mult, _k);
MatrixBlock ret1 = mb.scalarOperations(sop, new MatrixBlock());
// matrix-scalar compressed
MatrixBlock ret2 = cmb.scalarOperations(sop, new MatrixBlock());
- if(ret2 instanceof CompressedMatrixBlock)
- ret2 = ((CompressedMatrixBlock) ret2).decompress();
// compare result with input
double[][] d1 = DataConverter.convertToDoubleMatrix(ret1);
double[][] d2 = DataConverter.convertToDoubleMatrix(ret2);
- if(compressionSettings.lossy) {
- double modifiedTolerance = lossyTolerance * mult + lossyTolerance * 0.00001;
- TestUtils.compareMatrices(d1, d2, modifiedTolerance, compressionSettings.toString());
- }
- else {
- TestUtils.compareMatricesBitAvgDistance(d1, d2, 150, 1, compressionSettings.toString());
- }
+
+ if(compressionSettings.lossy)
+ TestUtils.compareMatrices(d1, d2, tolerance, this.toString());
+ else if(overlappingType == OverLapping.MATRIX_MULT_NEGATIVE || overlappingType == OverLapping.MATRIX_PLUS ||
+ overlappingType == OverLapping.MATRIX || overlappingType == OverLapping.COL)
+ TestUtils.compareMatricesBitAvgDistance(d1, d2, 50000, 1000, this.toString());
+ else
+ TestUtils.compareMatricesBitAvgDistance(d1, d2, 150, 1, this.toString());
+
}
catch(Exception e) {
e.printStackTrace();
@@ -722,33 +684,91 @@
}
@Test
- public void testScalarOperationsSparseUnsafe() {
+ public void testBinaryMVAddition() {
+
+ ValueFunction vf = Plus.getPlusFnObject();
+ testBinaryMV(vf);
+ }
+
+ @Test
+ public void testBinaryMVMultiply() {
+ ValueFunction vf = Multiply.getMultiplyFnObject();
+ testBinaryMV(vf);
+ }
+
+ @Test
+ public void testBinaryMVMinus() {
+ ValueFunction vf = Minus.getMinusFnObject();
+ testBinaryMV(vf);
+ }
+
+ @Test
+ public void testBinaryMVXor() {
+ ValueFunction vf = Xor.getXorFnObject();
+ testBinaryMV(vf);
+ }
+
+ public void testBinaryMV(ValueFunction vf) {
try {
if(!(cmb instanceof CompressedMatrixBlock))
return; // Input was not compressed then just pass test
- double addValue = 1000;
- // matrix-scalar uncompressed
- ScalarOperator sop = new RightScalarOperator(Plus.getPlusFnObject(), addValue);
- MatrixBlock ret1 = mb.scalarOperations(sop, new MatrixBlock());
+ BinaryOperator bop = new BinaryOperator(vf);
+ MatrixBlock vector = DataConverter
+ .convertToMatrixBlock(TestUtils.generateTestMatrix(1, cols, -1.0, 1.5, 1.0, 3));
- // matrix-scalar compressed
- MatrixBlock ret2 = cmb.scalarOperations(sop, new MatrixBlock());
+ MatrixBlock ret1 = mb.binaryOperations(bop, vector, new MatrixBlock());
+ MatrixBlock ret2 = cmb.binaryOperations(bop, vector, new MatrixBlock());
if(ret2 instanceof CompressedMatrixBlock)
ret2 = ((CompressedMatrixBlock) ret2).decompress();
-
- // compare result with input
double[][] d1 = DataConverter.convertToDoubleMatrix(ret1);
double[][] d2 = DataConverter.convertToDoubleMatrix(ret2);
- if(compressionSettings.lossy) {
- double modifiedTolerance = Math.max(TestConstants.getMaxRangeValue(valRange) + addValue,
- Math.abs(TestConstants.getMinRangeValue(valRange) + addValue)) * 2 / 127.0;
- TestUtils.compareMatrices(d1, d2, modifiedTolerance, compressionSettings.toString());
- }
- else {
- TestUtils.compareMatricesBitAvgDistance(d1, d2, 150, 1, compressionSettings.toString());
- }
+ if(compressionSettings.lossy)
+ TestUtils.compareMatrices(d1, d2, lossyTolerance * 2, this.toString());
+ else if(overlappingType == OverLapping.MATRIX_MULT_NEGATIVE || overlappingType == OverLapping.MATRIX_PLUS ||
+ overlappingType == OverLapping.MATRIX || overlappingType == OverLapping.COL)
+ TestUtils.compareMatricesBitAvgDistance(d1, d2, 65536, 512, this.toString());
+ else
+ TestUtils.compareMatricesBitAvgDistance(d1, d2, 150, 1, this.toString());
+
+ }
+ catch(Exception e) {
+ e.printStackTrace();
+ throw new RuntimeException(this.toString() + "\n" + e.getMessage(), e);
+ }
+ }
+
+ @Test
+ public void testBinaryVMMultiply() {
+ ValueFunction vf = Multiply.getMultiplyFnObject();
+ testBinaryVM(vf);
+ }
+
+ public void testBinaryVM(ValueFunction vf) {
+ try {
+ if(!(cmb instanceof CompressedMatrixBlock))
+ return; // Input was not compressed then just pass test
+ // NOTE THIS METHOD DECOMPRESSES AND MULTIPLIES
+
+ BinaryOperator bop = new BinaryOperator(vf);
+ MatrixBlock vector = DataConverter
+ .convertToMatrixBlock(TestUtils.generateTestMatrix(rows, 1, -1.0, 1.5, 1.0, 3));
+
+ MatrixBlock ret1 = mb.binaryOperations(bop, vector, new MatrixBlock());
+ MatrixBlock ret2 = cmb.binaryOperations(bop, vector, new MatrixBlock());
+
+ double[][] d1 = DataConverter.convertToDoubleMatrix(ret1);
+ double[][] d2 = DataConverter.convertToDoubleMatrix(ret2);
+
+ if(compressionSettings.lossy)
+ TestUtils.compareMatrices(d1, d2, lossyTolerance * 2, this.toString());
+ else if(overlappingType == OverLapping.MATRIX_MULT_NEGATIVE || overlappingType == OverLapping.MATRIX_PLUS ||
+ overlappingType == OverLapping.MATRIX || overlappingType == OverLapping.COL)
+ TestUtils.compareMatricesBitAvgDistance(d1, d2, 65536, 512, this.toString());
+ else
+ TestUtils.compareMatricesBitAvgDistance(d1, d2, 150, 1, this.toString());
+
}
catch(Exception e) {
e.printStackTrace();
diff --git a/src/test/java/org/apache/sysds/test/component/compress/CompressedVectorTest.java b/src/test/java/org/apache/sysds/test/component/compress/CompressedVectorTest.java
index a2090c0..1723d83 100644
--- a/src/test/java/org/apache/sysds/test/component/compress/CompressedVectorTest.java
+++ b/src/test/java/org/apache/sysds/test/component/compress/CompressedVectorTest.java
@@ -26,13 +26,13 @@
import org.apache.sysds.runtime.compress.CompressedMatrixBlock;
import org.apache.sysds.runtime.compress.CompressionSettings;
-import org.apache.sysds.runtime.controlprogram.parfor.stat.InfrastructureAnalyzer;
import org.apache.sysds.runtime.functionobjects.CM;
import org.apache.sysds.runtime.matrix.data.MatrixBlock;
import org.apache.sysds.runtime.matrix.operators.CMOperator;
import org.apache.sysds.runtime.matrix.operators.CMOperator.AggregateOperationTypes;
import org.apache.sysds.test.TestUtils;
import org.apache.sysds.test.component.compress.TestConstants.MatrixTypology;
+import org.apache.sysds.test.component.compress.TestConstants.OverLapping;
import org.apache.sysds.test.component.compress.TestConstants.SparsityType;
import org.apache.sysds.test.component.compress.TestConstants.ValueRange;
import org.apache.sysds.test.component.compress.TestConstants.ValueType;
@@ -45,37 +45,34 @@
public class CompressedVectorTest extends CompressedTestBase {
protected static MatrixTypology[] usedMatrixTypologyLocal = new MatrixTypology[] {// types
- MatrixTypology.SINGLE_COL,
+ MatrixTypology.SINGLE_COL,
// MatrixTypology.SINGLE_COL_L
};
@Parameters
public static Collection<Object[]> data() {
ArrayList<Object[]> tests = new ArrayList<>();
- for(SparsityType st : usedSparsityTypes) {
- for(ValueType vt : usedValueTypes) {
- for(ValueRange vr : usedValueRanges) {
- for(CompressionSettings cs : usedCompressionSettings) {
- for(MatrixTypology mt : usedMatrixTypologyLocal) {
- tests.add(new Object[] {st, vt, vr, cs, mt});
- }
- }
- }
- }
- }
+ for(SparsityType st : usedSparsityTypes)
+ for(ValueType vt : usedValueTypes)
+ for(ValueRange vr : usedValueRanges)
+ for(CompressionSettings cs : usedCompressionSettings)
+ for(MatrixTypology mt : usedMatrixTypologyLocal)
+ for(OverLapping ov : overLapping)
+ tests.add(new Object[] {st, vt, vr, cs, mt, ov});
+
return tests;
}
public CompressedVectorTest(SparsityType sparType, ValueType valType, ValueRange valRange,
- CompressionSettings compSettings, MatrixTypology matrixTypology) {
- super(sparType, valType, valRange, compSettings, matrixTypology, InfrastructureAnalyzer.getLocalParallelism());
+ CompressionSettings compSettings, MatrixTypology matrixTypology, OverLapping ov) {
+ super(sparType, valType, valRange, compSettings, matrixTypology, ov, 1);
}
@Test
public void testCentralMoment() throws Exception {
// TODO: Make Central Moment Test work on Multi dimensional Matrix
try {
- if(!(cmb instanceof CompressedMatrixBlock))
+ if(!(cmb instanceof CompressedMatrixBlock) || cols != 1)
return; // Input was not compressed then just pass test
// quantile uncompressed
@@ -106,6 +103,9 @@
@Test
public void testQuantile() {
try {
+ if(!(cmb instanceof CompressedMatrixBlock) || cols != 1)
+ return; // Input was not compressed then just pass test
+
// quantile uncompressed
MatrixBlock tmp1 = mb.sortOperations(null, new MatrixBlock());
double ret1 = tmp1.pickValue(0.95);
diff --git a/src/test/java/org/apache/sysds/test/component/compress/ParCompressedMatrixTest.java b/src/test/java/org/apache/sysds/test/component/compress/ParCompressedMatrixTest.java
index 57bf132..f9a1eb4 100644
--- a/src/test/java/org/apache/sysds/test/component/compress/ParCompressedMatrixTest.java
+++ b/src/test/java/org/apache/sysds/test/component/compress/ParCompressedMatrixTest.java
@@ -29,6 +29,7 @@
import org.apache.sysds.runtime.util.DataConverter;
import org.apache.sysds.test.TestUtils;
import org.apache.sysds.test.component.compress.TestConstants.MatrixTypology;
+import org.apache.sysds.test.component.compress.TestConstants.OverLapping;
import org.apache.sysds.test.component.compress.TestConstants.SparsityType;
import org.apache.sysds.test.component.compress.TestConstants.ValueRange;
import org.apache.sysds.test.component.compress.TestConstants.ValueType;
@@ -40,8 +41,8 @@
public class ParCompressedMatrixTest extends AbstractCompressedUnaryTests {
public ParCompressedMatrixTest(SparsityType sparType, ValueType valType, ValueRange valRange,
- CompressionSettings compressionSettings, MatrixTypology matrixTypology) {
- super(sparType, valType, valRange, compressionSettings, matrixTypology,
+ CompressionSettings compressionSettings, MatrixTypology matrixTypology, OverLapping ov) {
+ super(sparType, valType, valRange, compressionSettings, matrixTypology, ov,
InfrastructureAnalyzer.getLocalParallelism());
}
@@ -73,15 +74,18 @@
double[][] d1 = DataConverter.convertToDoubleMatrix(ret1);
double[][] d2 = DataConverter.convertToDoubleMatrix(ret2);
if(compressionSettings.lossy) {
- TestUtils.compareMatricesPercentageDistance(d1, d2, 0.25, 0.83, compressionSettings.toString());
+ TestUtils.compareMatricesPercentageDistance(d1, d2, 0.25, 0.83, this.toString());
}
else {
- if(rows > 65000) {
- TestUtils.compareMatricesPercentageDistance(d1, d2, 0.50, 0.99, compressionSettings.toString());
- }
- else {
- TestUtils.compareMatricesBitAvgDistance(d1, d2, 10000, 500, compressionSettings.toString());
- }
+ if(rows > 65000)
+ TestUtils.compareMatricesPercentageDistance(d1, d2, 0.50, 0.99, this.toString());
+ else if(overlappingType == OverLapping.MATRIX_MULT_NEGATIVE ||
+ overlappingType == OverLapping.MATRIX_PLUS || overlappingType == OverLapping.MATRIX ||
+ overlappingType == OverLapping.COL)
+ TestUtils.compareMatricesBitAvgDistance(d1, d2, 50000, 1000, this.toString());
+ else
+ TestUtils.compareMatricesBitAvgDistance(d1, d2, 15000, 500, this.toString());
+
}
}
catch(Exception e) {
diff --git a/src/test/java/org/apache/sysds/test/component/compress/TestBase.java b/src/test/java/org/apache/sysds/test/component/compress/TestBase.java
index f31ad48..2e9341f 100644
--- a/src/test/java/org/apache/sysds/test/component/compress/TestBase.java
+++ b/src/test/java/org/apache/sysds/test/component/compress/TestBase.java
@@ -28,6 +28,7 @@
import org.apache.sysds.runtime.util.DataConverter;
import org.apache.sysds.test.TestUtils;
import org.apache.sysds.test.component.compress.TestConstants.MatrixTypology;
+import org.apache.sysds.test.component.compress.TestConstants.OverLapping;
import org.apache.sysds.test.component.compress.TestConstants.SparsityType;
import org.apache.sysds.test.component.compress.TestConstants.ValueRange;
import org.apache.sysds.test.component.compress.TestConstants.ValueType;
@@ -46,13 +47,14 @@
protected double sparsity;
protected CompressionSettings compressionSettings;
+ protected OverLapping overlappingType;
// Input
protected double[][] input;
protected MatrixBlock mb;
public TestBase(SparsityType sparType, ValueType valType, ValueRange valueRange,
- CompressionSettings compressionSettings, MatrixTypology MatrixTypology) {
+ CompressionSettings compressionSettings, MatrixTypology MatrixTypology, OverLapping ov) {
this.sparsity = TestConstants.getSparsityValue(sparType);
this.rows = TestConstants.getNumberOfRows(MatrixTypology);
@@ -60,7 +62,7 @@
this.max = TestConstants.getMaxRangeValue(valueRange);
this.min = TestConstants.getMinRangeValue(valueRange);
-
+ this.overlappingType = ov;
try {
switch(valType) {
case CONST:
@@ -126,6 +128,7 @@
builder.append(String.format("%6s%12s", "Min:", min));
builder.append(String.format("%6s%12s", "Max:", max));
builder.append(String.format("%6s%5s", "Spar:", sparsity));
+ builder.append(String.format("%6s%5s", "OV:", overlappingType));
builder.append(String.format("%6s%8s", "CP:", compressionSettings));
return builder.toString();
diff --git a/src/test/java/org/apache/sysds/test/component/compress/TestConstants.java b/src/test/java/org/apache/sysds/test/component/compress/TestConstants.java
index 5846686..097b851 100644
--- a/src/test/java/org/apache/sysds/test/component/compress/TestConstants.java
+++ b/src/test/java/org/apache/sysds/test/component/compress/TestConstants.java
@@ -59,6 +59,10 @@
SMALL, LARGE, BYTE
}
+ public enum OverLapping{
+ COL, MATRIX, NONE, MATRIX_PLUS, MATRIX_MULT_NEGATIVE
+ }
+
public static double getSparsityValue(SparsityType sparsityType) {
switch(sparsityType) {
case DENSE:
