[SYSTEMDS-3490] Compressed Transformencode Binning

.gitignore

@@ -139,3 +139,4 @@ scripts/perftest/fed/temp
 src/test/scripts/functions/iogen/*.raw
 src/test/scripts/functions/pipelines/intermediates/regression/*.csv
 src/test/scripts/functions/pipelines/intermediates/regression/*.csv.mtd
+src/test/scripts/functions/pipelines/intermediates/classification/*

src/main/java/org/apache/sysds/lops/OperatorOrderingUtils.java

@@ -19,16 +19,15 @@
 
 package org.apache.sysds.lops;
 
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.Map;
+
 import org.apache.sysds.common.Types;
 import org.apache.sysds.hops.AggBinaryOp;
 import org.apache.sysds.parser.DMLProgram;
 import org.apache.sysds.parser.StatementBlock;
 
-import java.util.ArrayList;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-
 public class OperatorOrderingUtils
 {
 	// Return a list representation of all the lops in a SB

src/main/java/org/apache/sysds/runtime/compress/CompressedMatrixBlock.java

@@ -404,16 +404,18 @@ public void write(DataOutput out) throws IOException {
 		if(nonZeros > 0 && estDisk > estimateUncompressed) {
 			// If the size of this matrixBlock is smaller in uncompressed format, then
 			// decompress and save inside an uncompressed column group.
-			MatrixBlock uncompressed = getUncompressed(
-				"smaller serialization size: compressed: " + estDisk + " vs uncompressed: " + estimateUncompressed);
-			ColGroupUncompressed cg = (ColGroupUncompressed) ColGroupUncompressed.create(uncompressed);
-
-			if( estDisk / 10 > estimateUncompressed){
-				LOG.error(this);
-			}
+			final String message = "smaller serialization size: compressed: " + estDisk + " vs uncompressed: "
+				+ estimateUncompressed;
+			final MatrixBlock uncompressed = getUncompressed(message);
+			uncompressed.examSparsity(true);
+			// Here only Empty or Uncompressed should be returned.
+			AColGroup cg = ColGroupUncompressed.create(uncompressed);
 			allocateColGroup(cg);
-			nonZeros = cg.getNumberNonZeros(rlen);
-			// clear the soft reference to the decompressed version, since the one column group is perfectly,
+			// update non zeros, if not fully correct in compressed block
+			nonZeros = cg.getNumberNonZeros(rlen); 
+
+			// Clear the soft reference to the decompressed version,
+			// since the one column group is perfectly,
 			// representing the decompressed version.
 			clearSoftReferenceToDecompressed();
 		}
@@ -618,7 +620,7 @@ public MatrixBlock unaryOperations(UnaryOperator op, MatrixValue result) {
 	@Override
 	public boolean containsValue(double pattern) {
 		// Only if pattern is a finite value and overlapping then decompress.
-		if(isOverlapping() && Double.isFinite(pattern)) 
+		if(isOverlapping() && Double.isFinite(pattern))
 			return getUncompressed("ContainsValue").containsValue(pattern);
 		else {
 			for(AColGroup g : _colGroups)
@@ -1072,7 +1074,7 @@ public void clearSoftReferenceToDecompressed() {
 		decompressedVersion = null;
 	}
 
-	public void clearCounts(){
+	public void clearCounts() {
 		for(AColGroup a : _colGroups)
 			a.clear();
 	}

src/main/java/org/apache/sysds/runtime/compress/colgroup/ADictBasedColGroup.java

@@ -26,6 +26,7 @@
 
 import org.apache.sysds.runtime.compress.colgroup.dictionary.ADictionary;
 import org.apache.sysds.runtime.compress.colgroup.dictionary.Dictionary;
+import org.apache.sysds.runtime.compress.colgroup.dictionary.IdentityDictionary;
 import org.apache.sysds.runtime.compress.colgroup.dictionary.MatrixBlockDictionary;
 import org.apache.sysds.runtime.compress.colgroup.indexes.ColIndexFactory;
 import org.apache.sysds.runtime.compress.colgroup.indexes.IColIndex;
@@ -58,7 +59,17 @@ public ADictionary getDictionary() {
 
 	@Override
 	public final void decompressToDenseBlock(DenseBlock db, int rl, int ru, int offR, int offC) {
-		if(_dict instanceof MatrixBlockDictionary) {
+		if(_dict instanceof IdentityDictionary){
+
+			final MatrixBlockDictionary md = ((IdentityDictionary)_dict).getMBDict();
+			final MatrixBlock mb = md.getMatrixBlock();
+			// The dictionary is never empty.
+			if(mb.isInSparseFormat())
+				decompressToDenseBlockSparseDictionary(db, rl, ru, offR, offC, mb.getSparseBlock());
+			else
+				decompressToDenseBlockDenseDictionary(db, rl, ru, offR, offC, mb.getDenseBlockValues());
+		}
+		else if(_dict instanceof MatrixBlockDictionary) {
 			final MatrixBlockDictionary md = (MatrixBlockDictionary) _dict;
 			final MatrixBlock mb = md.getMatrixBlock();
 			// The dictionary is never empty.
@@ -73,7 +84,17 @@ public final void decompressToDenseBlock(DenseBlock db, int rl, int ru, int offR
 
 	@Override
 	public final void decompressToSparseBlock(SparseBlock sb, int rl, int ru, int offR, int offC) {
-		if(_dict instanceof MatrixBlockDictionary) {
+		if(_dict instanceof IdentityDictionary){
+
+			final MatrixBlockDictionary md = ((IdentityDictionary)_dict).getMBDict();
+			final MatrixBlock mb = md.getMatrixBlock();
+			// The dictionary is never empty.
+			if(mb.isInSparseFormat())
+				decompressToSparseBlockSparseDictionary(sb, rl, ru, offR, offC, mb.getSparseBlock());
+			else
+				decompressToSparseBlockDenseDictionary(sb, rl, ru, offR, offC, mb.getDenseBlockValues());
+		}
+		else if(_dict instanceof MatrixBlockDictionary) {
 			final MatrixBlockDictionary md = (MatrixBlockDictionary) _dict;
 			final MatrixBlock mb = md.getMatrixBlock();
 			// The dictionary is never empty.

src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupConst.java

@@ -26,6 +26,7 @@
 import org.apache.sysds.runtime.compress.colgroup.dictionary.ADictionary;
 import org.apache.sysds.runtime.compress.colgroup.dictionary.Dictionary;
 import org.apache.sysds.runtime.compress.colgroup.dictionary.DictionaryFactory;
+import org.apache.sysds.runtime.compress.colgroup.dictionary.IdentityDictionary;
 import org.apache.sysds.runtime.compress.colgroup.dictionary.MatrixBlockDictionary;
 import org.apache.sysds.runtime.compress.colgroup.indexes.ColIndexFactory;
 import org.apache.sysds.runtime.compress.colgroup.indexes.IColIndex;
@@ -305,7 +306,14 @@ public AColGroup binaryRowOpRight(BinaryOperator op, double[] v, boolean isRowSa
 	 * @param constV The output columns.
 	 */
 	public final void addToCommon(double[] constV) {
-		if(_dict instanceof MatrixBlockDictionary) {
+		if(_dict instanceof IdentityDictionary){
+			MatrixBlock mb = ((IdentityDictionary) _dict).getMBDict().getMatrixBlock();
+			if(mb.isInSparseFormat())
+				addToCommonSparse(constV, mb.getSparseBlock());
+			else
+				addToCommonDense(constV, mb.getDenseBlockValues());
+		}
+		else if(_dict instanceof MatrixBlockDictionary) {
 			MatrixBlock mb = ((MatrixBlockDictionary) _dict).getMatrixBlock();
 			if(mb.isInSparseFormat())
 				addToCommonSparse(constV, mb.getSparseBlock());

src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupSDCSingleZeros.java

@@ -96,10 +96,11 @@ protected void decompressToDenseBlockDenseDictionary(DenseBlock db, int rl, int
 		if(it == null)
 			return;
 		else if(it.value() >= ru)
-			_indexes.cacheIterator(it, ru);
+			return;
+			// _indexes.cacheIterator(it, ru);
 		else {
 			decompressToDenseBlockDenseDictionaryWithProvidedIterator(db, rl, ru, offR, offC, values, it);
-			_indexes.cacheIterator(it, ru);
+			// _indexes.cacheIterator(it, ru);
 		}
 	}
 

src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupSDCZeros.java

@@ -113,7 +113,7 @@ else if(it.value() >= ru)
 			_indexes.cacheIterator(it, ru);
 		else {
 			decompressToDenseBlockDenseDictionaryWithProvidedIterator(db, rl, ru, offR, offC, values, it);
-			_indexes.cacheIterator(it, ru);
+			// _indexes.cacheIterator(it, ru);
 		}
 	}
 
@@ -322,7 +322,7 @@ private final void decompressToDenseBlockSparseDictionaryPre(DenseBlock db, int
 
 			it.next();
 		}
-		_indexes.cacheIterator(it, ru);
+		// _indexes.cacheIterator(it, ru);
 	}
 
 	@Override

src/main/java/org/apache/sysds/runtime/compress/colgroup/dictionary/IdentityDictionary.java

@@ -26,10 +26,10 @@
 import java.util.Arrays;
 
 import org.apache.commons.lang.NotImplementedException;
-import org.apache.sysds.runtime.DMLRuntimeException;
 import org.apache.sysds.runtime.compress.DMLCompressionException;
 import org.apache.sysds.runtime.compress.colgroup.indexes.IColIndex;
 import org.apache.sysds.runtime.data.SparseBlock;
+import org.apache.sysds.runtime.data.SparseBlockFactory;
 import org.apache.sysds.runtime.functionobjects.Builtin;
 import org.apache.sysds.runtime.functionobjects.Builtin.BuiltinCode;
 import org.apache.sysds.runtime.functionobjects.ValueFunction;
@@ -61,13 +61,13 @@ public IdentityDictionary(int nRowCol) {
 
 	@Override
 	public double[] getValues() {
-		LOG.warn("Should not call getValues on Identity Dictionary");
-
-		double[] ret = new double[nRowCol * nRowCol];
-		for(int i = 0; i < nRowCol; i++) {
-			ret[(i * nRowCol) + i] = 1;
-		}
-		return ret;
+		throw new DMLCompressionException("Invalid to materialize identity Matrix Please Implement alternative");
+		// LOG.warn("Should not call getValues on Identity Dictionary");
+		// double[] ret = new double[nRowCol * nRowCol];
+		// for(int i = 0; i < nRowCol; i++) {
+		// 	ret[(i * nRowCol) + i] = 1;
+		// }
+		// return ret;
 	}
 
 	@Override
@@ -383,8 +383,7 @@ public ADictionary subtractTuple(double[] tuple) {
 	}
 
 	public MatrixBlockDictionary getMBDict() {
-		throw new DMLRuntimeException("Do not make MB Dict");
-		// return getMBDict(nRowCol);
+		return getMBDict(nRowCol);
 	}
 
 	@Override
@@ -400,10 +399,8 @@ public MatrixBlockDictionary getMBDict(int nCol) {
 	}
 
 	private MatrixBlockDictionary createMBDict() {
-		MatrixBlock identity = new MatrixBlock(nRowCol, nRowCol, true);
-		for(int i = 0; i < nRowCol; i++)
-			identity.quickSetValue(i, i, 1.0);
-
+		final SparseBlock sb = SparseBlockFactory.createIdentityMatrix(nRowCol);
+		final MatrixBlock identity = new MatrixBlock(nRowCol, nRowCol, nRowCol,  sb);
 		return new MatrixBlockDictionary(identity);
 	}
 

src/main/java/org/apache/sysds/runtime/compress/colgroup/dictionary/IdentityDictionarySlice.java

@@ -217,11 +217,6 @@ public long getNumberNonZeros(int[] counts, int nCol) {
 		return (long) sum(counts, nCol);
 	}
 
-
-	public MatrixBlockDictionary getMBDict() {
-		return getMBDict(nRowCol);
-	}
-
 	@Override
 	public MatrixBlockDictionary getMBDict(int nCol) {
 		if(cache != null) {

src/main/java/org/apache/sysds/runtime/compress/colgroup/offset/AOffset.java

@@ -95,7 +95,7 @@ public AIterator getIterator(int row) {
 			return getIterator();
 		else if(row > getOffsetToLast())
 			return null;
-		final OffsetCache c = cacheRow.get();
+		final OffsetCache c = getLength() < skipStride ? null :  cacheRow.get();
 		if(c != null && c.row == row)
 			return c.it.clone();
 		else if(getLength() < skipStride)
@@ -169,7 +169,7 @@ public AOffsetIterator getOffsetIterator(int row) {
 	 * @param row The row index to cache the iterator as.
 	 */
 	public void cacheIterator(AIterator it, int row) {
-		if(it == null)
+		if(it == null || getLength() < skipStride)
 			return;
 		cacheRow.set(new OffsetCache(it, row));
 	}
@@ -446,6 +446,12 @@ public boolean equals(AOffset b) {
 
 	protected abstract AOffset moveIndex(int m);
 
+	/**
+	 * Get the length of the underlying array. This does not reflect the number of contained elements, since some of the
+	 * elements can be skips.
+	 * 
+	 * @return The length of the underlying arrays
+	 */
 	protected abstract int getLength();
 
 	public OffsetSliceInfo slice(int l, int u) {

src/main/java/org/apache/sysds/runtime/compress/io/WriterCompressed.java

@@ -180,8 +180,10 @@ private void writeMultiBlockCompressed(String fname, MatrixBlock b, final int rl
 			for(int bc = 0; bc * blen < clen; bc++) {// column blocks
 				final int sC = bc * blen;
 				final int mC = Math.min(sC + blen, clen) - 1;
+				// slice out the current columns
 				final CompressedMatrixBlock mc = CLALibSlice.sliceColumns((CompressedMatrixBlock) b, sC, mC);
-				final List<MatrixBlock> blocks = CLALibSlice.sliceBlocks(mc, blen); // Slice compressed blocks
+				// slice out row blocks in this.
+				final List<MatrixBlock> blocks = CLALibSlice.sliceBlocks(mc, blen, k); // Slice compressed blocks
 				final int blocksPerThread = Math.max(1, blocks.size() / k);
 				for(int block = 0; block < blocks.size(); block += blocksPerThread, i++) {
 					final Path newPath = new Path(fname, IOUtilFunctions.getPartFileName(i));

src/main/java/org/apache/sysds/runtime/compress/lib/CLALibDecompress.java

@@ -177,20 +177,18 @@ private static MatrixBlock decompressExecute(CompressedMatrixBlock cmb, int k) {
 		if(k == 1) {
 			if(ret.isInSparseFormat()) {
 				decompressSparseSingleThread(ret, filteredGroups, nRows, blklen);
-				ret.setNonZeros(nonZeros);
 			}
 			else {
 				decompressDenseSingleThread(ret, filteredGroups, nRows, blklen, constV, eps, nonZeros, overlapping);
-				ret.recomputeNonZeros();
-				// ret.setNonZeros(nonZeros == -1 || overlapping ? ret.recomputeNonZeros() : nonZeros);
 			}
 		}
 		else if(ret.isInSparseFormat()) {
 			decompressSparseMultiThread(ret, filteredGroups, nRows, blklen, k);
-			ret.setNonZeros(nonZeros);
 		}
-		else
+		else{
 			decompressDenseMultiThread(ret, filteredGroups, nRows, blklen, constV, eps, k, overlapping);
+		}
+		ret.recomputeNonZeros();
 
 		ret.examSparsity();
 		return ret;