src/main/java/org/apache/sysds/runtime/compress/CompressedMatrixBlockFactory.java
/**
	 * The main method for compressing the input matrix.
	 * 
	 * SAMPLE-BASED DECISIONS: Decisions such as testing if a column is amenable to bitmap compression or evaluating
	 * co-coding potentials are made based on a subset of the rows. For large data sets, sampling might take a
	 * significant amount of time. So, we generate only one sample and use it for the entire compression process.
	 * 
	 * Once the compression plan is selected based on sampling, the plan is verified and decisions are overwritten by
	 * full estimates.
	 * 
	 * @param mb           The matrix block to compress
	 * @param k            The number of threads used to execute the compression
	 * @param compSettings The Compression settings used
	 * @return A compressed matrix block.
	 */
The main method for compressing the input matrix.

SAMPLE-BASED DECISIONS: Decisions such as testing if a column is amenable to bitmap compression or evaluating
co-coding potentials are made based on a subset of the rows. For large data sets, sampling might take a
significant amount of time. So, we generate only one sample and use it for the entire compression process.

Once the compression plan is selected based on sampling, the plan is verified and decisions are overwritten by
full estimates.

@param mb           The matrix block to compress
@param k            The number of threads used to execute the compression
@param compSettings The Compression settings used
@return A compressed matrix block.
The main method for compressing the input matrix . SAMPLE - BASED DECISIONS : Decisions such as testing if a column is amenable to bitmap compression or evaluating co - coding potentials are made based on a subset of the rows . For large data sets sampling might take a significant amount of time . So we generate only one sample and use it for the entire compression process . Once the compression plan is selected based on sampling the plan is verified and decisions are overwritten by full estimates . @param mb The matrix block to compress @param k The number of threads used to execute the compression @param compSettings The Compression settings used @return A compressed matrix block .
public static Pair<MatrixBlock, CompressionStatistics> compress(MatrixBlock mb, int k,
		CompressionSettings compSettings) {
		if(mb instanceof CompressedMatrixBlock) {
			throw new DMLRuntimeException("Redundant compression, block already compressed.");
		}
		Timing time = new Timing(true);
		CompressionStatistics _stats = new CompressionStatistics();
		int numRows = mb.getNumRows();
		int numCols = mb.getNumColumns();
		boolean sparse = mb.isInSparseFormat();
		MatrixBlock rawBlock = !compSettings.transposeInput ? new MatrixBlock(mb) : LibMatrixReorg
			.transpose(mb, new MatrixBlock(numCols, numRows, sparse), k);
		CompressedSizeEstimator sizeEstimator = CompressedSizeEstimatorFactory.getSizeEstimator(rawBlock, compSettings);
		CompressedSizeInfo sizeInfos = sizeEstimator.computeCompressedSizeInfos(k);
		if(compSettings.investigateEstimate)
			_stats.estimatedSizeCols = sizeInfos.memoryEstimate();
		_stats.setNextTimePhase(time.stop());
		if (DMLScript.STATISTICS ){
			DMLCompressionStatistics.addCompressionTime(_stats.getLastTimePhase(), 1);
		}
		if(LOG.isDebugEnabled()){
			LOG.debug("Compression statistics:");
			LOG.debug("--compression phase 1: " + _stats.getLastTimePhase());
		}
		if(sizeInfos.colsC.isEmpty()) {
			LOG.info("Abort block compression because all columns are incompressible.");
			return new ImmutablePair<>(new MatrixBlock().copyShallow(mb), _stats);
		}
		List<int[]> coCodeColGroups = PlanningCoCoder
			.findCoCodesByPartitioning(sizeEstimator, sizeInfos, numRows, k, compSettings);
		_stats.setNextTimePhase(time.stop());
		if (DMLScript.STATISTICS ){
			DMLCompressionStatistics.addCompressionTime(_stats.getLastTimePhase(), 2);
		}
		if(LOG.isDebugEnabled()) {
			LOG.debug("--compression phase 2: " + _stats.getLastTimePhase());
			StringBuilder sb = new StringBuilder();
			for(int[] group : coCodeColGroups)
				sb.append(Arrays.toString(group));
			LOG.debug(sb.toString());
		}
		ColGroup[] colGroups = ColGroupFactory
			.compressColGroups(rawBlock, sizeInfos.compRatios, coCodeColGroups, compSettings, k);
		CompressedMatrixBlock res = new CompressedMatrixBlock(mb);
		List<ColGroup> colGroupList = ColGroupFactory.assignColumns(numCols, colGroups, rawBlock, compSettings);
		res.allocateColGroupList(colGroupList);
		_stats.setNextTimePhase(time.stop());
		if (DMLScript.STATISTICS ){
			DMLCompressionStatistics.addCompressionTime(_stats.getLastTimePhase(), 3);
		}
		if(LOG.isDebugEnabled()) {
			LOG.debug("Hash overlap count:" + DblArrayIntListHashMap.hashMissCount);
			DblArrayIntListHashMap.hashMissCount = 0;
			LOG.debug("--compression phase 3: " + _stats.getLastTimePhase());
		}
		_stats.setNextTimePhase(time.stop());
		if (DMLScript.STATISTICS ){
			DMLCompressionStatistics.addCompressionTime(_stats.getLastTimePhase(), 4);
		}
		if(LOG.isDebugEnabled()) {
			LOG.debug("--compression phase 4: " + _stats.getLastTimePhase());
		}
		_stats.size = res.estimateCompressedSizeInMemory();
		_stats.originalSize = mb.estimateSizeInMemory();
		_stats.ratio = _stats.originalSize / (double) _stats.size;
		if(_stats.ratio < 1) {
			LOG.info("Abort block compression because compression ratio is less than 1.");
			return new ImmutablePair<>(new MatrixBlock().copyShallow(mb), _stats);
		}
		rawBlock.cleanupBlock(true, true);
		res.cleanupBlock(true, true);
		_stats.setNextTimePhase(time.stop());
		_stats.setColGroupsCounts(colGroupList);
		if (DMLScript.STATISTICS ){
			DMLCompressionStatistics.addCompressionTime(_stats.getLastTimePhase(), 5);
		}
		if(LOG.isDebugEnabled()) {
			LOG.debug("--num col groups: " + colGroupList.size() + ", -- num input cols: " + numCols);
			LOG.debug("--compression phase 5: " + _stats.getLastTimePhase());
			LOG.debug("--col groups types " + _stats.getGroupsTypesString());
			LOG.debug("--col groups sizes " + _stats.getGroupsSizesString());
			LOG.debug("--compressed size: " + _stats.size);
			LOG.debug("--compression ratio: " + _stats.ratio);
			if(LOG.isTraceEnabled()) {
				for(ColGroup colGroup : colGroupList) {
					LOG.trace("--colGroups colIndexes : " + Arrays.toString(colGroup.getColIndices()));
					LOG.trace("--colGroups type       : " + colGroup.getClass().getSimpleName());
					LOG.trace("--colGroups Values     : " + Arrays.toString(colGroup.getValues()));
				}
			}
		}
		return new ImmutablePair<>(res, _stats);
	}
