[SYSTEMDS-3799] Fix parfor result merge (all combinations)

This patch fixes recently discovered (see code coverage) issues of
parfor result merge for combinations of different result merge
implementations, dense/sparse inputs, with compare dense/sparse blocks,
and most importantly += accumulation into the output.

Author: mboehm7

src/main/java/org/apache/sysds/runtime/controlprogram/parfor/ResultMerge.java

@@ -36,6 +36,7 @@ public abstract class ResultMerge<T extends CacheableData<?>> implements Seriali
 	protected static final Log LOG = LogFactory.getLog(ResultMerge.class.getName());
 	protected static final String NAME_SUFFIX = "_rm";
 	protected static final BinaryOperator PLUS = InstructionUtils.parseBinaryOperator("+");
+	protected static final BinaryOperator MINUS = InstructionUtils.parseBinaryOperator("-");
 
 	//inputs to result merge
 	protected T       _output      = null;

src/main/java/org/apache/sysds/runtime/controlprogram/parfor/ResultMergeLocalFile.java

@@ -258,7 +258,10 @@ private void createBinaryBlockResultFile( String fnameStaging, String fnameStagi
 							DenseBlock compare = DataConverter.convertToDenseBlock(mb, true);
 							for( String lname : dir.list() ) {
 								MatrixBlock tmp = LocalFileUtils.readMatrixBlockFromLocal( dir+"/"+lname );
-								mergeWithComp(mb, tmp, compare);
+								if( _isAccum )
+									mergeWithoutComp(mb, tmp, compare, appendOnly);
+								else
+									mergeWithComp(mb, tmp, compare);
 							}
 
 							//sort sparse due to append-only
@@ -279,7 +282,7 @@ private void createBinaryBlockResultFile( String fnameStaging, String fnameStagi
 								}
 								else {
 									MatrixBlock tmp = LocalFileUtils.readMatrixBlockFromLocal( dir+"/"+lname );
-									mergeWithoutComp(mb, tmp, appendOnly);
+									mergeWithoutComp(mb, tmp, null, appendOnly);
 								}
 							}
 

src/main/java/org/apache/sysds/runtime/controlprogram/parfor/ResultMergeLocalMemory.java

@@ -73,7 +73,7 @@ public MatrixObject executeSerialMerge()
 
 			//create compare matrix if required (existing data in result)
 			_compare = getCompareMatrix(outMB);
-			if( _compare != null )
+			if( _compare != null || _isAccum )
 				outMBNew.copy(outMB);
 
 			//serial merge all inputs
@@ -90,7 +90,7 @@ public MatrixObject executeSerialMerge()
 					MatrixBlock inMB = in.acquireRead();
 
 					//core merge 
-					merge( outMBNew, inMB, appendOnly );
+					merge( outMBNew, inMB, _compare, appendOnly );
 
 					//unpin and clear in-memory input_i
 					in.release();
@@ -169,7 +169,7 @@ public MatrixObject executeParallelMerge( int par )
 
 				//create compare matrix if required (existing data in result)
 				_compare = getCompareMatrix(outMB);
-				if( _compare != null )
+				if( _compare != null || _isAccum )
 					outMBNew.copy(outMB);
 
 				//parallel merge of all inputs
@@ -215,7 +215,7 @@ public MatrixObject executeParallelMerge( int par )
 		return moNew;
 	}
 
