[MINOR] Minor cleanup in recent test commits

This commit adds:

- Util Wite MatrixBlock for tests with MTD
- New Parallel execution of test that catch dead threads.
  - This caught  https://issues.apache.org/jira/browse/SYSTEMDS-3716
- Fix minor spelling error.
- remove some commented code in tests
- remove some verbose printing in new tests

Closes #2059

src/test/java/org/apache/sysds/test/AutomatedTestBase.java

@@ -75,6 +75,7 @@
 import org.apache.sysds.runtime.io.FileFormatPropertiesCSV;
 import org.apache.sysds.runtime.io.FrameReader;
 import org.apache.sysds.runtime.io.FrameReaderFactory;
+import org.apache.sysds.runtime.io.MatrixWriterFactory;
 import org.apache.sysds.runtime.io.ReaderWriterFederated;
 import org.apache.sysds.runtime.matrix.data.MatrixBlock;
 import org.apache.sysds.runtime.matrix.data.MatrixValue;
@@ -591,6 +592,28 @@ protected double[][] writeInputMatrixWithMTD(String name, double[][] matrix, boo
 		return matrix;
 	}
 
+
+	protected void writeBinaryWithMTD(String name, MatrixBlock matrix) {
+		MatrixCharacteristics mc = new MatrixCharacteristics(matrix.getNumRows(), matrix.getNumColumns(),
+			OptimizerUtils.DEFAULT_BLOCKSIZE, matrix.getNonZeros());
+		writeBinaryWithMTD(name, matrix, mc);
+	}
+
+	protected void writeBinaryWithMTD(String name, MatrixBlock matrix, MatrixCharacteristics mc) {
+
+		try {
+			MatrixWriterFactory.createMatrixWriter(FileFormat.BINARY)//
+				.writeMatrixToHDFS(matrix, baseDirectory + INPUT_DIR + name, matrix.getNumRows(), 
+				matrix.getNumColumns(), mc.getBlocksize(), mc.getNonZeros());
+			String completeMTDPath = baseDirectory + INPUT_DIR + name + ".mtd";
+			HDFSTool.writeMetaDataFile(completeMTDPath, ValueType.FP64, mc, FileFormat.BINARY);
+		}
+		catch(IOException e) {
+			e.printStackTrace();
+			throw new RuntimeException(e);
+		}
+	}
+
 	protected void writeInputFederatedWithMTD(String name, MatrixObject fm){
 		writeFederatedInputMatrix(name, fm.getFedMapping());
 		MatrixCharacteristics mc = (MatrixCharacteristics)fm.getDataCharacteristics();
@@ -1356,15 +1379,25 @@ protected ByteArrayOutputStream runTest(Class<?> expectedException, String errMe
 	protected ByteArrayOutputStream runTest(boolean newWay, boolean exceptionExpected, Class<?> expectedException,
 		String errMessage, int maxSparkInst) {
 		try{
-			final List<ByteArrayOutputStream> out =  new ArrayList<>();
+			final List<ByteArrayOutputStream> out = new ArrayList<>();
 			Thread t = new Thread(
-				() -> out.add(runTestWithTimeout(newWay,exceptionExpected,expectedException,errMessage, maxSparkInst)),
+				() -> out.add(runTestWithTimeout(newWay, exceptionExpected, expectedException, errMessage, maxSparkInst)),
 				"TestRunner_main");
+			Thread.UncaughtExceptionHandler h = new Thread.UncaughtExceptionHandler() {
+				@Override
+				public void uncaughtException(Thread th, Throwable ex) {
+					fail("Thread Failed test with message: " +ex.getMessage());
+				}
+			};
+			t.setUncaughtExceptionHandler(h);
 			t.start();
-			
+
 			t.join(TEST_TIMEOUT * 1000);
 			if(t.isAlive())
 				throw new TimeoutException("Test failed to finish in time");
+			if(out.size() <= 0) // hack in case the test failed return empty string.
+				fail("test failed");
+
 			return out.get(0);
 		}
 		catch(TimeoutException e){
@@ -1463,8 +1496,6 @@ private ByteArrayOutputStream runTestWithTimeout(boolean newWay, boolean excepti
 				errorMessage.append("\nStandard Out:");
 				if(outputBuffering)
 					errorMessage.append("\n" + buff);
-				// errorMessage.append("\nStackTrace:");
-				// errorMessage.append(getStackTraceString(e, 0));
 				LOG.error(errorMessage);
 				e.printStackTrace();
 				fail(base);

src/test/java/org/apache/sysds/test/TestUtils.java

@@ -80,12 +80,15 @@
 import org.apache.sysds.runtime.io.FrameWriter;
 import org.apache.sysds.runtime.io.FrameWriterFactory;
 import org.apache.sysds.runtime.io.IOUtilFunctions;
+import org.apache.sysds.runtime.io.MatrixReaderFactory;
 import org.apache.sysds.runtime.matrix.data.MatrixBlock;
 import org.apache.sysds.runtime.matrix.data.MatrixCell;
 import org.apache.sysds.runtime.matrix.data.MatrixIndexes;
 import org.apache.sysds.runtime.matrix.data.MatrixValue.CellIndex;
 import org.apache.sysds.runtime.matrix.operators.UnaryOperator;
+import org.apache.sysds.runtime.meta.DataCharacteristics;
 import org.apache.sysds.runtime.meta.MatrixCharacteristics;
+import org.apache.sysds.runtime.meta.MetaDataAll;
 import org.apache.sysds.runtime.util.DataConverter;
 import org.apache.sysds.runtime.util.UtilFunctions;
 import org.junit.Assert;
@@ -528,6 +531,24 @@ public static HashMap<CellIndex, Double> readDMLMatrixFromHDFS(String filePath)
 		return expectedValues;
 	}
 
+	public static MatrixBlock readBinary(String path){
+		try{
+
+			MetaDataAll mba = new MetaDataAll(path + ".mtd", false, true);
+			if(!mba.mtdExists())
+				throw new IOException("File does not exist");
+			DataCharacteristics ds = mba.getDataCharacteristics();
+			FileFormat f = FileFormat.valueOf(mba.getFormatTypeString().toUpperCase());
+			return MatrixReaderFactory.createMatrixReader(f)
+				.readMatrixFromHDFS(path, ds.getRows(),ds.getCols(),ds.getBlocksize(),ds.getNonZeros());
+
+		}
+		catch(Exception e){
+			throw new RuntimeException(e);
+		}
+	}
+
+
 	/**
 	 * Reads values from a matrix file in OS's FS in R format
 	 *
@@ -851,12 +872,37 @@ public static void compareFrames(FrameBlock expected, FrameBlock actual, boolean
 		final int cols = expected.getNumColumns();
 		if(checkMeta)
 			checkMetadata(expected, actual);
+		if(expected.getNumRows() == 0){
+			if (expected.getColumns() != null)
+				fail();
+			if (actual.getColumns() != null) 
+				fail();
+		}
+		else{
+			for(int j = 0; j < cols; j++) {
+				Array<?> ec = expected.getColumn(j);
+				Array<?> ac = actual.getColumn(j);
+				if(ec.containsNull()) {
+					if(!ac.containsNull()) {
+						fail("Expected both columns to be containing null if one null:\n\nExpected containing null:\n"
+							+ ec.toString().substring(0, 1000) + "\n\nActual:\n" + ac.toString().substring(0, 1000));
+					}
+				}
+				else if(ac.containsNull()) {
+					fail("Expected both columns to be containing null if one null:\n\nExpected:\n"
+						+ ec.toString().substring(0, 1000) + "\n\nActual containing null:\n" + ac.toString().substring(0, 1000));
+				}
+			}
+		}
 
 		for(int i = 0; i < rows; i++) {
 			for(int j = 0; j < cols; j++) {
 				final Object a = expected.get(i, j);
 				final Object b = actual.get(i, j);
-				if(!(a == null && b == null)) {
+				if(a == null){
+					assertTrue(a == b);
+				}
+				else if(!(a == null && b == null)) {
 					try{
 						final String as = a.toString();
 						final String bs = b.toString();
@@ -1107,8 +1153,10 @@ private static void compareMatricesBitAvgDistanceSparse(SparseBlock sbe, SparseB
 		int countErrors = 0;
 		long sumDistance = 0;
 		for(int i = 0; i < rows && countErrors < 20; i++){
-			if( sbe.isEmpty(i) !=  sba.isEmpty(i))
-				fail(message +"\nBoth matrices are not equally empty on row : " + i);
+			if( sbe.isEmpty(i) !=  sba.isEmpty(i)){
+
+				fail(message +"\nBoth matrices are not equally empty on row : " + i + " :\n" + sbe.get(i) + " vs " + sba.get(i));
+			}
 
 			if(sbe.isEmpty(i))
 				continue;
@@ -3281,11 +3329,15 @@ public static double[][] round(double[][] data, int col) {
 	}
 
 	public static MatrixBlock round(MatrixBlock data) {
-		return data.unaryOperations(new UnaryOperator(Builtin.getBuiltinFnObject(BuiltinCode.ROUND),2, true), null);
+		return data.unaryOperations(new UnaryOperator(Builtin.getBuiltinFnObject(BuiltinCode.ROUND), 2, true), null);
+	}
+
+	public static MatrixBlock ceil(MatrixBlock data) {
+		return data.unaryOperations(new UnaryOperator(Builtin.getBuiltinFnObject(BuiltinCode.CEIL), 2, true), null);
 	}
 
-	public static MatrixBlock ceil(MatrixBlock data){
-		return data.unaryOperations(new UnaryOperator(Builtin.getBuiltinFnObject(BuiltinCode.CEIL),2, true), null);
+	public static MatrixBlock floor(MatrixBlock data) {
+		return data.unaryOperations(new UnaryOperator(Builtin.getBuiltinFnObject(BuiltinCode.FLOOR), 2, true), null);
 	}
 
 	public static double[][] floor(double[][] data) {
@@ -3302,6 +3354,12 @@ public static double[][] ceil(double[][] data) {
 		return data;
 	}
 
+	public static double[] ceil(double[] data) {
+		for(int i = 0; i < data.length; i++)
+			data[i] = Math.ceil(data[i]);
+		return data;
+	}
+
 	public static double[][] floor(double[][] data, int col) {
 		for(int i=0; i<data.length; i++)
 			data[i][col]=Math.floor(data[i][col]);

src/test/java/org/apache/sysds/test/component/compress/AbstractCompressedUnaryTests.java

@@ -304,15 +304,15 @@ else if(aggType == AggType.ROW_MEAN)
 							printedWarningForProduct.set(true);
 							LOG.warn("Product not equal because of double rounding upwards");
 						}
-						return;	
+						return;
 					}
 					if(Math.abs(ret2.get(0, 0)) <= 1E-16) {
 						if(!printedWarningForProduct.get()) {
 							printedWarningForProduct.set(true);
 							LOG.warn("Product not equal because of  double rounding downwards");
 						}
 
-						return; // Product is wierd around zero.
+						return; // Product is weird around zero.
 					}
 					TestUtils.compareMatricesPercentageDistance(ret1, ret2, 0.95, 0.98, css);
 				}

src/test/java/org/apache/sysds/test/component/compress/CompressedLoggingTests.java

@@ -272,7 +272,6 @@ public void compressedLoggingTest_TraceBigGroup() {
 			TestUtils.compareMatrices(mb, m2, 0.0);
 			final List<LoggingEvent> log = LoggingUtils.reinsert(appender);
 			for(LoggingEvent l : log) {
-				// LOG.error(l.getMessage());
 				if(l.getMessage().toString().contains("--colGroups type"))
 					return;
 			}
@@ -300,7 +299,6 @@ public void compressedLoggingTest_TraceBigGroupConst() {
 			TestUtils.compareMatrices(mb, m2, 0.0);
 			final List<LoggingEvent> log = LoggingUtils.reinsert(appender);
 			for(LoggingEvent l : log) {
-				// LOG.error(l.getMessage());
 				if(l.getMessage().toString().contains("--colGroups type"))
 					return;
 			}
@@ -328,7 +326,6 @@ public void compressedLoggingTestEmpty() {
 			TestUtils.compareMatrices(mb, m2, 0.0);
 			final List<LoggingEvent> log = LoggingUtils.reinsert(appender);
 			for(LoggingEvent l : log) {
-				// LOG.error(l.getMessage());
 				if(l.getMessage().toString().contains("Empty input to compress"))
 					return;
 			}
@@ -385,11 +382,10 @@ public void compressedLoggingTest_AbortEnd() {
 			CompressionSettingsBuilder sb = new CompressionSettingsBuilder();
 			sb.setMaxSampleSize(ss);
 			sb.setMinimumSampleSize(ss);
-			MatrixBlock cmb = CompressedMatrixBlockFactory.compress(mb, sb).getLeft();
+			CompressedMatrixBlockFactory.compress(mb, sb).getLeft();
 			final List<LoggingEvent> log = LoggingUtils.reinsert(appender);
-			LOG.error(cmb);
+
 			for(LoggingEvent l : log) {
-				// LOG.error(l.getMessage());
 				if(l.getMessage().toString().contains("Abort block compression"))
 					return;
 			}
@@ -414,7 +410,6 @@ public void compressionSettings() {
 			new CompressionSettingsBuilder().create();
 			final List<LoggingEvent> log = LoggingUtils.reinsert(appender);
 			for(LoggingEvent l : log) {
-				// LOG.error(l.getMessage());
 				if(l.getMessage().toString().contains("CompressionSettings"))
 					return;
 			}

src/test/java/org/apache/sysds/test/component/compress/CompressedMatrixTest.java

@@ -144,7 +144,7 @@ public void testUnaryOperators(AggType aggType, boolean inCP) {
 	public void testNonZeros() {
 		if(!(cmb instanceof CompressedMatrixBlock))
 			return; // Input was not compressed then just pass test
-		if(!(cmb.getNonZeros() >= mb.getNonZeros())) { // guarantee that the nnz is at least the nnz
+		if(!(cmb.getNonZeros() >= mb.getNonZeros() || cmb.getNonZeros() == -1)) { // guarantee that the nnz is at least the nnz
 			fail(bufferedToString + "\nIncorrect number of non Zeros should guarantee greater than or equals but are "
 				+ cmb.getNonZeros() + " and should be: " + mb.getNonZeros());
 		}
@@ -160,7 +160,6 @@ public void testSerialization() {
 			ByteArrayOutputStream bos = new ByteArrayOutputStream();
 			DataOutputStream fos = new DataOutputStream(bos);
 			cmb.write(fos);
-
 			// deserialize compressed matrix block
 			ByteArrayInputStream bis = new ByteArrayInputStream(bos.toByteArray());
 			DataInputStream fis = new DataInputStream(bis);
@@ -170,6 +169,8 @@ public void testSerialization() {
 			// decompress the compressed matrix block
 			MatrixBlock tmp = cmb2.decompress();
 
+			((CompressedMatrixBlock)cmb).clearSoftReferenceToDecompressed();
+
 			compareResultMatrices(mb, tmp, 1);
 		}
 		catch(Exception e) {
@@ -421,8 +422,7 @@ public void testAggregateTernaryOperation() {
 			MatrixBlock ret1 = MatrixBlock.aggregateTernaryOperations(cmb, m2, m3, null, op, true);
 			ucRet = MatrixBlock.aggregateTernaryOperations(mb, m2, m3, ucRet, op, true);
 
-			// LOG.error(ret1);
-			// LOG.error(ucRet);
+
 			compareResultMatrices(ucRet, ret1, 1);
 		}
 		catch(Exception e) {

src/test/java/org/apache/sysds/test/component/compress/cost/ACostTest.java

@@ -85,7 +85,6 @@ public void testCostEstimate() {
 				sb.append(String.format("\nActualCostIndividual:  %15.0f", actualCostIndividual));
 				sb.append(String.format("\nEstimateCoCodeCost:    %15.0f", estimatedCostCoCode));
 				sb.append(String.format("\nActualCoCodeCost:      %15.0f", actualCostCoCode));
-
 				LOG.error(sb);
 			}
 

src/test/java/org/apache/sysds/test/functions/builtin/part1/BuiltinFFNeuralNetworkTest.java

@@ -83,7 +83,7 @@ public void testClassificationFit() {
 
 		fullDMLScriptName = getScript();
 
-		LOG.error(runTest(null));
+		LOG.debug(runTest(null));
 
 		double[][] from_DML = TestUtils.convertHashMapToDoubleArray(readDMLScalarFromOutputDir(out_path));
 		double accuracy = from_DML[0][0];

src/test/java/org/apache/sysds/test/functions/builtin/part2/BuiltinMDTest.java

@@ -28,6 +28,7 @@
 import org.apache.sysds.test.AutomatedTestBase;
 import org.apache.sysds.test.TestConfiguration;
 import org.apache.sysds.test.TestUtils;
+import org.junit.Ignore;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
@@ -90,6 +91,8 @@ public void testMDCP() {
 	}
 
 	@Test
+	@Ignore
+	// https://issues.apache.org/jira/browse/SYSTEMDS-3716
 	public void testMDSP() {
 		double[][] D =  {
 			{7567, 231, 1231, 1232, 122, 321},

src/test/java/org/apache/sysds/test/functions/builtin/part2/BuiltinMeanImputationTest.java

@@ -81,7 +81,7 @@ private void runMeanMedianTest(String testname, boolean defaultProb, Types.ExecT
 			rCmd = "Rscript" + " " + fullRScriptName + " " + DATASET_DIR+"Salaries.csv" + " " + expectedDir();
 
 
-			LOG.error(runTest(null));
+			LOG.debug(runTest(null));
 			runRScript(true);
 
 			//compare matrices

src/test/java/org/apache/sysds/test/functions/builtin/part2/BuiltinRaJoinTest.java

@@ -282,23 +282,15 @@ private void runRaJoinTest(double [][] A, int colA, double [][] B, int colB, dou
 			programArgs = new String[]{"-stats", "-args",
 				input("A"), String.valueOf(colA), input("B"),
 				String.valueOf(colB), method, output("result") };
-			System.out.println(Arrays.deepToString(A));
-			System.out.println(colA);
-			//fullRScriptName = HOME + TEST_NAME + ".R";
-			//rCmd = "Rscript" + " " + fullRScriptName + " "
-			//	+ inputDir() + " " + col + " "  + expectedDir();
-
 			writeInputMatrixWithMTD("A", A, true);
 			writeInputMatrixWithMTD("B", B, true);
 
-			// run dmlScript and RScript
-			runTest(true, false, null, -1);
-			//runRScript(true);
+			// run dmlScript 
+			runTest(null);
 
 			//compare matrices
 			HashMap<CellIndex, Double> dmlfile = readDMLMatrixFromOutputDir("result");
 			HashMap<CellIndex, Double> expectedOutput = TestUtils.convert2DDoubleArrayToHashMap(Y);
-			//HashMap<CellIndex, Double> rfile  = readRMatrixFromExpectedDir("result");
 			TestUtils.compareMatrices(dmlfile, expectedOutput, eps, "Stat-DML", "Expected");
 		}
 		finally {

src/test/java/org/apache/sysds/test/functions/federated/codegen/FederatedRowwiseTmplTest.java

@@ -176,8 +176,6 @@ private void testFederatedCodegenRowwise(ExecMode exec_mode) {
 		HashMap<CellIndex, Double> refResults = readDMLMatrixFromExpectedDir(OUTPUT_NAME);
 		HashMap<CellIndex, Double> fedResults = readDMLMatrixFromOutputDir(OUTPUT_NAME);
 
-		// LOG.error(refResults);
-		// LOG.error(fedResults);
 		TestUtils.compareMatrices(fedResults, refResults, TOLERANCE, "Fed", "Ref");
 
 		TestUtils.shutdownThreads(thread1, thread2);