Project: S3779 LZW ColGroup | ASML

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

@@ -55,16 +55,16 @@ public class CompressionSettings {
 	/**
 	 * The sampling ratio used when choosing ColGroups. Note that, default behavior is to use exact estimator if the
 	 * number of elements is below 1000.
-	 * 
+	 *
 	 * DEPRECATED
 	 */
 	public final double samplingRatio;
 
 	/**
 	 * The sampling ratio power to use when choosing sample size. This is used in accordance to the function:
-	 * 
+	 *
 	 * sampleSize += nRows^samplePower;
-	 * 
+	 *
 	 * The value is bounded to be in the range of 0 to 1, 1 giving a sample size of everything, and 0 adding 1.
 	 */
 	public final double samplePower;
@@ -114,7 +114,7 @@ public class CompressionSettings {
 	/**
 	 * Transpose input matrix, to optimize access when extracting bitmaps. This setting is changed inside the script
 	 * based on the transposeInput setting.
-	 * 
+	 *
 	 * This is intentionally left as a mutable value, since the transposition of the input matrix is decided in phase 3.
 	 */
 	public boolean transposed = false;
@@ -135,6 +135,19 @@ public class CompressionSettings {
 
 	public final boolean preferDeltaEncoding;
 
+	// Handling Targetloss for piecewise linear Kompression
+
+	private double piecewiseTargetLoss = Double.NaN;
+
+	public void setPiecewiseTargetLoss(double piecewiseTargetLoss) {
+		this.piecewiseTargetLoss = piecewiseTargetLoss;
+
+	}
+
+	public double getPiecewiseTargetLoss() {
+		return piecewiseTargetLoss;
+	}
+
 	protected CompressionSettings(double samplingRatio, double samplePower, boolean allowSharedDictionary,
 		String transposeInput, int seed, boolean lossy, EnumSet<CompressionType> validCompressions,
 		boolean sortValuesByLength, PartitionerType columnPartitioner, int maxColGroupCoCode, double coCodePercentage,
@@ -161,7 +174,7 @@ protected CompressionSettings(double samplingRatio, double samplePower, boolean
 		this.sdcSortType = sdcSortType;
 		this.scaleFactors = scaleFactors;
 		this.preferDeltaEncoding = preferDeltaEncoding;
-		
+
 		if(!printedStatus && LOG.isDebugEnabled()) {
 			printedStatus = true;
 			LOG.debug(this.toString());

src/main/java/org/apache/sysds/runtime/compress/colgroup/AColGroup.java

@@ -66,7 +66,8 @@ public abstract class AColGroup implements Serializable {
 
 	/** Public super types of compression ColGroups supported */
 	public static enum CompressionType {
-		UNCOMPRESSED, RLE, OLE, DDC, CONST, EMPTY, SDC, SDCFOR, DDCFOR, DeltaDDC, DDCLZW, LinearFunctional;
+		UNCOMPRESSED, RLE, OLE, DDC, CONST, EMPTY, SDC, SDCFOR, DDCFOR, DeltaDDC, DDCLZW, LinearFunctional,
+		PiecewiseLinearCompressed;
 
 		public boolean isDense() {
 			return this == DDC || this == CONST || this == DDCFOR || this == DDCFOR || this == DDCLZW;
@@ -87,8 +88,8 @@ public boolean isSDC() {
 	 * Protected such that outside the ColGroup package it should be unknown which specific subtype is used.
 	 */
 	protected static enum ColGroupType {
-		UNCOMPRESSED, RLE, OLE, DDC, CONST, EMPTY, SDC, SDCSingle, SDCSingleZeros, SDCZeros, SDCFOR, DDCFOR, DDCLZW, DeltaDDC,
-		LinearFunctional;
+		UNCOMPRESSED, RLE, OLE, DDC, CONST, EMPTY, SDC, SDCSingle, SDCSingleZeros, SDCZeros, SDCFOR, DDCFOR, DeltaDDC,
+		LinearFunctional, PiecewiseLinearCompressed, DDCLZW;
 	}
 
 	/** The ColGroup indexes contained in the ColGroup */

src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupFactory.java

@@ -43,6 +43,7 @@
 import org.apache.sysds.runtime.compress.colgroup.dictionary.DictionaryFactory;
 import org.apache.sysds.runtime.compress.colgroup.dictionary.IDictionary;
 import org.apache.sysds.runtime.compress.colgroup.functional.LinearRegression;
+import org.apache.sysds.runtime.compress.colgroup.functional.PiecewiseLinearUtils;
 import org.apache.sysds.runtime.compress.colgroup.indexes.ColIndexFactory;
 import org.apache.sysds.runtime.compress.colgroup.indexes.IColIndex;
 import org.apache.sysds.runtime.compress.colgroup.insertionsort.AInsertionSorter;
@@ -106,7 +107,7 @@ private ColGroupFactory(MatrixBlock in, CompressedSizeInfo csi, CompressionSetti
 
 	/**
 	 * The actual compression method, that handles the logic of compressing multiple columns together.
-	 * 
+	 *
 	 * @param in  The input matrix, that could have been transposed. If it is transposed the compSettings should specify
 	 *            this.
 	 * @param csi The compression information extracted from the estimation, this contains which groups of columns to
@@ -120,7 +121,7 @@ public static List<AColGroup> compressColGroups(MatrixBlock in, CompressedSizeIn
 
 	/**
 	 * The actual compression method, that handles the logic of compressing multiple columns together.
-	 * 
+	 *
 	 * @param in  The input matrix, that could have been transposed. If it is transposed the compSettings should specify
 	 *            this.
 	 * @param csi The compression information extracted from the estimation, this contains which groups of columns to
@@ -135,7 +136,7 @@ public static List<AColGroup> compressColGroups(MatrixBlock in, CompressedSizeIn
 	}
 
 	/**
-	 * 
+	 *
 	 * @param in  The input matrix, that could have been transposed. If it is transposed the compSettings should specify
 	 *            this.
 	 * @param csi The compression information extracted from the estimation, this contains which groups of columns to
@@ -306,6 +307,9 @@ else if(ct == CompressionType.LinearFunctional) {
 				return compressLinearFunctional(colIndexes, in, cs);
 			}
 		}
+		else if(ct == CompressionType.PiecewiseLinearCompressed) {
+			return compressPiecewiseLinearFunctionalSuccessive(colIndexes, in, cs);
+		}
 		else if(ct == CompressionType.DDCFOR) {
 			AColGroup g = directCompressDDC(colIndexes, cg);
 			if(g instanceof ColGroupDDC)
@@ -710,7 +714,7 @@ private AColGroup directCompressDeltaDDC(IColIndex colIndexes, CompressedSizeInf
 		if(cs.scaleFactors != null) {
 			throw new NotImplementedException("Delta encoding with quantization not yet implemented");
 		}
-		
+
 		if(colIndexes.size() > 1) {
 			return directCompressDeltaDDCMultiCol(colIndexes, cg);
 		}
@@ -742,7 +746,7 @@ private AColGroup directCompressDeltaDDCSingleCol(IColIndex colIndexes, Compress
 
 		if(map.size() == 0)
 			return new ColGroupEmpty(colIndexes);
-		
+
 		final double[] dictValues = map.getDictionary();
 		IDictionary dict = new DeltaDictionary(dictValues, 1);
 
@@ -751,7 +755,8 @@ private AColGroup directCompressDeltaDDCSingleCol(IColIndex colIndexes, Compress
 		return ColGroupDeltaDDC.create(colIndexes, dict, resData, null);
 	}
 
-	private AColGroup directCompressDeltaDDCMultiCol(IColIndex colIndexes, CompressedSizeInfoColGroup cg) throws Exception {
+	private AColGroup directCompressDeltaDDCMultiCol(IColIndex colIndexes, CompressedSizeInfoColGroup cg)
+		throws Exception {
 		final AMapToData d = MapToFactory.create(nRow, Math.max(Math.min(cg.getNumOffs() + 1, nRow), 126));
 		final int fill = d.getUpperBoundValue();
 		d.fill(fill);
@@ -1078,6 +1083,40 @@ private static AColGroup compressLinearFunctional(IColIndex colIndexes, MatrixBl
 		return ColGroupLinearFunctional.create(colIndexes, coefficients, numRows);
 	}
 
+	/**
+	 * This method is the entry point to compress a matrix with successive piecewise linear compression method, which
+	 * compares each values in linear time and checks if the targetloss exceeded
+	 *
+	 * @param colIndexes the column indices to compress
+	 * @param in         the input Matrixblock containing the data
+	 * @param cs         compression settings to define the target loss, which should be considered
+	 * @return a piecewise linear compressed column group
+	 */
+
+	public static AColGroup compressPiecewiseLinearFunctionalSuccessive(IColIndex colIndexes, MatrixBlock in,
+		CompressionSettings cs) {
+
+		final int numRows = in.getNumRows();
+		final int numCols = colIndexes.size();
+		int[][] breakpointsPerCol = new int[numCols][];
+		double[][] slopesPerCol = new double[numCols][];
+		double[][] interceptsPerCol = new double[numCols][];
+
+		for(int col = 0; col < numCols; col++) {
+			final int colIdx = colIndexes.get(col);
+			double[] column = PiecewiseLinearUtils.getColumn(in, colIdx);
+			PiecewiseLinearUtils.SegmentedRegression fit = PiecewiseLinearUtils
+				.compressSuccessivePiecewiseLinear(column, cs);
+			breakpointsPerCol[col] = fit.getBreakpoints();
+			interceptsPerCol[col] = fit.getIntercepts();
+			slopesPerCol[col] = fit.getSlopes();
+
+		}
+		return ColGroupPiecewiseLinearCompressed.create(colIndexes, breakpointsPerCol, slopesPerCol, interceptsPerCol,
+			numRows);
+
+	}
+
 	private AColGroup compressSDCFromSparseTransposedBlock(IColIndex cols, int nrUniqueEstimate, double tupleSparsity) {
 		if(cols.size() > 1)
 			return compressMultiColSDCFromSparseTransposedBlock(cols, nrUniqueEstimate, tupleSparsity);

src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupIO.java

@@ -103,14 +103,16 @@ public static AColGroup readColGroup(DataInput in, int nRows) throws IOException
 		switch(ctype) {
 			case DDC:
 				return ColGroupDDC.read(in);
-		case DDCFOR:
-			return ColGroupDDCFOR.read(in);
-		case DeltaDDC:
-			return ColGroupDeltaDDC.read(in);
-		case DDCLZW:
-			return ColGroupDDCLZW.read(in);
-		case OLE:
-			return ColGroupOLE.read(in, nRows);
+			case DDCFOR:
+				return ColGroupDDCFOR.read(in);
+			case DeltaDDC:
+				return ColGroupDeltaDDC.read(in);
+			case DDCLZW:
+				return ColGroupDDCLZW.read(in);
+			case PiecewiseLinearCompressed:
+				return ColGroupPiecewiseLinearCompressed.read(in);
+			case OLE:
+				return ColGroupOLE.read(in, nRows);
 			case RLE:
 				return ColGroupRLE.read(in, nRows);
 			case CONST: