[MINOR] Fix robustness parameter server utils and tests

Author: mboehm7

src/main/java/org/apache/sysds/hops/rewrite/HopRewriteUtils.java

@@ -20,8 +20,6 @@
 package org.apache.sysds.hops.rewrite;
 
 import org.apache.commons.lang3.ArrayUtils;
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
 import org.apache.sysds.api.DMLScript;
 import org.apache.sysds.common.Opcodes;
 import org.apache.sysds.common.Types.DataType;
@@ -90,7 +88,6 @@
 import java.util.List;
 
 public class HopRewriteUtils {
-	private static final Log LOG = LogFactory.getLog(HopRewriteUtils.class.getName());
 
 	public static boolean isValueTypeCast( OpOp1 op ) {
 		return op == OpOp1.CAST_AS_BOOLEAN
@@ -1712,7 +1709,6 @@ public static boolean knownParamservFunctions(Hop hop, DMLProgram prog) {
 			// If the function keys are incorrect this exception is caught for robustness in error messages for users.
 			// Intensionally only catching the exception!
 			// For debugging if for some reason the error we encountered was something else we LOG the error.
-			LOG.error(ex);
 			return false;
 		}
 	}

src/main/java/org/apache/sysds/parser/DMLProgram.java

@@ -89,12 +89,14 @@ public FunctionDictionary<FunctionStatementBlock> createNamespace(String namespa
 	 */
 	public FunctionStatementBlock getFunctionStatementBlock(String fkey) {
 		String[] tmp = splitFunctionKey(fkey);
-		return getFunctionStatementBlock(tmp[0], tmp[1]);
+		return tmp.length < 2 ? null : 
+			getFunctionStatementBlock(tmp[0], tmp[1]);
 	}
 
 	public void removeFunctionStatementBlock(String fkey) {
 		String[] tmp = splitFunctionKey(fkey);
-		removeFunctionStatementBlock(tmp[0], tmp[1]);
+		if( tmp.length == 2 )
+			removeFunctionStatementBlock(tmp[0], tmp[1]);
 	}
 
 	public FunctionStatementBlock getFunctionStatementBlock(String namespaceKey, String functionName) {

src/main/java/org/apache/sysds/runtime/controlprogram/paramserv/dp/DCLocalScheme.java

@@ -53,7 +53,8 @@ public static List<MatrixBlock> partition(int k, MatrixBlock mb) {
 	}
 
 	private static List<MatrixObject> doPartitioning(int k, MatrixBlock mb) {
-		return partition(k, mb).stream().map(ParamservUtils::newMatrixObject).collect(Collectors.toList());
+		return partition(k, mb).stream()
+			.map(m -> ParamservUtils.newMatrixObject(m, false)).collect(Collectors.toList());
 	}
 
 	@Override

src/main/java/org/apache/sysds/runtime/controlprogram/paramserv/dp/DRLocalScheme.java

@@ -49,7 +49,8 @@ private static List<MatrixBlock> partition(int k, MatrixBlock mb, MatrixBlock pe
 	}
 
 	private static List<MatrixObject> internalDoPartitioning(int k, MatrixBlock mb, MatrixBlock permutation) {
-		return partition(k, mb, permutation).stream().map(ParamservUtils::newMatrixObject).collect(Collectors.toList());
+		return partition(k, mb, permutation).stream()
+			.map(m -> ParamservUtils.newMatrixObject(m, false)).collect(Collectors.toList());
 	}
 
 	@Override

src/main/java/org/apache/sysds/runtime/controlprogram/paramserv/dp/ORLocalScheme.java

@@ -46,7 +46,8 @@ public static List<MatrixBlock> partition(int k, MatrixBlock mb, List<MatrixBloc
 	}
 
 	private static List<MatrixObject> doPartitioning(int k, MatrixBlock mb, List<MatrixBlock> permutations) {
-		return partition(k, mb, permutations).stream().map(ParamservUtils::newMatrixObject).collect(Collectors.toList());
+		return partition(k, mb, permutations).stream()
+			.map(m -> ParamservUtils.newMatrixObject(m, false)).collect(Collectors.toList());
 	}
 
 	@Override

src/test/java/org/apache/sysds/test/functions/frame/FrameEqualTest.java

@@ -134,7 +134,7 @@ private void runComparisonTest(Types.ValueType[] schema1, Types.ValueType[] sche
 		Types.ExecMode platformOld = setExecMode(et);
 		boolean oldFlag = OptimizerUtils.ALLOW_ALGEBRAIC_SIMPLIFICATION;
 		boolean sparkConfigOld = DMLScript.USE_LOCAL_SPARK_CONFIG;
-		setOutputBuffering(true);
+		//setOutputBuffering(true);
 		try {
 			getAndLoadTestConfiguration(TEST_NAME);
 

src/test/java/org/apache/sysds/test/functions/paramserv/ParamservSyntaxTest.java

@@ -98,6 +98,5 @@ private void runDMLTest(String testname, boolean exceptionExpected, Class<?> exc
 		programArgs = new String[] { "-explain" };
 		fullDMLScriptName = HOME + testname + ".dml";
 		runTest(true, exceptionExpected, exceptionClass, errmsg, -1);
-		setOutputBuffering(false);
 	}
 }

src/test/java/org/apache/sysds/test/functions/paramserv/SparkDataPartitionerTest.java

@@ -49,7 +49,9 @@ public class SparkDataPartitionerTest extends BaseDataPartitionerTest {
 
 	private Map<Integer, Tuple2<MatrixBlock, MatrixBlock>> doPartitioning(Statement.PSScheme scheme) {
 		MatrixBlock[] mbs = generateData();
-		return SparkParamservUtils.doPartitionOnSpark(_sec, ParamservUtils.newMatrixObject(mbs[0]), ParamservUtils.newMatrixObject(mbs[1]), scheme, WORKER_NUM).collectAsMap();
+		return SparkParamservUtils.doPartitionOnSpark(_sec,
+			ParamservUtils.newMatrixObject(mbs[0], false), 
+			ParamservUtils.newMatrixObject(mbs[1]), scheme, WORKER_NUM).collectAsMap();
 	}
 
 	@Test