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,8 +114,9 @@ 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.
+	 *
+	 * 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 +136,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 +175,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
@@ -232,8 +233,9 @@ private void logEstVsActual(double time, AColGroup act, CompressedSizeInfoColGro
 					time, retType, estC, actC, act.getNumValues(), cols, wanted, warning));
 		}
 		else {
-			LOG.debug(String.format("time[ms]: %10.2f %25s est %10.0f -- act %10.0f distinct:%5d cols:%s wanted:%s",
-				time, retType, estC, actC, act.getNumValues(), cols, wanted));
+			LOG.debug(
+				String.format("time[ms]: %10.2f %25s est %10.0f -- act %10.0f distinct:%5d cols:%s wanted:%s", time,
+					retType, estC, actC, act.getNumValues(), cols, wanted));
 		}
 
 	}
@@ -306,6 +308,9 @@ else if(ct == CompressionType.LinearFunctional) {
 				return compressLinearFunctional(colIndexes, in, cs);
 			}
 		}
+		else if(ct == CompressionType.PiecewiseLinearCompressed) {
+			return compressPiecewiseLinearFunctional(colIndexes, in, cs);
+		}
 		else if(ct == CompressionType.DDCFOR) {
 			AColGroup g = directCompressDDC(colIndexes, cg);
 			if(g instanceof ColGroupDDC)
@@ -710,7 +715,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 +747,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 +756,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);
@@ -830,8 +836,8 @@ private boolean readToMapDDC(IColIndex colIndexes, DblArrayCountHashMap map, AMa
 		int fill) {
 
 		ReaderColumnSelection reader = (cs.scaleFactors == null) ? ReaderColumnSelection.createReader(in, colIndexes,
-			cs.transposed, rl,
-			ru) : ReaderColumnSelection.createQuantizedReader(in, colIndexes, cs.transposed, rl, ru, cs.scaleFactors);
+			cs.transposed, rl, ru) : ReaderColumnSelection.createQuantizedReader(in, colIndexes, cs.transposed, rl, ru,
+			cs.scaleFactors);
 
 		DblArray cellVals = reader.nextRow();
 		boolean extra = false;
@@ -1078,6 +1084,46 @@ private static AColGroup compressLinearFunctional(IColIndex colIndexes, MatrixBl
 		return ColGroupLinearFunctional.create(colIndexes, coefficients, numRows);
 	}
 
+	/**
+	 * This method is the entry point to compress a matrix with piecewise linear compression The first method uses a
+	 * segmented least squares with dynamic programming to compress the columns The second method uses a successive
+	 * 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 compressPiecewiseLinearFunctional(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);
+
+	}
+
+	public static AColGroup compressPiecewiseLinearFunctionalSuccessive(IColIndex colIndexes, MatrixBlock in,
+		CompressionSettings cs) {
+		return compressPiecewiseLinearFunctional(colIndexes, in, cs);
+	}
+
 	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

@@ -109,6 +109,8 @@ public static AColGroup readColGroup(DataInput in, int nRows) throws IOException
 			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: