mboehm7 commented on a change in pull request #1346:
URL: https://github.com/apache/systemds/pull/1346#discussion_r676151614
##########
File path: src/main/java/org/apache/sysds/hops/AggBinaryOp.java
##########
@@ -178,6 +181,11 @@ public Lop constructLops()
MMTSJType mmtsj = checkTransposeSelf(); //determine
tsmm pattern
ChainType chain = checkMapMultChain(); //determine
mmchain pattern
+ if(mmtsj == MMTSJType.LEFT &&
input2.isCompressedOutput()){
Review comment:
you might want to have an additional memory constraint on it, unless the
isCompressedOutput selection of input2 would already have done this implicitly;
otherwise good
##########
File path: src/main/java/org/apache/sysds/runtime/matrix/data/LibMatrixAgg.java
##########
@@ -1639,6 +1639,10 @@ else if( ixFn instanceof ReduceRow ) //COLVAR
out.quickSetValue(2, j,
in.rlen); //count
break;
}
+
+ case CUM_SUM_PROD:{
Review comment:
is a cum_sum_prod actually ever coming here?
##########
File path:
src/main/java/org/apache/sysds/runtime/matrix/data/LibMatrixReorg.java
##########
@@ -1754,14 +1754,14 @@ static void transposeRow( double[] a, double[] c, int
aix, int cix, int n2, int
//initial pass to determine capacity (this helps to prevent
//sparse row reallocations and mem inefficiency w/ skew
int[] cnt = null;
- if( in.sparse && in.clen <= 4096 ) { //16KB
+ // if( in.sparse && in.clen <= 4096 ) { //16KB
Review comment:
ensure some maximum bound on the temporary memory requirements. Also, I
will have a look into more efficient transpose operations for ultra-sparse
matrices.
##########
File path: src/main/java/org/apache/sysds/hops/AggBinaryOp.java
##########
@@ -306,9 +314,21 @@ protected double computeOutputMemEstimate( long dim1, long
dim2, long nnz )
sparsity = OptimizerUtils.getSparsity(dim1,
dim2, nnz);
}
*/
- //currently always estimated as dense in order to account for
dense intermediate without unnecessary overestimation
- double ret = OptimizerUtils.estimateSizeExactSparsity(dim1,
dim2, sparsity);
-
+ Hop input1 = getInput().get(0);
+ if(isMatrixMultiply() && hasCompressedInput() &&
!isRequiredDecompression() && input1.isCompressedOutput()) {
+
+ // right matrix multiplication ... compressed output
+ setCompressedOutput(true);
+ // conservatively set the size to a multiplication of
the compression size
+ // multiplied with the number of columns.
+ setCompressedSize(input1._compressedSize * dim2);
+ ret = getCompressedSize();
Review comment:
I would recommend moving the propagation of these properties into
refreshSizeInformation()
##########
File path:
src/main/java/org/apache/sysds/runtime/controlprogram/caching/CacheableData.java
##########
@@ -598,12 +616,16 @@ public T acquireModify(T newData) {
Statistics.addCPMemObject(System.identityHashCode(this), getDataSize());
}
+ if(newData instanceof CompressedMatrixBlock) {
+ setCompressedSize(newData.getInMemorySize());
+ }
+
return ret;
}
private synchronized T acquireModifyIntern(T newData) {
if (! isAvailableToModify ())
- throw new DMLRuntimeException("CacheableData not
available to modify.");
+ throw new DMLRuntimeException("CacheableData not available to
modify.");
Review comment:
indentation
##########
File path:
src/main/java/org/apache/sysds/runtime/matrix/data/LibMatrixReorg.java
##########
@@ -193,7 +193,7 @@ public static MatrixBlock transpose(MatrixBlock in,
MatrixBlock out, int k, bool
if(in.isEmptyBlock(false) || ((long) in.rlen * (long) in.clen <
PAR_NUMCELL_THRESHOLD) || k == 1 ||
(SHALLOW_COPY_REORG && !in.sparse && !out.sparse &&
(in.rlen == 1 || in.clen == 1)) ||
(in.sparse && !out.sparse && in.rlen == 1) ||
(!in.sparse && out.sparse && in.rlen == 1) ||
- (!in.sparse && out.sparse)) {
+ (!in.sparse && out.sparse) || (in.nonZeros != -1 &&
in.nonZeros > 1000)) {
Review comment:
Please double check this conditions - it looks like all large transpose
operations are forced into single-threaded execution.
##########
File path:
src/main/java/org/apache/sysds/runtime/compress/utils/DblArrayIntListHashMap.java
##########
@@ -20,33 +20,58 @@
package org.apache.sysds.runtime.compress.utils;
import java.util.ArrayList;
-import java.util.Comparator;
+import java.util.Arrays;
+import java.util.List;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
-/**
- * This class provides a memory-efficient replacement for {@code
HashMap<DblArray,IntArrayList>} for restricted use
- * cases.
- *
- */
-public class DblArrayIntListHashMap extends CustomHashMap {
+public class DblArrayIntListHashMap {
protected static final Log LOG =
LogFactory.getLog(DblArrayIntListHashMap.class.getName());
- private DArrayIListEntry[] _data = null;
+ protected static final int INIT_CAPACITY = 8;
+ protected static final int RESIZE_FACTOR = 2;
+ protected static final float LOAD_FACTOR = 0.5f;
public static int hashMissCount = 0;
+ protected int _size = -1;
+
+ protected DArrayIListEntry[] _data = null;
+
+ public int size() {
+ return _size;
+ }
+
public DblArrayIntListHashMap() {
_data = new DArrayIListEntry[INIT_CAPACITY];
_size = 0;
}
public DblArrayIntListHashMap(int init_capacity) {
- _data = new DArrayIListEntry[init_capacity];
+ _data = new DArrayIListEntry[getPow2(init_capacity)];
_size = 0;
}
+ private int getPow2(int x) {
+ x = x - 1;
+ x |= x >> 1;
+ x |= x >> 2;
+ x |= x >> 4;
+ x |= x >> 5;
+ x |= x >> 6;
+ x |= x >> 8;
+ x |= x >> 9;
+ x |= x >> 10;
+ x |= x >> 11;
+ x |= x >> 12;
+ x |= x >> 13;
+ x |= x >> 14;
+ x |= x >> 15;
+ x |= x >> 16;
+ return Math.max(x + 1, 4);
Review comment:
replace this implementation with something like UtilFunctions.nextIntPow2
##########
File path:
src/main/java/org/apache/sysds/runtime/instructions/cp/CompressionCPInstruction.java
##########
@@ -66,13 +66,17 @@ public void processInstruction(ExecutionContext ec) {
SingletonLookupHashMap m = SingletonLookupHashMap.getMap();
WTreeRoot root = (_singletonLookupID != 0) ? (WTreeRoot)
m.get(_singletonLookupID) : null;
+
// Compress the matrix block
- Pair<MatrixBlock, CompressionStatistics> compResult =
CompressedMatrixBlockFactory.compress(in,
OptimizerUtils.getConstrainedNumThreads(-1), root);
+ Pair<MatrixBlock, CompressionStatistics> compResult = (root ==
null ||
+ root.getDecompressList() == null) ?
CompressedMatrixBlockFactory.compress(in,
Review comment:
the formatting is a bit off - maybe use the ? : as natural splitting
points.
##########
File path: src/main/java/org/apache/sysds/hops/BinaryOp.java
##########
@@ -543,7 +549,12 @@ protected double computeOutputMemEstimate( long dim1, long
dim2, long nnz )
else //e.g., for append,pow or after inference
sparsity = OptimizerUtils.getSparsity(dim1,
dim2, nnz);
- ret = OptimizerUtils.estimateSizeExactSparsity(dim1,
dim2, sparsity);
+ ret = OptimizerUtils.estimateSizeExactSparsity(dim1,
dim2, sparsity);
+
+ if(hasCompressedInput()){
+
Review comment:
avoid such empty lines and move to refreshSizeInformation
##########
File path:
src/test/java/org/apache/sysds/test/component/compress/CompressedSingleTests.java
##########
@@ -0,0 +1,151 @@
+/*
+ * 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.test.component.compress;
+
+import static org.junit.Assert.assertEquals;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.util.ArrayList;
+
+import org.apache.commons.lang.NotImplementedException;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.sysds.runtime.DMLRuntimeException;
+import org.apache.sysds.runtime.compress.CompressedMatrixBlock;
+import org.apache.sysds.runtime.compress.CompressedMatrixBlockFactory;
+import org.apache.sysds.runtime.compress.CompressionSettings;
+import org.apache.sysds.runtime.compress.CompressionSettingsBuilder;
+import org.apache.sysds.runtime.compress.cocode.CoCoderFactory;
+import org.apache.sysds.runtime.compress.colgroup.AColGroup.CompressionType;
+import org.apache.sysds.runtime.compress.cost.CostEstimatorFactory;
+import
org.apache.sysds.runtime.compress.estim.sample.SampleEstimatorFactory.EstimationType;
+import org.apache.sysds.runtime.functionobjects.Multiply;
+import org.apache.sysds.runtime.instructions.spark.data.IndexedMatrixValue;
+import org.apache.sysds.runtime.matrix.data.MatrixBlock;
+import org.apache.sysds.runtime.matrix.data.MatrixValue;
+import org.apache.sysds.runtime.matrix.operators.BinaryOperator;
+import org.apache.sysds.runtime.util.DataConverter;
+import org.apache.sysds.test.TestUtils;
+import org.junit.Test;
+
+public class CompressedSingleTests {
+
+ protected static final Log LOG =
LogFactory.getLog(CompressedSingleTests.class.getName());
+ protected static final int rows = 1000;
+ protected static final int cols = 2;
+ protected static final double min = 0;
+ protected static final double max = 100;
+ protected static final double sparsity = 0.6;
+ protected static final int seed = 2;
+
+ protected final MatrixBlock mb;
+ protected final CompressedMatrixBlock cmb;
+
+ public CompressedSingleTests() {
+
+ double[][] input =
TestUtils.round(TestUtils.generateTestMatrix(rows, cols, min, max, sparsity,
seed));
+ mb = DataConverter.convertToMatrixBlock(input);
+ mb.recomputeNonZeros();
+ mb.examSparsity();
+
+ cmb = (CompressedMatrixBlock)
CompressedMatrixBlockFactory.compress(mb).getLeft();
+ }
+
+ @Test
+ public void binaryOperationsInPlace() {
+ MatrixBlock refmb = new MatrixBlock();
+ refmb.copy(mb, true);
+ MatrixValue thatValue = DataConverter
+
.convertToMatrixBlock(TestUtils.generateTestMatrix(rows, cols, min, max,
sparsity, seed + 13));
+ BinaryOperator op = new
BinaryOperator(Multiply.getMultiplyFnObject());
+
+ // reference
+ MatrixBlock ret1 = refmb.binaryOperationsInPlace(op, thatValue);
+ // compressed
+ MatrixBlock ret2 = cmb.binaryOperationsInPlace(op, thatValue);
+
+ double[][] d1 = DataConverter.convertToDoubleMatrix(ret1);
+ double[][] d2 = DataConverter.convertToDoubleMatrix(ret2);
+
+ TestUtils.compareMatricesBitAvgDistance(d1, d2, 1024, 1024,
"Binary Operations Inplace Failed.");
+ }
+
+ @Test(expected = DMLRuntimeException.class)
+ public void invalid_call_append_listOfMatrixBlock() {
+ cmb.append(new MatrixBlock[] {new MatrixBlock(10, 10, 3)},
null, false);
+ }
+
+ @Test(expected = NullPointerException.class)
+ public void invalid_call_appendArray() {
+ cmb.append(null, new ArrayList<IndexedMatrixValue>(), 1, false,
false, 1);
+ }
+
+ @Test
+ public void estimateIsEqualToActualSizeInMemory() {
+ assertEquals(cmb.getInMemorySize(), cmb.estimateSizeInMemory());
+ }
+
+ @Test
+ public void estimateIsEqualToActualSizeDisk() {
+ assertEquals(cmb.getExactSizeOnDisk(),
cmb.estimateSizeOnDisk());
+ }
+
+ @Test
+ public void testOnDiskSizeInBytes() {
+ try {
+ // Serialize out
+ ByteArrayOutputStream bos = new ByteArrayOutputStream();
+ DataOutputStream fos = new DataOutputStream(bos);
+ cmb.write(fos);
+ byte[] arr = bos.toByteArray();
+ int size = arr.length;
+ assertEquals(cmb.getClass().getSimpleName() + "\n" +
cmb.toString() + "\n", size, cmb.getExactSizeOnDisk());
+ }
+ catch(IOException e) {
+ throw new RuntimeException("Error in io", e);
+ }
+ catch(Exception e) {
+ e.printStackTrace();
+ throw e;
+ }
+ }
+
+ @Test(expected = NotImplementedException.class)
+ public void test_copyShallow() {
+ CompressedMatrixBlock copyIntoMe = new CompressedMatrixBlock();
+ copyIntoMe.copyShallow(cmb);
+ }
+
+ @Test
+ public void test_settingsBuilder() {
+ CompressionSettingsBuilder b = new CompressionSettingsBuilder();
+ b =
b.addValidCompression(CompressionType.CONST).setLossy(true).setLossy(false).setSortValuesByLength(true)
+
.setAllowSharedDictionary(true).setColumnPartitioner(CoCoderFactory.PartitionerType.BIN_PACKING)
+
.setMaxColGroupCoCode(3).setEstimationType(EstimationType.ShlosserJackknifeEstimator)
+
.clearValidCompression().setSamplingRatio(0.2).setSeed(1342).setCoCodePercentage(0.22)
+
.setMinimumSampleSize(1342).setCostType(CostEstimatorFactory.CostType.MEMORY);
+ CompressionSettings s = b.create();
+ b = b.copySettings(s);
+
Review comment:
remove the empty lines just before ending curly braces.
##########
File path:
src/main/java/org/apache/sysds/runtime/matrix/data/LibMatrixBincell.java
##########
@@ -278,25 +285,62 @@ public static void bincellOpInPlaceRight(MatrixBlock
m1ret, MatrixBlock m2, Bina
*
* m2 op m1ret
*
+ *
+ * Note that sometimes the returned matrix is not inplace, since it
might be optimal to return a new Sparse Matrix instead.
+ * Therefore to be safe use the returned matrix in continued operations
+ *
* @param m1ret result matrix updated in place
* @param m2 matrix block the other matrix to take values from
* @param op binary operator the operator that is placed in the middle
of m1ret and m2
+ * @return The result MatrixBlock
*/
- public static void bincellOpInPlaceLeft(MatrixBlock m1ret, MatrixBlock
m2, BinaryOperator op) {
- if(m1ret.isInSparseFormat() || m2.isInSparseFormat())
- throw new NotImplementedException("Not implemented
sparse inplace left binaryOperator for matrixBlocks");
-
- final double[] retV = m1ret.getDenseBlockValues();
- final double[] m2V = m2.getDenseBlockValues();
+ public static MatrixBlock bincellOpInPlaceLeft(MatrixBlock m1ret,
MatrixBlock m2, BinaryOperator op) {
+ final int nRows = m1ret.getNumRows();
+ final int nCols = m1ret.getNumColumns();
+ if(m1ret.isInSparseFormat()){
+ // not doing in place, since the m1ret is in sparse
format, and m2 might make it dense.
+ // this is not ideal either, but makes it work
+ LOG.warn("Inefficient bincell op in place left, because
output is materialized in new matrix");
+ MatrixBlock ret = new MatrixBlock(nRows, nCols, true);
Review comment:
please double check that compression is the only place calling this to
avoid creating implicitly new return blocks although the caller things this is
inplace.
##########
File path: src/main/java/org/apache/sysds/runtime/matrix/data/MatrixBlock.java
##########
@@ -1390,9 +1364,10 @@ public void copy(MatrixValue thatValue) {
public void copy(MatrixValue thatValue, boolean sp)
{
MatrixBlock that = checkType(thatValue);
-
if( this == that ) //prevent data loss (e.g., on sparse-dense
conversion)
throw new RuntimeException( "Copy must not overwrite
itself!" );
+ if(that instanceof CompressedMatrixBlock)
Review comment:
As discussed this is not very nice to have all these references to
CompressedMatrixBlock in the base class MatrixBlock. However, if
CompressedMatrixBlock is indeed moved into MatrixBlock it might be ok.
##########
File path:
src/main/java/org/apache/sysds/runtime/compress/utils/DoubleCountHashMap.java
##########
@@ -0,0 +1,207 @@
+/*
+ * 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.utils;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+public class DoubleCountHashMap {
+
+ protected static final Log LOG =
LogFactory.getLog(DoubleCountHashMap.class.getName());
+ protected static final int RESIZE_FACTOR = 2;
+ protected static final float LOAD_FACTOR = 0.80f;
+ public static int hashMissCount = 0;
+
+ protected int _size = -1;
+ private Bucket[] _data = null;
+
+ public DoubleCountHashMap(int init_capacity) {
+ _data = new Bucket[getPow2(init_capacity)];
+ _size = 0;
+ }
+
+ private int getPow2(int x) {
Review comment:
see above (and try to avoid code duplication)
##########
File path:
src/main/java/org/apache/sysds/runtime/matrix/operators/AggregateOperator.java
##########
@@ -31,7 +29,7 @@
import org.apache.sysds.runtime.functionobjects.ValueFunction;
-public class AggregateOperator extends Operator implements Serializable
Review comment:
this would need stay serializable - otherwise lazy evaluation can create
funny exceptions in Spark.
##########
File path: src/main/java/org/apache/sysds/hops/recompile/Recompiler.java
##########
@@ -300,6 +303,7 @@ public static void reinitRecompiler() {
private static ArrayList<Instruction> recompile(StatementBlock sb,
ArrayList<Hop> hops, ExecutionContext ec, RecompileStatus status,
boolean inplace, boolean replaceLit, boolean updateStats,
boolean forceEt, boolean pred, ExecType et, long tid )
{
+ LOG.trace("Recompiling");
Review comment:
avoid unnecessary tracing in the core recompilation
##########
File path: src/main/java/org/apache/sysds/runtime/matrix/data/MatrixBlock.java
##########
@@ -1332,52 +1352,6 @@ else if( !sparse && denseBlock!=null ) { //DENSE
return 0; //empty block
}
-
- /**
- * Basic debugging primitive to check correctness of nnz.
- * This method is not intended for production use.
- */
- public void checkNonZeros() {
Review comment:
where have these debugging primitives (check nnz, check sparse
representation) been moved? Please keep them in the code base.
##########
File path:
src/test/java/org/apache/sysds/test/component/compress/CompressedTestBase.java
##########
@@ -1220,14 +1244,55 @@ public void appendCBindTrue() {
@Test
public void appendCBindFalse() {
- if(!(cmb instanceof CompressedMatrixBlock))
+ if(!(cmb instanceof CompressedMatrixBlock) || rows * cols >
10000)
return;
MatrixBlock ap = new MatrixBlock(1, mb.getNumColumns(), false,
132);
MatrixBlock ret1 = mb.append(ap, new MatrixBlock(), false);
MatrixBlock ret2 = cmb.append(ap, new MatrixBlock(), false);
compareResultMatrices(ret1, ret2, 1);
}
+ @Test
+ public void aggregateTernaryOperations() {
+ if(!(cmb instanceof CompressedMatrixBlock) || rows * cols >
10000)
+ return;
+
+ MatrixBlock m1 = new MatrixBlock();
+ MatrixBlock m2 = new MatrixBlock();
+ MatrixBlock m3 = null;
+ CorrectionLocationType corr = CorrectionLocationType.LASTCOLUMN;
+ AggregateOperator agg = new AggregateOperator(0,
KahanPlus.getKahanPlusFnObject(), corr);
+ AggregateTernaryOperator op = new
AggregateTernaryOperator(Multiply.getMultiplyFnObject(), agg,
+ ReduceAll.getReduceAllFnObject(), _k);
+
+ boolean inCP = true;
+
+ MatrixBlock ret1 = mb.aggregateTernaryOperations(m1, m2, m3,
null, op, inCP);
+ MatrixBlock ret2 = cmb.aggregateTernaryOperations(m1, m2, m3,
null, op, inCP);
+
+ compareResultMatrices(ret1, ret2, 1);
+
+ }
+
+ @Test
+ public void unaryOperations() {
+ if(!(cmb instanceof CompressedMatrixBlock) ||
cmb.getNumColumns() < 2)
+ return;
+ try {
+
+ UnaryOperator op = new
UnaryOperator(Builtin.getBuiltinFnObject("ucumk+*"), _k, false);
+ MatrixBlock ret1 = mb.slice(0, rows - 1, 0,
1).unaryOperations(op, null);
+ MatrixBlock ret2 = cmb.slice(0, rows - 1, 0,
1).unaryOperations(op, null);
+
+ compareResultMatrices(ret1, ret2, 1);
+ }
+ catch(Exception e) {
+ e.printStackTrace();
+ throw e;
+ }
+
Review comment:
same as above (twice in this file)
##########
File path:
src/test/scripts/functions/compress/SystemDS-config-compress-workload.xml
##########
@@ -0,0 +1,25 @@
+<!--
+ * 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.
+-->
+
+<root>
+ <sysds.compressed.linalg>auto</sysds.compressed.linalg>
Review comment:
shouldn't the workload tests use the 'workload' instead of 'auto'
configuration?
--
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]
