[4/7] incubator-systemml git commit: [SYSTEMML-766][SYSTEMML-810] Fix missing licenses / unix line delimiters

[mb...@apache.org]

http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/da318739/src/main/java/org/apache/sysml/runtime/compress/CompressedMatrixBlock.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/sysml/runtime/compress/CompressedMatrixBlock.java b/src/main/java/org/apache/sysml/runtime/compress/CompressedMatrixBlock.java
index 6ac26c6..f5b2058 100644
--- a/src/main/java/org/apache/sysml/runtime/compress/CompressedMatrixBlock.java
+++ b/src/main/java/org/apache/sysml/runtime/compress/CompressedMatrixBlock.java
@@ -1,1342 +1,1342 @@
-/*
- * 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.sysml.runtime.compress;
-
-import java.io.DataInput;
-import java.io.DataOutput;
-import java.io.Externalizable;
-import java.io.IOException;
-import java.io.ObjectInput;
-import java.io.ObjectOutput;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.PriorityQueue;
-import java.util.concurrent.Callable;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.log4j.Level;
-import org.apache.log4j.Logger;
-import org.apache.sysml.hops.OptimizerUtils;
-import org.apache.sysml.lops.MMTSJ.MMTSJType;
-import org.apache.sysml.lops.MapMultChain.ChainType;
-import org.apache.sysml.runtime.DMLRuntimeException;
-import org.apache.sysml.runtime.compress.ColGroup.CompressionType;
-import org.apache.sysml.runtime.compress.estim.CompressedSizeEstimator;
-import org.apache.sysml.runtime.compress.estim.CompressedSizeInfo;
-import org.apache.sysml.runtime.compress.estim.SizeEstimatorFactory;
-import org.apache.sysml.runtime.compress.utils.ConverterUtils;
-import org.apache.sysml.runtime.compress.utils.LinearAlgebraUtils;
-import org.apache.sysml.runtime.controlprogram.parfor.stat.Timing;
-import org.apache.sysml.runtime.functionobjects.KahanPlus;
-import org.apache.sysml.runtime.functionobjects.KahanPlusSq;
-import org.apache.sysml.runtime.functionobjects.Multiply;
-import org.apache.sysml.runtime.functionobjects.ReduceRow;
-import org.apache.sysml.runtime.instructions.cp.KahanObject;
-import org.apache.sysml.runtime.matrix.data.LibMatrixBincell;
-import org.apache.sysml.runtime.matrix.data.LibMatrixReorg;
-import org.apache.sysml.runtime.matrix.data.MatrixBlock;
-import org.apache.sysml.runtime.matrix.data.MatrixIndexes;
-import org.apache.sysml.runtime.matrix.data.MatrixValue;
-import org.apache.sysml.runtime.matrix.data.SparseBlock;
-import org.apache.sysml.runtime.matrix.operators.AggregateBinaryOperator;
-import org.apache.sysml.runtime.matrix.operators.AggregateUnaryOperator;
-import org.apache.sysml.runtime.matrix.operators.BinaryOperator;
-import org.apache.sysml.runtime.matrix.operators.ScalarOperator;
-
-/**
- * Experimental version of MatrixBlock that allows a compressed internal
- * representation.
- */
-public class CompressedMatrixBlock extends MatrixBlock implements Externalizable
-{
-	private static final long serialVersionUID = 7319972089143154057L;
-	
-	//internal configuration
-	public static final int MAX_NUMBER_COCODING_COLUMNS = 1000;
-	public static final double MIN_COMPRESSION_RATIO = 2.0;
-	public static final double MIN_RLE_RATIO = 1.0; // Minimum additional compression (non-RLE size / RLE size) before we switch to run-length encoding.
-	public static final boolean TRANSPOSE_INPUT = true;
-	public static final boolean MATERIALIZE_ZEROS = false;
-	public static final long MIN_PAR_AGG_THRESHOLD = 16*1024*1024; //16MB
-	public static final boolean INVESTIGATE_ESTIMATES = false;
-	private static final boolean LDEBUG = false; //local debug flag
-	
-	private static final Log LOG = LogFactory.getLog(CompressedMatrixBlock.class.getName());
-	
-	static{
-		// for internal debugging only
-		if( LDEBUG ) {
-			Logger.getLogger("org.apache.sysml.runtime.compress")
-				  .setLevel((Level) Level.DEBUG);
-		}	
-	}
-	
-	protected ArrayList<ColGroup> _colGroups = null;
-	protected CompressionStatistics _stats = null;
-	
-	public CompressedMatrixBlock() {
-		super(-1, -1, true);
-	}
-	
-	/**
-	 * Main constructor for building a block from scratch.
-	 * 
-	 * @param rl
-	 *            number of rows in the block
-	 * @param cl
-	 *            number of columns
-	 * @param sparse
-	 *            true if the UNCOMPRESSED representation of the block should be
-	 *            sparse
-	 */
-	public CompressedMatrixBlock(int rl, int cl, boolean sparse) {
-		super(rl, cl, sparse);
-	}
-
-	/**
-	 * "Copy" constructor to populate this compressed block with the
-	 * uncompressed contents of a conventional block. Does <b>not</b> compress
-	 * the block.
-	 */
-	public CompressedMatrixBlock(MatrixBlock mb) {
-		super(mb.getNumRows(), mb.getNumColumns(), mb.isInSparseFormat());
-		
-		//shallow copy (deep copy on compression, prevents unnecessary copy) 
-		if( isInSparseFormat() )
-			sparseBlock = mb.getSparseBlock();
-		else
-			denseBlock = mb.getDenseBlock();
-		nonZeros = mb.getNonZeros();
-	}
-
-	/**
-	 * 
-	 * @return the column groups constructed by the compression process.
-	 * 
-	 */
-	public ArrayList<ColGroup> getColGroups() {
-		return _colGroups;
-	}
-
-	/**
-	 * @return true if this block is in compressed form; false if the block has
-	 *         not yet been compressed
-	 */
-	public boolean isCompressed() {
-		return (_colGroups != null);
-	}
-	
-	/**
-	 * 
-	 * @return
-	 */
-	public boolean isSingleUncompressedGroup(){
-		return (_colGroups!=null && _colGroups.size()==1 
-				&& _colGroups.get(0) instanceof ColGroupUncompressed);
-	}
-
-	private void allocateColGroupList() {
-		_colGroups = new ArrayList<ColGroup>();
-	}
-	
-	@Override
-	public boolean isEmptyBlock(boolean safe)  {
-		if( !isCompressed() )
-			return super.isEmptyBlock(safe);		
-		return (_colGroups == null || getNonZeros()==0);
-	}
-	
-	/**
-	 * Compress the contents of this matrix block. After compression, the
-	 * uncompressed data is discarded. Attempts to update this block after
-	 * calling this method currently result in INCORRECT BEHAVIOR, something
-	 * which should be fixed if we move ahead with this compression strategy.
-	 * 
-	 * +per column sparsity
-	 */
-	public void compress() 
-		throws DMLRuntimeException 
-	{
-		//check for redundant compression
-		if( isCompressed() ){
-			throw new DMLRuntimeException("Redundant compression, block already compressed.");
-		}
-
-		Timing time = new Timing(true);
-		_stats = new CompressionStatistics();
-		
-		// SAMPLE-BASED DECISIONS:
-		// Decisions such as testing if a column is amenable to bitmap
-		// compression or evaluating co-coding potentionls are made based on a
-		// subset of the rows. For large datasets, sampling might take a
-		// significant amount of time. So, we generate only one sample and use
-		// it for the entire compression process.
-
-		//prepare basic meta data and deep copy / transpose input
-		final int numRows = getNumRows();
-		final int numCols = getNumColumns();
-		final boolean sparse = isInSparseFormat();
-		MatrixBlock rawblock = this;
-		if( TRANSPOSE_INPUT )
-			rawblock = LibMatrixReorg.transpose(rawblock, new MatrixBlock(numCols, numRows, sparse));
-		else
-			rawblock = new MatrixBlock(this);
-		
-		//construct sample-based size estimator
-		CompressedSizeEstimator bitmapSizeEstimator = 
-				SizeEstimatorFactory.getSizeEstimator(rawblock, numRows);
-
-		//allocate list of column groups
-		allocateColGroupList();
-
-		// The current implementation of this method is written for correctness,
-		// not for performance or for minimal use of temporary space.
-
-		// We start with a full set of columns.
-		HashSet<Integer> remainingCols = new HashSet<Integer>();
-		for (int i = 0; i < numCols; i++) {
-			remainingCols.add(i);
-		}
-
-		// PHASE 1: Classify columns by compression type
-		// We start by determining which columns are amenable to bitmap
-		// compression
-
-		// It is correct to use the dense size as the uncompressed size
-		// FIXME not numRows but nnz / col otherwise too aggressive overestimation
-		// of uncompressed size and hence overestimation of compression potential
-		double uncompressedColumnSize = 8 * numRows;
-
-		// information about the bitmap amenable columns
-		List<Integer> bitmapCols = new ArrayList<Integer>();
-		List<Integer> uncompressedCols = new ArrayList<Integer>();
-		List<Integer> colsCardinalities = new ArrayList<Integer>();
-		List<Long> compressedSizes = new ArrayList<Long>();
-		HashMap<Integer, Double> compressionRatios = new HashMap<Integer, Double>();
-		
-		// Minimum ratio (size of uncompressed / size of compressed) that we
-		// will accept when encoding a field with a bitmap.
-		for (int col = 0; col < numCols; col++) 
-		{
-			CompressedSizeInfo compressedSizeInfo = bitmapSizeEstimator
-					.estimateCompressedColGroupSize(new int[] { col });
-			long compressedSize = compressedSizeInfo.getMinSize();
-			double compRatio = uncompressedColumnSize / compressedSize;
-			
-			//FIXME: compression ratio should be checked against 1 instead of min compression
-			//ratio; I think this threshold was only required because we overestimated the 
-			//the uncompressed column size with n\alpha instead of z\alpha
-			if (compRatio >= MIN_COMPRESSION_RATIO) {
-				bitmapCols.add(col);
-				compressionRatios.put(col, compRatio);
-				colsCardinalities.add(compressedSizeInfo.getEstCarinality());
-				compressedSizes.add(compressedSize);
-			}
-			else
-				uncompressedCols.add(col);
-		}
-
-		_stats.timePhase1 = time.stop();
-		if( LOG.isDebugEnabled() )
-			LOG.debug("compression phase 1: "+_stats.timePhase1);
-		
-		// Filters for additional types of compression should be inserted here.
-
-		// PHASE 2: Grouping columns
-		// Divide the bitmap columns into column groups.
-		List<int[]> bitmapColGrps = null;
-		if (bitmapCols.size() > MAX_NUMBER_COCODING_COLUMNS) {
-			// Too many columns to compute co-coding groups with current methods.
-			// Generate singleton groups.
-			bitmapColGrps = new ArrayList<int[]>(bitmapCols.size());
-			for (int col : bitmapCols) {
-				bitmapColGrps.add(new int[] { col });
-			}
-		} 
-		else {
-			bitmapColGrps = PlanningCoCoder.findCocodesByPartitioning(
-					bitmapSizeEstimator, bitmapCols, colsCardinalities,
-					compressedSizes, numRows, isInSparseFormat() ? 
-					OptimizerUtils.getSparsity(numRows, numCols, getNonZeros()): 1);
-		}
-
-		_stats.timePhase2 = time.stop();
-		if( LOG.isDebugEnabled() )
-			LOG.debug("compression phase 2: "+_stats.timePhase2);
-		
-		if( INVESTIGATE_ESTIMATES ) {
-			double est = 0;
-			for( int[] groupIndices : bitmapColGrps )
-				est += bitmapSizeEstimator.estimateCompressedColGroupSize(groupIndices).getMinSize();
-			est += uncompressedCols.size() * uncompressedColumnSize;
-			_stats.estSize = est;
-		}
-		
-		// PHASE 3: Compress and correct sample-based decisions
-		
-		for (int[] groupIndices : bitmapColGrps) 
-		{
-			int[] allGroupIndices = null;
-			int allColsCount = groupIndices.length;
-			CompressedSizeInfo bitmapSizeInfo;
-			// The compression type is decided based on a full bitmap since it
-			// will be reused for the actual compression step.
-			UncompressedBitmap ubm;
-			PriorityQueue<CompressedColumn> compRatioPQ = null;
-			boolean skipGroup = false;
-			while (true) 
-			{
-				ubm = BitmapEncoder.extractBitmap(groupIndices, rawblock); 
-				bitmapSizeInfo = bitmapSizeEstimator
-						.estimateCompressedColGroupSize(ubm);
-				double compRatio = uncompressedColumnSize * groupIndices.length
-						/ bitmapSizeInfo.getMinSize();
-				if (compRatio >= MIN_COMPRESSION_RATIO) {
-					// we have a good group
-					for( Integer col : groupIndices )
-						remainingCols.remove(col);
-					break;
-				} else {
-					// modify the group
-					if (compRatioPQ == null) {
-						// first modification
-						allGroupIndices = Arrays.copyOf(groupIndices, groupIndices.length);
-						compRatioPQ = new PriorityQueue<CompressedMatrixBlock.CompressedColumn>();
-						for (int i = 0; i < groupIndices.length; i++)
-							compRatioPQ.add(new CompressedColumn(i,
-									compressionRatios.get(groupIndices[i])));
-					}
-
-					// index in allGroupIndices
-					int removeIx = compRatioPQ.poll().colIx;
-					allGroupIndices[removeIx] = -1;
-					allColsCount--;
-					if (allColsCount == 0) {
-						skipGroup = true;
-						break;
-					}
-					groupIndices = new int[allColsCount];
-					// copying the values that do not equal -1
-					int ix = 0;
-					for (int col : allGroupIndices) {
-						if (col != -1) {
-							groupIndices[ix++] = col;
-						}
-					}
-
-				}
-			}
-
-			if (skipGroup)
-				continue;
-			long rleNumBytes = bitmapSizeInfo.getRLESize();
-			long offsetNumBytes = bitmapSizeInfo.getOLESize();
-			double rleRatio = (double) offsetNumBytes / (double) rleNumBytes;
-
-			if (rleRatio > MIN_RLE_RATIO) {
-				ColGroupRLE compressedGroup = new ColGroupRLE(groupIndices,
-						numRows, ubm);
-				_colGroups.add(compressedGroup);
-			} 
-			else {
-				ColGroupOLE compressedGroup = new ColGroupOLE(
-						groupIndices, numRows, ubm);
-				_colGroups.add(compressedGroup);
-			}
-
-		}
-		
-		_stats.timePhase3 = time.stop();
-		if( LOG.isDebugEnabled() )
-			LOG.debug("compression phase 3: "+_stats.timePhase3);
-		
-		// Phase 4: Cleanup
-		// The remaining columns are stored uncompressed as one big column group
-		if (remainingCols.size() > 0) {
-			ArrayList<Integer> list = new ArrayList<Integer>(remainingCols);
-			ColGroupUncompressed ucgroup = new ColGroupUncompressed(list, rawblock);
-			_colGroups.add(ucgroup);
-		}
-
-		//final cleanup (discard uncompressed block)
-		rawblock.cleanupBlock(true, true);
-		this.cleanupBlock(true, true);
-		
-		_stats.timePhase4 = time.stop();
-		if( LOG.isDebugEnabled() )
-			LOG.debug("compression phase 4: "+_stats.timePhase4);
-	}
-
-	/**
-	 * @return a new uncompressed matrix block containing the contents of this
-	 *         block
-	 */
-	public MatrixBlock decompress() throws DMLRuntimeException 
-	{
-		//early abort for not yet compressed blocks
-		if( !isCompressed() )
-			return new MatrixBlock(this); 
-		
-		//preallocation sparse rows to avoid repeated reallocations		
-		MatrixBlock ret = new MatrixBlock(getNumRows(), getNumColumns(), isInSparseFormat(), getNonZeros());
-		if( ret.isInSparseFormat() ) {
-			int[] rnnz = new int[rlen];
-			for (ColGroup grp : _colGroups)
-				grp.countNonZerosPerRow(rnnz);
-			ret.allocateSparseRowsBlock();
-			SparseBlock rows = ret.getSparseBlock();
-			for( int i=0; i<rlen; i++ )
-				rows.allocate(i, rnnz[i]);
-		}
-		
-		//core decompression (append if sparse)
-		for (ColGroup grp : _colGroups)
-			grp.decompressToBlock(ret);
-		
-		//post-processing (for append in decompress)
-		if( isInSparseFormat() )
-			ret.sortSparseRows();
-
-		return ret;
-	}
-
-	public CompressionStatistics getCompressionStatistics(){
-		return _stats;
-	}
-
-	/**
-	 * 
-	 * @return an upper bound on the memory used to store this compressed block
-	 *         considering class overhead.
-	 */
-	public long estimateCompressedSizeInMemory() {
-		if (!isCompressed())
-			return 0;
-		// basic data inherited from MatrixBlock
-		long total = MatrixBlock.estimateSizeInMemory(0, 0, 0);
-		// adding the size of colGroups ArrayList overhead
-		// object overhead (32B) + int size (4B) + int modCount (4B) + Object[]
-		// elementData overhead + reference (32+8)B +reference ofr each Object (8B)
-		total += 80 + 8 * _colGroups.size();
-		for (ColGroup grp : _colGroups)
-			total += grp.estimateInMemorySize();
-		return total;
-	}
-
-	private class CompressedColumn implements Comparable<CompressedColumn> {
-		int colIx;
-		double compRatio;
-
-		public CompressedColumn(int colIx, double compRatio) {
-			this.colIx = colIx;
-			this.compRatio = compRatio;
-		}
-
-		@Override
-		public int compareTo(CompressedColumn o) {
-			return (int) Math.signum(compRatio - o.compRatio);
-		}
-	}
-	
-	public static class CompressionStatistics {
-		public double timePhase1 = -1;
-		public double timePhase2 = -1;
-		public double timePhase3 = -1;
-		public double timePhase4 = -1;
-		public double estSize = -1;
-		
-		public CompressionStatistics() {
-			//do nothing
-		}
-		
-		public CompressionStatistics(double t1, double t2, double t3, double t4){
-			timePhase1 = t1;
-			timePhase2 = t2;
-			timePhase3 = t3;
-			timePhase4 = t4;
-		}
-	} 
-
-	
-	//////////////////////////////////////////
-	// Serialization / Deserialization
-
-	@Override
-	public long getExactSizeOnDisk() 
-	{
-		//header information
-		long ret = 12;
-		
-		for( ColGroup grp : _colGroups ) {
-			ret += 1; //type info
-			ret += grp.getExactSizeOnDisk();
-		}
-		
-		return ret;
-	}
-	
-	@Override
-	public void readFields(DataInput in) 
-		throws IOException 
-	{
-		boolean compressed = in.readBoolean();
-		
-		//deserialize uncompressed block
-		if( !compressed ) {
-			super.readFields(in);
-			return;
-		}
-		
-		//deserialize compressed block
-		rlen = in.readInt();
-		clen = in.readInt();
-		nonZeros = in.readLong();
-		int ncolGroups = in.readInt();
-		
-		_colGroups = new ArrayList<ColGroup>(ncolGroups);
-		for( int i=0; i<ncolGroups; i++ ) 
-		{
-			CompressionType ctype = CompressionType.values()[in.readByte()];
-			ColGroup grp = null;
-			
-			//create instance of column group
-			switch( ctype ) {
-				case UNCOMPRESSED:
-					grp = new ColGroupUncompressed(); break;
-				case OLE_BITMAP:
-					grp = new ColGroupOLE(); break;
-				case RLE_BITMAP:
-					grp = new ColGroupRLE(); break;
-			}
-			
-			//deserialize and add column group
-			grp.readFields(in);
-			_colGroups.add(grp);
-		}
-	}
-	
-	@Override
-	public void write(DataOutput out) 
-		throws IOException 
-	{
-		out.writeBoolean( isCompressed() );
-		
-		//serialize uncompressed block
-		if( !isCompressed() ) {
-			super.write(out);
-			return;
-		}
-		
-		//serialize compressed matrix block
-		out.writeInt(rlen);
-		out.writeInt(clen);
-		out.writeLong(nonZeros);
-		out.writeInt(_colGroups.size());
-		
-		for( ColGroup grp : _colGroups ) {
-			out.writeByte( grp.getCompType().ordinal() );
-			grp.write(out); //delegate serialization
-		}
-	}
-	
-	
-	/**
-	 * Redirects the default java serialization via externalizable to our default 
-	 * hadoop writable serialization for efficient broadcast/rdd deserialization. 
-	 * 
-	 * @param is
-	 * @throws IOException
-	 */
-	@Override
-	public void readExternal(ObjectInput is) 
-		throws IOException
-	{
-		readFields(is);
-	}
-	
-	/**
-	 * Redirects the default java serialization via externalizable to our default 
-	 * hadoop writable serialization for efficient broadcast/rdd serialization. 
-	 * 
-	 * @param is
-	 * @throws IOException
-	 */
-	@Override
-	public void writeExternal(ObjectOutput os) 
-		throws IOException
-	{
-		write(os);	
-	}
-	
-	
-	//////////////////////////////////////////
-	// Operations (overwrite existing ops for seamless integration)
-
-	@Override
-	public MatrixValue scalarOperations(ScalarOperator sop, MatrixValue result) 
-		throws DMLRuntimeException
-	{
-		//call uncompressed matrix scalar if necessary
-		if( !isCompressed() ) {
-			return super.scalarOperations(sop, result);
-		}
-		
-		//allocate the output matrix block
-		CompressedMatrixBlock ret = null;
-		if( result==null || !(result instanceof CompressedMatrixBlock) )
-			ret = new CompressedMatrixBlock(getNumRows(), getNumColumns(), sparse);
-		else {
-			ret = (CompressedMatrixBlock) result;
-			ret.reset(rlen, clen);
-		}
-		
-		// Apply the operation recursively to each of the column groups.
-		// Most implementations will only modify metadata.
-		ArrayList<ColGroup> newColGroups = new ArrayList<ColGroup>();
-		for (ColGroup grp : _colGroups) {
-			newColGroups.add(grp.scalarOperation(sop));
-		}
-		ret._colGroups = newColGroups;
-		ret.setNonZeros(rlen*clen);
-		
-		return ret;
-	}
-
-	@Override
-	public MatrixBlock appendOperations(MatrixBlock that, MatrixBlock ret) 
-		throws DMLRuntimeException
-	{
-		//call uncompressed matrix append if necessary
-		if( !isCompressed() ) {
-			if( that instanceof CompressedMatrixBlock )
-				that = ((CompressedMatrixBlock) that).decompress();
-			return super.appendOperations(that, ret);
-		}
-		
-		final int m = rlen;
-		final int n = clen+that.getNumColumns();
-		final long nnz = nonZeros+that.getNonZeros();		
-		
-		//init result matrix 
-		CompressedMatrixBlock ret2 = null;
-		if( ret == null || !(ret instanceof CompressedMatrixBlock) ) {
-			ret2 = new CompressedMatrixBlock(m, n, isInSparseFormat());
-		}
-		else {
-			ret2 = (CompressedMatrixBlock) ret;
-			ret2.reset(m, n);
-		}
-			
-		//shallow copy of lhs column groups
-		ret2.allocateColGroupList();
-		ret2._colGroups.addAll(_colGroups);
-		
-		//copy of rhs column groups w/ col index shifting
-		if( !(that instanceof CompressedMatrixBlock) ) {
-			that = new CompressedMatrixBlock(that);
-			((CompressedMatrixBlock)that).compress();
-		}
-		ArrayList<ColGroup> inColGroups = ((CompressedMatrixBlock) that)._colGroups;
-		for( ColGroup group : inColGroups ) {
-			ColGroup tmp = ConverterUtils.copyColGroup(group);
-			tmp.shiftColIndices(clen);
-			ret2._colGroups.add(tmp);
-		}
-		
-		//meta data maintenance
-		ret2.setNonZeros(nnz);		
-		return ret2;
-	}
-	
-	@Override
-	public MatrixBlock chainMatrixMultOperations(MatrixBlock v, MatrixBlock w, MatrixBlock out, ChainType ctype) 
-		throws DMLRuntimeException 
-	{
-		//call uncompressed matrix mult if necessary
-		if( !isCompressed() ) {
-			return super.chainMatrixMultOperations(v, w, out, ctype);
-		}
-		
-		//single-threaded mmchain of single uncompressed colgroup
-		if( isSingleUncompressedGroup() ){
-			return ((ColGroupUncompressed)_colGroups.get(0))
-				.getData().chainMatrixMultOperations(v, w, out, ctype);
-		}
-		
-		//Timing time = new Timing(true);
-		
-		//prepare result
-		if( out != null )
-			out.reset(clen, 1, false);
-		else 
-			out = new MatrixBlock(clen, 1, false);
-		
-		//empty block handling
-		if( isEmptyBlock(false) ) 
-			return out;
-			
-		//compute matrix mult
-		MatrixBlock tmp = new MatrixBlock(rlen, 1, false);
-		rightMultByVector(v, tmp);
-		if( ctype == ChainType.XtwXv ) {
-			BinaryOperator bop = new BinaryOperator(Multiply.getMultiplyFnObject());
-			LibMatrixBincell.bincellOpInPlace(tmp, w, bop);
-		}
-		leftMultByVectorTranspose(_colGroups, tmp, out, true);
-		
-		//System.out.println("Compressed MMChain in "+time.stop());
-		
-		return out;
-	}
-
-	@Override
-	public MatrixBlock chainMatrixMultOperations(MatrixBlock v, MatrixBlock w, MatrixBlock out, ChainType ctype, int k) 
-		throws DMLRuntimeException 
-	{
-		//call uncompressed matrix mult if necessary
-		if( !isCompressed() ){
-			return super.chainMatrixMultOperations(v, w, out, ctype, k);
-		}
-		
-		//multi-threaded mmchain of single uncompressed colgroup
-		if( isSingleUncompressedGroup() ){
-			return ((ColGroupUncompressed)_colGroups.get(0))
-				.getData().chainMatrixMultOperations(v, w, out, ctype, k);
-		}
-
-		Timing time = LOG.isDebugEnabled() ? new Timing(true) : null;
-		
-		//prepare result
-		if( out != null )
-			out.reset(clen, 1, false);
-		else 
-			out = new MatrixBlock(clen, 1, false);
-		
-		//empty block handling
-		if( isEmptyBlock(false) ) 
-			return out;
-		
-		//compute matrix mult
-		MatrixBlock tmp = new MatrixBlock(rlen, 1, false);
-		rightMultByVector(v, tmp, k);
-		if( ctype == ChainType.XtwXv ) {
-			BinaryOperator bop = new BinaryOperator(Multiply.getMultiplyFnObject());
-			LibMatrixBincell.bincellOpInPlace(tmp, w, bop);
-		}
-		leftMultByVectorTranspose(_colGroups, tmp, out, true, k);
-		
-		if( LOG.isDebugEnabled() )
-			LOG.debug("Compressed MMChain k="+k+" in "+time.stop());
-		
-		return out;
-	}
-	
-	@Override
-	public MatrixValue aggregateBinaryOperations(MatrixValue mv1, MatrixValue mv2, MatrixValue result, AggregateBinaryOperator op)
-			throws DMLRuntimeException 
-	{
-		//call uncompressed matrix mult if necessary
-		if( !isCompressed() ) {
-			return super.aggregateBinaryOperations(mv1, mv2, result, op);
-		}
-	
-		//multi-threaded mm of single uncompressed colgroup
-		if( isSingleUncompressedGroup() ){
-			MatrixBlock tmp = ((ColGroupUncompressed)_colGroups.get(0)).getData();
-			return tmp.aggregateBinaryOperations(this==mv1?tmp:mv1, this==mv2?tmp:mv2, result, op);
-		}
-		
-		Timing time = LOG.isDebugEnabled() ? new Timing(true) : null;
-		
-		//setup meta data (dimensions, sparsity)
-		int rl = mv1.getNumRows();
-		int cl = mv2.getNumColumns();
-		
-		//create output matrix block
-		MatrixBlock ret = (MatrixBlock) result;
-		if( ret==null )
-			ret = new MatrixBlock(rl, cl, false, rl*cl);
-		else
-			ret.reset(rl, cl, false, rl*cl);
-		
-		//compute matrix mult
-		if( mv1.getNumRows()>1 && mv2.getNumColumns()==1 ) { //MV right
-			CompressedMatrixBlock cmb = (CompressedMatrixBlock)mv1;
-			MatrixBlock mb = (MatrixBlock) mv2;
-			if( op.getNumThreads()>1 )
-				cmb.rightMultByVector(mb, ret, op.getNumThreads());
-			else
-				cmb.rightMultByVector(mb, ret);
-		}
-		else if( mv1.getNumRows()==1 && mv2.getNumColumns()>1 ) { //MV left
-			MatrixBlock mb = (MatrixBlock) mv1;
-			if( op.getNumThreads()>1 )
-				leftMultByVectorTranspose(_colGroups, mb, ret, false, op.getNumThreads());
-			else
-				leftMultByVectorTranspose(_colGroups, mb, ret, false);
-		}
-		else {
-			//NOTE: we could decompress and invoke super.aggregateBinary but for now
-			//we want to have an eager fail if this happens
-			throw new DMLRuntimeException("Unsupported matrix-matrix multiplication over compressed matrix block.");
-		}
-		
-		if( LOG.isDebugEnabled() )
-			LOG.debug("Compressed MM in "+time.stop());
-		
-		return ret;
-	}
-	
-	@Override
-	public MatrixValue aggregateUnaryOperations(AggregateUnaryOperator op, MatrixValue result, 
-			int blockingFactorRow, int blockingFactorCol, MatrixIndexes indexesIn, boolean inCP) 
-		throws DMLRuntimeException
-	{
-		//call uncompressed matrix mult if necessary
-		if( !isCompressed() ) {
-			return super.aggregateUnaryOperations(op, result, blockingFactorRow, blockingFactorCol, indexesIn, inCP);
-		}
-		
-		//check for supported operations
-		if( !(op.aggOp.increOp.fn instanceof KahanPlus || op.aggOp.increOp.fn instanceof KahanPlusSq) ){
-			throw new DMLRuntimeException("Unary aggregates other than sums not supported yet.");
-		}
-		
-		//prepare output dimensions
-		CellIndex tempCellIndex = new CellIndex(-1,-1);
-		op.indexFn.computeDimension(rlen, clen, tempCellIndex);
-		if(op.aggOp.correctionExists) {
-			switch(op.aggOp.correctionLocation)
-			{
-				case LASTROW: tempCellIndex.row++;  break;
-				case LASTCOLUMN: tempCellIndex.column++; break;
-				case LASTTWOROWS: tempCellIndex.row+=2; break;
-				case LASTTWOCOLUMNS: tempCellIndex.column+=2; break;
-				default:
-					throw new DMLRuntimeException("unrecognized correctionLocation: "+op.aggOp.correctionLocation);	
-			}
-		}
-		
-		//prepare output
-		if(result==null)
-			result=new MatrixBlock(tempCellIndex.row, tempCellIndex.column, false);
-		else
-			result.reset(tempCellIndex.row, tempCellIndex.column, false);
-		
-		MatrixBlock ret = (MatrixBlock) result;
-		
-		//core unary aggregate
-		if(    op.getNumThreads() > 1 
-			&& getExactSizeOnDisk() > MIN_PAR_AGG_THRESHOLD ) 
-		{
-			// initialize and allocate the result
-			ret.allocateDenseBlock();
-			
-			//multi-threaded execution of all groups 
-			ArrayList<ColGroup>[] grpParts = createStaticTaskPartitioning(op.getNumThreads(), false);
-			ColGroupUncompressed uc = getUncompressedColGroup();
-			try {
-				//compute all compressed column groups
-				ExecutorService pool = Executors.newFixedThreadPool( op.getNumThreads() );
-				ArrayList<UnaryAggregateTask> tasks = new ArrayList<UnaryAggregateTask>();
-				for( ArrayList<ColGroup> grp : grpParts )
-					tasks.add(new UnaryAggregateTask(grp, ret, op));
-				pool.invokeAll(tasks);	
-				pool.shutdown();
-				//compute uncompressed column group in parallel (otherwise bottleneck)
-				if( uc != null )
-					 ret = (MatrixBlock)uc.getData().aggregateUnaryOperations(op, ret, blockingFactorRow, blockingFactorCol, indexesIn, false);					
-				//aggregate partial results
-				if( !(op.indexFn instanceof ReduceRow) ){
-					KahanObject kbuff = new KahanObject(0,0);
-					KahanPlus kplus = KahanPlus.getKahanPlusFnObject();
-					for( int i=0; i<ret.getNumRows(); i++ ) {
-						kbuff.set(ret.quickGetValue(i, 0), ret.quickGetValue(i, 0));
-						for( UnaryAggregateTask task : tasks )
-							kplus.execute2(kbuff, task.getResult().quickGetValue(i, 0));
-						ret.quickSetValue(i, 0, kbuff._sum);
-						ret.quickSetValue(i, 1, kbuff._correction);
-					}
-				}		
-			}
-			catch(Exception ex) {
-				throw new DMLRuntimeException(ex);
-			}
-		}
-		else {
-			for (ColGroup grp : _colGroups) {
-				grp.unaryAggregateOperations(op, ret);
-			}
-		}
-		
-		//drop correction if necessary
-		if(op.aggOp.correctionExists && inCP)
-			ret.dropLastRowsOrColums(op.aggOp.correctionLocation);
-	
-		//post-processing
-		ret.recomputeNonZeros();
-		
-		return ret;
-	}
-	
-	@Override
-	public MatrixBlock transposeSelfMatrixMultOperations(MatrixBlock out, MMTSJType tstype) 
-		throws DMLRuntimeException 
-	{
-		//call uncompressed matrix mult if necessary
-		if( !isCompressed() ) {
-			return super.transposeSelfMatrixMultOperations(out, tstype);
-		}
-				
-		//single-threaded tsmm of single uncompressed colgroup
-		if( isSingleUncompressedGroup() ){
-			return ((ColGroupUncompressed)_colGroups.get(0))
-				.getData().transposeSelfMatrixMultOperations(out, tstype);
-		}
-
-		Timing time = LOG.isDebugEnabled() ? new Timing(true) : null;
-
-		//check for transpose type
-		if( tstype != MMTSJType.LEFT ) //right not supported yet
-			throw new DMLRuntimeException("Invalid MMTSJ type '"+tstype.toString()+"'.");
-		
-		//create output matrix block
-		if( out == null )
-			out = new MatrixBlock(clen, clen, false);
-		else
-			out.reset(clen, clen, false);
-		out.allocateDenseBlock();
-		
-		if( !isEmptyBlock(false) ) {
-			//compute matrix mult
-			leftMultByTransposeSelf(_colGroups, out, 0, _colGroups.size());
-			
-			// post-processing
-			out.recomputeNonZeros();
-		}
-		
-		if( LOG.isDebugEnabled() )
-			LOG.debug("Compressed TSMM in "+time.stop());
-		
-		return out;
-	}
-
-	
-	@Override
-	public MatrixBlock transposeSelfMatrixMultOperations(MatrixBlock out, MMTSJType tstype, int k) 
-		throws DMLRuntimeException 
-	{
-		//call uncompressed matrix mult if necessary
-		if( !isCompressed() ){
-			return super.transposeSelfMatrixMultOperations(out, tstype, k);
-		}
-		
-		//multi-threaded tsmm of single uncompressed colgroup
-		if( isSingleUncompressedGroup() ){
-			return ((ColGroupUncompressed)_colGroups.get(0))
-				.getData().transposeSelfMatrixMultOperations(out, tstype, k);
-		}
-		
-		Timing time = LOG.isDebugEnabled() ? new Timing(true) : null;
-		
-		//check for transpose type
-		if( tstype != MMTSJType.LEFT ) //right not supported yet
-			throw new DMLRuntimeException("Invalid MMTSJ type '"+tstype.toString()+"'.");
-		
-		//create output matrix block
-		if( out == null )
-			out = new MatrixBlock(clen, clen, false);
-		else
-			out.reset(clen, clen, false);
-		out.allocateDenseBlock();
-		
-		if( !isEmptyBlock(false) ) {
-			//compute matrix mult
-			try {
-				ExecutorService pool = Executors.newFixedThreadPool( k );
-				ArrayList<MatrixMultTransposeTask> tasks = new ArrayList<MatrixMultTransposeTask>();
-				int blklen = (int)(Math.ceil((double)clen/(2*k)));
-				for( int i=0; i<2*k & i*blklen<clen; i++ )
-					tasks.add(new MatrixMultTransposeTask(_colGroups, out, i*blklen, Math.min((i+1)*blklen, clen)));
-				List<Future<Object>> ret = pool.invokeAll(tasks);
-				for( Future<Object> tret : ret )
-					tret.get(); //check for errors
-				pool.shutdown();
-			}
-			catch(Exception ex) {
-				throw new DMLRuntimeException(ex);
-			}
-			
-			// post-processing
-			out.recomputeNonZeros();
-		}
-		
-		if( LOG.isDebugEnabled() )
-			LOG.debug("Compressed TSMM k="+k+" in "+time.stop());
-		
-		return out;
-	}
-
-	
-	/**
-	 * Multiply this matrix block by a column vector on the right.
-	 * 
-	 * @param vector
-	 *            right-hand operand of the multiplication
-	 * @param result
-	 *            buffer to hold the result; must have the appropriate size
-	 *            already
-	 */
-	private void rightMultByVector(MatrixBlock vector, MatrixBlock result)
-		throws DMLRuntimeException 
-	{
-		// initialize and allocate the result
-		result.allocateDenseBlock();
-
-		// delegate matrix-vector operation to each column group
-		for( ColGroup grp : _colGroups )
-			if( grp instanceof ColGroupUncompressed ) //overwrites output
-				grp.rightMultByVector(vector, result, 0, result.getNumRows());
-		for( ColGroup grp : _colGroups )
-			if( !(grp instanceof ColGroupUncompressed) ) //adds to output
-				grp.rightMultByVector(vector, result, 0, result.getNumRows());
-		
-		// post-processing
-		result.recomputeNonZeros();
-	}
-
-	/**
-	 * Multi-threaded version of rightMultByVector.
-	 * 
-	 * @param vector
-	 * @param result
-	 * @param k
-	 * @throws DMLRuntimeException
-	 */
-	private void rightMultByVector(MatrixBlock vector, MatrixBlock result, int k)
-		throws DMLRuntimeException 
-	{
-		// initialize and allocate the result
-		result.allocateDenseBlock();
-
-		//multi-threaded execution of all groups
-		try {
-			ExecutorService pool = Executors.newFixedThreadPool( k );
-			int rlen = getNumRows();
-			int seqsz = BitmapEncoder.BITMAP_BLOCK_SZ;
-			int blklen = (int)(Math.ceil((double)rlen/k));
-			blklen += (blklen%seqsz != 0)?seqsz-blklen%seqsz:0;
-			ArrayList<RightMatrixMultTask> tasks = new ArrayList<RightMatrixMultTask>();
-			for( int i=0; i<k & i*blklen<getNumRows(); i++ )
-				tasks.add(new RightMatrixMultTask(_colGroups, vector, result, i*blklen, Math.min((i+1)*blklen,rlen)));
-			pool.invokeAll(tasks);	
-			pool.shutdown();
-		}
-		catch(Exception ex) {
-			throw new DMLRuntimeException(ex);
-		}
-		
-		// post-processing
-		result.recomputeNonZeros();
-	}
-	
-	/**
-	 * Multiply this matrix block by the transpose of a column vector (i.e.
-	 * t(v)%*%X)
-	 * 
-	 * @param vector
-	 *            left-hand operand of the multiplication
-	 * @param result
-	 *            buffer to hold the result; must have the appropriate size
-	 *            already
-	 */
-	private static void leftMultByVectorTranspose(List<ColGroup> colGroups, MatrixBlock vector, MatrixBlock result, boolean doTranspose) 
-		throws DMLRuntimeException 
-	{
-		//transpose vector if required
-		MatrixBlock rowVector = vector;
-		if (doTranspose) {
-			rowVector = new MatrixBlock(1, vector.getNumRows(), false);
-			LibMatrixReorg.transpose(vector, rowVector);
-		}
-		
-		// initialize and allocate the result
-		result.reset();
-		result.allocateDenseBlock();
-		
-		// delegate matrix-vector operation to each column group
-		for (ColGroup grp : colGroups) {			
-			grp.leftMultByRowVector(rowVector, result);
-		}
-
-		// post-processing
-		result.recomputeNonZeros();
-	}
-	
-	/**
-	 * Multi-thread version of leftMultByVectorTranspose.
-	 * 
-	 * @param vector
-	 * @param result
-	 * @param doTranspose
-	 * @param k
-	 * @throws DMLRuntimeException
-	 */
-	private static void leftMultByVectorTranspose(List<ColGroup> colGroups,MatrixBlock vector, MatrixBlock result, boolean doTranspose, int k) 
-		throws DMLRuntimeException 
-	{
-		int kuc = Math.max(1, k - colGroups.size() + 1);
-		
-		//transpose vector if required
-		MatrixBlock rowVector = vector;
-		if (doTranspose) {
-			rowVector = new MatrixBlock(1, vector.getNumRows(), false);
-			LibMatrixReorg.transpose(vector, rowVector);
-		}
-		
-		// initialize and allocate the result
-		result.reset();
-		result.allocateDenseBlock();
-
-		//multi-threaded execution
-		try {
-			ExecutorService pool = Executors.newFixedThreadPool( Math.min(colGroups.size(), k) );
-			ArrayList<LeftMatrixMultTask> tasks = new ArrayList<LeftMatrixMultTask>();
-			for( ColGroup grp : colGroups )
-				tasks.add(new LeftMatrixMultTask(grp, rowVector, result, kuc));
-			pool.invokeAll(tasks);	
-			pool.shutdown();
-		}
-		catch(Exception ex) {
-			throw new DMLRuntimeException(ex);
-		}
-
-		// post-processing
-		result.recomputeNonZeros();
-	}
-
-	/**
-	 * 
-	 * @param result
-	 * @throws DMLRuntimeException
-	 */
-	private static void leftMultByTransposeSelf(ArrayList<ColGroup> groups, MatrixBlock result, int cl, int cu)
-		throws DMLRuntimeException 
-	{
-		final int numRows = groups.get(0).getNumRows();
-		final int numGroups = groups.size();		
-		
-		//preallocated dense matrix block
-		MatrixBlock lhs = new MatrixBlock(numRows, 1, false);
-		lhs.allocateDenseBlock();
-		
-		//approach: for each colgroup, extract uncompressed columns one at-a-time
-		//vector-matrix multiplies against remaining col groups
-		for( int i=cl; i<cu; i++ ) 
-		{
-			//get current group and relevant col groups
-			ColGroup group = groups.get(i);	
-			int[] ixgroup = group.getColIndices();
-			List<ColGroup> tmpList = groups.subList(i, numGroups);
-			
-			//for all uncompressed lhs columns vectors
-			for( int j=0; j<ixgroup.length; j++ ) {
-				//decompress single column
-				lhs.reset(numRows, 1, false);				
-				group.decompressToBlock(lhs, j);
-				
-				if( !lhs.isEmptyBlock(false) ) {
-					//compute vector-matrix partial result
-					MatrixBlock tmpret = new MatrixBlock(1,result.getNumColumns(),false);
-					leftMultByVectorTranspose(tmpList, lhs, tmpret, true);								
-					
-					//write partial results (disjoint non-zeros)
-					LinearAlgebraUtils.copyNonZerosToRowCol(result, tmpret, ixgroup[j]);	
-				}
-			}
-		}
-	}
-	
-	/**
-	 * 
-	 * @param k
-	 * @return
-	 */
-	@SuppressWarnings("unchecked")
-	private ArrayList<ColGroup>[] createStaticTaskPartitioning(int k, boolean inclUncompressed)
-	{
-		// special case: single uncompressed col group
-		if( _colGroups.size()==1 && _colGroups.get(0) instanceof ColGroupUncompressed ){
-			return new ArrayList[0];
-		}
-		
-		// initialize round robin col group distribution
-		// (static task partitioning to reduce mem requirements/final agg)
-		int numTasks = Math.min(k, _colGroups.size());
-		ArrayList<ColGroup>[] grpParts = new ArrayList[numTasks];
-		int pos = 0;
-		for( ColGroup grp : _colGroups ){
-			if( grpParts[pos]==null )
-				grpParts[pos] = new ArrayList<ColGroup>();
-			if( inclUncompressed || !(grp instanceof ColGroupUncompressed) ) {
-				grpParts[pos].add(grp);
-				pos = (pos==numTasks-1) ? 0 : pos+1;
-			}
-		}
-		
-		return grpParts;
-	}
-	
-	/**
-	 * 
-	 * @return
-	 */
-	private ColGroupUncompressed getUncompressedColGroup()
-	{
-		for( ColGroup grp : _colGroups )
-			if( grp instanceof ColGroupUncompressed ) 
-				return (ColGroupUncompressed)grp;
-		
-		return null;
-	}
-	
-	/**
-	 * 
-	 */
-	private static class LeftMatrixMultTask implements Callable<Object> 
-	{
-		private ColGroup _group = null;
-		private MatrixBlock _vect = null;
-		private MatrixBlock _ret = null;
-		private int _kuc = 1;
-		
-		protected LeftMatrixMultTask( ColGroup group, MatrixBlock vect, MatrixBlock ret, int kuc)  {
-			_group = group;
-			_vect = vect;
-			_ret = ret;
-			_kuc = kuc;
-		}
-		
-		@Override
-		public Object call() throws DMLRuntimeException 
-		{
-			// delegate matrix-vector operation to each column group
-			if( _group instanceof ColGroupUncompressed && _kuc >1 && ColGroupBitmap.LOW_LEVEL_OPT )
-				((ColGroupUncompressed)_group).leftMultByRowVector(_vect, _ret, _kuc);
-			else
-				_group.leftMultByRowVector(_vect, _ret);
-			return null;
-		}
-	}
-	
-	/**
-	 * 
-	 */
-	private static class RightMatrixMultTask implements Callable<Object> 
-	{
-		private ArrayList<ColGroup> _groups = null;
-		private MatrixBlock _vect = null;
-		private MatrixBlock _ret = null;
-		private int _rl = -1;
-		private int _ru = -1;
-		
-		protected RightMatrixMultTask( ArrayList<ColGroup> groups, MatrixBlock vect, MatrixBlock ret, int rl, int ru)  {
-			_groups = groups;
-			_vect = vect;
-			_ret = ret;
-			_rl = rl;
-			_ru = ru;
-		}
-		
-		@Override
-		public Object call() throws DMLRuntimeException 
-		{
-			// delegate vector-matrix operation to each column group
-			for( ColGroup grp : _groups )
-				if( grp instanceof ColGroupUncompressed ) //overwrites output
-					grp.rightMultByVector(_vect, _ret, _rl, _ru);
-			for( ColGroup grp : _groups )
-				if( !(grp instanceof ColGroupUncompressed) ) //adds to output
-					grp.rightMultByVector(_vect, _ret, _rl, _ru);
-			return null;
-		}
-	}
-	
-	private static class MatrixMultTransposeTask implements Callable<Object> 
-	{
-		private ArrayList<ColGroup> _groups = null;
-		private MatrixBlock _ret = null;
-		private int _cl = -1;
-		private int _cu = -1;
-		
-		protected MatrixMultTransposeTask(ArrayList<ColGroup> groups, MatrixBlock ret, int cl, int cu)  {
-			_groups = groups;
-			_ret = ret;
-			_cl = cl;
-			_cu = cu;
-		}
-		
-		@Override
-		public Object call() throws DMLRuntimeException {
-			leftMultByTransposeSelf(_groups, _ret, _cl, _cu);
-			return null;
-		}
-	}
-	
-	private static class UnaryAggregateTask implements Callable<Object> 
-	{
-		private ArrayList<ColGroup> _groups = null;
-		private MatrixBlock _ret = null;
-		private AggregateUnaryOperator _op = null;
-		
-		protected UnaryAggregateTask( ArrayList<ColGroup> groups, MatrixBlock ret, AggregateUnaryOperator op)  {
-			_groups = groups;
-			_op = op;
-			
-			if( !(_op.indexFn instanceof ReduceRow) ) { //sum/rowSums
-				_ret = new MatrixBlock(ret.getNumRows(), ret.getNumColumns(), false);
-				_ret.allocateDenseBlock();
-			}
-			else { //colSums
-				_ret = ret;
-			}
-		}
-		
-		@Override
-		public Object call() throws DMLRuntimeException 
-		{
-			// delegate vector-matrix operation to each column group
-			for( ColGroup grp : _groups )
-				grp.unaryAggregateOperations(_op, _ret);
-			return null;
-		}
-		
-		public MatrixBlock getResult(){
-			return _ret;
-		}
-	}
-}
+/*
+ * 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.sysml.runtime.compress;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.Externalizable;
+import java.io.IOException;
+import java.io.ObjectInput;
+import java.io.ObjectOutput;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.PriorityQueue;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.log4j.Level;
+import org.apache.log4j.Logger;
+import org.apache.sysml.hops.OptimizerUtils;
+import org.apache.sysml.lops.MMTSJ.MMTSJType;
+import org.apache.sysml.lops.MapMultChain.ChainType;
+import org.apache.sysml.runtime.DMLRuntimeException;
+import org.apache.sysml.runtime.compress.ColGroup.CompressionType;
+import org.apache.sysml.runtime.compress.estim.CompressedSizeEstimator;
+import org.apache.sysml.runtime.compress.estim.CompressedSizeInfo;
+import org.apache.sysml.runtime.compress.estim.SizeEstimatorFactory;
+import org.apache.sysml.runtime.compress.utils.ConverterUtils;
+import org.apache.sysml.runtime.compress.utils.LinearAlgebraUtils;
+import org.apache.sysml.runtime.controlprogram.parfor.stat.Timing;
+import org.apache.sysml.runtime.functionobjects.KahanPlus;
+import org.apache.sysml.runtime.functionobjects.KahanPlusSq;
+import org.apache.sysml.runtime.functionobjects.Multiply;
+import org.apache.sysml.runtime.functionobjects.ReduceRow;
+import org.apache.sysml.runtime.instructions.cp.KahanObject;
+import org.apache.sysml.runtime.matrix.data.LibMatrixBincell;
+import org.apache.sysml.runtime.matrix.data.LibMatrixReorg;
+import org.apache.sysml.runtime.matrix.data.MatrixBlock;
+import org.apache.sysml.runtime.matrix.data.MatrixIndexes;
+import org.apache.sysml.runtime.matrix.data.MatrixValue;
+import org.apache.sysml.runtime.matrix.data.SparseBlock;
+import org.apache.sysml.runtime.matrix.operators.AggregateBinaryOperator;
+import org.apache.sysml.runtime.matrix.operators.AggregateUnaryOperator;
+import org.apache.sysml.runtime.matrix.operators.BinaryOperator;
+import org.apache.sysml.runtime.matrix.operators.ScalarOperator;
+
+/**
+ * Experimental version of MatrixBlock that allows a compressed internal
+ * representation.
+ */
+public class CompressedMatrixBlock extends MatrixBlock implements Externalizable
+{
+	private static final long serialVersionUID = 7319972089143154057L;
+	
+	//internal configuration
+	public static final int MAX_NUMBER_COCODING_COLUMNS = 1000;
+	public static final double MIN_COMPRESSION_RATIO = 2.0;
+	public static final double MIN_RLE_RATIO = 1.0; // Minimum additional compression (non-RLE size / RLE size) before we switch to run-length encoding.
+	public static final boolean TRANSPOSE_INPUT = true;
+	public static final boolean MATERIALIZE_ZEROS = false;
+	public static final long MIN_PAR_AGG_THRESHOLD = 16*1024*1024; //16MB
+	public static final boolean INVESTIGATE_ESTIMATES = false;
+	private static final boolean LDEBUG = false; //local debug flag
+	
+	private static final Log LOG = LogFactory.getLog(CompressedMatrixBlock.class.getName());
+	
+	static{
+		// for internal debugging only
+		if( LDEBUG ) {
+			Logger.getLogger("org.apache.sysml.runtime.compress")
+				  .setLevel((Level) Level.DEBUG);
+		}	
+	}
+	
+	protected ArrayList<ColGroup> _colGroups = null;
+	protected CompressionStatistics _stats = null;
+	
+	public CompressedMatrixBlock() {
+		super(-1, -1, true);
+	}
+	
+	/**
+	 * Main constructor for building a block from scratch.
+	 * 
+	 * @param rl
+	 *            number of rows in the block
+	 * @param cl
+	 *            number of columns
+	 * @param sparse
+	 *            true if the UNCOMPRESSED representation of the block should be
+	 *            sparse
+	 */
+	public CompressedMatrixBlock(int rl, int cl, boolean sparse) {
+		super(rl, cl, sparse);
+	}
+
+	/**
+	 * "Copy" constructor to populate this compressed block with the
+	 * uncompressed contents of a conventional block. Does <b>not</b> compress
+	 * the block.
+	 */
+	public CompressedMatrixBlock(MatrixBlock mb) {
+		super(mb.getNumRows(), mb.getNumColumns(), mb.isInSparseFormat());
+		
+		//shallow copy (deep copy on compression, prevents unnecessary copy) 
+		if( isInSparseFormat() )
+			sparseBlock = mb.getSparseBlock();
+		else
+			denseBlock = mb.getDenseBlock();
+		nonZeros = mb.getNonZeros();
+	}
+
+	/**
+	 * 
+	 * @return the column groups constructed by the compression process.
+	 * 
+	 */
+	public ArrayList<ColGroup> getColGroups() {
+		return _colGroups;
+	}
+
+	/**
+	 * @return true if this block is in compressed form; false if the block has
+	 *         not yet been compressed
+	 */
+	public boolean isCompressed() {
+		return (_colGroups != null);
+	}
+	
+	/**
+	 * 
+	 * @return
+	 */
+	public boolean isSingleUncompressedGroup(){
+		return (_colGroups!=null && _colGroups.size()==1 
+				&& _colGroups.get(0) instanceof ColGroupUncompressed);
+	}
+
+	private void allocateColGroupList() {
+		_colGroups = new ArrayList<ColGroup>();
+	}
+	
+	@Override
+	public boolean isEmptyBlock(boolean safe)  {
+		if( !isCompressed() )
+			return super.isEmptyBlock(safe);		
+		return (_colGroups == null || getNonZeros()==0);
+	}
+	
+	/**
+	 * Compress the contents of this matrix block. After compression, the
+	 * uncompressed data is discarded. Attempts to update this block after
+	 * calling this method currently result in INCORRECT BEHAVIOR, something
+	 * which should be fixed if we move ahead with this compression strategy.
+	 * 
+	 * +per column sparsity
+	 */
+	public void compress() 
+		throws DMLRuntimeException 
+	{
+		//check for redundant compression
+		if( isCompressed() ){
+			throw new DMLRuntimeException("Redundant compression, block already compressed.");
+		}
+
+		Timing time = new Timing(true);
+		_stats = new CompressionStatistics();
+		
+		// SAMPLE-BASED DECISIONS:
+		// Decisions such as testing if a column is amenable to bitmap
+		// compression or evaluating co-coding potentionls are made based on a
+		// subset of the rows. For large datasets, sampling might take a
+		// significant amount of time. So, we generate only one sample and use
+		// it for the entire compression process.
+
+		//prepare basic meta data and deep copy / transpose input
+		final int numRows = getNumRows();
+		final int numCols = getNumColumns();
+		final boolean sparse = isInSparseFormat();
+		MatrixBlock rawblock = this;
+		if( TRANSPOSE_INPUT )
+			rawblock = LibMatrixReorg.transpose(rawblock, new MatrixBlock(numCols, numRows, sparse));
+		else
+			rawblock = new MatrixBlock(this);
+		
+		//construct sample-based size estimator
+		CompressedSizeEstimator bitmapSizeEstimator = 
+				SizeEstimatorFactory.getSizeEstimator(rawblock, numRows);
+
+		//allocate list of column groups
+		allocateColGroupList();
+
+		// The current implementation of this method is written for correctness,
+		// not for performance or for minimal use of temporary space.
+
+		// We start with a full set of columns.
+		HashSet<Integer> remainingCols = new HashSet<Integer>();
+		for (int i = 0; i < numCols; i++) {
+			remainingCols.add(i);
+		}
+
+		// PHASE 1: Classify columns by compression type
+		// We start by determining which columns are amenable to bitmap
+		// compression
+
+		// It is correct to use the dense size as the uncompressed size
+		// FIXME not numRows but nnz / col otherwise too aggressive overestimation
+		// of uncompressed size and hence overestimation of compression potential
+		double uncompressedColumnSize = 8 * numRows;
+
+		// information about the bitmap amenable columns
+		List<Integer> bitmapCols = new ArrayList<Integer>();
+		List<Integer> uncompressedCols = new ArrayList<Integer>();
+		List<Integer> colsCardinalities = new ArrayList<Integer>();
+		List<Long> compressedSizes = new ArrayList<Long>();
+		HashMap<Integer, Double> compressionRatios = new HashMap<Integer, Double>();
+		
+		// Minimum ratio (size of uncompressed / size of compressed) that we
+		// will accept when encoding a field with a bitmap.
+		for (int col = 0; col < numCols; col++) 
+		{
+			CompressedSizeInfo compressedSizeInfo = bitmapSizeEstimator
+					.estimateCompressedColGroupSize(new int[] { col });
+			long compressedSize = compressedSizeInfo.getMinSize();
+			double compRatio = uncompressedColumnSize / compressedSize;
+			
+			//FIXME: compression ratio should be checked against 1 instead of min compression
+			//ratio; I think this threshold was only required because we overestimated the 
+			//the uncompressed column size with n\alpha instead of z\alpha
+			if (compRatio >= MIN_COMPRESSION_RATIO) {
+				bitmapCols.add(col);
+				compressionRatios.put(col, compRatio);
+				colsCardinalities.add(compressedSizeInfo.getEstCarinality());
+				compressedSizes.add(compressedSize);
+			}
+			else
+				uncompressedCols.add(col);
+		}
+
+		_stats.timePhase1 = time.stop();
+		if( LOG.isDebugEnabled() )
+			LOG.debug("compression phase 1: "+_stats.timePhase1);
+		
+		// Filters for additional types of compression should be inserted here.
+
+		// PHASE 2: Grouping columns
+		// Divide the bitmap columns into column groups.
+		List<int[]> bitmapColGrps = null;
+		if (bitmapCols.size() > MAX_NUMBER_COCODING_COLUMNS) {
+			// Too many columns to compute co-coding groups with current methods.
+			// Generate singleton groups.
+			bitmapColGrps = new ArrayList<int[]>(bitmapCols.size());
+			for (int col : bitmapCols) {
+				bitmapColGrps.add(new int[] { col });
+			}
+		} 
+		else {
+			bitmapColGrps = PlanningCoCoder.findCocodesByPartitioning(
+					bitmapSizeEstimator, bitmapCols, colsCardinalities,
+					compressedSizes, numRows, isInSparseFormat() ? 
+					OptimizerUtils.getSparsity(numRows, numCols, getNonZeros()): 1);
+		}
+
+		_stats.timePhase2 = time.stop();
+		if( LOG.isDebugEnabled() )
+			LOG.debug("compression phase 2: "+_stats.timePhase2);
+		
+		if( INVESTIGATE_ESTIMATES ) {
+			double est = 0;
+			for( int[] groupIndices : bitmapColGrps )
+				est += bitmapSizeEstimator.estimateCompressedColGroupSize(groupIndices).getMinSize();
+			est += uncompressedCols.size() * uncompressedColumnSize;
+			_stats.estSize = est;
+		}
+		
+		// PHASE 3: Compress and correct sample-based decisions
+		
+		for (int[] groupIndices : bitmapColGrps) 
+		{
+			int[] allGroupIndices = null;
+			int allColsCount = groupIndices.length;
+			CompressedSizeInfo bitmapSizeInfo;
+			// The compression type is decided based on a full bitmap since it
+			// will be reused for the actual compression step.
+			UncompressedBitmap ubm;
+			PriorityQueue<CompressedColumn> compRatioPQ = null;
+			boolean skipGroup = false;
+			while (true) 
+			{
+				ubm = BitmapEncoder.extractBitmap(groupIndices, rawblock); 
+				bitmapSizeInfo = bitmapSizeEstimator
+						.estimateCompressedColGroupSize(ubm);
+				double compRatio = uncompressedColumnSize * groupIndices.length
+						/ bitmapSizeInfo.getMinSize();
+				if (compRatio >= MIN_COMPRESSION_RATIO) {
+					// we have a good group
+					for( Integer col : groupIndices )
+						remainingCols.remove(col);
+					break;
+				} else {
+					// modify the group
+					if (compRatioPQ == null) {
+						// first modification
+						allGroupIndices = Arrays.copyOf(groupIndices, groupIndices.length);
+						compRatioPQ = new PriorityQueue<CompressedMatrixBlock.CompressedColumn>();
+						for (int i = 0; i < groupIndices.length; i++)
+							compRatioPQ.add(new CompressedColumn(i,
+									compressionRatios.get(groupIndices[i])));
+					}
+
+					// index in allGroupIndices
+					int removeIx = compRatioPQ.poll().colIx;
+					allGroupIndices[removeIx] = -1;
+					allColsCount--;
+					if (allColsCount == 0) {
+						skipGroup = true;
+						break;
+					}
+					groupIndices = new int[allColsCount];
+					// copying the values that do not equal -1
+					int ix = 0;
+					for (int col : allGroupIndices) {
+						if (col != -1) {
+							groupIndices[ix++] = col;
+						}
+					}
+
+				}
+			}
+
+			if (skipGroup)
+				continue;
+			long rleNumBytes = bitmapSizeInfo.getRLESize();
+			long offsetNumBytes = bitmapSizeInfo.getOLESize();
+			double rleRatio = (double) offsetNumBytes / (double) rleNumBytes;
+
+			if (rleRatio > MIN_RLE_RATIO) {
+				ColGroupRLE compressedGroup = new ColGroupRLE(groupIndices,
+						numRows, ubm);
+				_colGroups.add(compressedGroup);
+			} 
+			else {
+				ColGroupOLE compressedGroup = new ColGroupOLE(
+						groupIndices, numRows, ubm);
+				_colGroups.add(compressedGroup);
+			}
+
+		}
+		
+		_stats.timePhase3 = time.stop();
+		if( LOG.isDebugEnabled() )
+			LOG.debug("compression phase 3: "+_stats.timePhase3);
+		
+		// Phase 4: Cleanup
+		// The remaining columns are stored uncompressed as one big column group
+		if (remainingCols.size() > 0) {
+			ArrayList<Integer> list = new ArrayList<Integer>(remainingCols);
+			ColGroupUncompressed ucgroup = new ColGroupUncompressed(list, rawblock);
+			_colGroups.add(ucgroup);
+		}
+
+		//final cleanup (discard uncompressed block)
+		rawblock.cleanupBlock(true, true);
+		this.cleanupBlock(true, true);
+		
+		_stats.timePhase4 = time.stop();
+		if( LOG.isDebugEnabled() )
+			LOG.debug("compression phase 4: "+_stats.timePhase4);
+	}
+
+	/**
+	 * @return a new uncompressed matrix block containing the contents of this
+	 *         block
+	 */
+	public MatrixBlock decompress() throws DMLRuntimeException 
+	{
+		//early abort for not yet compressed blocks
+		if( !isCompressed() )
+			return new MatrixBlock(this); 
+		
+		//preallocation sparse rows to avoid repeated reallocations		
+		MatrixBlock ret = new MatrixBlock(getNumRows(), getNumColumns(), isInSparseFormat(), getNonZeros());
+		if( ret.isInSparseFormat() ) {
+			int[] rnnz = new int[rlen];
+			for (ColGroup grp : _colGroups)
+				grp.countNonZerosPerRow(rnnz);
+			ret.allocateSparseRowsBlock();
+			SparseBlock rows = ret.getSparseBlock();
+			for( int i=0; i<rlen; i++ )
+				rows.allocate(i, rnnz[i]);
+		}
+		
+		//core decompression (append if sparse)
+		for (ColGroup grp : _colGroups)
+			grp.decompressToBlock(ret);
+		
+		//post-processing (for append in decompress)
+		if( isInSparseFormat() )
+			ret.sortSparseRows();
+
+		return ret;
+	}
+
+	public CompressionStatistics getCompressionStatistics(){
+		return _stats;
+	}
+
+	/**
+	 * 
+	 * @return an upper bound on the memory used to store this compressed block
+	 *         considering class overhead.
+	 */
+	public long estimateCompressedSizeInMemory() {
+		if (!isCompressed())
+			return 0;
+		// basic data inherited from MatrixBlock
+		long total = MatrixBlock.estimateSizeInMemory(0, 0, 0);
+		// adding the size of colGroups ArrayList overhead
+		// object overhead (32B) + int size (4B) + int modCount (4B) + Object[]
+		// elementData overhead + reference (32+8)B +reference ofr each Object (8B)
+		total += 80 + 8 * _colGroups.size();
+		for (ColGroup grp : _colGroups)
+			total += grp.estimateInMemorySize();
+		return total;
+	}
+
+	private class CompressedColumn implements Comparable<CompressedColumn> {
+		int colIx;
+		double compRatio;
+
+		public CompressedColumn(int colIx, double compRatio) {
+			this.colIx = colIx;
+			this.compRatio = compRatio;
+		}
+
+		@Override
+		public int compareTo(CompressedColumn o) {
+			return (int) Math.signum(compRatio - o.compRatio);
+		}
+	}
+	
+	public static class CompressionStatistics {
+		public double timePhase1 = -1;
+		public double timePhase2 = -1;
+		public double timePhase3 = -1;
+		public double timePhase4 = -1;
+		public double estSize = -1;
+		
+		public CompressionStatistics() {
+			//do nothing
+		}
+		
+		public CompressionStatistics(double t1, double t2, double t3, double t4){
+			timePhase1 = t1;
+			timePhase2 = t2;
+			timePhase3 = t3;
+			timePhase4 = t4;
+		}
+	} 
+
+	
+	//////////////////////////////////////////
+	// Serialization / Deserialization
+
+	@Override
+	public long getExactSizeOnDisk() 
+	{
+		//header information
+		long ret = 12;
+		
+		for( ColGroup grp : _colGroups ) {
+			ret += 1; //type info
+			ret += grp.getExactSizeOnDisk();
+		}
+		
+		return ret;
+	}
+	
+	@Override
+	public void readFields(DataInput in) 
+		throws IOException 
+	{
+		boolean compressed = in.readBoolean();
+		
+		//deserialize uncompressed block
+		if( !compressed ) {
+			super.readFields(in);
+			return;
+		}
+		
+		//deserialize compressed block
+		rlen = in.readInt();
+		clen = in.readInt();
+		nonZeros = in.readLong();
+		int ncolGroups = in.readInt();
+		
+		_colGroups = new ArrayList<ColGroup>(ncolGroups);
+		for( int i=0; i<ncolGroups; i++ ) 
+		{
+			CompressionType ctype = CompressionType.values()[in.readByte()];
+			ColGroup grp = null;
+			
+			//create instance of column group
+			switch( ctype ) {
+				case UNCOMPRESSED:
+					grp = new ColGroupUncompressed(); break;
+				case OLE_BITMAP:
+					grp = new ColGroupOLE(); break;
+				case RLE_BITMAP:
+					grp = new ColGroupRLE(); break;
+			}
+			
+			//deserialize and add column group
+			grp.readFields(in);
+			_colGroups.add(grp);
+		}
+	}
+	
+	@Override
+	public void write(DataOutput out) 
+		throws IOException 
+	{
+		out.writeBoolean( isCompressed() );
+		
+		//serialize uncompressed block
+		if( !isCompressed() ) {
+			super.write(out);
+			return;
+		}
+		
+		//serialize compressed matrix block
+		out.writeInt(rlen);
+		out.writeInt(clen);
+		out.writeLong(nonZeros);
+		out.writeInt(_colGroups.size());
+		
+		for( ColGroup grp : _colGroups ) {
+			out.writeByte( grp.getCompType().ordinal() );
+			grp.write(out); //delegate serialization
+		}
+	}
+	
+	
+	/**
+	 * Redirects the default java serialization via externalizable to our default 
+	 * hadoop writable serialization for efficient broadcast/rdd deserialization. 
+	 * 
+	 * @param is
+	 * @throws IOException
+	 */
+	@Override
+	public void readExternal(ObjectInput is) 
+		throws IOException
+	{
+		readFields(is);
+	}
+	
+	/**
+	 * Redirects the default java serialization via externalizable to our default 
+	 * hadoop writable serialization for efficient broadcast/rdd serialization. 
+	 * 
+	 * @param is
+	 * @throws IOException
+	 */
+	@Override
+	public void writeExternal(ObjectOutput os) 
+		throws IOException
+	{
+		write(os);	
+	}
+	
+	
+	//////////////////////////////////////////
+	// Operations (overwrite existing ops for seamless integration)
+
+	@Override
+	public MatrixValue scalarOperations(ScalarOperator sop, MatrixValue result) 
+		throws DMLRuntimeException
+	{
+		//call uncompressed matrix scalar if necessary
+		if( !isCompressed() ) {
+			return super.scalarOperations(sop, result);
+		}
+		
+		//allocate the output matrix block
+		CompressedMatrixBlock ret = null;
+		if( result==null || !(result instanceof CompressedMatrixBlock) )
+			ret = new CompressedMatrixBlock(getNumRows(), getNumColumns(), sparse);
+		else {
+			ret = (CompressedMatrixBlock) result;
+			ret.reset(rlen, clen);
+		}
+		
+		// Apply the operation recursively to each of the column groups.
+		// Most implementations will only modify metadata.
+		ArrayList<ColGroup> newColGroups = new ArrayList<ColGroup>();
+		for (ColGroup grp : _colGroups) {
+			newColGroups.add(grp.scalarOperation(sop));
+		}
+		ret._colGroups = newColGroups;
+		ret.setNonZeros(rlen*clen);
+		
+		return ret;
+	}
+
+	@Override
+	public MatrixBlock appendOperations(MatrixBlock that, MatrixBlock ret) 
+		throws DMLRuntimeException
+	{
+		//call uncompressed matrix append if necessary
+		if( !isCompressed() ) {
+			if( that instanceof CompressedMatrixBlock )
+				that = ((CompressedMatrixBlock) that).decompress();
+			return super.appendOperations(that, ret);
+		}
+		
+		final int m = rlen;
+		final int n = clen+that.getNumColumns();
+		final long nnz = nonZeros+that.getNonZeros();		
+		
+		//init result matrix 
+		CompressedMatrixBlock ret2 = null;
+		if( ret == null || !(ret instanceof CompressedMatrixBlock) ) {
+			ret2 = new CompressedMatrixBlock(m, n, isInSparseFormat());
+		}
+		else {
+			ret2 = (CompressedMatrixBlock) ret;
+			ret2.reset(m, n);
+		}
+			
+		//shallow copy of lhs column groups
+		ret2.allocateColGroupList();
+		ret2._colGroups.addAll(_colGroups);
+		
+		//copy of rhs column groups w/ col index shifting
+		if( !(that instanceof CompressedMatrixBlock) ) {
+			that = new CompressedMatrixBlock(that);
+			((CompressedMatrixBlock)that).compress();
+		}
+		ArrayList<ColGroup> inColGroups = ((CompressedMatrixBlock) that)._colGroups;
+		for( ColGroup group : inColGroups ) {
+			ColGroup tmp = ConverterUtils.copyColGroup(group);
+			tmp.shiftColIndices(clen);
+			ret2._colGroups.add(tmp);
+		}
+		
+		//meta data maintenance
+		ret2.setNonZeros(nnz);		
+		return ret2;
+	}
+	
+	@Override
+	public MatrixBlock chainMatrixMultOperations(MatrixBlock v, MatrixBlock w, MatrixBlock out, ChainType ctype) 
+		throws DMLRuntimeException 
+	{
+		//call uncompressed matrix mult if necessary
+		if( !isCompressed() ) {
+			return super.chainMatrixMultOperations(v, w, out, ctype);
+		}
+		
+		//single-threaded mmchain of single uncompressed colgroup
+		if( isSingleUncompressedGroup() ){
+			return ((ColGroupUncompressed)_colGroups.get(0))
+				.getData().chainMatrixMultOperations(v, w, out, ctype);
+		}
+		
+		//Timing time = new Timing(true);
+		
+		//prepare result
+		if( out != null )
+			out.reset(clen, 1, false);
+		else 
+			out = new MatrixBlock(clen, 1, false);
+		
+		//empty block handling
+		if( isEmptyBlock(false) ) 
+			return out;
+			
+		//compute matrix mult
+		MatrixBlock tmp = new MatrixBlock(rlen, 1, false);
+		rightMultByVector(v, tmp);
+		if( ctype == ChainType.XtwXv ) {
+			BinaryOperator bop = new BinaryOperator(Multiply.getMultiplyFnObject());
+			LibMatrixBincell.bincellOpInPlace(tmp, w, bop);
+		}
+		leftMultByVectorTranspose(_colGroups, tmp, out, true);
+		
+		//System.out.println("Compressed MMChain in "+time.stop());
+		
+		return out;
+	}
+
+	@Override
+	public MatrixBlock chainMatrixMultOperations(MatrixBlock v, MatrixBlock w, MatrixBlock out, ChainType ctype, int k) 
+		throws DMLRuntimeException 
+	{
+		//call uncompressed matrix mult if necessary
+		if( !isCompressed() ){
+			return super.chainMatrixMultOperations(v, w, out, ctype, k);
+		}
+		
+		//multi-threaded mmchain of single uncompressed colgroup
+		if( isSingleUncompressedGroup() ){
+			return ((ColGroupUncompressed)_colGroups.get(0))
+				.getData().chainMatrixMultOperations(v, w, out, ctype, k);
+		}
+
+		Timing time = LOG.isDebugEnabled() ? new Timing(true) : null;
+		
+		//prepare result
+		if( out != null )
+			out.reset(clen, 1, false);
+		else 
+			out = new MatrixBlock(clen, 1, false);
+		
+		//empty block handling
+		if( isEmptyBlock(false) ) 
+			return out;
+		
+		//compute matrix mult
+		MatrixBlock tmp = new MatrixBlock(rlen, 1, false);
+		rightMultByVector(v, tmp, k);
+		if( ctype == ChainType.XtwXv ) {
+			BinaryOperator bop = new BinaryOperator(Multiply.getMultiplyFnObject());
+			LibMatrixBincell.bincellOpInPlace(tmp, w, bop);
+		}
+		leftMultByVectorTranspose(_colGroups, tmp, out, true, k);
+		
+		if( LOG.isDebugEnabled() )
+			LOG.debug("Compressed MMChain k="+k+" in "+time.stop());
+		
+		return out;
+	}
+	
+	@Override
+	public MatrixValue aggregateBinaryOperations(MatrixValue mv1, MatrixValue mv2, MatrixValue result, AggregateBinaryOperator op)
+			throws DMLRuntimeException 
+	{
+		//call uncompressed matrix mult if necessary
+		if( !isCompressed() ) {
+			return super.aggregateBinaryOperations(mv1, mv2, result, op);
+		}
+	
+		//multi-threaded mm of single uncompressed colgroup
+		if( isSingleUncompressedGroup() ){
+			MatrixBlock tmp = ((ColGroupUncompressed)_colGroups.get(0)).getData();
+			return tmp.aggregateBinaryOperations(this==mv1?tmp:mv1, this==mv2?tmp:mv2, result, op);
+		}
+		
+		Timing time = LOG.isDebugEnabled() ? new Timing(true) : null;
+		
+		//setup meta data (dimensions, sparsity)
+		int rl = mv1.getNumRows();
+		int cl = mv2.getNumColumns();
+		
+		//create output matrix block
+		MatrixBlock ret = (MatrixBlock) result;
+		if( ret==null )
+			ret = new MatrixBlock(rl, cl, false, rl*cl);
+		else
+			ret.reset(rl, cl, false, rl*cl);
+		
+		//compute matrix mult
+		if( mv1.getNumRows()>1 && mv2.getNumColumns()==1 ) { //MV right
+			CompressedMatrixBlock cmb = (CompressedMatrixBlock)mv1;
+			MatrixBlock mb = (MatrixBlock) mv2;
+			if( op.getNumThreads()>1 )
+				cmb.rightMultByVector(mb, ret, op.getNumThreads());
+			else
+				cmb.rightMultByVector(mb, ret);
+		}
+		else if( mv1.getNumRows()==1 && mv2.getNumColumns()>1 ) { //MV left
+			MatrixBlock mb = (MatrixBlock) mv1;
+			if( op.getNumThreads()>1 )
+				leftMultByVectorTranspose(_colGroups, mb, ret, false, op.getNumThreads());
+			else
+				leftMultByVectorTranspose(_colGroups, mb, ret, false);
+		}
+		else {
+			//NOTE: we could decompress and invoke super.aggregateBinary but for now
+			//we want to have an eager fail if this happens
+			throw new DMLRuntimeException("Unsupported matrix-matrix multiplication over compressed matrix block.");
+		}
+		
+		if( LOG.isDebugEnabled() )
+			LOG.debug("Compressed MM in "+time.stop());
+		
+		return ret;
+	}
+	
+	@Override
+	public MatrixValue aggregateUnaryOperations(AggregateUnaryOperator op, MatrixValue result, 
+			int blockingFactorRow, int blockingFactorCol, MatrixIndexes indexesIn, boolean inCP) 
+		throws DMLRuntimeException
+	{
+		//call uncompressed matrix mult if necessary
+		if( !isCompressed() ) {
+			return super.aggregateUnaryOperations(op, result, blockingFactorRow, blockingFactorCol, indexesIn, inCP);
+		}
+		
+		//check for supported operations
+		if( !(op.aggOp.increOp.fn instanceof KahanPlus || op.aggOp.increOp.fn instanceof KahanPlusSq) ){
+			throw new DMLRuntimeException("Unary aggregates other than sums not supported yet.");
+		}
+		
+		//prepare output dimensions
+		CellIndex tempCellIndex = new CellIndex(-1,-1);
+		op.indexFn.computeDimension(rlen, clen, tempCellIndex);
+		if(op.aggOp.correctionExists) {
+			switch(op.aggOp.correctionLocation)
+			{
+				case LASTROW: tempCellIndex.row++;  break;
+				case LASTCOLUMN: tempCellIndex.column++; break;
+				case LASTTWOROWS: tempCellIndex.row+=2; break;
+				case LASTTWOCOLUMNS: tempCellIndex.column+=2; break;
+				default:
+					throw new DMLRuntimeException("unrecognized correctionLocation: "+op.aggOp.correctionLocation);	
+			}
+		}
+		
+		//prepare output
+		if(result==null)
+			result=new MatrixBlock(tempCellIndex.row, tempCellIndex.column, false);
+		else
+			result.reset(tempCellIndex.row, tempCellIndex.column, false);
+		
+		MatrixBlock ret = (MatrixBlock) result;
+		
+		//core unary aggregate
+		if(    op.getNumThreads() > 1 
+			&& getExactSizeOnDisk() > MIN_PAR_AGG_THRESHOLD ) 
+		{
+			// initialize and allocate the result
+			ret.allocateDenseBlock();
+			
+			//multi-threaded execution of all groups 
+			ArrayList<ColGroup>[] grpParts = createStaticTaskPartitioning(op.getNumThreads(), false);
+			ColGroupUncompressed uc = getUncompressedColGroup();
+			try {
+				//compute all compressed column groups
+				ExecutorService pool = Executors.newFixedThreadPool( op.getNumThreads() );
+				ArrayList<UnaryAggregateTask> tasks = new ArrayList<UnaryAggregateTask>();
+				for( ArrayList<ColGroup> grp : grpParts )
+					tasks.add(new UnaryAggregateTask(grp, ret, op));
+				pool.invokeAll(tasks);	
+				pool.shutdown();
+				//compute uncompressed column group in parallel (otherwise bottleneck)
+				if( uc != null )
+					 ret = (MatrixBlock)uc.getData().aggregateUnaryOperations(op, ret, blockingFactorRow, blockingFactorCol, indexesIn, false);					
+				//aggregate partial results
+				if( !(op.indexFn instanceof ReduceRow) ){
+					KahanObject kbuff = new KahanObject(0,0);
+					KahanPlus kplus = KahanPlus.getKahanPlusFnObject();
+					for( int i=0; i<ret.getNumRows(); i++ ) {
+						kbuff.set(ret.quickGetValue(i, 0), ret.quickGetValue(i, 0));
+						for( UnaryAggregateTask task : tasks )
+							kplus.execute2(kbuff, task.getResult().quickGetValue(i, 0));
+						ret.quickSetValue(i, 0, kbuff._sum);
+						ret.quickSetValue(i, 1, kbuff._correction);
+					}
+				}		
+			}
+			catch(Exception ex) {
+				throw new DMLRuntimeException(ex);
+			}
+		}
+		else {
+			for (ColGroup grp : _colGroups) {
+				grp.unaryAggregateOperations(op, ret);
+			}
+		}
+		
+		//drop correction if necessary
+		if(op.aggOp.correctionExists && inCP)
+			ret.dropLastRowsOrColums(op.aggOp.correctionLocation);
+	
+		//post-processing
+		ret.recomputeNonZeros();
+		
+		return ret;
+	}
+	
+	@Override
+	public MatrixBlock transposeSelfMatrixMultOperations(MatrixBlock out, MMTSJType tstype) 
+		throws DMLRuntimeException 
+	{
+		//call uncompressed matrix mult if necessary
+		if( !isCompressed() ) {
+			return super.transposeSelfMatrixMultOperations(out, tstype);
+		}
+				
+		//single-threaded tsmm of single uncompressed colgroup
+		if( isSingleUncompressedGroup() ){
+			return ((ColGroupUncompressed)_colGroups.get(0))
+				.getData().transposeSelfMatrixMultOperations(out, tstype);
+		}
+
+		Timing time = LOG.isDebugEnabled() ? new Timing(true) : null;
+
+		//check for transpose type
+		if( tstype != MMTSJType.LEFT ) //right not supported yet
+			throw new DMLRuntimeException("Invalid MMTSJ type '"+tstype.toString()+"'.");
+		
+		//create output matrix block
+		if( out == null )
+			out = new MatrixBlock(clen, clen, false);
+		else
+			out.reset(clen, clen, false);
+		out.allocateDenseBlock();
+		
+		if( !isEmptyBlock(false) ) {
+			//compute matrix mult
+			leftMultByTransposeSelf(_colGroups, out, 0, _colGroups.size());
+			
+			// post-processing
+			out.recomputeNonZeros();
+		}
+		
+		if( LOG.isDebugEnabled() )
+			LOG.debug("Compressed TSMM in "+time.stop());
+		
+		return out;
+	}
+
+	
+	@Override
+	public MatrixBlock transposeSelfMatrixMultOperations(MatrixBlock out, MMTSJType tstype, int k) 
+		throws DMLRuntimeException 
+	{
+		//call uncompressed matrix mult if necessary
+		if( !isCompressed() ){
+			return super.transposeSelfMatrixMultOperations(out, tstype, k);
+		}
+		
+		//multi-threaded tsmm of single uncompressed colgroup
+		if( isSingleUncompressedGroup() ){
+			return ((ColGroupUncompressed)_colGroups.get(0))
+				.getData().transposeSelfMatrixMultOperations(out, tstype, k);
+		}
+		
+		Timing time = LOG.isDebugEnabled() ? new Timing(true) : null;
+		
+		//check for transpose type
+		if( tstype != MMTSJType.LEFT ) //right not supported yet
+			throw new DMLRuntimeException("Invalid MMTSJ type '"+tstype.toString()+"'.");
+		
+		//create output matrix block
+		if( out == null )
+			out = new MatrixBlock(clen, clen, false);
+		else
+			out.reset(clen, clen, false);
+		out.allocateDenseBlock();
+		
+		if( !isEmptyBlock(false) ) {
+			//compute matrix mult
+			try {
+				ExecutorService pool = Executors.newFixedThreadPool( k );
+				ArrayList<MatrixMultTransposeTask> tasks = new ArrayList<MatrixMultTransposeTask>();
+				int blklen = (int)(Math.ceil((double)clen/(2*k)));
+				for( int i=0; i<2*k & i*blklen<clen; i++ )
+					tasks.add(new MatrixMultTransposeTask(_colGroups, out, i*blklen, Math.min((i+1)*blklen, clen)));
+				List<Future<Object>> ret = pool.invokeAll(tasks);
+				for( Future<Object> tret : ret )
+					tret.get(); //check for errors
+				pool.shutdown();
+			}
+			catch(Exception ex) {
+				throw new DMLRuntimeException(ex);
+			}
+			
+			// post-processing
+			out.recomputeNonZeros();
+		}
+		
+		if( LOG.isDebugEnabled() )
+			LOG.debug("Compressed TSMM k="+k+" in "+time.stop());
+		
+		return out;
+	}
+
+	
+	/**
+	 * Multiply this matrix block by a column vector on the right.
+	 * 
+	 * @param vector
+	 *            right-hand operand of the multiplication
+	 * @param result
+	 *            buffer to hold the result; must have the appropriate size
+	 *            already
+	 */
+	private void rightMultByVector(MatrixBlock vector, MatrixBlock result)
+		throws DMLRuntimeException 
+	{
+		// initialize and allocate the result
+		result.allocateDenseBlock();
+
+		// delegate matrix-vector operation to each column group
+		for( ColGroup grp : _colGroups )
+			if( grp instanceof ColGroupUncompressed ) //overwrites output
+				grp.rightMultByVector(vector, result, 0, result.getNumRows());
+		for( ColGroup grp : _colGroups )
+			if( !(grp instanceof ColGroupUncompressed) ) //adds to output
+				grp.rightMultByVector(vector, result, 0, result.getNumRows());
+		
+		// post-processing
+		result.recomputeNonZeros();
+	}
+
+	/**
+	 * Multi-threaded version of rightMultByVector.
+	 * 
+	 * @param vector
+	 * @param result
+	 * @param k
+	 * @throws DMLRuntimeException
+	 */
+	private void rightMultByVector(MatrixBlock vector, MatrixBlock result, int k)
+		throws DMLRuntimeException 
+	{
+		// initialize and allocate the result
+		result.allocateDenseBlock();
+
+		//multi-threaded execution of all groups
+		try {
+			ExecutorService pool = Executors.newFixedThreadPool( k );
+			int rlen = getNumRows();
+			int seqsz = BitmapEncoder.BITMAP_BLOCK_SZ;
+			int blklen = (int)(Math.ceil((double)rlen/k));
+			blklen += (blklen%seqsz != 0)?seqsz-blklen%seqsz:0;
+			ArrayList<RightMatrixMultTask> tasks = new ArrayList<RightMatrixMultTask>();
+			for( int i=0; i<k & i*blklen<getNumRows(); i++ )
+				tasks.add(new RightMatrixMultTask(_colGroups, vector, result, i*blklen, Math.min((i+1)*blklen,rlen)));
+			pool.invokeAll(tasks);	
+			pool.shutdown();
+		}
+		catch(Exception ex) {
+			throw new DMLRuntimeException(ex);
+		}
+		
+		// post-processing
+		result.recomputeNonZeros();
+	}
+	
+	/**
+	 * Multiply this matrix block by the transpose of a column vector (i.e.
+	 * t(v)%*%X)
+	 * 
+	 * @param vector
+	 *            left-hand operand of the multiplication
+	 * @param result
+	 *            buffer to hold the result; must have the appropriate size
+	 *            already
+	 */
+	private static void leftMultByVectorTranspose(List<ColGroup> colGroups, MatrixBlock vector, MatrixBlock result, boolean doTranspose) 
+		throws DMLRuntimeException 
+	{
+		//transpose vector if required
+		MatrixBlock rowVector = vector;
+		if (doTranspose) {
+			rowVector = new MatrixBlock(1, vector.getNumRows(), false);
+			LibMatrixReorg.transpose(vector, rowVector);
+		}
+		
+		// initialize and allocate the result
+		result.reset();
+		result.allocateDenseBlock();
+		
+		// delegate matrix-vector operation to each column group
+		for (ColGroup grp : colGroups) {			
+			grp.leftMultByRowVector(rowVector, result);
+		}
+
+		// post-processing
+		result.recomputeNonZeros();
+	}
+	
+	/**
+	 * Multi-thread version of leftMultByVectorTranspose.
+	 * 
+	 * @param vector
+	 * @param result
+	 * @param doTranspose
+	 * @param k
+	 * @throws DMLRuntimeException
+	 */
+	private static void leftMultByVectorTranspose(List<ColGroup> colGroups,MatrixBlock vector, MatrixBlock result, boolean doTranspose, int k) 
+		throws DMLRuntimeException 
+	{
+		int kuc = Math.max(1, k - colGroups.size() + 1);
+		
+		//transpose vector if required
+		MatrixBlock rowVector = vector;
+		if (doTranspose) {
+			rowVector = new MatrixBlock(1, vector.getNumRows(), false);
+			LibMatrixReorg.transpose(vector, rowVector);
+		}
+		
+		// initialize and allocate the result
+		result.reset();
+		result.allocateDenseBlock();
+
+		//multi-threaded execution
+		try {
+			ExecutorService pool = Executors.newFixedThreadPool( Math.min(colGroups.size(), k) );
+			ArrayList<LeftMatrixMultTask> tasks = new ArrayList<LeftMatrixMultTask>();
+			for( ColGroup grp : colGroups )
+				tasks.add(new LeftMatrixMultTask(grp, rowVector, result, kuc));
+			pool.invokeAll(tasks);	
+			pool.shutdown();
+		}
+		catch(Exception ex) {
+			throw new DMLRuntimeException(ex);
+		}
+
+		// post-processing
+		result.recomputeNonZeros();
+	}
+
+	/**
+	 * 
+	 * @param result
+	 * @throws DMLRuntimeException
+	 */
+	private static void leftMultByTransposeSelf(ArrayList<ColGroup> groups, MatrixBlock result, int cl, int cu)
+		throws DMLRuntimeException 
+	{
+		final int numRows = groups.get(0).getNumRows();
+		final int numGroups = groups.size();		
+		
+		//preallocated dense matrix block
+		MatrixBlock lhs = new MatrixBlock(numRows, 1, false);
+		lhs.allocateDenseBlock();
+		
+		//approach: for each colgroup, extract uncompressed columns one at-a-time
+		//vector-matrix multiplies against remaining col groups
+		for( int i=cl; i<cu; i++ ) 
+		{
+			//get current group and relevant col groups
+			ColGroup group = groups.get(i);	
+			int[] ixgroup = group.getColIndices();
+			List<ColGroup> tmpList = groups.subList(i, numGroups);
+			
+			//for all uncompressed lhs columns vectors
+			for( int j=0; j<ixgroup.length; j++ ) {
+				//decompress single column
+				lhs.reset(numRows, 1, false);				
+				group.decompressToBlock(lhs, j);
+				
+				if( !lhs.isEmptyBlock(false) ) {
+					//compute vector-matrix partial result
+					MatrixBlock tmpret = new MatrixBlock(1,result.getNumColumns(),false);
+					leftMultByVectorTranspose(tmpList, lhs, tmpret, true);								
+					
+					//write partial results (disjoint non-zeros)
+					LinearAlgebraUtils.copyNonZerosToRowCol(result, tmpret, ixgroup[j]);	
+				}
+			}
+		}
+	}
+	
+	/**
+	 * 
+	 * @param k
+	 * @return
+	 */
+	@SuppressWarnings("unchecked")
+	private ArrayList<ColGroup>[] createStaticTaskPartitioning(int k, boolean inclUncompressed)
+	{
+		// special case: single uncompressed col group
+		if( _colGroups.size()==1 && _colGroups.get(0) instanceof ColGroupUncompressed ){
+			return new ArrayList[0];
+		}
+		
+		// initialize round robin col group distribution
+		// (static task partitioning to reduce mem requirements/final agg)
+		int numTasks = Math.min(k, _colGroups.size());
+		ArrayList<ColGroup>[] grpParts = new ArrayList[numTasks];
+		int pos = 0;
+		for( ColGroup grp : _colGroups ){
+			if( grpParts[pos]==null )
+				grpParts[pos] = new ArrayList<ColGroup>();
+			if( inclUncompressed || !(grp instanceof ColGroupUncompressed) ) {
+				grpParts[pos].add(grp);
+				pos = (pos==numTasks-1) ? 0 : pos+1;
+			}
+		}
+		
+		return grpParts;
+	}
+	
+	/**
+	 * 
+	 * @return
+	 */
+	private ColGroupUncompressed getUncompressedColGroup()
+	{
+		for( ColGroup grp : _colGroups )
+			if( grp instanceof ColGroupUncompressed ) 
+				return (ColGroupUncompressed)grp;
+		
+		return null;
+	}
+	
+	/**
+	 * 
+	 */
+	private static class LeftMatrixMultTask implements Callable<Object> 
+	{
+		private ColGroup _group = null;
+		private MatrixBlock _vect = null;
+		private MatrixBlock _ret = null;
+		private int _kuc = 1;
+		
+		protected LeftMatrixMultTask( ColGroup group, MatrixBlock vect, MatrixBlock ret, int kuc)  {
+			_group = group;
+			_vect = vect;
+			_ret = ret;
+			_kuc = kuc;
+		}
+		
+		@Override
+		public Object call() throws DMLRuntimeException 
+		{
+			// delegate matrix-vector operation to each column group
+			if( _group instanceof ColGroupUncompressed && _kuc >1 && ColGroupBitmap.LOW_LEVEL_OPT )
+				((ColGroupUncompressed)_group).leftMultByRowVector(_vect, _ret, _kuc);
+			else
+				_group.leftMultByRowVector(_vect, _ret);
+			return null;
+		}
+	}
+	
+	/**
+	 * 
+	 */
+	private static class RightMatrixMultTask implements Callable<Object> 
+	{
+		private ArrayList<ColGroup> _groups = null;
+		private MatrixBlock _vect = null;
+		private MatrixBlock _ret = null;
+		private int _rl = -1;
+		private int _ru = -1;
+		
+		protected RightMatrixMultTask( ArrayList<ColGroup> groups, MatrixBlock vect, MatrixBlock ret, int rl, int ru)  {
+			_groups = groups;
+			_vect = vect;
+			_ret = ret;
+			_rl = rl;
+			_ru = ru;
+		}
+		
+		@Override
+		public Object call() throws DMLRuntimeException 
+		{
+			// delegate vector-matrix operation to each column group
+			for( ColGroup grp : _groups )
+				if( grp instanceof ColGroupUncompressed ) //overwrites output
+					grp.rightMultByVector(_vect, _ret, _rl, _ru);
+			for( ColGroup grp : _groups )
+				if( !(grp instanceof ColGroupUncompressed) ) //adds to output
+					grp.rightMultByVector(_vect, _ret, _rl, _ru);
+			return null;
+		}
+	}
+	
+	private static class MatrixMultTransposeTask implements Callable<Object> 
+	{
+		private ArrayList<ColGroup> _groups = null;
+		private MatrixBlock _ret = null;
+		private int _cl = -1;
+		private int _cu = -1;
+		
+		protected MatrixMultTransposeTask(ArrayList<ColGroup> groups, MatrixBlock ret, int cl, int cu)  {
+			_groups = groups;
+			_ret = ret;
+			_cl = cl;
+			_cu = cu;
+		}
+		
+		@Override
+		public Object call() throws DMLRuntimeException {
+			leftMultByTransposeSelf(_groups, _ret, _cl, _cu);
+			return null;
+		}
+	}
+	
+	private static class UnaryAggregateTask implements Callable<Object> 
+	{
+		private ArrayList<ColGroup> _groups = null;
+		private MatrixBlock _ret = null;
+		private AggregateUnaryOperator _op = null;
+		
+		protected UnaryAggregateTask( ArrayList<ColGroup> groups, MatrixBlock ret, AggregateUnaryOperator op)  {
+			_groups = groups;
+			_op = op;
+			
+			if( !(_op.indexFn instanceof ReduceRow) ) { //sum/rowSums
+				_ret = new MatrixBlock(ret.getNumRows(), ret.getNumColumns(), false);
+				_ret.allocateDenseBlock();
+			}
+			else { //colSums
+				_ret = ret;
+			}
+		}
+		
+		@Override
+		public Object call() throws DMLRuntimeException 
+		{
+			// delegate vector-matrix operation to each column group
+			for( ColGroup grp : _groups )
+				grp.unaryAggregateOperations(_op, _ret);
+			return null;
+		}
+		
+		public MatrixBlock getResult(){
+			return _ret;
+		}
+	}
+}