diff --git a/src/main/java/org/apache/sysds/hops/AggBinaryOp.java b/src/main/java/org/apache/sysds/hops/AggBinaryOp.java
index 2cf651f1894..85ce9882ecc 100644
--- a/src/main/java/org/apache/sysds/hops/AggBinaryOp.java
+++ b/src/main/java/org/apache/sysds/hops/AggBinaryOp.java
@@ -439,8 +439,7 @@ private boolean isApplicableForTransitiveSparkExecType(boolean left)
 				|| (left && !isLeftTransposeRewriteApplicable(true)))
 			&& getInput(index).getParent().size()==1 //bagg is only parent
 			&& !getInput(index).areDimsBelowThreshold() 
-			&& (getInput(index).optFindExecType() == ExecType.SPARK
-				|| (getInput(index) instanceof DataOp && ((DataOp)getInput(index)).hasOnlyRDD()))
+			&& getInput(index).hasSparkOutput()
 			&& getInput(index).getOutputMemEstimate()>getOutputMemEstimate();
 	}
 	
diff --git a/src/main/java/org/apache/sysds/hops/BinaryOp.java b/src/main/java/org/apache/sysds/hops/BinaryOp.java
index 8d2b00c1aa8..cbb154df74f 100644
--- a/src/main/java/org/apache/sysds/hops/BinaryOp.java
+++ b/src/main/java/org/apache/sysds/hops/BinaryOp.java
@@ -747,8 +747,8 @@ protected ExecType optFindExecType(boolean transitive) {
 		
 		checkAndSetForcedPlatform();
 		
-		DataType dt1 = getInput().get(0).getDataType();
-		DataType dt2 = getInput().get(1).getDataType();
+		final DataType dt1 = getInput(0).getDataType();
+		final DataType dt2 = getInput(1).getDataType();
 		
 		if( _etypeForced != null ) {
 			setExecType(_etypeForced);
@@ -796,18 +796,28 @@ else if ( dt1 == DataType.SCALAR && dt2 == DataType.MATRIX ) {
 			checkAndSetInvalidCPDimsAndSize();
 		}
 
-		//spark-specific decision refinement (execute unary scalar w/ spark input and
+		// spark-specific decision refinement (execute unary scalar w/ spark input and
 		// single parent also in spark because it's likely cheap and reduces intermediates)
-		if(transitive && _etype == ExecType.CP && _etypeForced != ExecType.CP && _etypeForced != ExecType.FED &&
-			getDataType().isMatrix() // output should be a matrix
-			&& (dt1.isScalar() || dt2.isScalar()) // one side should be scalar
-			&& supportsMatrixScalarOperations() // scalar operations
-			&& !(getInput().get(dt1.isScalar() ? 1 : 0) instanceof DataOp) // input is not checkpoint
-			&& getInput().get(dt1.isScalar() ? 1 : 0).getParent().size() == 1 // unary scalar is only parent
-			&& !HopRewriteUtils.isSingleBlock(getInput().get(dt1.isScalar() ? 1 : 0)) // single block triggered exec
-			&& getInput().get(dt1.isScalar() ? 1 : 0).optFindExecType() == ExecType.SPARK) {
-			// pull unary scalar operation into spark
-			_etype = ExecType.SPARK;
+		if(transitive // we allow transitive Spark operations. continue sequences of spark operations
+			&& _etype == ExecType.CP // The instruction is currently in CP
+			&& _etypeForced != ExecType.CP // not forced CP
+			&& _etypeForced != ExecType.FED // not federated
+			&& (getDataType().isMatrix() || getDataType().isFrame()) // output should be a matrix or frame
+		) {
+			final boolean v1 = getInput(0).isScalarOrVectorBellowBlockSize();
+			final boolean v2 = getInput(1).isScalarOrVectorBellowBlockSize();
+			final boolean left = v1 == true; // left side is the vector or scalar
+			final Hop sparkIn = getInput(left ? 1 : 0);
+			if((v1 ^ v2) // XOR only one side is allowed to be a vector or a scalar.
+				&& (supportsMatrixScalarOperations() || op == OpOp2.APPLY_SCHEMA)  // supported operation
+				&& sparkIn.getParent().size() == 1 // only one parent
+				&& !HopRewriteUtils.isSingleBlock(sparkIn) // single block triggered exec
+				&& sparkIn.optFindExecType() == ExecType.SPARK // input was spark op.
+				&& !(sparkIn instanceof DataOp) // input is not checkpoint
+			) {
+				// pull operation into spark
+				_etype = ExecType.SPARK;
+			}
 		}
 
 		if( OptimizerUtils.ALLOW_BINARY_UPDATE_IN_PLACE &&
@@ -837,7 +847,7 @@ else if( (op == OpOp2.CBIND && getDataType().isList())
 				|| (op == OpOp2.RBIND && getDataType().isList())) {
 			_etype = ExecType.CP;
 		}
-		
+
 		//mark for recompile (forever)
 		setRequiresRecompileIfNecessary();
 		
diff --git a/src/main/java/org/apache/sysds/hops/Hop.java b/src/main/java/org/apache/sysds/hops/Hop.java
index b32a1a74aab..4a842c69b0f 100644
--- a/src/main/java/org/apache/sysds/hops/Hop.java
+++ b/src/main/java/org/apache/sysds/hops/Hop.java
@@ -1040,6 +1040,12 @@ public final String toString() {
 	// ========================================================================================
 
 	
+	protected boolean isScalarOrVectorBellowBlockSize(){
+		return getDataType().isScalar() || (dimsKnown() && 
+			(( _dc.getRows() == 1 && _dc.getCols() < ConfigurationManager.getBlocksize())
+			|| _dc.getCols() == 1 && _dc.getRows() < ConfigurationManager.getBlocksize()));
+	}
+
 	protected boolean isVector() {
 		return (dimsKnown() && (_dc.getRows() == 1 || _dc.getCols() == 1) );
 	}
@@ -1624,6 +1630,11 @@ protected void setMemoryAndComputeEstimates(Lop lop) {
 		lop.setComputeEstimate(ComputeCost.getHOPComputeCost(this));
 	}
 
+	protected boolean hasSparkOutput(){
+		return (this.optFindExecType() == ExecType.SPARK
+			|| (this instanceof DataOp && ((DataOp)this).hasOnlyRDD()));
+	}
+
 	/**
 	 * Set parse information.
 	 *
diff --git a/src/main/java/org/apache/sysds/hops/UnaryOp.java b/src/main/java/org/apache/sysds/hops/UnaryOp.java
index 91f3a5ec584..e16896b869b 100644
--- a/src/main/java/org/apache/sysds/hops/UnaryOp.java
+++ b/src/main/java/org/apache/sysds/hops/UnaryOp.java
@@ -366,7 +366,11 @@ protected double computeOutputMemEstimate( long dim1, long dim2, long nnz )
 		} else {
 			sparsity = OptimizerUtils.getSparsity(dim1, dim2, nnz);
 		}
-		return OptimizerUtils.estimateSizeExactSparsity(dim1, dim2, sparsity);
+
+		if(getDataType() == DataType.FRAME)
+			return OptimizerUtils.estimateSizeExactFrame(dim1, dim2);
+		else
+			return OptimizerUtils.estimateSizeExactSparsity(dim1, dim2, sparsity);
 	}
 	
 	@Override
@@ -463,6 +467,13 @@ public boolean isMetadataOperation() {
 			|| _op == OpOp1.CAST_AS_LIST;
 	}
 	
+	private boolean isDisallowedSparkOps(){
+		return isCumulativeUnaryOperation() 
+			|| isCastUnaryOperation()
+			|| _op==OpOp1.MEDIAN
+			|| _op==OpOp1.IQM;
+	}
+
 	@Override
 	protected ExecType optFindExecType(boolean transitive) 
 	{
@@ -493,19 +504,22 @@ else if ( getInput().get(0).areDimsBelowThreshold() || getInput().get(0).isVecto
 			checkAndSetInvalidCPDimsAndSize();
 		}
 	
+
 		//spark-specific decision refinement (execute unary w/ spark input and 
 		//single parent also in spark because it's likely cheap and reduces intermediates)
-		if( _etype == ExecType.CP && _etypeForced != ExecType.CP
-			&& getInput().get(0).optFindExecType() == ExecType.SPARK 
-			&& getDataType().isMatrix() 
-			&& !isCumulativeUnaryOperation() && !isCastUnaryOperation()
-			&& _op!=OpOp1.MEDIAN && _op!=OpOp1.IQM
-			&& !(getInput().get(0) instanceof DataOp)    //input is not checkpoint
-			&& getInput().get(0).getParent().size()==1 ) //unary is only parent
-		{
+		if(_etype == ExecType.CP // currently CP instruction
+			&& _etype != ExecType.SPARK /// currently not SP.
+			&& _etypeForced != ExecType.CP // not forced as CP instruction
+			&& getInput(0).hasSparkOutput() // input is a spark instruction
+			&& (getDataType().isMatrix() || getDataType().isFrame()) // output is a matrix or frame
+			&& !isDisallowedSparkOps() // is invalid spark instruction
+			// && !(getInput().get(0) instanceof DataOp) // input is not checkpoint
+			// && getInput(0).getParent().size() <= 1// unary is only parent
+		) {
 			//pull unary operation into spark 
 			_etype = ExecType.SPARK;
 		}
+
 		
 		//mark for recompile (forever)
 		setRequiresRecompileIfNecessary();
@@ -520,7 +534,7 @@ && getInput().get(0).getParent().size()==1 ) //unary is only parent
 		} else {
 			setRequiresRecompileIfNecessary();
 		}
-		
+
 		return _etype;
 	}
 	
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 48637595741..fd451805a28 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/CompressedMatrixBlock.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/CompressedMatrixBlock.java
@@ -58,8 +58,8 @@
 import org.apache.sysds.runtime.compress.lib.CLALibMMChain;
 import org.apache.sysds.runtime.compress.lib.CLALibMatrixMult;
 import org.apache.sysds.runtime.compress.lib.CLALibMerge;
-import org.apache.sysds.runtime.compress.lib.CLALibReplace;
 import org.apache.sysds.runtime.compress.lib.CLALibReorg;
+import org.apache.sysds.runtime.compress.lib.CLALibReplace;
 import org.apache.sysds.runtime.compress.lib.CLALibReshape;
 import org.apache.sysds.runtime.compress.lib.CLALibRexpand;
 import org.apache.sysds.runtime.compress.lib.CLALibScalar;
@@ -1202,8 +1202,8 @@ public void examSparsity(boolean allowCSR, int k) {
 	}
 
 	@Override
-	public void sparseToDense(int k) {
-		// do nothing
+	public MatrixBlock sparseToDense(int k) {
+		return this; // do nothing
 	}
 
 	@Override
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/spark/utils/FrameRDDConverterUtils.java b/src/main/java/org/apache/sysds/runtime/instructions/spark/utils/FrameRDDConverterUtils.java
index 9371d43094c..a5974640cc5 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/spark/utils/FrameRDDConverterUtils.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/spark/utils/FrameRDDConverterUtils.java
@@ -90,10 +90,7 @@ public static JavaPairRDD<Long, FrameBlock> csvToBinaryBlock(JavaSparkContext sc
  			JavaRDD<String> tmp = input.values()
 					.map(new TextToStringFunction());
 			String tmpStr = tmp.first();
-			boolean metaHeader = tmpStr.startsWith(TfUtils.TXMTD_MVPREFIX) 
-					|| tmpStr.startsWith(TfUtils.TXMTD_NDPREFIX);
-			tmpStr = (metaHeader) ? tmpStr.substring(tmpStr.indexOf(delim)+1) : tmpStr;
-			long rlen = tmp.count() - (hasHeader ? 1 : 0) - (metaHeader ? 2 : 0);
+			long rlen = tmp.count() ;
 			long clen = IOUtilFunctions.splitCSV(tmpStr, delim).length;
 			mc.set(rlen, clen, mc.getBlocksize(), -1);
 		}
@@ -582,14 +579,14 @@ public Iterator<Tuple2<Long, FrameBlock>> call(Iterator<Tuple2<Text,Long>> arg0)
 					_colnames = row.split(_delim);
 					continue;
 				}
-				if( row.startsWith(TfUtils.TXMTD_MVPREFIX) ) {
-					_mvMeta = Arrays.asList(Arrays.copyOfRange(IOUtilFunctions.splitCSV(row, _delim), 1, (int)_clen+1));
-					continue;
-				}
-				else if( row.startsWith(TfUtils.TXMTD_NDPREFIX) ) {
-					_ndMeta = Arrays.asList(Arrays.copyOfRange(IOUtilFunctions.splitCSV(row, _delim), 1, (int)_clen+1));
-					continue;
-				}
+				// if( row.startsWith(TfUtils.TXMTD_MVPREFIX) ) {
+				// 	_mvMeta = Arrays.asList(Arrays.copyOfRange(IOUtilFunctions.splitCSV(row, _delim), 1, (int)_clen+1));
+				// 	continue;
+				// }
+				// else if( row.startsWith(TfUtils.TXMTD_NDPREFIX) ) {
+				// 	_ndMeta = Arrays.asList(Arrays.copyOfRange(IOUtilFunctions.splitCSV(row, _delim), 1, (int)_clen+1));
+				// 	continue;
+				// }
 				
 				//adjust row index for header and meta data
 				rowix += (_hasHeader ? 0 : 1) - ((_mvMeta == null) ? 0 : 2);
@@ -670,18 +667,18 @@ public Iterator<String> call(Tuple2<Long, FrameBlock> arg0)
 					ret.add(sb.toString());
 					sb.setLength(0); //reset
 				}
-				if( !blk.isColumnMetadataDefault() ) {
-					sb.append(TfUtils.TXMTD_MVPREFIX + _props.getDelim());
-					for( int j=0; j<blk.getNumColumns(); j++ )
-						sb.append(blk.getColumnMetadata(j).getMvValue() + ((j<blk.getNumColumns()-1)?_props.getDelim():""));
-					ret.add(sb.toString());
-					sb.setLength(0); //reset
-					sb.append(TfUtils.TXMTD_NDPREFIX + _props.getDelim());
-					for( int j=0; j<blk.getNumColumns(); j++ )
-						sb.append(blk.getColumnMetadata(j).getNumDistinct() + ((j<blk.getNumColumns()-1)?_props.getDelim():""));
-					ret.add(sb.toString());
-					sb.setLength(0); //reset		
-				}
+				// if( !blk.isColumnMetadataDefault() ) {
+				// 	sb.append(TfUtils.TXMTD_MVPREFIX + _props.getDelim());
+				// 	for( int j=0; j<blk.getNumColumns(); j++ )
+				// 		sb.append(blk.getColumnMetadata(j).getMvValue() + ((j<blk.getNumColumns()-1)?_props.getDelim():""));
+				// 	ret.add(sb.toString());
+				// 	sb.setLength(0); //reset
+				// 	sb.append(TfUtils.TXMTD_NDPREFIX + _props.getDelim());
+				// 	for( int j=0; j<blk.getNumColumns(); j++ )
+				// 		sb.append(blk.getColumnMetadata(j).getNumDistinct() + ((j<blk.getNumColumns()-1)?_props.getDelim():""));
+				// 	ret.add(sb.toString());
+				// 	sb.setLength(0); //reset		
+				// }
 			}
 		
 			//handle Frame block data
diff --git a/src/main/java/org/apache/sysds/runtime/io/FrameWriterTextCSV.java b/src/main/java/org/apache/sysds/runtime/io/FrameWriterTextCSV.java
index 5815ff231ea..f14cdf7ae28 100644
--- a/src/main/java/org/apache/sysds/runtime/io/FrameWriterTextCSV.java
+++ b/src/main/java/org/apache/sysds/runtime/io/FrameWriterTextCSV.java
@@ -31,7 +31,6 @@
 import org.apache.sysds.runtime.DMLRuntimeException;
 import org.apache.sysds.runtime.frame.data.FrameBlock;
 import org.apache.sysds.runtime.frame.data.iterators.IteratorFactory;
-import org.apache.sysds.runtime.transform.TfUtils;
 import org.apache.sysds.runtime.util.HDFSTool;
 
 /**
@@ -107,17 +106,7 @@ protected static void writeCSVFrameToFile( Path path, JobConf job, FileSystem fs
 					}
 					sb.append('\n');
 				}
-				//append meta data
-				if( !src.isColumnMetadataDefault() ) {
-					sb.append(TfUtils.TXMTD_MVPREFIX + delim);
-					for( int j=0; j<cols; j++ )
-						sb.append(src.getColumnMetadata(j).getMvValue() + ((j<cols-1)?delim:""));
-					sb.append("\n");
-					sb.append(TfUtils.TXMTD_NDPREFIX + delim);
-					for( int j=0; j<cols; j++ )
-						sb.append(src.getColumnMetadata(j).getNumDistinct() + ((j<cols-1)?delim:""));
-					sb.append("\n");
-				}
+
 				br.write( sb.toString() );
 	            sb.setLength(0);
 			}
diff --git a/src/main/java/org/apache/sysds/runtime/io/IOUtilFunctions.java b/src/main/java/org/apache/sysds/runtime/io/IOUtilFunctions.java
index c6085cb8960..e4feb3ed756 100644
--- a/src/main/java/org/apache/sysds/runtime/io/IOUtilFunctions.java
+++ b/src/main/java/org/apache/sysds/runtime/io/IOUtilFunctions.java
@@ -76,7 +76,6 @@
 import org.apache.sysds.runtime.matrix.data.MatrixBlock;
 import org.apache.sysds.runtime.matrix.data.MatrixCell;
 import org.apache.sysds.runtime.matrix.data.MatrixIndexes;
-import org.apache.sysds.runtime.transform.TfUtils;
 import org.apache.sysds.runtime.util.LocalFileUtils;
 
 import io.airlift.compress.lzo.LzoCodec;
@@ -696,10 +695,10 @@ public static int countNumColumnsCSV(InputSplit[] splits, InputFormat informat,
 			try {
 				if( reader.next(key, value) ) {
 					boolean hasValue = true;
-					if( value.toString().startsWith(TfUtils.TXMTD_MVPREFIX) )
-						hasValue = reader.next(key, value);
-					if( value.toString().startsWith(TfUtils.TXMTD_NDPREFIX) )
-						hasValue = reader.next(key, value);
+					// if( value.toString().startsWith(TfUtils.TXMTD_MVPREFIX) )
+					// 	hasValue = reader.next(key, value);
+					// if( value.toString().startsWith(TfUtils.TXMTD_NDPREFIX) )
+					// 	hasValue = reader.next(key, value);
 					String row = value.toString().trim();
 					if( hasValue && !row.isEmpty() ) {
 						ncol = IOUtilFunctions.countTokensCSV(row, delim);
diff --git a/src/main/java/org/apache/sysds/runtime/matrix/data/LibAggregateUnarySpecialization.java b/src/main/java/org/apache/sysds/runtime/matrix/data/LibAggregateUnarySpecialization.java
new file mode 100644
index 00000000000..79f08cb353a
--- /dev/null
+++ b/src/main/java/org/apache/sysds/runtime/matrix/data/LibAggregateUnarySpecialization.java
@@ -0,0 +1,148 @@
+/*
+ * 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.matrix.data;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.sysds.common.Types.CorrectionLocationType;
+import org.apache.sysds.runtime.DMLRuntimeException;
+import org.apache.sysds.runtime.data.DenseBlock;
+import org.apache.sysds.runtime.data.SparseBlock;
+import org.apache.sysds.runtime.instructions.cp.KahanObject;
+import org.apache.sysds.runtime.matrix.data.MatrixValue.CellIndex;
+import org.apache.sysds.runtime.matrix.operators.AggregateOperator;
+import org.apache.sysds.runtime.matrix.operators.AggregateUnaryOperator;
+
+public class LibAggregateUnarySpecialization {
+	protected static final Log LOG = LogFactory.getLog(LibAggregateUnarySpecialization.class.getName());
+
+	public static void aggregateUnary(final MatrixBlock mb, AggregateUnaryOperator op, MatrixBlock result, int blen,
+		MatrixIndexes indexesIn) {
+		if(op.sparseSafe)
+			sparseAggregateUnaryHelp(mb, op, result, blen, indexesIn);
+		else
+			denseAggregateUnaryHelp(mb, op, result, blen, indexesIn);
+	}
+
+	private static void sparseAggregateUnaryHelp(final MatrixBlock mb, AggregateUnaryOperator op, MatrixBlock result,
+		int blen, MatrixIndexes indexesIn) {
+		// initialize result
+		if(op.aggOp.initialValue != 0)
+			result.reset(result.rlen, result.clen, op.aggOp.initialValue);
+		CellIndex tempCellIndex = new CellIndex(-1, -1);
+		KahanObject buffer = new KahanObject(0, 0);
+
+		if(mb.sparse && mb.sparseBlock != null) {
+			SparseBlock a = mb.sparseBlock;
+			for(int r = 0; r < Math.min(mb.rlen, a.numRows()); r++) {
+				if(a.isEmpty(r))
+					continue;
+				int apos = a.pos(r);
+				int alen = a.size(r);
+				int[] aix = a.indexes(r);
+				double[] aval = a.values(r);
+				for(int i = apos; i < apos + alen; i++) {
+					tempCellIndex.set(r, aix[i]);
+					op.indexFn.execute(tempCellIndex, tempCellIndex);
+					incrementalAggregateUnaryHelp(op.aggOp, result, tempCellIndex.row, tempCellIndex.column, aval[i],
+						buffer);
+				}
+			}
+		}
+		else if(!mb.sparse && mb.denseBlock != null) {
+			DenseBlock a = mb.getDenseBlock();
+			for(int i = 0; i < mb.rlen; i++)
+				for(int j = 0; j < mb.clen; j++) {
+					tempCellIndex.set(i, j);
+					op.indexFn.execute(tempCellIndex, tempCellIndex);
+					incrementalAggregateUnaryHelp(op.aggOp, result, tempCellIndex.row, tempCellIndex.column, a.get(i, j),
+						buffer);
+				}
+		}
+	}
+
+	private static void denseAggregateUnaryHelp(MatrixBlock mb, AggregateUnaryOperator op, MatrixBlock result, int blen,
+		MatrixIndexes indexesIn) {
+		if(op.aggOp.initialValue != 0)
+			result.reset(result.rlen, result.clen, op.aggOp.initialValue);
+		CellIndex tempCellIndex = new CellIndex(-1, -1);
+		KahanObject buffer = new KahanObject(0, 0);
+		for(int i = 0; i < mb.rlen; i++)
+			for(int j = 0; j < mb.clen; j++) {
+				tempCellIndex.set(i, j);
+				op.indexFn.execute(tempCellIndex, tempCellIndex);
+				incrementalAggregateUnaryHelp(op.aggOp, result, tempCellIndex.row, tempCellIndex.column,
+					mb.get(i, j), buffer);
+			}
+	}
+
+	private static void incrementalAggregateUnaryHelp(AggregateOperator aggOp, MatrixBlock result, int row, int column,
+		double newvalue, KahanObject buffer) {
+		if(aggOp.existsCorrection()) {
+			if(aggOp.correction == CorrectionLocationType.LASTROW ||
+				aggOp.correction == CorrectionLocationType.LASTCOLUMN) {
+				int corRow = row, corCol = column;
+				if(aggOp.correction == CorrectionLocationType.LASTROW)// extra row
+					corRow++;
+				else if(aggOp.correction == CorrectionLocationType.LASTCOLUMN)
+					corCol++;
+				else
+					throw new DMLRuntimeException("unrecognized correctionLocation: " + aggOp.correction);
+
+				buffer._sum = result.get(row, column);
+				buffer._correction = result.get(corRow, corCol);
+				buffer = (KahanObject) aggOp.increOp.fn.execute(buffer, newvalue);
+				result.set(row, column, buffer._sum);
+				result.set(corRow, corCol, buffer._correction);
+			}
+			else if(aggOp.correction == CorrectionLocationType.NONE) {
+				throw new DMLRuntimeException("unrecognized correctionLocation: " + aggOp.correction);
+			}
+			else// for mean
+			{
+				int corRow = row, corCol = column;
+				int countRow = row, countCol = column;
+				if(aggOp.correction == CorrectionLocationType.LASTTWOROWS) {
+					countRow++;
+					corRow += 2;
+				}
+				else if(aggOp.correction == CorrectionLocationType.LASTTWOCOLUMNS) {
+					countCol++;
+					corCol += 2;
+				}
+				else
+					throw new DMLRuntimeException("unrecognized correctionLocation: " + aggOp.correction);
+				buffer._sum = result.get(row, column);
+				buffer._correction = result.get(corRow, corCol);
+				double count = result.get(countRow, countCol) + 1.0;
+				buffer = (KahanObject) aggOp.increOp.fn.execute(buffer, newvalue, count);
+				result.set(row, column, buffer._sum);
+				result.set(corRow, corCol, buffer._correction);
+				result.set(countRow, countCol, count);
+			}
+
+		}
+		else {
+			newvalue = aggOp.increOp.fn.execute(result.get(row, column), newvalue);
+			result.set(row, column, newvalue);
+		}
+	}
+
+}
diff --git a/src/main/java/org/apache/sysds/runtime/matrix/data/LibMatrixMult.java b/src/main/java/org/apache/sysds/runtime/matrix/data/LibMatrixMult.java
index 66f7c3c9445..ba7553cc501 100644
--- a/src/main/java/org/apache/sysds/runtime/matrix/data/LibMatrixMult.java
+++ b/src/main/java/org/apache/sysds/runtime/matrix/data/LibMatrixMult.java
@@ -60,6 +60,7 @@
 import org.apache.sysds.utils.NativeHelper;
 import org.apache.sysds.utils.stats.InfrastructureAnalyzer;
 
+
 /**
  * MB: Library for matrix multiplications including MM, MV, VV for all
  * combinations of dense, sparse, ultrasparse representations and special
@@ -78,6 +79,7 @@ public class LibMatrixMult
 	public static final int L2_CACHESIZE = 256 * 1024; //256KB (common size)
 	public static final int L3_CACHESIZE = 16 * 1024 * 1024; //16MB (common size)
 	private static final Log LOG = LogFactory.getLog(LibMatrixMult.class.getName());
+	
 
 	private LibMatrixMult() {
 		//prevent instantiation via private constructor
@@ -266,7 +268,7 @@ private static void singleThreadedMatrixMult(MatrixBlock m1, MatrixBlock m2, Mat
 		else if(!m1.sparse && !m2.sparse && !ret.sparse)
 			matrixMultDenseDense(m1, m2, ret, tm2, pm2, 0, ru2, 0, m2.clen);
 		else if(m1.sparse && m2.sparse)
-			matrixMultSparseSparse(m1, m2, ret, pm2, sparse, 0, ru2);
+			matrixMultSparseSparse(m1, m2, ret, pm2, ret.sparse, 0, ru2);
 		else if(m1.sparse)
 			matrixMultSparseDense(m1, m2, ret, pm2, 0, ru2);
 		else
@@ -831,7 +833,7 @@ public static void matrixMultWDivMM(MatrixBlock mW, MatrixBlock mU, MatrixBlock
 		//check for empty result 
 		if(   mW.isEmptyBlock(false) 
 		   || (wt.isLeft() && mU.isEmptyBlock(false))
-		   || (wt.isRight() && mV.isEmptyBlock(false)) 
+		   || (wt.isRight() && mV.isEmptyBlock(false))
 		   || (wt.isBasic() && mW.isEmptyBlock(false)))  {
 			ret.examSparsity(); //turn empty dense into sparse
 			return; 
@@ -2003,7 +2005,6 @@ private static void matrixMultUltraSparseSparseSparseLeftRowGeneric(int i, int a
 		}
 	}
 
-	
 	private static void matrixMultUltraSparseRight(MatrixBlock m1, MatrixBlock m2, MatrixBlock ret, int rl, int ru) {
 		if(ret.isInSparseFormat()){
 			if(m1.isInSparseFormat())
@@ -3227,8 +3228,13 @@ private static void matrixMultWDivMMDense(MatrixBlock mW, MatrixBlock mU, Matrix
 		DenseBlock u = mU.getDenseBlock();
 		DenseBlock v = mV.getDenseBlock();
 		DenseBlock x = (mX==null) ? null : mX.getDenseBlock();
-		DenseBlock c = ret.getDenseBlock();
-		
+		DenseBlock c = ret.getDenseBlock();		
+
+		if(c == null){
+			ret.allocateDenseBlock();
+			c = ret.getDenseBlock();
+		}
+
 		//approach: iterate over non-zeros of w, selective mm computation
 		//cache-conscious blocking: due to blocksize constraint (default 1000),
 		//a blocksize of 16 allows to fit blocks of UV into L2 cache (256KB) 
@@ -3375,7 +3381,11 @@ private static void matrixMultWDivMMGeneric(MatrixBlock mW, MatrixBlock mU, Matr
 
 		//output always in dense representation
 		DenseBlock c = ret.getDenseBlock();
-		
+		if(c == null){
+			ret.allocateDenseBlock();
+			c = ret.getDenseBlock();
+		}
+
 		//approach: iterate over non-zeros of w, selective mm computation
 		if( mW.sparse ) //SPARSE
 		{
@@ -4082,7 +4092,8 @@ public static void vectAdd( double[] a, double bval, double[] c, int ai, int ci,
 			c[ ci+7 ] += a[ ai+7 ] + bval;
 		}
 	}
-	
+
+
 	//note: public for use by codegen for consistency
 	public static void vectAdd( double[] a, double[] c, int ai, int ci, final int len )
 	{
diff --git a/src/main/java/org/apache/sysds/runtime/matrix/data/MatrixBlock.java b/src/main/java/org/apache/sysds/runtime/matrix/data/MatrixBlock.java
index 7bc516588a5..7264100cd7d 100644
--- a/src/main/java/org/apache/sysds/runtime/matrix/data/MatrixBlock.java
+++ b/src/main/java/org/apache/sysds/runtime/matrix/data/MatrixBlock.java
@@ -1312,7 +1312,7 @@ public void examSparsity(boolean allowCSR, int k) {
 		else if( !sparse && sparseDst )
 			denseToSparse(allowCSR, k);
 	}
-	
+
 	public static boolean evalSparseFormatInMemory(DataCharacteristics dc) {
 		return evalSparseFormatInMemory(dc.getRows(), dc.getCols(), dc.getNonZeros());
 	}
@@ -1384,12 +1384,13 @@ public void denseToSparse(boolean allowCSR, int k){
 		LibMatrixDenseToSparse.denseToSparse(this, allowCSR, k);
 	}
 
-	public final void sparseToDense() {
-		sparseToDense(1);
+	public final MatrixBlock sparseToDense() {
+		return sparseToDense(1);
 	}
 
-	public void sparseToDense(int k) {
+	public MatrixBlock sparseToDense(int k) {
 		LibMatrixSparseToDense.sparseToDense(this, k);
+		return this;
 	}
 
 	/**
@@ -2951,13 +2952,14 @@ public boolean isShallowSerialize(boolean inclConvert) {
 		boolean sparseDst = evalSparseFormatOnDisk();
 		return !sparse || !sparseDst
 			|| (sparse && sparseBlock instanceof SparseBlockCSR)
-			|| (sparse && sparseBlock instanceof SparseBlockMCSR
-				&& getInMemorySize() / MAX_SHALLOW_SERIALIZE_OVERHEAD 
-				<= getExactSerializedSize())
-			|| (sparse && sparseBlock instanceof SparseBlockMCSR
-				&& nonZeros < Integer.MAX_VALUE //CSR constraint
-				&& inclConvert && CONVERT_MCSR_TO_CSR_ON_DEEP_SERIALIZE
-				&& !isUltraSparseSerialize(sparseDst));
+			|| (sparse && sparseBlock instanceof SparseBlockMCSR);
+			// || (sparse && sparseBlock instanceof SparseBlockMCSR
+			// 	&& getInMemorySize() / MAX_SHALLOW_SERIALIZE_OVERHEAD 
+			// 	<= getExactSerializedSize())
+			// || (sparse && sparseBlock instanceof SparseBlockMCSR
+			// 	&& nonZeros < Integer.MAX_VALUE //CSR constraint
+			// 	&& inclConvert && CONVERT_MCSR_TO_CSR_ON_DEEP_SERIALIZE
+			// 	&& !isUltraSparseSerialize(sparseDst));
 	}
 	
 	@Override 
@@ -5254,8 +5256,8 @@ public MatrixBlock ctableSeqOperations(MatrixValue thatMatrix, double thatScalar
 	 * (i1,j1,v2) from input2 (that)
 	 * (w)  from scalar_input3 (scalarThat2)
 	 *
-	 * @param thatMatrix matrix value
-	 * @param thatScalar scalar double
+	 * @param thatMatrix matrix value, the vector to encode via table
+	 * @param thatScalar scalar double, w, that is the weight to multiply on the encoded values
 	 * @param resultBlock result matrix block
 	 * @return resultBlock
 	 */
diff --git a/src/test/java/org/apache/sysds/test/component/compress/estim/encoding/EncodeSampleMultiColTest.java b/src/test/java/org/apache/sysds/test/component/compress/estim/encoding/EncodeSampleMultiColTest.java
index 194f581121a..a5bd3cebfb0 100644
--- a/src/test/java/org/apache/sysds/test/component/compress/estim/encoding/EncodeSampleMultiColTest.java
+++ b/src/test/java/org/apache/sysds/test/component/compress/estim/encoding/EncodeSampleMultiColTest.java
@@ -21,6 +21,7 @@
 
 import static org.junit.Assert.fail;
 
+import org.apache.sysds.runtime.compress.CompressionSettingsBuilder;
 import org.apache.sysds.runtime.compress.estim.encoding.IEncode;
 import org.apache.sysds.runtime.matrix.data.MatrixBlock;
 import org.junit.Test;
@@ -115,6 +116,8 @@ public void testJoinWithSecondSubpartLeft() {
 	private void partJoinVerification(IEncode er) {
 		boolean incorrectUnique = e.getUnique() != er.getUnique();
 
+		er.extractFacts(10000, 1.0, 1.0, new CompressionSettingsBuilder().create());
+
 		if(incorrectUnique) {
 			StringBuilder sb = new StringBuilder();
 			sb.append("\nFailed joining sub parts to recreate whole.");
diff --git a/src/test/java/org/apache/sysds/test/component/compress/estim/encoding/EncodeSampleUnbalancedTest.java b/src/test/java/org/apache/sysds/test/component/compress/estim/encoding/EncodeSampleUnbalancedTest.java
index 182bd7fa37e..5a298f145ec 100644
--- a/src/test/java/org/apache/sysds/test/component/compress/estim/encoding/EncodeSampleUnbalancedTest.java
+++ b/src/test/java/org/apache/sysds/test/component/compress/estim/encoding/EncodeSampleUnbalancedTest.java
@@ -81,6 +81,10 @@ public static Collection<Object[]> data() {
 		// Both Sparse and end dense joined
 		tests.add(createT(1, 0.2, 10, 10, 0.1, 2, 1000, 1231521));
 
+
+		tests.add(createT(1, 1.0, 100, 1, 1.0, 10,  10000, 132));
+		tests.add(createT(1, 1.0, 1000, 1, 1.0, 10,  10000, 132));
+
 		return tests;
 	}
 
diff --git a/src/test/java/org/apache/sysds/test/functions/compress/configuration/CompressForce.java b/src/test/java/org/apache/sysds/test/functions/compress/configuration/CompressForce.java
index c6d52a70a51..872ec79c1f1 100644
--- a/src/test/java/org/apache/sysds/test/functions/compress/configuration/CompressForce.java
+++ b/src/test/java/org/apache/sysds/test/functions/compress/configuration/CompressForce.java
@@ -49,7 +49,7 @@ protected String getTestDir() {
 
 	@Test
 	public void testTranspose_CP() {
-		runTest(1500, 20, 1, 1, ExecType.CP, "transpose");
+		runTest(1500, 20, 2, 1, ExecType.CP, "transpose");
 	}
 
 	@Test