-	private static DenseBlock getCompareMatrix( MatrixBlock output ) {
+	private DenseBlock getCompareMatrix( MatrixBlock output ) {
 		//create compare matrix only if required
 		if( !output.isEmptyBlock(false) )
 			return DataConverter.convertToDenseBlock(output, false);
@@ -253,11 +253,12 @@ private MatrixObject createNewMatrixObject( MatrixBlock data ) {
 	 * 
 	 * @param out output matrix block
 	 * @param in input matrix block
+	 * @param compare initialized output
 	 * @param appendOnly ?
 	 */
-	private void merge( MatrixBlock out, MatrixBlock in, boolean appendOnly ) {
-		if( _compare == null )
-			mergeWithoutComp(out, in, appendOnly, true);
+	private void merge( MatrixBlock out, MatrixBlock in, DenseBlock compare, boolean appendOnly ) {
+		if( _compare == null || _isAccum )
+			mergeWithoutComp(out, in, _compare, appendOnly, true);
 		else
 			mergeWithComp(out, in, _compare);
 	}
@@ -304,7 +305,7 @@ public void run()
 				LOG.trace("ResultMerge (local, in-memory): Merge input "+_inMO.hashCode()+" (fname="+_inMO.getFileName()+")");
 
 				MatrixBlock inMB = _inMO.acquireRead(); //incl. implicit read from HDFS
-				merge( _outMB, inMB, false );
+				merge( _outMB, inMB, _compare, false );
 				_inMO.release();
 				_inMO.clearData();
 			}

src/main/java/org/apache/sysds/runtime/controlprogram/parfor/ResultMergeMatrix.java

@@ -22,7 +22,6 @@
 import java.util.List;
 
 import org.apache.sysds.runtime.DMLRuntimeException;
-import org.apache.sysds.runtime.compress.utils.Util;
 import org.apache.sysds.runtime.controlprogram.caching.MatrixObject;
 import org.apache.sysds.runtime.data.DenseBlock;
 import org.apache.sysds.runtime.data.SparseBlock;
@@ -58,17 +57,21 @@ public ResultMergeMatrix(MatrixObject out, MatrixObject[] in, String outputFilen
 		super(out, in, outputFilename, accum);
 	}
 
-	protected void mergeWithoutComp(MatrixBlock out, MatrixBlock in, boolean appendOnly) {
-		mergeWithoutComp(out, in, appendOnly, false);
+	protected void mergeWithoutComp(MatrixBlock out, MatrixBlock in, DenseBlock compare, boolean appendOnly) {
+		mergeWithoutComp(out, in, compare, appendOnly, false);
 	}
 
-	protected void mergeWithoutComp(MatrixBlock out, MatrixBlock in, boolean appendOnly, boolean par) {
+	protected void mergeWithoutComp(MatrixBlock out, MatrixBlock in, DenseBlock compare, boolean appendOnly, boolean par) {
 		// pass through to matrix block operations
-		if(_isAccum)
+		if(_isAccum) {
 			out.binaryOperationsInPlace(PLUS, in);
+			//compare block used for compensation here
+			if( compare != null )
+				out.binaryOperationsInPlace(MINUS, 
+					new MatrixBlock(out.getNumRows(),out.getNumColumns(), compare));
+		}
 		else {
 			MatrixBlock out2 = out.merge(in, appendOnly, par);
-
 			if(out2 != out)
 				throw new DMLRuntimeException("Failed merge need to allow returned MatrixBlock to be used");
 		}
@@ -90,18 +93,13 @@ protected void mergeWithComp(MatrixBlock out, MatrixBlock in, DenseBlock compare
 		// NaNs, since NaN != NaN, otherwise we would potentially overwrite results
 		// * For the case of accumulation, we add out += (new-old) to ensure correct results
 		// because all inputs have the old values replicated
-		final int rows = in.getNumRows();
-		final int cols = in.getNumColumns();
-		if(in.isEmptyBlock(false)) {
-			if(_isAccum)
-				return; // nothing to do
+		int rows = in.getNumRows();
+		int cols = in.getNumColumns();
+		if(in.isEmptyBlock(false))
 			mergeWithCompEmpty(out, rows, cols, compare);
-		}
-		else if(in.isInSparseFormat() && _isAccum)
-			mergeSparseAccumulative(out, in, rows, cols, compare);
 		else if(in.isInSparseFormat())
 			mergeSparse(out, in, rows, cols, compare);
-		else // SPARSE/DENSE
+		else // DENSE
 			mergeGeneric(out, in, rows, cols, compare);
 	}
 
@@ -111,90 +109,62 @@ private void mergeWithCompEmpty(MatrixBlock out, int m, int n, DenseBlock compar
 	}
 
 	private void mergeWithCompEmptyRow(MatrixBlock out, int m, int n, DenseBlock compare, int i) {
-
 		for(int j = 0; j < n; j++) {
 			final double valOld = compare.get(i, j);
-			if(!Util.eq(0.0, valOld)) // NaN awareness
+			if(!equals(0.0, valOld)) // NaN awareness
 				out.set(i, j, 0);
 		}
 	}
 
-	private void mergeSparseAccumulative(MatrixBlock out, MatrixBlock in, int m, int n, DenseBlock compare) {
-		final SparseBlock a = in.getSparseBlock();
-		for(int i = 0; i < m; i++) {
-			if(a.isEmpty(i))
-				continue;
-			final int apos = a.pos(i);
-			final int alen = a.size(i) + apos;
-			final int[] aix = a.indexes(i);
-			final double[] aval = a.values(i);
-			mergeSparseRowAccumulative(out, apos, alen, aix, aval, compare, n, i);
-		}
-	}
-
-	private void mergeSparseRowAccumulative(MatrixBlock out, int apos, int alen, int[] aix, double[] aval,
-		DenseBlock compare, int n, int i) {
-		for(; apos < alen; apos++) { // inside
-			final double valOld = compare.get(i, aix[apos]);
-			final double valNew = aval[apos];
-			if(!Util.eq(valNew, valOld)) { // NaN awareness
-				double value = out.get(i, aix[apos]) + (valNew - valOld);
-				out.set(i, aix[apos], value);
-			}
-		}
-	}
-
 	private void mergeSparse(MatrixBlock out, MatrixBlock in, int m, int n, DenseBlock compare) {
 		final SparseBlock a = in.getSparseBlock();
 		for(int i = 0; i < m; i++) {
 			if(a.isEmpty(i))
 				mergeWithCompEmptyRow(out, m, n, compare, i);
 			else {
-				final int apos = a.pos(i);
-				final int alen = a.size(i) + apos;
-				final int[] aix = a.indexes(i);
-				final double[] aval = a.values(i);
-				mergeSparseRow(out, apos, alen, aix, aval, compare, n, i);
-			}
-		}
-	}
-
-	private void mergeSparseRow(MatrixBlock out, int apos, int alen, int[] aix, double[] aval, DenseBlock compare, int n,
-		int i) {
-		int j = 0;
-		for(; j < n && apos < alen; j++) { // inside
-			final boolean aposValid = aix[apos] == j;
-			final double valOld = compare.get(i, j);
-			final double valNew = aix[apos] == j ? aval[apos] : 0.0;
-			if(!Util.eq(valNew, valOld)) { // NaN awareness
-				double value = !_isAccum ? valNew : (out.get(i, j) + (valNew - valOld));
-				out.set(i, j, value);
-			}
-			if(aposValid)
-				apos++;
-		}
-		for(; j < n; j++) {
-			final double valOld = compare.get(i, j);
-			if(valOld != 0) {
-				double value = (out.get(i, j) - valOld);
-				out.set(i, j, value);
+				int apos = a.pos(i);
+				int alen = a.size(i) + apos;
+				int[] aix = a.indexes(i);
+				double[] avals = a.values(i);
+				int j = 0;
+				for(; j < n && apos < alen; j++) { // inside
+					boolean aposValid = (aix[apos] == j);
+					double valOld = compare.get(i, j);
+					double valNew = aposValid ? avals[apos] : 0.0;
+					if(!equals(valNew, valOld)) { // NaN awareness
+						double value = !_isAccum ? valNew : (out.get(i, j) + (valNew - valOld));
+						out.set(i, j, value);
+					}
+					if(aposValid)
+						apos++;
+				}
+				for(; j < n; j++) {
+					double valOld = compare.get(i, j);
+					if(valOld != 0) {
+						double value = (out.get(i, j) - valOld);
+						out.set(i, j, value);
+					}
+				}
 			}
 		}
-
 	}
 
 	private void mergeGeneric(MatrixBlock out, MatrixBlock in, int m, int n, DenseBlock compare) {
 		for(int i = 0; i < m; i++) {
 			for(int j = 0; j < n; j++) {
 				final double valOld = compare.get(i, j);
 				final double valNew = in.get(i, j); // input value
-				if(!Util.eq(valNew, valOld)) { // NaN awareness
-					double value = !_isAccum ? valNew : (out.get(i, j) + (valNew - valOld));
-					out.set(i, j, value);
+				if(!equals(valNew, valOld)) { // NaN awareness
+					out.set(i, j, valNew);
 				}
 			}
 		}
 	}
+	
+	private boolean equals(double valNew, double valOld) {
+		return (valNew == valOld && !Double.isNaN(valNew))     //for changed values 
+			|| (Double.isNaN(valNew) && Double.isNaN(valOld)); //NaN awareness 
+	}
 
 	protected long computeNonZeros(MatrixObject out, List<MatrixObject> in) {
 		// sum of nnz of input (worker result) - output var existing nnz

src/main/java/org/apache/sysds/runtime/controlprogram/parfor/ResultMergeRemoteSparkWCompare.java

@@ -52,8 +52,12 @@ public Tuple2<MatrixIndexes, MatrixBlock> call(Tuple2<MatrixIndexes, Tuple2<Iter
 
 		//merge all blocks into compare block
 		MatrixBlock out = new MatrixBlock(cin);
-		while( din.hasNext() )
-			mergeWithComp(out, din.next(), compare);
+		while( din.hasNext() ) {
+			if( _isAccum )
+				mergeWithoutComp(out, din.next(), compare, false);
+			else
+				mergeWithComp(out, din.next(), compare);
+		}
 
 		//create output tuple
 		return new Tuple2<>(new MatrixIndexes(ixin), out);