public static Pair < MatrixBlock , CompressionStatistics > compress ( MatrixBlock mb , int k , CompressionSettings compSettings ) { if ( mb instanceof CompressedMatrixBlock ) { throw new DMLRuntimeException ( "Redundant compression, block already compressed." ) ; } Timing time = new Timing ( true ) ; CompressionStatistics _stats = new CompressionStatistics ( ) ; int numRows = mb . getNumRows ( ) ; int numCols = mb . getNumColumns ( ) ; boolean sparse = mb . isInSparseFormat ( ) ; MatrixBlock rawBlock = ! compSettings . transposeInput ? new MatrixBlock ( mb ) : LibMatrixReorg . transpose ( mb , new MatrixBlock ( numCols , numRows , sparse ) , k ) ; CompressedSizeEstimator sizeEstimator = CompressedSizeEstimatorFactory . getSizeEstimator ( rawBlock , compSettings ) ; CompressedSizeInfo sizeInfos = sizeEstimator . computeCompressedSizeInfos ( k ) ; if ( compSettings . investigateEstimate ) _stats . estimatedSizeCols = sizeInfos . memoryEstimate ( ) ; _stats . setNextTimePhase ( time . stop ( ) ) ; if ( DMLScript . STATISTICS ) { DMLCompressionStatistics . addCompressionTime ( _stats . getLastTimePhase ( ) , 1 ) ; } if ( LOG . isDebugEnabled ( ) ) { LOG . debug ( "Compression statistics:" ) ; LOG . debug ( "--compression phase 1: " + _stats . getLastTimePhase ( ) ) ; } if ( sizeInfos . colsC . isEmpty ( ) ) { LOG . info ( "Abort block compression because all columns are incompressible." ) ; return new ImmutablePair < > ( new MatrixBlock ( ) . copyShallow ( mb ) , _stats ) ; } List < int [ ] > coCodeColGroups = PlanningCoCoder . findCoCodesByPartitioning ( sizeEstimator , sizeInfos , numRows , k , compSettings ) ; _stats . setNextTimePhase ( time . stop ( ) ) ; if ( DMLScript . STATISTICS ) { DMLCompressionStatistics . addCompressionTime ( _stats . getLastTimePhase ( ) , 2 ) ; } if ( LOG . isDebugEnabled ( ) ) { LOG . debug ( "--compression phase 2: " + _stats . getLastTimePhase ( ) ) ; StringBuilder sb = new StringBuilder ( ) ; for ( int [ ] group : coCodeColGroups ) sb . append ( Arrays . toString ( group ) ) ; LOG . debug ( sb . toString ( ) ) ; } ColGroup [ ] colGroups = ColGroupFactory . compressColGroups ( rawBlock , sizeInfos . compRatios , coCodeColGroups , compSettings , k ) ; CompressedMatrixBlock res = new CompressedMatrixBlock ( mb ) ; List < ColGroup > colGroupList = ColGroupFactory . assignColumns ( numCols , colGroups , rawBlock , compSettings ) ; res . allocateColGroupList ( colGroupList ) ; _stats . setNextTimePhase ( time . stop ( ) ) ; if ( DMLScript . STATISTICS ) { DMLCompressionStatistics . addCompressionTime ( _stats . getLastTimePhase ( ) , 3 ) ; } if ( LOG . isDebugEnabled ( ) ) { LOG . debug ( "Hash overlap count:" + DblArrayIntListHashMap . hashMissCount ) ; DblArrayIntListHashMap . hashMissCount = 0 ; LOG . debug ( "--compression phase 3: " + _stats . getLastTimePhase ( ) ) ; } _stats . setNextTimePhase ( time . stop ( ) ) ; if ( DMLScript . STATISTICS ) { DMLCompressionStatistics . addCompressionTime ( _stats . getLastTimePhase ( ) , 4 ) ; } if ( LOG . isDebugEnabled ( ) ) { LOG . debug ( "--compression phase 4: " + _stats . getLastTimePhase ( ) ) ; } _stats . size = res . estimateCompressedSizeInMemory ( ) ; _stats . originalSize = mb . estimateSizeInMemory ( ) ; _stats . ratio = _stats . originalSize / ( double ) _stats . size ; if ( _stats . ratio < 1 ) { LOG . info ( "Abort block compression because compression ratio is less than 1." ) ; return new ImmutablePair < > ( new MatrixBlock ( ) . copyShallow ( mb ) , _stats ) ; } rawBlock . cleanupBlock ( true , true ) ; res . cleanupBlock ( true , true ) ; _stats . setNextTimePhase ( time . stop ( ) ) ; _stats . setColGroupsCounts ( colGroupList ) ; if ( DMLScript . STATISTICS ) { DMLCompressionStatistics . addCompressionTime ( _stats . getLastTimePhase ( ) , 5 ) ; } if ( LOG . isDebugEnabled ( ) ) { LOG . debug ( "--num col groups: " + colGroupList . size ( ) + ", -- num input cols: " + numCols ) ; LOG . debug ( "--compression phase 5: " + _stats . getLastTimePhase ( ) ) ; LOG . debug ( "--col groups types " + _stats . getGroupsTypesString ( ) ) ; LOG . debug ( "--col groups sizes " + _stats . getGroupsSizesString ( ) ) ; LOG . debug ( "--compressed size: " + _stats . size ) ; LOG . debug ( "--compression ratio: " + _stats . ratio ) ; if ( LOG . isTraceEnabled ( ) ) { for ( ColGroup colGroup : colGroupList ) { LOG . trace ( "--colGroups colIndexes : " + Arrays . toString ( colGroup . getColIndices ( ) ) ) ; LOG . trace ( "--colGroups type       : " + colGroup . getClass ( ) . getSimpleName ( ) ) ; LOG . trace ( "--colGroups Values     : " + Arrays . toString ( colGroup . getValues ( ) ) ) ; } } } return new ImmutablePair < > ( res , _stats ) ; }
The main method for compressing the input matrix.
The main method for compressing the input matrix .
// Check for redundant compression // Prepare basic meta data and deep copy / transpose input // Transpose the MatrixBlock if the TransposeInput flag is set. // This gives better cache consciousness, at a small upfront cost. // Construct sample-based size estimator // -------------------------------------------------- // PHASE 1: Classify columns by compression type // Start by determining which columns are amenable to compression // Classify columns according to ratio (size uncompressed / size compressed), // where a column is compressible if ratio > 1. // -------------------------------------------------- // -------------------------------------------------- // PHASE 2: Grouping columns // Divide the columns into column groups. // TODO: Make second estimate of memory usage if the ColGroups are as above? // This should already be done inside the PlanningCoCoder, and therefore this information // should be returned there, and not estimated twice. // if(INVESTIGATE_ESTIMATES) { // _stats.estimatedSizeColGroups = memoryEstimateIfColsAre(coCodeColGroups); // } // -------------------------------------------------- // -------------------------------------------------- // PHASE 3: Compress and correct sample-based decisions // Make Compression happen! // -------------------------------------------------- // -------------------------------------------------- // PHASE 4: Best-effort dictionary sharing for DDC1 single-col groups // Dictionary dict = (!(compSettings.validCompressions.contains(CompressionType.DDC)) || // !(compSettings.allowSharedDDCDictionary)) ? null : createSharedDDC1Dictionary(colGroupList); // if(dict != null) { // applySharedDDC1Dictionary(colGroupList, dict); // res._sharedDDC1Dict = true; // } // -------------------------------------------------- // -------------------------------------------------- // Phase 5: Cleanup // The remaining columns are stored uncompressed as one big column group // Final cleanup (discard uncompressed block) // --------------------------------------------------