added config option for aggressive compression

extended the compression workload analyzer to pick up aggregation operations if the input is compressed as a single column group

extended the compression workload analyzer to pick up binary matrix-vec op, if of both inputs are single column group compressed and and row encodings are the same

updated cost estimation for compression on already compressed inputs with single column group

removed scalars from possible compressible intermediate candidates

added support for double compressed binary matrix-matrix op

single-threaded compressed binary matrix-vector operation with single column group encoding

multithreaded compressed binary matrix-vector operation with single column group encoding

Author: e-strauss

src/main/java/org/apache/sysds/conf/DMLConfig.java

@@ -79,6 +79,7 @@ public class DMLConfig
 	public static final String PARALLEL_TOKENIZE = "sysds.parallel.tokenize";
 	public static final String PARALLEL_TOKENIZE_NUM_BLOCKS = "sysds.parallel.tokenize.numBlocks";
 	public static final String COMPRESSED_LINALG    = "sysds.compressed.linalg";
+	public static final String COMPRESSED_LINALG_INTERMEDIATE    = "sysds.compressed.linalg.intermediate";
 	public static final String COMPRESSED_LOSSY     = "sysds.compressed.lossy";
 	public static final String COMPRESSED_VALID_COMPRESSIONS = "sysds.compressed.valid.compressions";
 	public static final String COMPRESSED_OVERLAPPING = "sysds.compressed.overlapping"; 

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

@@ -171,11 +171,12 @@ public static boolean satisfiesAggressiveCompressionCondition(Hop hop) {
 			satisfies |= HopRewriteUtils.isTernary(hop, OpOp3.CTABLE) 
 				&& hop.getInput(0).getDataType().isMatrix() 
 				&& hop.getInput(1).getDataType().isMatrix();
-			satisfies |= HopRewriteUtils.isData(hop, OpOpData.PERSISTENTREAD) && !hop.isScalar();
+			satisfies |= HopRewriteUtils.isData(hop, OpOpData.PERSISTENTREAD);
 			satisfies |= HopRewriteUtils.isUnary(hop, OpOp1.ROUND, OpOp1.FLOOR, OpOp1.NOT, OpOp1.CEIL);
 			satisfies |= HopRewriteUtils.isBinary(hop, OpOp2.EQUAL, OpOp2.NOTEQUAL, OpOp2.LESS,
 				OpOp2.LESSEQUAL, OpOp2.GREATER, OpOp2.GREATEREQUAL, OpOp2.AND, OpOp2.OR, OpOp2.MODULUS);
 			satisfies |= HopRewriteUtils.isTernary(hop, OpOp3.CTABLE);
+			satisfies &= !hop.isScalar();
 		}
 		if(LOG.isDebugEnabled() && satisfies)
 			LOG.debug("Operation Satisfies: " + hop);

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

@@ -344,7 +344,8 @@ private void classifyPhase() {
 		// final int nRows = mb.getNumRows();
 		final int nCols = mb.getNumColumns();
 		// Assume the scaling of cocoding is at maximum square root good relative to number of columns.
-		final double scale = Math.sqrt(nCols);
+		final double scale = mb instanceof CompressedMatrixBlock &&
+				((CompressedMatrixBlock) mb).getColGroups().size() == 1 ? 1 : Math.sqrt(nCols);
 		final double threshold = _stats.estimatedCostCols / scale;
 
 		if(threshold < _stats.originalCost *
@@ -381,7 +382,7 @@ private void coCodePhase() {
 
 		logPhase();
 		// if cocode is estimated larger than uncompressed abort compression.
-		if(_stats.estimatedCostCoCoded > _stats.originalCost) {
+		if(_stats.estimatedCostCoCoded > _stats.originalCost*2) {
 			// abort compression
 			compressionGroups = null;
 			if(LOG.isInfoEnabled()) {
@@ -458,7 +459,7 @@ private void finalizePhase() {
 		_stats.compressedCost = costEstimator.getCost(res.getColGroups(), res.getNumRows());
 		_stats.setColGroupsCounts(res.getColGroups());
 
-		if(_stats.compressedCost > _stats.originalCost) {
+		if(_stats.compressedCost > _stats.originalCost*2) {
 			LOG.info("--dense size:        " + _stats.denseSize);
 			LOG.info("--original size:     " + _stats.originalSize);
 			LOG.info("--compressed size:   " + _stats.compressedSize);