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 8a6478f..5074a69 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/CompressedMatrixBlockFactory.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/CompressedMatrixBlockFactory.java
@@ -267,10 +267,13 @@
 	private Pair<MatrixBlock, CompressionStatistics> compressMatrix() {
 		if(mb.getNonZeros() < 0)
 			throw new DMLCompressionException("Invalid to compress matrices with unknown nonZeros");
-		else if(mb instanceof CompressedMatrixBlock) // Redundant compression
-			return recompress((CompressedMatrixBlock) mb);
+		else if(mb instanceof CompressedMatrixBlock && ((CompressedMatrixBlock) mb).isOverlapping()) {
+			LOG.warn("Unsupported recompression of overlapping compression");
+			return new ImmutablePair<>(mb, null);
+		}
 
 		_stats.denseSize = MatrixBlock.estimateSizeInMemory(mb.getNumRows(), mb.getNumColumns(), 1.0);
+		_stats.sparseSize = MatrixBlock.estimateSizeSparseInMemory(mb.getNumRows(), mb.getNumColumns(), mb.getSparsity());
 		_stats.originalSize = mb.getInMemorySize();
 		_stats.originalCost = costEstimator.getCost(mb);
 
@@ -427,11 +430,15 @@
 		final double ratio = _stats.getRatio();
 		final double denseRatio = _stats.getDenseRatio();
 
+		_stats.setColGroupsCounts(res.getColGroups());
 		if(ratio < 1 && denseRatio < 100.0) {
 			LOG.info("--dense size:        " + _stats.denseSize);
 			LOG.info("--original size:     " + _stats.originalSize);
 			LOG.info("--compressed size:   " + _stats.compressedSize);
 			LOG.info("--compression ratio: " + ratio);
+			LOG.debug("--col groups types   " + _stats.getGroupsTypesString());
+			LOG.debug("--col groups sizes   " + _stats.getGroupsSizesString());
+			logLengths();
 			LOG.info("Abort block compression because compression ratio is less than 1.");
 			res = null;
 			setNextTimePhase(time.stop());
@@ -439,8 +446,6 @@
 			return;
 		}
 
-		_stats.setColGroupsCounts(res.getColGroups());
-
 		if(compSettings.isInSparkInstruction)
 			res.clearSoftReferenceToDecompressed();
 
@@ -454,30 +459,6 @@
 		return new ImmutablePair<>(mb, _stats);
 	}
 
-	private Pair<MatrixBlock, CompressionStatistics> recompress(CompressedMatrixBlock cmb) {
-		LOG.debug("Recompressing an already compressed MatrixBlock");
-		LOG.warn("Not Implemented Recompress yet");
-
-		classifyPhase();
-		// informationExtractor = ComEstFactory.createEstimator(mb, compSettings, k);
-
-		// compressionGroups = informationExtractor.computeCompressedSizeInfos(k);
-
-		// _stats.estimatedSizeCols = compressionGroups.memoryEstimate();
-		// _stats.estimatedCostCols = costEstimator.getCost(compressionGroups);
-
-		// logPhase();
-
-
-
-		return new ImmutablePair<>(cmb, null);
-		// _stats.originalSize = cmb.getInMemorySize();
-		// CompressedMatrixBlock combined = CLALibCombineGroups.combine(cmb, k);
-		// CompressedMatrixBlock squashed = CLALibSquash.squash(combined, k);
-		// _stats.compressedSize = squashed.getInMemorySize();
-		// return new ImmutablePair<>(squashed, _stats);
-	}
-
 	private void logPhase() {
 		setNextTimePhase(time.stop());
 		DMLCompressionStatistics.addCompressionTime(getLastTimePhase(), phase);
@@ -492,6 +473,9 @@
 						LOG.debug("--Seed used for comp : " + compSettings.seed);
 						LOG.debug("--compression phase " + phase + " Classify  : " + getLastTimePhase());
 						LOG.debug("--Individual Columns Estimated Compression: " + _stats.estimatedSizeCols);
+						if(mb instanceof CompressedMatrixBlock) {
+							LOG.debug("--Recompressing already compressed MatrixBlock");
+						}
 						break;
 					case 1:
 						LOG.debug("--compression phase " + phase + " Grouping  : " + getLastTimePhase());
@@ -521,7 +505,9 @@
 						LOG.debug("--compression phase  " + phase + " Cleanup   : " + getLastTimePhase());
 						LOG.debug("--col groups types   " + _stats.getGroupsTypesString());
 						LOG.debug("--col groups sizes   " + _stats.getGroupsSizesString());
+						LOG.debug("--input was compressed " + (mb instanceof CompressedMatrixBlock));
 						LOG.debug(String.format("--dense size:        %16d", _stats.denseSize));
+						LOG.debug(String.format("--sparse size:       %16d", _stats.sparseSize));
 						LOG.debug(String.format("--original size:     %16d", _stats.originalSize));
 						LOG.debug(String.format("--compressed size:   %16d", _stats.compressedSize));
 						LOG.debug(String.format("--compression ratio: %4.3f", _stats.getRatio()));
@@ -534,35 +520,39 @@
 							LOG.debug(
 								String.format("--relative cost:     %1.4f", (_stats.compressedCost / _stats.originalCost)));
 						}
-						if(compressionGroups != null && compressionGroups.getInfo().size() < 1000) {
-							int[] lengths = new int[res.getColGroups().size()];
-							int i = 0;
-							for(AColGroup colGroup : res.getColGroups())
-								lengths[i++] = colGroup.getNumValues();
-
-							LOG.debug("--compressed colGroup dictionary sizes: " + Arrays.toString(lengths));
-							LOG.debug(
-								"--compressed colGroup nr columns      : " + constructNrColumnString(res.getColGroups()));
-						}
-						if(LOG.isTraceEnabled()) {
-							for(AColGroup colGroup : res.getColGroups()) {
-								if(colGroup.estimateInMemorySize() < 1000)
-									LOG.trace(colGroup);
-								else {
-									LOG.trace("--colGroups type       : " + colGroup.getClass().getSimpleName() + " size: "
-										+ colGroup.estimateInMemorySize()
-										+ ((colGroup instanceof AColGroupValue) ? "  numValues :"
-											+ ((AColGroupValue) colGroup).getNumValues() : "")
-										+ "  colIndexes : " + colGroup.getColIndices());
-								}
-							}
-						}
+						logLengths();
 				}
 			}
 		}
 		phase++;
 	}
 
+	private void logLengths() {
+		if(compressionGroups != null && compressionGroups.getInfo().size() < 1000) {
+			int[] lengths = new int[res.getColGroups().size()];
+			int i = 0;
+			for(AColGroup colGroup : res.getColGroups())
+				lengths[i++] = colGroup.getNumValues();
+
+			LOG.debug("--compressed colGroup dictionary sizes: " + Arrays.toString(lengths));
+			LOG.debug("--compressed colGroup nr columns      : " + constructNrColumnString(res.getColGroups()));
+		}
+		if(LOG.isTraceEnabled()) {
+			for(AColGroup colGroup : res.getColGroups()) {
+				if(colGroup.estimateInMemorySize() < 1000)
+					LOG.trace(colGroup);
+				else {
+					LOG.trace(
+						"--colGroups type       : " + colGroup.getClass().getSimpleName() + " size: "
+							+ colGroup.estimateInMemorySize()
+							+ ((colGroup instanceof AColGroupValue) ? "  numValues :"
+								+ ((AColGroupValue) colGroup).getNumValues() : "")
+							+ "  colIndexes : " + colGroup.getColIndices());
+				}
+			}
+		}
+	}
+
 	private void setNextTimePhase(double time) {
 		lastPhase = time;
 	}
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 e9abe83..d54eb2c 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/CompressionStatistics.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/CompressionStatistics.java
@@ -35,6 +35,8 @@
 	public long originalSize;
 	/** Size if the input is dense */
 	public long denseSize;
+	/** Size if the input is sparse */
+	public long sparseSize;
 	/** Estimated size of compressing individual columns */
 	public long estimatedSizeCols;
 	/** Estimated size of compressing after co-coding */
