janniklinde commented on code in PR #2398: URL: https://github.com/apache/systemds/pull/2398#discussion_r2839754001
########## src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupDDCLZW.java: ########## @@ -0,0 +1,1011 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.sysds.runtime.compress.colgroup; + +import org.apache.commons.lang3.NotImplementedException; +import org.apache.sysds.runtime.DMLRuntimeException; +import org.apache.sysds.runtime.compress.CompressedMatrixBlock; +import org.apache.sysds.runtime.compress.DMLCompressionException; +import org.apache.sysds.runtime.compress.colgroup.ColGroupUtils.P; +import org.apache.sysds.runtime.compress.colgroup.dictionary.Dictionary; +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.dictionary.MatrixBlockDictionary; +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.mapping.AMapToData; +import org.apache.sysds.runtime.compress.colgroup.mapping.MapToFactory; +import org.apache.sysds.runtime.compress.colgroup.scheme.DDCLZWScheme; +import org.apache.sysds.runtime.compress.colgroup.scheme.ICLAScheme; +import org.apache.sysds.runtime.compress.cost.ComputationCostEstimator; +import org.apache.sysds.runtime.compress.estim.CompressedSizeInfoColGroup; +import org.apache.sysds.runtime.compress.estim.EstimationFactors; +import org.apache.sysds.runtime.compress.estim.encoding.IEncode; +import org.apache.sysds.runtime.data.DenseBlock; +import org.apache.sysds.runtime.data.SparseBlock; +import org.apache.sysds.runtime.data.SparseBlockMCSR; +import org.apache.sysds.runtime.functionobjects.Builtin; +import org.apache.sysds.runtime.functionobjects.Minus; +import org.apache.sysds.runtime.functionobjects.Plus; +import org.apache.sysds.runtime.matrix.data.MatrixBlock; +import org.apache.sysds.runtime.matrix.operators.BinaryOperator; +import org.apache.sysds.runtime.matrix.operators.RightScalarOperator; +import org.apache.sysds.runtime.matrix.operators.ScalarOperator; +import org.apache.sysds.runtime.matrix.operators.UnaryOperator; +import org.apache.sysds.utils.MemoryEstimates; +import shaded.parquet.it.unimi.dsi.fastutil.ints.IntArrayList; +import shaded.parquet.it.unimi.dsi.fastutil.longs.Long2IntLinkedOpenHashMap; + +import java.io.DataInput; +import java.io.DataOutput; +import java.io.IOException; +import java.util.Arrays; +import java.util.Map; +import java.util.Stack; +import java.util.HashMap; +import java.util.NoSuchElementException; + +/** + * Class to encapsulate information about a column group that is encoded with dense dictionary encoding (DDC) whose + * mapping vector is additionally lzw compressed. Idea: - DDCLZW stores the mapping vector exclusively in compressed + * form. - No persistent MapToData cache is maintained. - Sequential operations decode on-the-fly, while operations + * requiring random access explicitly materialize and fall back to DDC. + */ +public class ColGroupDDCLZW extends APreAgg implements IMapToDataGroup { + private static final long serialVersionUID = -5769772089913918987L; + + /** + * Stores the LZW-compressed representation of data mapping. + */ + private final int[] _dataLZW; + private final int _nRows; + private final int _nUnique; + + private ColGroupDDCLZW(IColIndex colIndexes, IDictionary dict, AMapToData data, int[] cachedCounts) { + super(colIndexes, dict, cachedCounts); + _nRows = data.size(); + _nUnique = dict.getNumberOfValues(colIndexes.size()); + _dataLZW = compress(data); + + if(CompressedMatrixBlock.debug) { + if(getNumValues() == 0) + throw new DMLCompressionException("Invalid construction with empty dictionary"); + if(_nRows == 0) + throw new DMLCompressionException("Invalid length of the data. is zero"); + if(data.getUnique() != dict.getNumberOfValues(colIndexes.size())) + throw new DMLCompressionException( + "Invalid map to dict Map has:" + data.getUnique() + " while dict has " + + dict.getNumberOfValues(colIndexes.size())); + int[] c = getCounts(); + if(c.length != dict.getNumberOfValues(colIndexes.size())) + throw new DMLCompressionException("Invalid DDC Construction"); + data.verify(); + } + } + + private ColGroupDDCLZW(IColIndex colIndexes, IDictionary dict, int[] dataLZW, int nRows, int nUnique, + int[] cachedCounts) { + super(colIndexes, dict, cachedCounts); + _dataLZW = dataLZW; + _nRows = nRows; + _nUnique = nUnique; + + if(CompressedMatrixBlock.debug) { + if(getNumValues() == 0) + throw new DMLCompressionException("Invalid construction with empty dictionary"); + if(_nRows <= 0) + throw new DMLCompressionException("Invalid length of the data. is zero"); + if(_nUnique != dict.getNumberOfValues(colIndexes.size())) + throw new DMLCompressionException("Invalid map to dict Map has:" + _nUnique + " while dict has " + + dict.getNumberOfValues(colIndexes.size())); + int[] c = getCounts(); + if(c.length != dict.getNumberOfValues(colIndexes.size())) + throw new DMLCompressionException("Invalid DDC Construction"); + } + } + + public static AColGroup create(IColIndex colIndexes, IDictionary dict, AMapToData data, int[] cachedCounts) { + if(dict == null) + return new ColGroupEmpty(colIndexes); + else if(data.getUnique() == 1) + return ColGroupConst.create(colIndexes, dict); + else + return new ColGroupDDCLZW(colIndexes, dict, data, cachedCounts); + } + + /** + * Compresses the given data using the Lempel-Ziv-Welch (LZW) compression algorithm. The compression is performed on + * integer dictionary indices stored in the provided AMapToData object. + * + * @param data The input data to be compressed, represented as an AMapToData object. The data must not be null, must + * have at least one row, and contain valid dictionary indices. + * @return An array of integers representing the compressed data, with each integer corresponding to a dictionary + * code. + * @throws IllegalArgumentException If the input data is null, has no rows, contains no unique values, or has + * symbols that exceed the valid dictionary index range. + */ + private static int[] compress(AMapToData data) { + if(data == null) + throw new IllegalArgumentException("Invalid input: data is null"); + + final int nRows = data.size(); + if(nRows <= 0) { + throw new IllegalArgumentException("Invalid input: data has no rows"); + } + + final int nUnique = data.getUnique(); + if(nUnique <= 0) { + throw new IllegalArgumentException("Invalid input: data has no unique values"); + } + + if(nRows == 1) + return new int[] {data.getIndex(0)}; + + final Long2IntLinkedOpenHashMap dict = new Long2IntLinkedOpenHashMap(1 << 16); + dict.defaultReturnValue(-1); + + final IntArrayList out = new IntArrayList(Math.max(16, nRows / 2)); + + int nextCode = nUnique; + + int w = data.getIndex(0); + + for(int i = 1; i < nRows; i++) { + final int k = data.getIndex(i); + + if(k < 0 || k >= nUnique) + throw new IllegalArgumentException("Symbol out of range: " + k + " (nUnique=" + nUnique + ")"); + + final long key = packKey(w, k); + + int wk = dict.get(key); + if(wk != -1) { + w = wk; + } + else { + out.add(w); + dict.put(key, nextCode++); + w = k; + } + } + + out.add(w); + return out.toIntArray(); + } + + /** + * Decompresses the given lzw-encoded data into a full representation based on the provided parameters. + * + * @param codes an array of integers representing the lzw-compressed data + * @param nUnique the number of unique values in dictionary _dict + * @param nRows the total number of rows in the original data + * @return a fully decompressed AMapToData object representing the complete decompressed data + */ + private static AMapToData decompressFull(int[] codes, int nUnique, int nRows) { + if(codes == null) + throw new IllegalArgumentException("codes is null"); + if(codes.length == 0) + throw new IllegalArgumentException("codes is empty"); + if(nUnique <= 0) + throw new IllegalArgumentException("Invalid alphabet size: " + nUnique); + if(nRows <= 0) { + throw new IllegalArgumentException("Invalid nRows: " + nRows); + } + + final Map<Integer, Long> dict = new HashMap<>(); + + AMapToData out = MapToFactory.create(nRows, nUnique); + int outPos = 0; + + int old = codes[0]; + int[] oldPhrase = unpack(old, nUnique, dict); + + for(int v : oldPhrase) { + out.set(outPos++, v); + } + + int nextCode = nUnique; + + for(int i = 1; i < codes.length; i++) { + int key = codes[i]; + + int[] next; + if(key < nUnique || dict.containsKey(key)) { + next = unpack(key, nUnique, dict); + } + else { + int first = oldPhrase[0]; + next = packint(oldPhrase, first); + } + + for(int v : next) { + out.set(outPos++, v); + } + + final int first = next[0]; + dict.put(nextCode++, packKey(old, first)); + + old = key; + oldPhrase = next; + } + + if(outPos != nRows) + throw new IllegalStateException("Decompression length mismatch: got " + outPos + " expected " + nRows); + + return out; + } + + /** + * The LZWMappingIterator class is responsible for decoding and iterating through an LZW (Lempel-Ziv-Welch) + * compressed mapping. This iterator is primarily used for reconstructing symbols and phrases from the compressed + * mapping data, maintaining the internal state of the LZW decompression process. + * + * The decoding process maintains an LZW dictionary, tracks the current and previous phrases, handles new LZW codes, + * and provides methods to retrieve or skip mapping symbols. + */ + private final class LZWMappingIterator { + private final Map<Integer, Long> dict = new HashMap<>(); + private int lzwIndex = 0; + private int mapIndex = 0; + private int nextCode = _nUnique; + private int[] currentPhrase = null; + private int currentPhraseIndex = 0; + private int[] oldPhrase = null; + private int oldCode = -1; + + LZWMappingIterator() { + lzwIndex = 1; + oldCode = _dataLZW[0]; + oldPhrase = unpack(oldCode, _nUnique, dict); + currentPhrase = oldPhrase; + currentPhraseIndex = 0; + mapIndex = 0; + } + + boolean hasNext() { + return mapIndex < _nRows; + } + + void skip(int k) { + for(int i = 0; i < k; i++) + next(); + } + + int next() { + if(!hasNext()) + throw new NoSuchElementException(); + + if(currentPhraseIndex < currentPhrase.length) { + mapIndex++; + return currentPhrase[currentPhraseIndex++]; + } + + if(lzwIndex >= _dataLZW.length) + throw new IllegalStateException("Invalid LZW index: " + lzwIndex); + + final int key = _dataLZW[lzwIndex++]; + + final int[] next; + if(key < _nUnique || dict.containsKey(key)) { + next = unpack(key, _nUnique, dict); + } + else { + next = packint(oldPhrase, oldPhrase[0]); + } + + dict.put(nextCode++, packKey(oldCode, next[0])); + + oldCode = key; + oldPhrase = next; + + currentPhrase = next; + currentPhraseIndex = 0; + + mapIndex++; + return currentPhrase[currentPhraseIndex++]; + } + } + + /** + * Builds a packed 64-bit key from a prefix code and a next symbol, typically used in an LZW dictionary. + * + * @return a 64-bit packed key representing the (prefixCode, nextSymbol) pair + */ + private static long packKey(int prefixCode, int nextSymbol) { + return (((long) prefixCode) << 32) | (nextSymbol & 0xffffffffL); + } + + /** + * Extracts and returns the higher 32 bits of the given long key as an integer. + * + * @param key the long value from which the higher 32 bits will be unpacked + * @return the higher 32 bits of the given key as an integer + */ + private static int unpackfirst(long key) { + return (int) (key >>> 32); + } + + /** + * Extracts and returns the second component from the given long value key. This method assumes the key encodes the + * second component in its least significant bits. + * + * @param key the long value containing the encoded second component + * @return the extracted second component as an integer + */ + private static int unpacksecond(long key) { + return (int) (key); + } + + /** + * Creates a new array by appending the specified integer to the end of the given array. + * + * @param arr the original array to which the integer will be added + * @param last the integer value to be appended to the end of the array + * @return a new array containing all elements of the original array followed by the specified integer + */ + private static int[] packint(int[] arr, int last) { + int[] result = Arrays.copyOf(arr, arr.length + 1); + result[arr.length] = last; + return result; + } + + /** + * Decodes a given code into an array of integers based on a dictionary mapping. If the code is less than the number + * of unique symbols, it directly returns the code as a single-element array. Otherwise, it iteratively unpacks the + * code using a dictionary until the base symbols are resolved. + * + * @param code the encoded integer value to be unpacked + * @param nUnique the number of unique symbols; codes less than this are directly returned + * @param dict a mapping of integer codes to packed values represented as {@code Long}, used for unpacking + * @return an array of integers representing the unpacked sequence for the input code + * @throws IllegalStateException if the provided code does not have a corresponding entry in the dictionary + */ + private static int[] unpack(int code, int nUnique, Map<Integer, Long> dict) { + if(code < nUnique) + return new int[] {code}; + + Stack<Integer> stack = new Stack<>(); + int c = code; + + while(c >= nUnique) { + Long key = dict.get(c); + if(key == null) + throw new IllegalStateException("Missing dictionary entry for code: " + c); + + int symbol = unpacksecond(key); + stack.push(symbol); + c = unpackfirst(key); + } + + stack.push(c); + int[] outarray = new int[stack.size()]; + int i = 0; + while(!stack.isEmpty()) { + outarray[i++] = stack.pop(); + } + return outarray; + } + + /** + * Converts the current ColGroupDDCLZW instance to a ColGroupDDC instance. The method decompresses the + * LZW-compressed data of this instance, reconstructs the mapping to the decompressed data, and creates a new + * ColGroupDDC instance with the decompressed mapping, dictionary, and column indexes of this instance. + * + * @return an AColGroup instance representing the decoded ColGroup in DDC format + */ + public AColGroup convertToDDC() { + final AMapToData map = decompressFull(_dataLZW, _nUnique, _nRows); + final int[] counts = getCounts(); + return ColGroupDDC.create(_colIndexes, _dict, map, counts); + } + + public static ColGroupDDCLZW read(DataInput in) throws IOException { + final IColIndex colIndexes = ColIndexFactory.read(in); + final IDictionary dict = DictionaryFactory.read(in); + + final int nRows = in.readInt(); + final int nUnique = in.readInt(); + + final int len = in.readInt(); + if(len < 0) + throw new IOException("Invalid LZW data length: " + len); + + final int[] dataLZW = new int[len]; + for(int i = 0; i < len; i++) + dataLZW[i] = in.readInt(); + + return new ColGroupDDCLZW(colIndexes, dict, dataLZW, nRows, nUnique, null); + } + + @Override + public void write(DataOutput out) throws IOException { + super.write(out); + out.writeInt(_nRows); + out.writeInt(_nUnique); + out.writeInt(_dataLZW.length); + for(int i : _dataLZW) + out.writeInt(i); + } + + @Override + public double getIdx(int r, int colIdx) { + if(r < 0 || r >= _nRows) + throw new DMLRuntimeException("Row index out of bounds"); + + if(colIdx < 0 || colIdx >= _colIndexes.size()) + throw new DMLRuntimeException("Column index out of bounds"); + + final LZWMappingIterator it = new LZWMappingIterator(); + int dictIdx = -1; + for(int i = 0; i <= r; i++) { + dictIdx = it.next(); + } + return _dict.getValue(dictIdx, colIdx, _colIndexes.size()); + } + + @Override + public CompressionType getCompType() { + return CompressionType.DDCLZW; + } + + @Override + protected ColGroupType getColGroupType() { + return ColGroupType.DDCLZW; + } + + @Override + public boolean containsValue(double pattern) { + return _dict.containsValue(pattern); + } + + @Override + public double getCost(ComputationCostEstimator e, int nRows) { + final int nVals = getNumValues(); + final int nCols = getNumCols(); + return e.getCost(nRows, nRows, nCols, nVals, _dict.getSparsity()); + } + + @Override + public ICLAScheme getCompressionScheme() { + return DDCLZWScheme.create(this); + } + + @Override + protected int numRowsToMultiply() { + return _nRows; + } + + @Override + protected AColGroup copyAndSet(IColIndex colIndexes, IDictionary newDictionary) { + return new ColGroupDDCLZW(colIndexes, newDictionary, _dataLZW, _nRows, _nUnique, getCachedCounts()); + } + + @Override + public long estimateInMemorySize() { + long size = super.estimateInMemorySize(); + + if(_dataLZW != null) + size += (long) MemoryEstimates.intArrayCost(_dataLZW.length); + + size += 2L * Long.BYTES; // _nRows, _nUnique, _dataLZW.length + return size; + } + + @Override + public long getExactSizeOnDisk() { + long ret = super.getExactSizeOnDisk(); + ret += 12; + ret += (long) _dataLZW.length * 4; + return ret; + } + + @Override + public AMapToData getMapToData() { + return decompressFull(_dataLZW, _nUnique, _nRows); + } + + @Override + public boolean sameIndexStructure(AColGroupCompressed that) { + return that instanceof ColGroupDDCLZW && ((ColGroupDDCLZW) that)._dataLZW == this._dataLZW; + } + + @Override + protected double computeMxx(double c, Builtin builtin) { + return _dict.aggregate(c, builtin); + } + + @Override + protected void computeColMxx(double[] c, Builtin builtin) { + _dict.aggregateCols(c, builtin, _colIndexes); + } + + @Override + public AColGroup sliceRows(int rl, int ru) { + try { + if(rl < 0 || ru > _nRows) + throw new DMLRuntimeException("Invalid slice range: " + rl + " - " + ru); + + final int len = ru - rl; + if(len == 0) + return new ColGroupEmpty(_colIndexes); + + final int[] slicedMapping = new int[len]; + + final LZWMappingIterator it = new LZWMappingIterator(); + + for(int i = 0; i < rl; i++) + it.next(); + + for(int i = rl; i < ru; i++) + slicedMapping[i - rl] = it.next(); + + AMapToData slicedMappingAMapToData = MapToFactory.create(len, _nUnique); + for(int i = 0; i < len; i++) { + slicedMappingAMapToData.set(i, slicedMapping[i]); + } + + return new ColGroupDDCLZW(_colIndexes, _dict, slicedMappingAMapToData, null); + } + catch(Exception e) { + throw new DMLRuntimeException("Failed to slice out sub part DDCLZW: " + rl + ", " + ru, e); + } + } + + @Override + protected void decompressToDenseBlockTransposedSparseDictionary(DenseBlock db, int rl, int ru, SparseBlock sb) { + LZWMappingIterator it = new LZWMappingIterator(); + for(int i = 0; i < rl; i++) { + it.next(); + } + + for(int i = rl; i < ru; i++) { + final int vr = it.next(); + if(sb.isEmpty(vr)) + continue; + final int apos = sb.pos(vr); + final int alen = sb.size(vr) + apos; + final int[] aix = sb.indexes(vr); + final double[] aval = sb.values(vr); + for(int j = apos; j < alen; j++) { + final int rowOut = _colIndexes.get(aix[j]); + final double[] c = db.values(rowOut); + final int off = db.pos(rowOut); + c[off + i] += aval[j]; + } + } + } + + @Override + protected void decompressToDenseBlockTransposedDenseDictionary(DenseBlock db, int rl, int ru, double[] dict) { + ColGroupDDC g = (ColGroupDDC) convertToDDC(); + g.decompressToDenseBlockTransposedDenseDictionary(db, rl, ru, dict); + + } + + @Override + protected void decompressToSparseBlockTransposedSparseDictionary(SparseBlockMCSR sbr, SparseBlock sb, int nColOut) { + + int[] colCounts = _dict.countNNZZeroColumns(getCounts()); + for(int j = 0; j < _colIndexes.size(); j++) + sbr.allocate(_colIndexes.get(j), colCounts[j]); + + LZWMappingIterator it = new LZWMappingIterator(); + + for(int i = 0; i < _nRows; i++) { + int di = it.next(); + if(sb.isEmpty(di)) + continue; + + final int apos = sb.pos(di); + final int alen = sb.size(di) + apos; + final int[] aix = sb.indexes(di); + final double[] aval = sb.values(di); + + for(int j = apos; j < alen; j++) { + sbr.append(_colIndexes.get(aix[j]), i, aval[apos]); Review Comment: I think you mean `aval[j]` and not `aval[apos]`, right? ########## src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupDDCLZW.java: ########## @@ -0,0 +1,1011 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.sysds.runtime.compress.colgroup; + +import org.apache.commons.lang3.NotImplementedException; +import org.apache.sysds.runtime.DMLRuntimeException; +import org.apache.sysds.runtime.compress.CompressedMatrixBlock; +import org.apache.sysds.runtime.compress.DMLCompressionException; +import org.apache.sysds.runtime.compress.colgroup.ColGroupUtils.P; +import org.apache.sysds.runtime.compress.colgroup.dictionary.Dictionary; +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.dictionary.MatrixBlockDictionary; +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.mapping.AMapToData; +import org.apache.sysds.runtime.compress.colgroup.mapping.MapToFactory; +import org.apache.sysds.runtime.compress.colgroup.scheme.DDCLZWScheme; +import org.apache.sysds.runtime.compress.colgroup.scheme.ICLAScheme; +import org.apache.sysds.runtime.compress.cost.ComputationCostEstimator; +import org.apache.sysds.runtime.compress.estim.CompressedSizeInfoColGroup; +import org.apache.sysds.runtime.compress.estim.EstimationFactors; +import org.apache.sysds.runtime.compress.estim.encoding.IEncode; +import org.apache.sysds.runtime.data.DenseBlock; +import org.apache.sysds.runtime.data.SparseBlock; +import org.apache.sysds.runtime.data.SparseBlockMCSR; +import org.apache.sysds.runtime.functionobjects.Builtin; +import org.apache.sysds.runtime.functionobjects.Minus; +import org.apache.sysds.runtime.functionobjects.Plus; +import org.apache.sysds.runtime.matrix.data.MatrixBlock; +import org.apache.sysds.runtime.matrix.operators.BinaryOperator; +import org.apache.sysds.runtime.matrix.operators.RightScalarOperator; +import org.apache.sysds.runtime.matrix.operators.ScalarOperator; +import org.apache.sysds.runtime.matrix.operators.UnaryOperator; +import org.apache.sysds.utils.MemoryEstimates; +import shaded.parquet.it.unimi.dsi.fastutil.ints.IntArrayList; +import shaded.parquet.it.unimi.dsi.fastutil.longs.Long2IntLinkedOpenHashMap; + +import java.io.DataInput; +import java.io.DataOutput; +import java.io.IOException; +import java.util.Arrays; +import java.util.Map; +import java.util.Stack; +import java.util.HashMap; +import java.util.NoSuchElementException; + +/** + * Class to encapsulate information about a column group that is encoded with dense dictionary encoding (DDC) whose + * mapping vector is additionally lzw compressed. Idea: - DDCLZW stores the mapping vector exclusively in compressed + * form. - No persistent MapToData cache is maintained. - Sequential operations decode on-the-fly, while operations + * requiring random access explicitly materialize and fall back to DDC. + */ +public class ColGroupDDCLZW extends APreAgg implements IMapToDataGroup { + private static final long serialVersionUID = -5769772089913918987L; + + /** + * Stores the LZW-compressed representation of data mapping. + */ + private final int[] _dataLZW; + private final int _nRows; + private final int _nUnique; + + private ColGroupDDCLZW(IColIndex colIndexes, IDictionary dict, AMapToData data, int[] cachedCounts) { + super(colIndexes, dict, cachedCounts); + _nRows = data.size(); + _nUnique = dict.getNumberOfValues(colIndexes.size()); + _dataLZW = compress(data); + + if(CompressedMatrixBlock.debug) { + if(getNumValues() == 0) + throw new DMLCompressionException("Invalid construction with empty dictionary"); + if(_nRows == 0) + throw new DMLCompressionException("Invalid length of the data. is zero"); + if(data.getUnique() != dict.getNumberOfValues(colIndexes.size())) + throw new DMLCompressionException( + "Invalid map to dict Map has:" + data.getUnique() + " while dict has " + + dict.getNumberOfValues(colIndexes.size())); + int[] c = getCounts(); + if(c.length != dict.getNumberOfValues(colIndexes.size())) + throw new DMLCompressionException("Invalid DDC Construction"); + data.verify(); + } + } + + private ColGroupDDCLZW(IColIndex colIndexes, IDictionary dict, int[] dataLZW, int nRows, int nUnique, + int[] cachedCounts) { + super(colIndexes, dict, cachedCounts); + _dataLZW = dataLZW; + _nRows = nRows; + _nUnique = nUnique; + + if(CompressedMatrixBlock.debug) { + if(getNumValues() == 0) + throw new DMLCompressionException("Invalid construction with empty dictionary"); + if(_nRows <= 0) + throw new DMLCompressionException("Invalid length of the data. is zero"); + if(_nUnique != dict.getNumberOfValues(colIndexes.size())) + throw new DMLCompressionException("Invalid map to dict Map has:" + _nUnique + " while dict has " + + dict.getNumberOfValues(colIndexes.size())); + int[] c = getCounts(); + if(c.length != dict.getNumberOfValues(colIndexes.size())) + throw new DMLCompressionException("Invalid DDC Construction"); + } + } + + public static AColGroup create(IColIndex colIndexes, IDictionary dict, AMapToData data, int[] cachedCounts) { + if(dict == null) + return new ColGroupEmpty(colIndexes); + else if(data.getUnique() == 1) + return ColGroupConst.create(colIndexes, dict); + else + return new ColGroupDDCLZW(colIndexes, dict, data, cachedCounts); + } + + /** + * Compresses the given data using the Lempel-Ziv-Welch (LZW) compression algorithm. The compression is performed on + * integer dictionary indices stored in the provided AMapToData object. + * + * @param data The input data to be compressed, represented as an AMapToData object. The data must not be null, must + * have at least one row, and contain valid dictionary indices. + * @return An array of integers representing the compressed data, with each integer corresponding to a dictionary + * code. + * @throws IllegalArgumentException If the input data is null, has no rows, contains no unique values, or has + * symbols that exceed the valid dictionary index range. + */ + private static int[] compress(AMapToData data) { + if(data == null) + throw new IllegalArgumentException("Invalid input: data is null"); + + final int nRows = data.size(); + if(nRows <= 0) { + throw new IllegalArgumentException("Invalid input: data has no rows"); + } + + final int nUnique = data.getUnique(); + if(nUnique <= 0) { + throw new IllegalArgumentException("Invalid input: data has no unique values"); + } + + if(nRows == 1) + return new int[] {data.getIndex(0)}; + + final Long2IntLinkedOpenHashMap dict = new Long2IntLinkedOpenHashMap(1 << 16); + dict.defaultReturnValue(-1); + + final IntArrayList out = new IntArrayList(Math.max(16, nRows / 2)); + + int nextCode = nUnique; + + int w = data.getIndex(0); + + for(int i = 1; i < nRows; i++) { + final int k = data.getIndex(i); + + if(k < 0 || k >= nUnique) + throw new IllegalArgumentException("Symbol out of range: " + k + " (nUnique=" + nUnique + ")"); + + final long key = packKey(w, k); + + int wk = dict.get(key); + if(wk != -1) { + w = wk; + } + else { + out.add(w); + dict.put(key, nextCode++); + w = k; + } + } + + out.add(w); + return out.toIntArray(); + } + + /** + * Decompresses the given lzw-encoded data into a full representation based on the provided parameters. + * + * @param codes an array of integers representing the lzw-compressed data + * @param nUnique the number of unique values in dictionary _dict + * @param nRows the total number of rows in the original data + * @return a fully decompressed AMapToData object representing the complete decompressed data + */ + private static AMapToData decompressFull(int[] codes, int nUnique, int nRows) { + if(codes == null) + throw new IllegalArgumentException("codes is null"); + if(codes.length == 0) + throw new IllegalArgumentException("codes is empty"); + if(nUnique <= 0) + throw new IllegalArgumentException("Invalid alphabet size: " + nUnique); + if(nRows <= 0) { + throw new IllegalArgumentException("Invalid nRows: " + nRows); + } + + final Map<Integer, Long> dict = new HashMap<>(); + + AMapToData out = MapToFactory.create(nRows, nUnique); + int outPos = 0; + + int old = codes[0]; + int[] oldPhrase = unpack(old, nUnique, dict); + + for(int v : oldPhrase) { + out.set(outPos++, v); + } + + int nextCode = nUnique; + + for(int i = 1; i < codes.length; i++) { + int key = codes[i]; + + int[] next; + if(key < nUnique || dict.containsKey(key)) { + next = unpack(key, nUnique, dict); + } + else { + int first = oldPhrase[0]; + next = packint(oldPhrase, first); + } + + for(int v : next) { + out.set(outPos++, v); + } + + final int first = next[0]; + dict.put(nextCode++, packKey(old, first)); + + old = key; + oldPhrase = next; + } + + if(outPos != nRows) + throw new IllegalStateException("Decompression length mismatch: got " + outPos + " expected " + nRows); + + return out; + } + + /** + * The LZWMappingIterator class is responsible for decoding and iterating through an LZW (Lempel-Ziv-Welch) + * compressed mapping. This iterator is primarily used for reconstructing symbols and phrases from the compressed + * mapping data, maintaining the internal state of the LZW decompression process. + * + * The decoding process maintains an LZW dictionary, tracks the current and previous phrases, handles new LZW codes, + * and provides methods to retrieve or skip mapping symbols. + */ + private final class LZWMappingIterator { + private final Map<Integer, Long> dict = new HashMap<>(); + private int lzwIndex = 0; + private int mapIndex = 0; + private int nextCode = _nUnique; + private int[] currentPhrase = null; + private int currentPhraseIndex = 0; + private int[] oldPhrase = null; + private int oldCode = -1; + + LZWMappingIterator() { + lzwIndex = 1; + oldCode = _dataLZW[0]; + oldPhrase = unpack(oldCode, _nUnique, dict); + currentPhrase = oldPhrase; + currentPhraseIndex = 0; + mapIndex = 0; + } + + boolean hasNext() { + return mapIndex < _nRows; + } + + void skip(int k) { + for(int i = 0; i < k; i++) + next(); + } + + int next() { + if(!hasNext()) + throw new NoSuchElementException(); + + if(currentPhraseIndex < currentPhrase.length) { + mapIndex++; + return currentPhrase[currentPhraseIndex++]; + } + + if(lzwIndex >= _dataLZW.length) + throw new IllegalStateException("Invalid LZW index: " + lzwIndex); + + final int key = _dataLZW[lzwIndex++]; + + final int[] next; + if(key < _nUnique || dict.containsKey(key)) { + next = unpack(key, _nUnique, dict); + } + else { + next = packint(oldPhrase, oldPhrase[0]); + } + + dict.put(nextCode++, packKey(oldCode, next[0])); + + oldCode = key; + oldPhrase = next; + + currentPhrase = next; + currentPhraseIndex = 0; + + mapIndex++; + return currentPhrase[currentPhraseIndex++]; + } + } + + /** + * Builds a packed 64-bit key from a prefix code and a next symbol, typically used in an LZW dictionary. + * + * @return a 64-bit packed key representing the (prefixCode, nextSymbol) pair + */ + private static long packKey(int prefixCode, int nextSymbol) { + return (((long) prefixCode) << 32) | (nextSymbol & 0xffffffffL); + } + + /** + * Extracts and returns the higher 32 bits of the given long key as an integer. + * + * @param key the long value from which the higher 32 bits will be unpacked + * @return the higher 32 bits of the given key as an integer + */ + private static int unpackfirst(long key) { + return (int) (key >>> 32); + } + + /** + * Extracts and returns the second component from the given long value key. This method assumes the key encodes the + * second component in its least significant bits. + * + * @param key the long value containing the encoded second component + * @return the extracted second component as an integer + */ + private static int unpacksecond(long key) { + return (int) (key); + } + + /** + * Creates a new array by appending the specified integer to the end of the given array. + * + * @param arr the original array to which the integer will be added + * @param last the integer value to be appended to the end of the array + * @return a new array containing all elements of the original array followed by the specified integer + */ + private static int[] packint(int[] arr, int last) { + int[] result = Arrays.copyOf(arr, arr.length + 1); + result[arr.length] = last; + return result; + } + + /** + * Decodes a given code into an array of integers based on a dictionary mapping. If the code is less than the number + * of unique symbols, it directly returns the code as a single-element array. Otherwise, it iteratively unpacks the + * code using a dictionary until the base symbols are resolved. + * + * @param code the encoded integer value to be unpacked + * @param nUnique the number of unique symbols; codes less than this are directly returned + * @param dict a mapping of integer codes to packed values represented as {@code Long}, used for unpacking + * @return an array of integers representing the unpacked sequence for the input code + * @throws IllegalStateException if the provided code does not have a corresponding entry in the dictionary + */ + private static int[] unpack(int code, int nUnique, Map<Integer, Long> dict) { + if(code < nUnique) + return new int[] {code}; + + Stack<Integer> stack = new Stack<>(); + int c = code; + + while(c >= nUnique) { + Long key = dict.get(c); + if(key == null) + throw new IllegalStateException("Missing dictionary entry for code: " + c); + + int symbol = unpacksecond(key); + stack.push(symbol); + c = unpackfirst(key); + } + + stack.push(c); + int[] outarray = new int[stack.size()]; + int i = 0; + while(!stack.isEmpty()) { + outarray[i++] = stack.pop(); + } + return outarray; + } + + /** + * Converts the current ColGroupDDCLZW instance to a ColGroupDDC instance. The method decompresses the + * LZW-compressed data of this instance, reconstructs the mapping to the decompressed data, and creates a new + * ColGroupDDC instance with the decompressed mapping, dictionary, and column indexes of this instance. + * + * @return an AColGroup instance representing the decoded ColGroup in DDC format + */ + public AColGroup convertToDDC() { + final AMapToData map = decompressFull(_dataLZW, _nUnique, _nRows); + final int[] counts = getCounts(); + return ColGroupDDC.create(_colIndexes, _dict, map, counts); + } + + public static ColGroupDDCLZW read(DataInput in) throws IOException { + final IColIndex colIndexes = ColIndexFactory.read(in); + final IDictionary dict = DictionaryFactory.read(in); + + final int nRows = in.readInt(); + final int nUnique = in.readInt(); + + final int len = in.readInt(); + if(len < 0) + throw new IOException("Invalid LZW data length: " + len); + + final int[] dataLZW = new int[len]; + for(int i = 0; i < len; i++) + dataLZW[i] = in.readInt(); + + return new ColGroupDDCLZW(colIndexes, dict, dataLZW, nRows, nUnique, null); + } + + @Override + public void write(DataOutput out) throws IOException { + super.write(out); + out.writeInt(_nRows); + out.writeInt(_nUnique); + out.writeInt(_dataLZW.length); + for(int i : _dataLZW) + out.writeInt(i); + } + + @Override + public double getIdx(int r, int colIdx) { + if(r < 0 || r >= _nRows) + throw new DMLRuntimeException("Row index out of bounds"); + + if(colIdx < 0 || colIdx >= _colIndexes.size()) + throw new DMLRuntimeException("Column index out of bounds"); + + final LZWMappingIterator it = new LZWMappingIterator(); + int dictIdx = -1; + for(int i = 0; i <= r; i++) { + dictIdx = it.next(); + } + return _dict.getValue(dictIdx, colIdx, _colIndexes.size()); + } + + @Override + public CompressionType getCompType() { + return CompressionType.DDCLZW; + } + + @Override + protected ColGroupType getColGroupType() { + return ColGroupType.DDCLZW; + } + + @Override + public boolean containsValue(double pattern) { + return _dict.containsValue(pattern); + } + + @Override + public double getCost(ComputationCostEstimator e, int nRows) { + final int nVals = getNumValues(); + final int nCols = getNumCols(); + return e.getCost(nRows, nRows, nCols, nVals, _dict.getSparsity()); + } + + @Override + public ICLAScheme getCompressionScheme() { + return DDCLZWScheme.create(this); + } + + @Override + protected int numRowsToMultiply() { + return _nRows; + } + + @Override + protected AColGroup copyAndSet(IColIndex colIndexes, IDictionary newDictionary) { + return new ColGroupDDCLZW(colIndexes, newDictionary, _dataLZW, _nRows, _nUnique, getCachedCounts()); + } + + @Override + public long estimateInMemorySize() { + long size = super.estimateInMemorySize(); + + if(_dataLZW != null) + size += (long) MemoryEstimates.intArrayCost(_dataLZW.length); + + size += 2L * Long.BYTES; // _nRows, _nUnique, _dataLZW.length + return size; + } + + @Override + public long getExactSizeOnDisk() { + long ret = super.getExactSizeOnDisk(); + ret += 12; + ret += (long) _dataLZW.length * 4; + return ret; + } + + @Override + public AMapToData getMapToData() { + return decompressFull(_dataLZW, _nUnique, _nRows); + } + + @Override + public boolean sameIndexStructure(AColGroupCompressed that) { + return that instanceof ColGroupDDCLZW && ((ColGroupDDCLZW) that)._dataLZW == this._dataLZW; + } + + @Override + protected double computeMxx(double c, Builtin builtin) { + return _dict.aggregate(c, builtin); + } + + @Override + protected void computeColMxx(double[] c, Builtin builtin) { + _dict.aggregateCols(c, builtin, _colIndexes); + } + + @Override + public AColGroup sliceRows(int rl, int ru) { + try { + if(rl < 0 || ru > _nRows) + throw new DMLRuntimeException("Invalid slice range: " + rl + " - " + ru); + + final int len = ru - rl; + if(len == 0) + return new ColGroupEmpty(_colIndexes); + + final int[] slicedMapping = new int[len]; + + final LZWMappingIterator it = new LZWMappingIterator(); + + for(int i = 0; i < rl; i++) + it.next(); + + for(int i = rl; i < ru; i++) + slicedMapping[i - rl] = it.next(); + + AMapToData slicedMappingAMapToData = MapToFactory.create(len, _nUnique); + for(int i = 0; i < len; i++) { + slicedMappingAMapToData.set(i, slicedMapping[i]); + } + + return new ColGroupDDCLZW(_colIndexes, _dict, slicedMappingAMapToData, null); + } + catch(Exception e) { + throw new DMLRuntimeException("Failed to slice out sub part DDCLZW: " + rl + ", " + ru, e); + } + } + + @Override + protected void decompressToDenseBlockTransposedSparseDictionary(DenseBlock db, int rl, int ru, SparseBlock sb) { + LZWMappingIterator it = new LZWMappingIterator(); + for(int i = 0; i < rl; i++) { + it.next(); + } + + for(int i = rl; i < ru; i++) { + final int vr = it.next(); + if(sb.isEmpty(vr)) + continue; + final int apos = sb.pos(vr); + final int alen = sb.size(vr) + apos; + final int[] aix = sb.indexes(vr); + final double[] aval = sb.values(vr); + for(int j = apos; j < alen; j++) { + final int rowOut = _colIndexes.get(aix[j]); + final double[] c = db.values(rowOut); + final int off = db.pos(rowOut); + c[off + i] += aval[j]; + } + } + } + + @Override + protected void decompressToDenseBlockTransposedDenseDictionary(DenseBlock db, int rl, int ru, double[] dict) { + ColGroupDDC g = (ColGroupDDC) convertToDDC(); + g.decompressToDenseBlockTransposedDenseDictionary(db, rl, ru, dict); + + } + + @Override + protected void decompressToSparseBlockTransposedSparseDictionary(SparseBlockMCSR sbr, SparseBlock sb, int nColOut) { + + int[] colCounts = _dict.countNNZZeroColumns(getCounts()); + for(int j = 0; j < _colIndexes.size(); j++) + sbr.allocate(_colIndexes.get(j), colCounts[j]); + + LZWMappingIterator it = new LZWMappingIterator(); + + for(int i = 0; i < _nRows; i++) { + int di = it.next(); + if(sb.isEmpty(di)) + continue; + + final int apos = sb.pos(di); + final int alen = sb.size(di) + apos; + final int[] aix = sb.indexes(di); + final double[] aval = sb.values(di); + + for(int j = apos; j < alen; j++) { + sbr.append(_colIndexes.get(aix[j]), i, aval[apos]); + } + } + } + + @Override + protected void decompressToSparseBlockTransposedDenseDictionary(SparseBlockMCSR db, double[] dict, int nColOut) { + ColGroupDDC g = (ColGroupDDC) convertToDDC(); + g.decompressToSparseBlockTransposedDenseDictionary(db, dict, nColOut); + } + + @Override + protected void decompressToDenseBlockSparseDictionary(DenseBlock db, int rl, int ru, int offR, int offC, + SparseBlock sb) { + LZWMappingIterator it = new LZWMappingIterator(); + for(int i = 0; i < rl; i++) { + it.next(); + } + + for(int r = rl, offT = rl + offR; r < ru; r++, offT++) { + final int vr = it.next(); + if(sb.isEmpty(vr)) + continue; + final double[] c = db.values(offT); + final int off = db.pos(offT) + offC; + _colIndexes.decompressToDenseFromSparse(sb, vr, off, c); + } + } + + @Override + protected void decompressToDenseBlockDenseDictionary(DenseBlock db, int rl, int ru, int offR, int offC, + double[] values) { + final int nCol = _colIndexes.size(); + final LZWMappingIterator it = new LZWMappingIterator(); + + for(int i = 0; i < rl; i++) { + it.next(); + } + + if(db.isContiguous() && nCol == db.getDim(1) && offC == 0) { + final int nColOut = db.getDim(1); + final double[] c = db.values(0); + + for(int i = rl; i < ru; i++) { + final int dictIdx = it.next(); + final int rowIndex = dictIdx * nCol; + final int rowBaseOff = (i + offR) * nColOut; + + for(int j = 0; j < nCol; j++) + c[rowBaseOff + j] = values[rowIndex + j]; + } + } + else { + for(int i = rl, offT = rl + offR; i < ru; i++, offT++) { + final double[] c = db.values(offT); + final int off = db.pos(offT) + offC; + final int dictIdx = it.next(); + final int rowIndex = dictIdx * nCol; + + for(int j = 0; j < nCol; j++) { + final int colIdx = _colIndexes.get(j); + c[off + colIdx] = values[rowIndex + j]; Review Comment: Setting instead of using `+=` could be problematic, at least `AColIndex.decrompressVec` uses `+=` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