diff --git a/src/main/java/org/apache/sysds/runtime/compress/colgroup/AColGroup.java b/src/main/java/org/apache/sysds/runtime/compress/colgroup/AColGroup.java
index b3356db..e85b220 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/colgroup/AColGroup.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/colgroup/AColGroup.java
@@ -36,6 +36,7 @@
 import org.apache.sysds.runtime.compress.lib.CLALibCombineGroups;
 import org.apache.sysds.runtime.data.DenseBlock;
 import org.apache.sysds.runtime.data.SparseBlock;
+import org.apache.sysds.runtime.functionobjects.Plus;
 import org.apache.sysds.runtime.instructions.cp.CM_COV_Object;
 import org.apache.sysds.runtime.matrix.data.MatrixBlock;
 import org.apache.sysds.runtime.matrix.operators.AggregateUnaryOperator;
@@ -418,17 +419,29 @@
 	 * 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 v         The vector of values to apply the values contained should be at least the length of the highest
+	 *                  value in the column index
 	 * @param isRowSafe True if the binary op is applied to an entire zero row and all results are zero
 	 * @return A updated column group with the new values.
 	 */
 	public abstract AColGroup binaryRowOpLeft(BinaryOperator op, double[] v, boolean isRowSafe);
 
 	/**
+	 * Short hand add operator call on column group to add a row vector to the column group
+	 * 
+	 * @param v The vector to add
+	 * @return A new column group where the vector is added.
+	 */
+	public AColGroup addVector(double[] v) {
+		return binaryRowOpRight(new BinaryOperator(Plus.getPlusFnObject(), 1), v, false);
+	}
+
+	/**
 	 * 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 v         The vector of values to apply the values contained should be at least the length of the highest
+	 *                  value in the column index
 	 * @param isRowSafe True if the binary op is applied to an entire zero row and all results are zero
 	 * @return A updated column group with the new values.
 	 */
diff --git a/src/main/java/org/apache/sysds/runtime/compress/colgroup/AMorphingMMColGroup.java b/src/main/java/org/apache/sysds/runtime/compress/colgroup/AMorphingMMColGroup.java
index 7f18843..bb85e78 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/colgroup/AMorphingMMColGroup.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/colgroup/AMorphingMMColGroup.java
@@ -197,7 +197,6 @@
 	protected abstract AColGroup allocateRightMultiplicationCommon(double[] common, IColIndex colIndexes,
 		ADictionary preAgg);
 
-	/** extract common value from group and return non morphing group */
 	/**
 	 * extract common value from group and return non morphing group
 	 * 
diff --git a/src/main/java/org/apache/sysds/runtime/compress/colgroup/ASDC.java b/src/main/java/org/apache/sysds/runtime/compress/colgroup/ASDC.java
index 964dc08..96c3dda 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/colgroup/ASDC.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/colgroup/ASDC.java
@@ -60,6 +60,6 @@
 	@Override
 	public final CompressedSizeInfoColGroup getCompressionInfo(int nRow) {
 		EstimationFactors ef = new EstimationFactors(getNumValues(), _numRows, getNumberOffsets(), _dict.getSparsity());
-		return new CompressedSizeInfoColGroup(_colIndexes, ef, nRow, getCompType());
+		return new CompressedSizeInfoColGroup(_colIndexes, ef, nRow, getCompType(),getEncoding());
 	}
 }
diff --git a/src/main/java/org/apache/sysds/runtime/compress/colgroup/ASDCZero.java b/src/main/java/org/apache/sysds/runtime/compress/colgroup/ASDCZero.java
index 4ce3946..0414586 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/colgroup/ASDCZero.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/colgroup/ASDCZero.java
@@ -228,6 +228,6 @@
 	@Override
 	public final CompressedSizeInfoColGroup getCompressionInfo(int nRow) {
 		EstimationFactors ef = new EstimationFactors(getNumValues(), _numRows, getNumberOffsets(), _dict.getSparsity());
-		return new CompressedSizeInfoColGroup(_colIndexes, ef, nRow, getCompType());
+		return new CompressedSizeInfoColGroup(_colIndexes, ef, nRow, getCompType(),getEncoding());
 	}
 }
diff --git a/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupConst.java b/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupConst.java
index 8704429..e345488 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupConst.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupConst.java
@@ -34,6 +34,7 @@
 import org.apache.sysds.runtime.compress.colgroup.scheme.ICLAScheme;
 import org.apache.sysds.runtime.compress.cost.ComputationCostEstimator;
 import org.apache.sysds.runtime.compress.estim.CompressedSizeInfoColGroup;
+import org.apache.sysds.runtime.compress.estim.EstimationFactors;
 import org.apache.sysds.runtime.compress.estim.encoding.EncodingFactory;
 import org.apache.sysds.runtime.compress.estim.encoding.IEncode;
 import org.apache.sysds.runtime.compress.lib.CLALibLeftMultBy;
@@ -576,7 +577,8 @@
 
 	@Override
 	public CompressedSizeInfoColGroup getCompressionInfo(int nRow) {
-		return new CompressedSizeInfoColGroup(_colIndexes, 1, nRow, CompressionType.CONST);
+		EstimationFactors ef = new EstimationFactors(1, 1, 1, _dict.getSparsity());
+		return new CompressedSizeInfoColGroup(_colIndexes, ef, estimateInMemorySize(), CompressionType.CONST, getEncoding());
 	}
 
 	@Override
@@ -594,8 +596,8 @@
 		return _dict.getValues();
 	}
 
-	@Override 
-	protected AColGroup fixColIndexes(IColIndex newColIndex, int[] reordering){
+	@Override
+	protected AColGroup fixColIndexes(IColIndex newColIndex, int[] reordering) {
 		return ColGroupConst.create(newColIndex, _dict.reorder(reordering));
 	}
 
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 f885790..78a0bdb 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
@@ -555,8 +555,9 @@
 
 	@Override
 	public CompressedSizeInfoColGroup getCompressionInfo(int nRow) {
+		IEncode enc = getEncoding();
 		EstimationFactors ef = new EstimationFactors(getNumValues(), _data.size(), _data.size(), _dict.getSparsity());
-		return new CompressedSizeInfoColGroup(_colIndexes, ef, nRow, getCompType());
+		return new CompressedSizeInfoColGroup(_colIndexes, ef, estimateInMemorySize(), getCompType(), enc);
 	}
 
 	@Override
diff --git a/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupDDCFOR.java b/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupDDCFOR.java
index 39399b5..c1f99d0 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupDDCFOR.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupDDCFOR.java
@@ -54,7 +54,7 @@
 /**
  * Class to encapsulate information about a column group that is encoded with dense dictionary encoding (DDC).
  */
-public class ColGroupDDCFOR extends AMorphingMMColGroup {
+public class ColGroupDDCFOR extends AMorphingMMColGroup implements IFrameOfReferenceGroup {
 	private static final long serialVersionUID = -5769772089913918987L;
 
 	/** Pointers to row indexes in the dictionary */
@@ -469,8 +469,9 @@
 
 	@Override
 	public CompressedSizeInfoColGroup getCompressionInfo(int nRow) {
+		IEncode enc = getEncoding();
 		EstimationFactors ef = new EstimationFactors(getNumValues(), _data.size(), _data.size(), _dict.getSparsity());
-		return new CompressedSizeInfoColGroup(_colIndexes, ef, nRow, getCompType());
+		return new CompressedSizeInfoColGroup(_colIndexes, ef, estimateInMemorySize(), getCompType(), enc);
 	}
 
 	@Override
diff --git a/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupEmpty.java b/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupEmpty.java
index 1e1b847..c908b26 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupEmpty.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupEmpty.java
@@ -33,6 +33,7 @@
 import org.apache.sysds.runtime.compress.colgroup.scheme.ICLAScheme;
 import org.apache.sysds.runtime.compress.cost.ComputationCostEstimator;
 import org.apache.sysds.runtime.compress.estim.CompressedSizeInfoColGroup;
+import org.apache.sysds.runtime.compress.estim.EstimationFactors;
 import org.apache.sysds.runtime.compress.estim.encoding.EncodingFactory;
 import org.apache.sysds.runtime.compress.estim.encoding.IEncode;
 import org.apache.sysds.runtime.data.DenseBlock;
@@ -350,7 +351,8 @@
 
 	@Override
 	public CompressedSizeInfoColGroup getCompressionInfo(int nRow) {
-		return new CompressedSizeInfoColGroup(_colIndexes, 0, nRow, CompressionType.CONST);
+		EstimationFactors ef = new EstimationFactors(getNumValues(), 1, 0, 0.0);
+		return new CompressedSizeInfoColGroup(_colIndexes, ef, estimateInMemorySize(), CompressionType.CONST, getEncoding());
 	}
 
 	@Override
diff --git a/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupFactory.java b/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupFactory.java
index bf4900e..3358517 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupFactory.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupFactory.java
@@ -32,6 +32,7 @@
 import org.apache.commons.lang3.NotImplementedException;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.apache.sysds.runtime.compress.CompressedMatrixBlock;
 import org.apache.sysds.runtime.compress.CompressionSettings;
 import org.apache.sysds.runtime.compress.DMLCompressionException;
 import org.apache.sysds.runtime.compress.bitmap.ABitmap;
@@ -53,6 +54,7 @@
 import org.apache.sysds.runtime.compress.cost.ACostEstimate;
 import org.apache.sysds.runtime.compress.estim.CompressedSizeInfo;
 import org.apache.sysds.runtime.compress.estim.CompressedSizeInfoColGroup;
+import org.apache.sysds.runtime.compress.lib.CLALibCombineGroups;
 import org.apache.sysds.runtime.compress.readers.ReaderColumnSelection;
 import org.apache.sysds.runtime.compress.utils.ACount.DCounts;
 import org.apache.sysds.runtime.compress.utils.DblArray;
@@ -149,15 +151,17 @@
 
 	private List<AColGroup> compress() {
 		try {
-			List<AColGroup> ret = compressExecute();
-			if(pool != null)
-				pool.shutdown();
-			return ret;
+			if(in instanceof CompressedMatrixBlock)
+				return CLALibCombineGroups.combine((CompressedMatrixBlock) in, csi, pool);
+			else
+				return compressExecute();
 		}
 		catch(Exception e) {
+			throw new DMLCompressionException("Compression Failed", e);
+		}
+		finally {
 			if(pool != null)
 				pool.shutdown();
-			throw new DMLCompressionException("Compression Failed", e);
 		}
 	}
 
diff --git a/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupSDCFOR.java b/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupSDCFOR.java
index d05930d..d2e7b54 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupSDCFOR.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupSDCFOR.java
@@ -62,7 +62,7 @@
  * with no modifications.
  * 
  */
-public class ColGroupSDCFOR extends ASDC implements IMapToDataGroup {
+public class ColGroupSDCFOR extends ASDC implements IMapToDataGroup , IFrameOfReferenceGroup{
 
 	private static final long serialVersionUID = 3883228464052204203L;
 
diff --git a/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupSDCSingle.java b/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupSDCSingle.java
index f4c11f2..7f43df5 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupSDCSingle.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupSDCSingle.java
@@ -24,7 +24,6 @@
 import java.io.IOException;
 import java.util.Arrays;
 
-import org.apache.commons.lang.NotImplementedException;
 import org.apache.sysds.runtime.DMLRuntimeException;
 import org.apache.sysds.runtime.compress.colgroup.dictionary.ADictionary;
 import org.apache.sysds.runtime.compress.colgroup.dictionary.Dictionary;
@@ -117,12 +116,6 @@
 	}
 
 	@Override
-	public ADictionary getDictionary() {
-		throw new NotImplementedException(
-			"Not implemented getting the dictionary out, and i think we should consider removing the option");
-	}
-
-	@Override
 	protected double[] preAggSumRows() {
 		return _dict.sumAllRowsToDoubleWithDefault(_defaultTuple);
 	}
diff --git a/src/main/java/org/apache/sysds/runtime/compress/colgroup/IFrameOfReferenceGroup.java b/src/main/java/org/apache/sysds/runtime/compress/colgroup/IFrameOfReferenceGroup.java
new file mode 100644
index 0000000..dc55b56
--- /dev/null
+++ b/src/main/java/org/apache/sysds/runtime/compress/colgroup/IFrameOfReferenceGroup.java
@@ -0,0 +1,33 @@
+/*
+ * 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;
+
+/**
+ * Interface for frame of reference groups.
+ */
+public interface IFrameOfReferenceGroup {
+	/**
+	 * extract common value from group and return non morphing group
+	 * 
+	 * @param constV a vector to contain all values, note length = nCols in total matrix.
+	 * @return A non morphing column group with decompression instructions.
+	 */
+	public AColGroup extractCommon(double[] constV);
+}
diff --git a/src/main/java/org/apache/sysds/runtime/compress/colgroup/dictionary/DictionaryFactory.java b/src/main/java/org/apache/sysds/runtime/compress/colgroup/dictionary/DictionaryFactory.java
index da5c317..37ed289 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/colgroup/dictionary/DictionaryFactory.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/colgroup/dictionary/DictionaryFactory.java
@@ -22,6 +22,7 @@
 import java.io.DataInput;
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.Map;
 
 import org.apache.commons.lang.NotImplementedException;
 import org.apache.commons.logging.Log;
@@ -238,6 +239,11 @@
 	}
 
 	public static ADictionary combineDictionaries(AColGroupCompressed a, AColGroupCompressed b) {
+		return combineDictionaries(a, b, null);
+	}
+
+	public static ADictionary combineDictionaries(AColGroupCompressed a, AColGroupCompressed b,
+		Map<Integer, Integer> filter) {
 		if(a instanceof ColGroupEmpty && b instanceof ColGroupEmpty)
 			return null; // null return is handled elsewhere.
 
@@ -266,23 +272,31 @@
 					return combineSparseConstSparseRet(ad, a.getNumCols(), bt);
 				}
 				else if(bc.isDense())
-					return combineFullDictionaries(ad, a.getNumCols(), bd, b.getNumCols());
+					return combineFullDictionaries(ad, a.getNumCols(), bd, b.getNumCols(), filter);
 				else if(bc.isSDC()) {
 					double[] tuple = ((IContainDefaultTuple) b).getDefaultTuple();
-					return combineSDCRight(ad, a.getNumCols(), bd, tuple);
+					return combineSDCRight(ad, a.getNumCols(), bd, tuple, filter);
 				}
 			}
 			else if(ac.isSDC()) {
 				if(bc.isSDC()) {
 					final double[] at = ((IContainDefaultTuple) a).getDefaultTuple();
 					final double[] bt = ((IContainDefaultTuple) b).getDefaultTuple();
-					return combineSDC(ad, at, bd, bt);
+					return combineSDC(ad, at, bd, bt, filter);
 				}
 			}
 		}
 		throw new NotImplementedException("Not supporting combining dense: " + a + " " + b);
 	}
 
+	/**
+	 * Combine the dictionaries assuming a sparse combination where each dictionary can be a SDC containing a default
+	 * element that have to be introduced into the combined dictionary.
+	 * 
+	 * @param a A Dictionary can be SDC or const
+	 * @param b A Dictionary can be Const or SDC.
+	 * @return The combined dictionary
+	 */
 	public static ADictionary combineDictionariesSparse(AColGroupCompressed a, AColGroupCompressed b) {
 		CompressionType ac = a.getCompType();
 		CompressionType bc = b.getCompType();
@@ -298,9 +312,7 @@
 				if(a.sameIndexStructure(b)) {
 					return ad.cbind(bd, b.getNumCols());
 				}
-
 				// real combine extract default and combine like dense but with default before.
-
 			}
 		}
 		else if(ac.isConst()) {
@@ -315,7 +327,7 @@
 	}
 
 	/**
-	 * Combine the dictionaries as if the dictionaries contain the full spectrum of the data contained.
+	 * Combine the dictionaries as if the dictionaries contain the full spectrum of the combined data.
 	 * 
 	 * @param a   Left side dictionary
 	 * @param nca Number of columns left dictionary
@@ -324,6 +336,22 @@
 	 * @return A combined dictionary
 	 */
 	public static ADictionary combineFullDictionaries(ADictionary a, int nca, ADictionary b, int ncb) {
+		return combineFullDictionaries(a, nca, b, ncb, null);
+	}
+
+	/**
+	 * Combine the dictionaries as if the dictionaries only contain the values in the specified filter.
+	 * 
+	 * @param a      Left side dictionary
+	 * @param nca    Number of columns left dictionary
+	 * @param b      Right side dictionary
+	 * @param ncb    Number of columns right dictionary
+	 * @param filter The mapping filter to not include all possible combinations in the output, this filter is allowed to
+	 *               be null, that means the output is defaulting back to a full combine
+	 * @return A combined dictionary
+	 */
+	public static ADictionary combineFullDictionaries(ADictionary a, int nca, ADictionary b, int ncb,
+		Map<Integer, Integer> filter) {
 		final int ra = a.getNumberOfValues(nca);
 		final int rb = b.getNumberOfValues(ncb);
 
@@ -333,24 +361,45 @@
 		if(ra == 1 && rb == 1)
 			return new MatrixBlockDictionary(ma.append(mb));
 
-		MatrixBlock out = new MatrixBlock(ra * rb, nca + ncb, false);
+		MatrixBlock out = new MatrixBlock(filter != null ? filter.size() : ra * rb, nca + ncb, false);
 
 		out.allocateBlock();
 
-		for(int r = 0; r < out.getNumRows(); r++) {
-			int ia = r % ra;
-			int ib = r / ra;
-			for(int c = 0; c < nca; c++)
-				out.quickSetValue(r, c, ma.quickGetValue(ia, c));
+		if(filter != null) {
+			for(int r : filter.keySet()) {
+				int o = filter.get(r);
+				int ia = r % ra;
+				int ib = r / ra;
+				for(int c = 0; c < nca; c++)
+					out.quickSetValue(o, c, ma.quickGetValue(ia, c));
 
-			for(int c = 0; c < ncb; c++)
-				out.quickSetValue(r, c + nca, mb.quickGetValue(ib, c));
+				for(int c = 0; c < ncb; c++)
+					out.quickSetValue(o, c + nca, mb.quickGetValue(ib, c));
 
+			}
+		}
+		else {
+
+			for(int r = 0; r < out.getNumRows(); r++) {
+				int ia = r % ra;
+				int ib = r / ra;
+				for(int c = 0; c < nca; c++)
+					out.quickSetValue(r, c, ma.quickGetValue(ia, c));
+
+				for(int c = 0; c < ncb; c++)
+					out.quickSetValue(r, c + nca, mb.quickGetValue(ib, c));
+
+			}
 		}
 		return new MatrixBlockDictionary(out);
 	}
 
 	public static ADictionary combineSDCRight(ADictionary a, int nca, ADictionary b, double[] tub) {
+		return combineSDCRight(a, nca, b, tub, null);
+	}
+
+	public static ADictionary combineSDCRight(ADictionary a, int nca, ADictionary b, double[] tub,
+		Map<Integer, Integer> filter) {
 		final int ncb = tub.length;
 		final int ra = a.getNumberOfValues(nca);
 		final int rb = b.getNumberOfValues(ncb);
@@ -384,6 +433,13 @@
 	}
 
 	public static ADictionary combineSDC(ADictionary a, double[] tua, ADictionary b, double[] tub) {
+		return combineSDC(a, tua, b, tub, null);
+	}
+
+	public static ADictionary combineSDC(ADictionary a, double[] tua, ADictionary b, double[] tub,
+		Map<Integer, Integer> filter) {
+		if(filter != null)
+			throw new NotImplementedException();
 		final int nca = tua.length;
 		final int ncb = tub.length;
 		final int ra = a.getNumberOfValues(nca);
diff --git a/src/main/java/org/apache/sysds/runtime/compress/colgroup/indexes/AColIndex.java b/src/main/java/org/apache/sysds/runtime/compress/colgroup/indexes/AColIndex.java
index f8eb0bf..cf22ba0 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/colgroup/indexes/AColIndex.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/colgroup/indexes/AColIndex.java
@@ -66,4 +66,13 @@
 			res = 31 * res + it.next();
 		return res;
 	}
+
+	@Override
+	public boolean containsAny(IColIndex idx) {
+		IIterate it = idx.iterator();
+		while(it.hasNext())
+			if(contains(it.next()))
+				return true;
+		return false;
+	}
 }
diff --git a/src/main/java/org/apache/sysds/runtime/compress/colgroup/indexes/IColIndex.java b/src/main/java/org/apache/sysds/runtime/compress/colgroup/indexes/IColIndex.java
index 8b73abf..5163998 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/colgroup/indexes/IColIndex.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/colgroup/indexes/IColIndex.java
@@ -178,6 +178,14 @@
 	 */
 	public boolean contains(int i);
 
+	/**
+	 * Analyze if this column group contain any of the given column Ids.
+	 * 
+	 * @param idx A List of indexes
+	 * @return If it is contained
+	 */
+	public boolean containsAny(IColIndex idx);
+
 	/** A Class for slice results containing indexes for the slicing of dictionaries, and the resulting column index */
 	public static class SliceResult {
 		/** Start index to slice inside the dictionary */
diff --git a/src/main/java/org/apache/sysds/runtime/compress/estim/AComEst.java b/src/main/java/org/apache/sysds/runtime/compress/estim/AComEst.java
index 145682f..cf55e61 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/estim/AComEst.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/estim/AComEst.java
@@ -184,7 +184,6 @@
 		final int worstCase = worstCaseUpperBound(combinedColumns);
 		// Get max number of tuples based on the above.
 		final long max = Math.min((long) g1V * g2V, worstCase);
-
 		if(max > 1000000) // set the max combination to a million distinct
 			return null; // This combination is clearly not a good idea return null to indicate that.
 		else if(g1.getMap() == null || g2.getMap() == null)
diff --git a/src/main/java/org/apache/sysds/runtime/compress/estim/ComEstCompressed.java b/src/main/java/org/apache/sysds/runtime/compress/estim/ComEstCompressed.java
index 7f8ccfe..5a1a133 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/estim/ComEstCompressed.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/estim/ComEstCompressed.java
@@ -22,11 +22,12 @@
 import java.util.ArrayList;
 import java.util.List;
 
-import org.apache.commons.lang.NotImplementedException;
 import org.apache.sysds.runtime.compress.CompressedMatrixBlock;
 import org.apache.sysds.runtime.compress.CompressionSettings;
 import org.apache.sysds.runtime.compress.colgroup.AColGroup;
 import org.apache.sysds.runtime.compress.colgroup.indexes.IColIndex;
+import org.apache.sysds.runtime.compress.estim.encoding.IEncode;
+import org.apache.sysds.runtime.compress.lib.CLALibCombineGroups;
 
 public class ComEstCompressed extends AComEst {
 
@@ -66,14 +67,26 @@
 			AColGroup g = cData.getColGroupForColumn(id);
 			return g.getNumValues();
 		}
-		else
-			throw new UnsupportedOperationException("Unimplemented method 'worstCaseUpperBound'");
+		else {
+			List<AColGroup> groups = CLALibCombineGroups.findGroupsInIndex(columns, cData.getColGroups());
+			int nVals = 1;
+			for(AColGroup g : groups)
+				nVals *= g.getNumValues();
+
+			return Math.min(_data.getNumRows(), nVals);
+		}
 	}
 
 	@Override
 	protected CompressedSizeInfoColGroup combine(IColIndex combinedColumns, CompressedSizeInfoColGroup g1,
 		CompressedSizeInfoColGroup g2, int maxDistinct) {
-		throw new UnsupportedOperationException("Unimplemented method 'combine'");
+		final IEncode map = g1.getMap().combine(g2.getMap());
+		return getFacts(map, combinedColumns);
 	}
 
+	protected CompressedSizeInfoColGroup getFacts(IEncode map, IColIndex colIndexes) {
+		final int _numRows = getNumRows();
+		final EstimationFactors em = map.extractFacts(_numRows, _data.getSparsity(), _data.getSparsity(), _cs);
+		return new CompressedSizeInfoColGroup(colIndexes, em, _cs.validCompressions, map);
+	}
 }
diff --git a/src/main/java/org/apache/sysds/runtime/compress/estim/ComEstExact.java b/src/main/java/org/apache/sysds/runtime/compress/estim/ComEstExact.java
index e9ac7a9..63af720 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/estim/ComEstExact.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/estim/ComEstExact.java
@@ -56,7 +56,7 @@
 		return getFacts(map, combinedColumns);
 	}
 
-	private CompressedSizeInfoColGroup getFacts(IEncode map, IColIndex colIndexes) {
+	protected CompressedSizeInfoColGroup getFacts(IEncode map, IColIndex colIndexes) {
 		final int _numRows = getNumRows();
 		final EstimationFactors em = map.extractFacts(_numRows, _data.getSparsity(), _data.getSparsity(), _cs);	
 		return new CompressedSizeInfoColGroup(colIndexes, em, _cs.validCompressions, map);
diff --git a/src/main/java/org/apache/sysds/runtime/compress/estim/CompressedSizeInfoColGroup.java b/src/main/java/org/apache/sysds/runtime/compress/estim/CompressedSizeInfoColGroup.java
index 8de7360..b9a1f24 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/estim/CompressedSizeInfoColGroup.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/estim/CompressedSizeInfoColGroup.java
@@ -79,6 +79,16 @@
 		_sizes.put(bestCompressionType, _minSize);
 	}
 
+	public CompressedSizeInfoColGroup(IColIndex columns, EstimationFactors facts, long minSize, CompressionType bestCompression, IEncode map){
+		_cols = columns;
+		_facts = facts;
+		_minSize = minSize;
+		_bestCompressionType = bestCompression;
+		_sizes = new EnumMap<>(CompressionType.class);
+		_sizes.put(bestCompression, _minSize);
+		_map = map;
+	}
+
 	public CompressedSizeInfoColGroup(IColIndex columns, EstimationFactors facts,
 		Set<CompressionType> validCompressionTypes, IEncode map) {
 		_cols = columns;
@@ -260,12 +270,11 @@
 	public String toString() {
 		StringBuilder sb = new StringBuilder();
 		sb.append(this.getClass().getSimpleName());
-		sb.append("cols: " + _cols);
+		sb.append(" cols: " + _cols);
 		sb.append(String.format(" common: %4.3f", getMostCommonFraction()));
-		sb.append(" Sizes: ");
-		sb.append(_sizes);
+		sb.append(" Sizes: " + _sizes);
 		sb.append(" facts: " + _facts);
-		// sb.append("\n" + _map);
+		sb.append(" mapIsNull: " + (_map == null));
 		return sb.toString();
 	}
 
diff --git a/src/main/java/org/apache/sysds/runtime/compress/estim/encoding/ConstEncoding.java b/src/main/java/org/apache/sysds/runtime/compress/estim/encoding/ConstEncoding.java
index b3839c9..1a12077 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/estim/encoding/ConstEncoding.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/estim/encoding/ConstEncoding.java
@@ -19,6 +19,10 @@
 
 package org.apache.sysds.runtime.compress.estim.encoding;
 
+import java.util.Map;
+
+import org.apache.commons.lang3.tuple.ImmutablePair;
+import org.apache.commons.lang3.tuple.Pair;
 import org.apache.sysds.runtime.compress.CompressionSettings;
 import org.apache.sysds.runtime.compress.estim.EstimationFactors;
 
@@ -37,8 +41,8 @@
 	}
 
 	@Override
-	public IEncode combineNoResize(IEncode e){
-		return e;
+	public Pair<IEncode, Map<Integer, Integer>> combineWithMap(IEncode e) {
+		return new ImmutablePair<>(e, null);
 	}
 
 	@Override
diff --git a/src/main/java/org/apache/sysds/runtime/compress/estim/encoding/DenseEncoding.java b/src/main/java/org/apache/sysds/runtime/compress/estim/encoding/DenseEncoding.java
index 5504e1c..f68dd3d 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/estim/encoding/DenseEncoding.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/estim/encoding/DenseEncoding.java
@@ -22,12 +22,17 @@
 import java.util.HashMap;
 import java.util.Map;
 
+import org.apache.commons.lang3.tuple.ImmutablePair;
+import org.apache.commons.lang3.tuple.Pair;
 import org.apache.sysds.runtime.compress.CompressionSettings;
 import org.apache.sysds.runtime.compress.colgroup.mapping.AMapToData;
 import org.apache.sysds.runtime.compress.colgroup.mapping.MapToFactory;
 import org.apache.sysds.runtime.compress.colgroup.offset.AIterator;
 import org.apache.sysds.runtime.compress.estim.EstimationFactors;
 
+/**
+ * An Encoding that contains a value on each row of the input.
+ */
 public class DenseEncoding implements IEncode {
 
 	private final AMapToData map;
@@ -38,7 +43,7 @@
 	}
 
 	@Override
-	public DenseEncoding combine(IEncode e) {
+	public IEncode combine(IEncode e) {
 		if(e instanceof EmptyEncoding || e instanceof ConstEncoding)
 			return this;
 		else if(e instanceof SparseEncoding)
@@ -48,16 +53,16 @@
 	}
 
 	@Override
-	public IEncode combineNoResize(IEncode e) {
+	public Pair<IEncode, Map<Integer, Integer>> combineWithMap(IEncode e) {
 		if(e instanceof EmptyEncoding || e instanceof ConstEncoding)
-			return this;
+			return new ImmutablePair<>(this, null);
 		else if(e instanceof SparseEncoding)
 			return combineSparseNoResize((SparseEncoding) e);
 		else
 			return combineDenseNoResize((DenseEncoding) e);
 	}
 
-	protected DenseEncoding combineSparse(SparseEncoding e) {
+	protected IEncode combineSparse(SparseEncoding e) {
 		final int maxUnique = e.getUnique() * getUnique();
 		final int size = map.size();
 		final int nVl = getUnique();
@@ -66,7 +71,7 @@
 		final AMapToData ret = assignSparse(e);
 		// Iteration 2 reassign indexes.
 		if(maxUnique + nVl > size)
-			return combineSparseHashMap(ret);
+			return combineSparseHashMap(ret).getLeft();
 		else
 			return combineSparseMapToData(ret, maxUnique, nVl);
 	}
@@ -92,7 +97,7 @@
 		return ret;
 	}
 
-	private final DenseEncoding combineSparseHashMap(final AMapToData ret) {
+	private final Pair<IEncode, Map<Integer, Integer>> combineSparseHashMap(final AMapToData ret) {
 		final int size = ret.size();
 		final Map<Integer, Integer> m = new HashMap<>(size);
 		for(int r = 0; r < size; r++) {
@@ -104,7 +109,7 @@
 			else
 				ret.set(r, mv);
 		}
-		return new DenseEncoding(MapToFactory.resize(ret, m.size()));
+		return new ImmutablePair<>(new DenseEncoding(MapToFactory.resize(ret, m.size())), m);
 	}
 
 	private final DenseEncoding combineSparseMapToData(final AMapToData ret, final int maxUnique, final int nVl) {
@@ -136,15 +141,20 @@
 
 		final AMapToData ret = MapToFactory.create(size, maxUnique);
 
-		if(maxUnique > size)
-			return combineDenseWithHashMap(lm, rm, size, nVL, ret);
-		else
-			return combineDenseWithMapToData(lm, rm, size, nVL, ret, maxUnique);
+		if(maxUnique > size) {
+			// aka there is more maxUnique than rows.
+			final Map<Integer, Integer> m = new HashMap<>(size);
+			return combineDenseWithHashMap(lm, rm, size, nVL, ret, m);
+		}
+		else {
+			final AMapToData m = MapToFactory.create(maxUnique, maxUnique + 1);
+			return combineDenseWithMapToData(lm, rm, size, nVL, ret, maxUnique, m);
+		}
 	}
 
-	private DenseEncoding combineDenseNoResize(final DenseEncoding other) {
+	private Pair<IEncode, Map<Integer, Integer>> combineDenseNoResize(final DenseEncoding other) {
 		if(map == other.map)
-			return this; // same object
+			return new ImmutablePair<>(this, null); // same object
 
 		final AMapToData lm = map;
 		final AMapToData rm = other.map;
@@ -156,20 +166,21 @@
 
 		final AMapToData ret = MapToFactory.create(size, maxUnique);
 
-		for(int r = 0; r < size; r++)
-			ret.set(r, lm.getIndex(r) + rm.getIndex(r) * nVL);
+		final Map<Integer, Integer> m = new HashMap<>(Math.min(size, maxUnique));
+		return new ImmutablePair<>(combineDenseWithHashMap(lm, rm, size, nVL, ret, m), m);
+
 		// there can be less unique.
 
-		return new DenseEncoding(ret);
+		// return new DenseEncoding(ret);
 	}
 
-	private DenseEncoding combineSparseNoResize(final SparseEncoding other) {
-		return new DenseEncoding(assignSparse(other));
+	private Pair<IEncode, Map<Integer, Integer>> combineSparseNoResize(final SparseEncoding other) {
+		final AMapToData a = assignSparse(other);
+		return combineSparseHashMap(a);
 	}
 
 	protected final DenseEncoding combineDenseWithHashMap(final AMapToData lm, final AMapToData rm, final int size,
-		final int nVL, final AMapToData ret) {
-		final Map<Integer, Integer> m = new HashMap<>(size);
+		final int nVL, final AMapToData ret, Map<Integer, Integer> m) {
 
 		for(int r = 0; r < size; r++)
 			addValHashMap(lm.getIndex(r) + rm.getIndex(r) * nVL, r, m, ret);
@@ -177,8 +188,7 @@
 	}
 
 	protected final DenseEncoding combineDenseWithMapToData(final AMapToData lm, final AMapToData rm, final int size,
-		final int nVL, final AMapToData ret, final int maxUnique) {
-		final AMapToData m = MapToFactory.create(maxUnique, maxUnique + 1);
+		final int nVL, final AMapToData ret, final int maxUnique, final AMapToData m) {
 		int newUID = 1;
 		for(int r = 0; r < size; r++)
 			newUID = addValMapToData(lm.getIndex(r) + rm.getIndex(r) * nVL, r, m, newUID, ret);
diff --git a/src/main/java/org/apache/sysds/runtime/compress/estim/encoding/EmptyEncoding.java b/src/main/java/org/apache/sysds/runtime/compress/estim/encoding/EmptyEncoding.java
index 6aa9f91..9e12654 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/estim/encoding/EmptyEncoding.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/estim/encoding/EmptyEncoding.java
@@ -19,10 +19,16 @@
 
 package org.apache.sysds.runtime.compress.estim.encoding;
 
+import java.util.Map;
+
+import org.apache.commons.lang3.tuple.ImmutablePair;
+import org.apache.commons.lang3.tuple.Pair;
 import org.apache.sysds.runtime.compress.CompressionSettings;
 import org.apache.sysds.runtime.compress.estim.EstimationFactors;
 
-/** Empty encoding for cases where the entire group of columns is zero */
+/**
+ * Empty encoding for cases where the entire group of columns is zero
+ */
 public class EmptyEncoding implements IEncode {
 
 	// empty constructor
@@ -35,8 +41,8 @@
 	}
 
 	@Override
-	public IEncode combineNoResize(IEncode e){
-		return e;
+	public Pair<IEncode, Map<Integer, Integer>> combineWithMap(IEncode e) {
+		return new ImmutablePair<>(e, null);
 	}
 
 	@Override
diff --git a/src/main/java/org/apache/sysds/runtime/compress/estim/encoding/IEncode.java b/src/main/java/org/apache/sysds/runtime/compress/estim/encoding/IEncode.java
index dcff899..e7202a1 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/estim/encoding/IEncode.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/estim/encoding/IEncode.java
@@ -19,6 +19,9 @@
 
 package org.apache.sysds.runtime.compress.estim.encoding;
 
+import java.util.Map;
+
+import org.apache.commons.lang3.tuple.Pair;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.sysds.runtime.compress.CompressionSettings;
@@ -42,13 +45,13 @@
 	public IEncode combine(IEncode e);
 
 	/**
-	 * Combine two encodings without resizing the output. meaning the mapping of the indexes should be consistant with 
+	 * Combine two encodings without resizing the output. meaning the mapping of the indexes should be consistent with 
 	 * left hand side Dictionary indexes and right hand side indexes.
 	 * 
 	 * @param e The other side to combine with 
 	 * @return The combined encoding
 	 */
-	public IEncode combineNoResize(IEncode e);
+	public Pair<IEncode, Map<Integer,Integer>> combineWithMap(IEncode e);
 
 	/**
 	 * Get the number of unique values in this encoding
diff --git a/src/main/java/org/apache/sysds/runtime/compress/estim/encoding/SparseEncoding.java b/src/main/java/org/apache/sysds/runtime/compress/estim/encoding/SparseEncoding.java
index ff3b328..872e512 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/estim/encoding/SparseEncoding.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/estim/encoding/SparseEncoding.java
@@ -19,6 +19,10 @@
 
 package org.apache.sysds.runtime.compress.estim.encoding;
 
+import java.util.Map;
+
+import org.apache.commons.lang3.tuple.ImmutablePair;
+import org.apache.commons.lang3.tuple.Pair;
 import org.apache.sysds.runtime.compress.CompressionSettings;
 import org.apache.sysds.runtime.compress.DMLCompressionException;
 import org.apache.sysds.runtime.compress.colgroup.mapping.AMapToData;
@@ -29,7 +33,10 @@
 import org.apache.sysds.runtime.compress.estim.EstimationFactors;
 import org.apache.sysds.runtime.compress.utils.IntArrayList;
 
-/** Most common is zero encoding */
+/**
+ * A Encoding that contain a default value that is not encoded and every other value is encoded in the map. The logic is
+ * similar to the SDC column group.
+ */
 public class SparseEncoding implements IEncode {
 
 	/** A map to the distinct values contained */
@@ -63,14 +70,14 @@
 	}
 
 	@Override
-	public IEncode combineNoResize(IEncode e) {
+	public Pair<IEncode, Map<Integer, Integer>> combineWithMap(IEncode e) {
 		if(e instanceof EmptyEncoding || e instanceof ConstEncoding)
-			return this;
+			return new ImmutablePair<>(this, null);
 		else if(e instanceof SparseEncoding) {
 			SparseEncoding es = (SparseEncoding) e;
 			if(es.off == off && es.map == map)
-				return this;
-			return combineSparseNoResize(es);
+				return new ImmutablePair<>(this, null);
+			return new ImmutablePair<>(combineSparseNoResize(es), null);
 		}
 		else
 			throw new DMLCompressionException("Not allowed other to be dense");
diff --git a/src/main/java/org/apache/sysds/runtime/compress/io/ReaderCompressed.java b/src/main/java/org/apache/sysds/runtime/compress/io/ReaderCompressed.java
index f1f45ab..c3493df 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/io/ReaderCompressed.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/io/ReaderCompressed.java
@@ -35,6 +35,7 @@
 import org.apache.sysds.runtime.DMLRuntimeException;
 import org.apache.sysds.runtime.compress.CompressedMatrixBlock;
 import org.apache.sysds.runtime.compress.CompressedMatrixBlockFactory;
+import org.apache.sysds.runtime.compress.DMLCompressionException;
 import org.apache.sysds.runtime.compress.lib.CLALibStack;
 import org.apache.sysds.runtime.io.IOUtilFunctions;
 import org.apache.sysds.runtime.io.MatrixReader;
@@ -78,6 +79,8 @@
 		for(Path subPath : IOUtilFunctions.getSequenceFilePaths(fs, path)){
 			read(subPath, job, data);
 		}
+		if(data.containsValue(null))
+			throw new DMLCompressionException("Invalid read data contains null:");
 
 		if(data.size() == 1)
 			return data.entrySet().iterator().next().getValue();
diff --git a/src/main/java/org/apache/sysds/runtime/compress/lib/CLALibCombineGroups.java b/src/main/java/org/apache/sysds/runtime/compress/lib/CLALibCombineGroups.java
index 27ea3f5..3d112a5 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/lib/CLALibCombineGroups.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/lib/CLALibCombineGroups.java
@@ -19,10 +19,17 @@
 
 package org.apache.sysds.runtime.compress.lib;
 
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutorService;
+
 import org.apache.commons.lang.NotImplementedException;
+import org.apache.commons.lang3.tuple.Pair;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.sysds.runtime.compress.CompressedMatrixBlock;
+import org.apache.sysds.runtime.compress.DMLCompressionException;
 import org.apache.sysds.runtime.compress.colgroup.AColGroup;
 import org.apache.sysds.runtime.compress.colgroup.AColGroupCompressed;
 import org.apache.sysds.runtime.compress.colgroup.ColGroupConst;
@@ -31,10 +38,13 @@
 import org.apache.sysds.runtime.compress.colgroup.ColGroupSDC;
 import org.apache.sysds.runtime.compress.colgroup.ColGroupUncompressed;
 import org.apache.sysds.runtime.compress.colgroup.IContainDefaultTuple;
+import org.apache.sysds.runtime.compress.colgroup.IFrameOfReferenceGroup;
 import org.apache.sysds.runtime.compress.colgroup.dictionary.ADictionary;
 import org.apache.sysds.runtime.compress.colgroup.dictionary.DictionaryFactory;
 import org.apache.sysds.runtime.compress.colgroup.indexes.ColIndexFactory;
 import org.apache.sysds.runtime.compress.colgroup.indexes.IColIndex;
+import org.apache.sysds.runtime.compress.estim.CompressedSizeInfo;
+import org.apache.sysds.runtime.compress.estim.CompressedSizeInfoColGroup;
 import org.apache.sysds.runtime.compress.estim.encoding.ConstEncoding;
 import org.apache.sysds.runtime.compress.estim.encoding.DenseEncoding;
 import org.apache.sysds.runtime.compress.estim.encoding.EmptyEncoding;
@@ -42,6 +52,7 @@
 import org.apache.sysds.runtime.compress.estim.encoding.SparseEncoding;
 import org.apache.sysds.runtime.data.DenseBlock;
 import org.apache.sysds.runtime.matrix.data.MatrixBlock;
+import org.apache.sysds.runtime.util.CommonThreadPool;
 
 /**
  * Library functions to combine column groups inside a compressed matrix.
@@ -53,8 +64,66 @@
 		// private constructor
 	}
 
-	public static CompressedMatrixBlock combine(CompressedMatrixBlock cmb, int k) {
-		throw new NotImplementedException();
+	public static List<AColGroup> combine(CompressedMatrixBlock cmb, int k) {
+		ExecutorService pool = null;
+		try {
+			pool = (k > 1) ? CommonThreadPool.get(k) : null;
+			return combine(cmb, null, pool);
+		}
+		catch(Exception e) {
+			throw new DMLCompressionException("Compression Failed", e);
+		}
+		finally {
+			if(pool != null)
+				pool.shutdown();
+		}
+	}
+
+	public static List<AColGroup> combine(CompressedMatrixBlock cmb, CompressedSizeInfo csi, ExecutorService pool) {
+
+		List<AColGroup> input = cmb.getColGroups();
+		boolean filterFor = CLALibUtils.shouldFilterFOR(input);
+		double[] c = filterFor ? new double[cmb.getNumColumns()] : null;
+		if(filterFor)
+			input = CLALibUtils.filterFOR(input, c);
+
+
+		List<List<AColGroup>> combinations = new ArrayList<>();
+		for(CompressedSizeInfoColGroup gi : csi.getInfo()) {
+			combinations.add(findGroupsInIndex(gi.getColumns(), input));
+		}
+
+		List<AColGroup> ret = new ArrayList<>();
+		if(filterFor)
+			for(List<AColGroup> combine : combinations)
+				ret.add(combineN(combine).addVector(c));
+		else
+			for(List<AColGroup> combine : combinations) 
+				ret.add(combineN(combine));
+		
+
+
+		return ret;
+	}
+
+	public static List<AColGroup> findGroupsInIndex(IColIndex idx, List<AColGroup> groups) {
+		List<AColGroup> ret = new ArrayList<>();
+		for(AColGroup g : groups) {
+			if(g.getColIndices().containsAny(idx))
+				ret.add(g);
+		}
+		return ret;
+	}
+
+	public static AColGroup combineN(List<AColGroup> groups) {
+
+		AColGroup base = groups.get(0);
+		// Inefficient combine N but base line
+		for(int i = 1; i < groups.size(); i++) {
+			base = combine(base, groups.get(i));
+		}
+
+		return base;
 	}
 
 	/**
@@ -62,11 +131,16 @@
 	 * 
 	 * The number of rows should be equal, and it is not verified so there will be unexpected behavior in such cases.
 	 * 
+	 * It is assumed that this method is not called with FOR groups
+	 * 
 	 * @param a The first group to combine.
 	 * @param b The second group to combine.
 	 * @return A new column group containing the two.
 	 */
 	public static AColGroup combine(AColGroup a, AColGroup b) {
+		if(a instanceof IFrameOfReferenceGroup || b instanceof IFrameOfReferenceGroup)
+			throw new DMLCompressionException("Invalid call with frame of reference group to combine");
+
 		IColIndex combinedColumns = ColIndexFactory.combine(a, b);
 
 		// try to recompress a and b if uncompressed
@@ -96,18 +170,19 @@
 			return combineCompressed(combinedColumns, bc, ac);
 		}
 
-		IEncode ce = ae.combineNoResize(be);
-
+		Pair<IEncode, Map<Integer,Integer>> cec = ae.combineWithMap(be);
+		IEncode ce = cec.getLeft();
+		Map<Integer,Integer> filter = cec.getRight();
 		if(ce instanceof DenseEncoding) {
-			DenseEncoding ced = (DenseEncoding) ce;
-			ADictionary cd = DictionaryFactory.combineDictionaries(ac, bc);
+			DenseEncoding ced = (DenseEncoding) (ce);
+			ADictionary cd = DictionaryFactory.combineDictionaries(ac, bc, filter);
 			return ColGroupDDC.create(combinedColumns, cd, ced.getMap(), null);
 		}
 		else if(ce instanceof EmptyEncoding) {
 			return new ColGroupEmpty(combinedColumns);
 		}
 		else if(ce instanceof ConstEncoding) {
-			ADictionary cd = DictionaryFactory.combineDictionaries(ac, bc);
+			ADictionary cd = DictionaryFactory.combineDictionaries(ac, bc, filter);
 			return ColGroupConst.create(combinedColumns, cd);
 		}
 		else if(ce instanceof SparseEncoding) {
@@ -145,12 +220,12 @@
 
 	public static double[] constructDefaultTuple(AColGroupCompressed ac, AColGroupCompressed bc) {
 		double[] ret = new double[ac.getNumCols() + bc.getNumCols()];
-		if(ac instanceof IContainDefaultTuple ){
-			double[] defa = ((IContainDefaultTuple)ac).getDefaultTuple();
+		if(ac instanceof IContainDefaultTuple) {
+			double[] defa = ((IContainDefaultTuple) ac).getDefaultTuple();
 			System.arraycopy(defa, 0, ret, 0, defa.length);
 		}
-		if(bc instanceof IContainDefaultTuple){
-			double[] defb = ((IContainDefaultTuple)bc).getDefaultTuple();
+		if(bc instanceof IContainDefaultTuple) {
+			double[] defb = ((IContainDefaultTuple) bc).getDefaultTuple();
 			System.arraycopy(defb, 0, ret, ac.getNumCols(), defb.length);
 		}
 		return ret;
diff --git a/src/main/java/org/apache/sysds/runtime/compress/lib/CLALibStack.java b/src/main/java/org/apache/sysds/runtime/compress/lib/CLALibStack.java
index adce71a..1dd7648 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/lib/CLALibStack.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/lib/CLALibStack.java
@@ -192,6 +192,7 @@
 		final AColGroup[][] finalCols = new AColGroup[clen][]; // temp array for combining
 		final int blocksInColumn = (rlen - 1) / blen + 1;
 
+
 		// Add all the blocks into linear structure.
 		for(int br = 0; br * blen < rlen; br++) {
 			for(int bc = 0; bc * blen < clen; bc++) {
@@ -208,24 +209,28 @@
 						return combineViaDecompression(m, rlen, clen, blen, k);
 					}
 					finalCols[c][br] = gc;
+					if(br != 0 && (finalCols[c][0] == null || !finalCols[c][br].getColIndices().equals(finalCols[c][0].getColIndices()))){
+						LOG.warn("Combining via decompression. There was an column with different index");
+						return combineViaDecompression(m, rlen, clen, blen, k);
+					}
 
 				}
 			}
 		}
+
+
 		final ExecutorService pool = CommonThreadPool.get(Math.max(Math.min(clen / 500, k), 1));
 		try {
 
 			List<AColGroup> finalGroups = pool.submit(() -> {
 				return Arrays//
 					.stream(finalCols)//
-					.filter(x -> x != null)//
+					.filter(x -> x != null)// filter all columns that are contained in other groups.
 					.parallel()//
 					.map(x -> {
 						return combineN(x);
 					}).collect(Collectors.toList());
 			}).get();
-
-			pool.shutdown();
 			if(finalGroups.contains(null)) {
 				LOG.warn("Combining via decompression. There was a column group that did not append ");
 				return combineViaDecompression(m, rlen, clen, blen, k);
@@ -233,12 +238,20 @@
 			return new CompressedMatrixBlock(rlen, clen, -1, false, finalGroups);
 		}
 		catch(InterruptedException | ExecutionException e) {
-			pool.shutdown();
 			throw new DMLRuntimeException("Failed to combine column groups", e);
 		}
+		finally {
+			pool.shutdown();
+		}
 	}
 
 	private static AColGroup combineN(AColGroup[] groups) {
-		return AColGroup.appendN(groups);
+		try {
+			return AColGroup.appendN(groups);
+
+		}
+		catch(Exception e) {
+			throw new DMLCompressionException("Failed to combine groups:\n" + Arrays.toString(groups), e);
+		}
 	}
 }
diff --git a/src/main/java/org/apache/sysds/runtime/compress/lib/CLALibUtils.java b/src/main/java/org/apache/sysds/runtime/compress/lib/CLALibUtils.java
index 06da63d..20262e7 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/lib/CLALibUtils.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/lib/CLALibUtils.java
@@ -34,6 +34,7 @@
 import org.apache.sysds.runtime.compress.colgroup.APreAgg;
 import org.apache.sysds.runtime.compress.colgroup.ColGroupConst;
 import org.apache.sysds.runtime.compress.colgroup.ColGroupEmpty;
+import org.apache.sysds.runtime.compress.colgroup.IFrameOfReferenceGroup;
 import org.apache.sysds.runtime.compress.colgroup.indexes.ColIndexFactory;
 import org.apache.sysds.runtime.compress.colgroup.indexes.IColIndex;
 import org.apache.sysds.runtime.compress.colgroup.indexes.IIterate;
@@ -94,6 +95,29 @@
 	}
 
 	/**
+	 * Detect if the list of groups contains FOR.
+	 * 
+	 * @param groups the groups
+	 * @return If it contains FOR.
+	 */
+	protected static boolean shouldFilterFOR(List<AColGroup> groups) {
+		for(AColGroup g : groups)
+			if(g instanceof IFrameOfReferenceGroup)
+				return true;
+		return false;
+	}
+
+	protected static List<AColGroup> filterFOR(List<AColGroup> groups, double[] constV) {
+		if(constV == null)
+			return groups;
+		final List<AColGroup> filteredGroups = new ArrayList<>();
+		for(AColGroup g : groups)
+			if(g instanceof IFrameOfReferenceGroup)
+				filteredGroups.add(((IFrameOfReferenceGroup) g).extractCommon(constV));
+		return filteredGroups;
+	}
+
+	/**
 	 * Helper method to filter out SDC Groups and remove all constant groups, to reduce computation.
 	 * 
 	 * @param groups The Column Groups
@@ -166,7 +190,7 @@
 			final double[] colVals = cg.getValues();
 			for(int i = 0; i < colIdx.size(); i++) {
 				// Find the index in the result columns to add the value into.
-				int outId = resCols.findIndex(colIdx.get(i)); 
+				int outId = resCols.findIndex(colIdx.get(i));
 				values[outId] = colVals[i];
 			}
 		}
diff --git a/src/test/java/org/apache/sysds/test/component/compress/CompressedLoggingTests.java b/src/test/java/org/apache/sysds/test/component/compress/CompressedLoggingTests.java
index 4373aa4..650ad80 100644
--- a/src/test/java/org/apache/sysds/test/component/compress/CompressedLoggingTests.java
+++ b/src/test/java/org/apache/sysds/test/component/compress/CompressedLoggingTests.java
@@ -275,7 +275,7 @@
 				if(l.getMessage().toString().contains("--colGroups type"))
 					return;
 			}
-			fail("Log did not contain Compressed Size");
+			fail("Log did not contain colgroups type");
 		}
 		catch(Exception e) {
 			e.printStackTrace();
@@ -303,7 +303,7 @@
 				if(l.getMessage().toString().contains("--colGroups type"))
 					return;
 			}
-			fail("Log did not contain Compressed Size");
+			fail("Log did not contain colgroups type");
 		}
 		catch(Exception e) {
 			e.printStackTrace();
@@ -331,7 +331,7 @@
 				if(l.getMessage().toString().contains("Empty input to compress"))
 					return;
 			}
-			fail("Log Did not contain Empty");
+			fail("Log did not contain Empty");
 		}
 		catch(Exception e) {
 			e.printStackTrace();
@@ -349,17 +349,15 @@
 
 		try {
 			Logger.getLogger(CompressedMatrixBlockFactory.class).setLevel(Level.DEBUG);
-			MatrixBlock mb = TestUtils.generateTestMatrixBlock(10, 1000, 1, 1, 0.0, 235);
-			mb = TestUtils.round(mb);
+			MatrixBlock mb = TestUtils.generateTestMatrixBlock(100, 3, 1, 1, 0.5, 235);
 			MatrixBlock m2 = CompressedMatrixBlockFactory.compress(mb).getLeft();
 			CompressedMatrixBlockFactory.compress(m2).getLeft();
 			final List<LoggingEvent> log = LoggingUtils.reinsert(appender);
 			for(LoggingEvent l : log) {
-				// LOG.error(l.getMessage());
 				if(l.getMessage().toString().contains("Recompressing"))
 					return;
 			}
-			fail("Log Did not contain Recompressing");
+			fail("Log did not contain Recompressing");
 		}
 		catch(Exception e) {
 			e.printStackTrace();
@@ -390,7 +388,7 @@
 				if(l.getMessage().toString().contains("Abort block compression"))
 					return;
 			}
-			fail("Log Did not contain Recompressing");
+			fail("Log did not contain abort block compression");
 		}
 		catch(Exception e) {
 			e.printStackTrace();
@@ -415,7 +413,7 @@
 				if(l.getMessage().toString().contains("CompressionSettings"))
 					return;
 			}
-			fail("failed to get Compressionsetting to log");
+			fail("failed to get Compression setting to log");
 		}
 		catch(Exception e) {
 			e.printStackTrace();
@@ -439,7 +437,7 @@
 				if(l.getMessage().toString().contains("Estimation Type"))
 					return;
 			}
-			fail("failed to get Compressionsetting to log");
+			fail("failed to get estimation type");
 		}
 		catch(Exception e) {
 			e.printStackTrace();