You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@systemds.apache.org by mb...@apache.org on 2024/08/17 11:43:59 UTC

(systemds) branch main updated: [MINOR] Additional parfor tests, and minor cleanups of unused code

This is an automated email from the ASF dual-hosted git repository.

mboehm7 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/systemds.git


The following commit(s) were added to refs/heads/main by this push:
     new 78b340fe9e [MINOR] Additional parfor tests, and minor cleanups of unused code
78b340fe9e is described below

commit 78b340fe9eb84adbc1e236f543366b5d53653faf
Author: Matthias Boehm <mb...@gmail.com>
AuthorDate: Sat Aug 17 13:42:11 2024 +0200

    [MINOR] Additional parfor tests, and minor cleanups of unused code
---
 src/main/java/org/apache/sysds/hops/LiteralOp.java |   2 +-
 .../sysds/parser/BuiltinFunctionExpression.java    |   1 +
 .../runtime/controlprogram/ParForProgramBlock.java |  47 +-----
 .../parfor/ResultMergeRemoteGrouping.java          |  42 -----
 .../parfor/ResultMergeRemoteSorting.java           |  46 -----
 .../parfor/ResultMergeTaggedMatrixIndexes.java     |  94 -----------
 .../controlprogram/parfor/TaskPartitioner.java     |   4 +-
 .../parfor/TaskPartitionerFactory.java             |  50 ++++++
 .../parfor/TaskPartitionerNaive.java               |   6 +-
 .../parfor/TaskPartitionerStatic.java              |   6 +-
 .../parfor/opt/OptTreePlanChecker.java             | 185 ---------------------
 .../parfor/opt/OptimizationWrapper.java            |  13 --
 .../sysds/runtime/util/BinaryBlockInputFormat.java |  48 ------
 .../runtime/util/BinaryBlockRecordReader.java      |  68 --------
 .../apache/sysds/utils/SystemDSLoaderUtils.java    |  41 -----
 .../sysds/test/component/misc/OpTypeTest.java      |   4 -
 .../test/component/parfor/TaskPartitionerTest.java |  85 ++++++++++
 .../functions/builtin/part2/BuiltinRaJoinTest.java |   1 -
 .../parfor/misc/ParForRecursiveFunctionTest.java   |  84 ++++++++++
 .../scripts/functions/parfor/parfor_recursive.dml  |  37 +++++
 20 files changed, 270 insertions(+), 594 deletions(-)

diff --git a/src/main/java/org/apache/sysds/hops/LiteralOp.java b/src/main/java/org/apache/sysds/hops/LiteralOp.java
index 9c44ef1187..1d2911f1fa 100644
--- a/src/main/java/org/apache/sysds/hops/LiteralOp.java
+++ b/src/main/java/org/apache/sysds/hops/LiteralOp.java
@@ -254,8 +254,8 @@ public class LiteralOp extends Hop
 			case FP64:
 				return String.valueOf(value_double);
 			case STRING:
+			case HASH32:
 			case HASH64:
-				return value_string;
 			case CHARACTER:
 				return value_string;
 			case UNKNOWN:
diff --git a/src/main/java/org/apache/sysds/parser/BuiltinFunctionExpression.java b/src/main/java/org/apache/sysds/parser/BuiltinFunctionExpression.java
index ec9b4a4bbd..de10e090f5 100644
--- a/src/main/java/org/apache/sysds/parser/BuiltinFunctionExpression.java
+++ b/src/main/java/org/apache/sysds/parser/BuiltinFunctionExpression.java
@@ -965,6 +965,7 @@ public class BuiltinFunctionExpression extends DataIdentifier {
 				case CHARACTER:
 				case FP64:
 				case FP32:
+				case HASH32:
 				case HASH64: //default
 					output.setValueType(ValueType.FP64);
 					break;
diff --git a/src/main/java/org/apache/sysds/runtime/controlprogram/ParForProgramBlock.java b/src/main/java/org/apache/sysds/runtime/controlprogram/ParForProgramBlock.java
index c4c75c35e7..01896cb73e 100644
--- a/src/main/java/org/apache/sysds/runtime/controlprogram/ParForProgramBlock.java
+++ b/src/main/java/org/apache/sysds/runtime/controlprogram/ParForProgramBlock.java
@@ -75,12 +75,7 @@ import org.apache.sysds.runtime.controlprogram.parfor.ResultMergeLocalMemory;
 import org.apache.sysds.runtime.controlprogram.parfor.ResultMergeRemoteSpark;
 import org.apache.sysds.runtime.controlprogram.parfor.Task;
 import org.apache.sysds.runtime.controlprogram.parfor.TaskPartitioner;
-import org.apache.sysds.runtime.controlprogram.parfor.TaskPartitionerFactoring;
-import org.apache.sysds.runtime.controlprogram.parfor.TaskPartitionerFactoringCmax;
-import org.apache.sysds.runtime.controlprogram.parfor.TaskPartitionerFactoringCmin;
-import org.apache.sysds.runtime.controlprogram.parfor.TaskPartitionerFixedsize;
-import org.apache.sysds.runtime.controlprogram.parfor.TaskPartitionerNaive;
-import org.apache.sysds.runtime.controlprogram.parfor.TaskPartitionerStatic;
+import org.apache.sysds.runtime.controlprogram.parfor.TaskPartitionerFactory;
 import org.apache.sysds.runtime.controlprogram.parfor.opt.OptTreeConverter;
 import org.apache.sysds.runtime.controlprogram.parfor.opt.OptimizationWrapper;
 import org.apache.sysds.runtime.controlprogram.parfor.opt.OptimizerRuleBased;
@@ -1279,43 +1274,9 @@ public class ParForProgramBlock extends ForProgramBlock {
 	 * @param incr ?
 	 * @return task partitioner
 	 */
-	private TaskPartitioner createTaskPartitioner( IntObject from, IntObject to, IntObject incr ) 
-	{
-		TaskPartitioner tp;
-		
-		switch( _taskPartitioner ) {
-			case FIXED:
-				tp = new TaskPartitionerFixedsize(
-					_taskSize, _iterPredVar, from, to, incr);
-				break;
-			case NAIVE:
-				tp = new TaskPartitionerNaive(
-					_taskSize, _iterPredVar, from, to, incr);
-				break;
-			case STATIC:
-				tp = new TaskPartitionerStatic(
-					_taskSize, _numThreads, _iterPredVar, from, to, incr);
-				break;
-			case FACTORING:
-				tp = new TaskPartitionerFactoring(
-					_taskSize,_numThreads, _iterPredVar, from, to, incr);
-				break;
-			case FACTORING_CMIN:
-				//for constrained factoring the tasksize is used as the minimum constraint
-				tp = new TaskPartitionerFactoringCmin(_taskSize,_numThreads, 
-					_taskSize, _iterPredVar, from, to, incr);
-				break;
-
-			case FACTORING_CMAX:
-				//for constrained factoring the tasksize is used as the minimum constraint
-				tp = new TaskPartitionerFactoringCmax(_taskSize,_numThreads, 
-					_taskSize, _iterPredVar, from, to, incr);
-				break;	
-			default:
-				throw new DMLRuntimeException("Undefined task partitioner: '"+_taskPartitioner+"'.");
-		}
-		
-		return tp;
+	private TaskPartitioner createTaskPartitioner( IntObject from, IntObject to, IntObject incr ) {
+		return TaskPartitionerFactory.createTaskPartitioner(
+			_taskPartitioner, from, to, incr, _taskSize, _numThreads, _iterPredVar);
 	}
 	
 	/**
diff --git a/src/main/java/org/apache/sysds/runtime/controlprogram/parfor/ResultMergeRemoteGrouping.java b/src/main/java/org/apache/sysds/runtime/controlprogram/parfor/ResultMergeRemoteGrouping.java
deleted file mode 100644
index ccf27c9a8f..0000000000
--- a/src/main/java/org/apache/sysds/runtime/controlprogram/parfor/ResultMergeRemoteGrouping.java
+++ /dev/null
@@ -1,42 +0,0 @@
-/*
- * 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.controlprogram.parfor;
-
-
-import org.apache.hadoop.io.WritableComparable;
-import org.apache.hadoop.io.WritableComparator;
-
-public class ResultMergeRemoteGrouping extends WritableComparator
-{
-	protected ResultMergeRemoteGrouping() {
-		super(ResultMergeTaggedMatrixIndexes.class,true);
-	}
-	
-	@Override
-	@SuppressWarnings("rawtypes")
-	public int compare(WritableComparable k1, WritableComparable k2) 
-	{
-		ResultMergeTaggedMatrixIndexes key1 = (ResultMergeTaggedMatrixIndexes)k1;
-		ResultMergeTaggedMatrixIndexes key2 = (ResultMergeTaggedMatrixIndexes)k2;
-		
-		//group by matrix indexes only (including all tags)
-		return key1.getIndexes().compareTo(key2.getIndexes());
-	}
-}
diff --git a/src/main/java/org/apache/sysds/runtime/controlprogram/parfor/ResultMergeRemoteSorting.java b/src/main/java/org/apache/sysds/runtime/controlprogram/parfor/ResultMergeRemoteSorting.java
deleted file mode 100644
index 5d44903958..0000000000
--- a/src/main/java/org/apache/sysds/runtime/controlprogram/parfor/ResultMergeRemoteSorting.java
+++ /dev/null
@@ -1,46 +0,0 @@
-/*
- * 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.controlprogram.parfor;
-
-import org.apache.hadoop.io.WritableComparable;
-import org.apache.hadoop.io.WritableComparator;
-
-public class ResultMergeRemoteSorting extends WritableComparator
-{
-	protected ResultMergeRemoteSorting() {
-		super(ResultMergeTaggedMatrixIndexes.class, true);
-	}
-	
-	@Override
-	@SuppressWarnings("rawtypes")
-	public int compare(WritableComparable k1, WritableComparable k2) 
-	{
-		ResultMergeTaggedMatrixIndexes key1 = (ResultMergeTaggedMatrixIndexes)k1;
-		ResultMergeTaggedMatrixIndexes key2 = (ResultMergeTaggedMatrixIndexes)k2;
-
-		int ret = key1.getIndexes().compareTo(key2.getIndexes());
-		if( ret == 0 ) //same indexes, secondary sort
-		{
-			ret = ((key1.getTag() == key2.getTag()) ? 0 : 
-				(key1.getTag() < key2.getTag())? -1 : 1);
-		}	
-		return ret;
-	}
-}
diff --git a/src/main/java/org/apache/sysds/runtime/controlprogram/parfor/ResultMergeTaggedMatrixIndexes.java b/src/main/java/org/apache/sysds/runtime/controlprogram/parfor/ResultMergeTaggedMatrixIndexes.java
deleted file mode 100644
index c0437dce35..0000000000
--- a/src/main/java/org/apache/sysds/runtime/controlprogram/parfor/ResultMergeTaggedMatrixIndexes.java
+++ /dev/null
@@ -1,94 +0,0 @@
-/*
- * 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.controlprogram.parfor;
-
-import java.io.DataInput;
-import java.io.DataOutput;
-import java.io.IOException;
-
-import org.apache.hadoop.io.WritableComparable;
-import org.apache.sysds.runtime.matrix.data.MatrixIndexes;
-
-/**
- * This class serves as composite key for the remote result merge job
- * (for any data format) in order to sort on both matrix indexes and tag
- * but group all blocks according to matrix indexes only. This prevents
- * us from doing an 2pass out-of-core algorithm at the reducer since we
- * can guarantee that the compare block (tag 0) will be the first element
- * in the iterator.
- * 
- */
-public class ResultMergeTaggedMatrixIndexes implements WritableComparable<ResultMergeTaggedMatrixIndexes>
-{
-	private MatrixIndexes _ix;
-	private byte _tag = -1;
-	
-	public ResultMergeTaggedMatrixIndexes() {
-		_ix = new MatrixIndexes();
-	}
-
-	public MatrixIndexes getIndexes() {
-		return _ix;
-	}
-	
-	public byte getTag() {
-		return _tag;
-	}
-	
-	public void setTag(byte tag) {
-		_tag = tag;
-	}
-
-	@Override
-	public void readFields(DataInput in) throws IOException {
-		if( _ix == null )
-			_ix = new MatrixIndexes();
-		_ix.readFields(in);
-		_tag = in.readByte();
-	}
-
-	@Override
-	public void write(DataOutput out) throws IOException {
-		_ix.write(out);
-		out.writeByte(_tag);
-	}
-
-	@Override
-	public int compareTo(ResultMergeTaggedMatrixIndexes that) {
-		int ret = _ix.compareTo(that._ix);
-		if( ret == 0 )
-			ret = ((_tag == that._tag) ? 0 : 
-				(_tag < that._tag)? -1 : 1);
-		return ret;
-	}
-	
-	@Override
-	public boolean equals(Object other) {
-		if( !(other instanceof ResultMergeTaggedMatrixIndexes) )
-			return false;
-		ResultMergeTaggedMatrixIndexes that = (ResultMergeTaggedMatrixIndexes)other;
-		return (_ix.equals(that._ix) && _tag == that._tag);
-	}
-	
-	@Override
-	public int hashCode() {
-		throw new RuntimeException("hashCode() should never be called on instances of this class.");
-	}
-}
diff --git a/src/main/java/org/apache/sysds/runtime/controlprogram/parfor/TaskPartitioner.java b/src/main/java/org/apache/sysds/runtime/controlprogram/parfor/TaskPartitioner.java
index abd0279077..9d46947ba6 100644
--- a/src/main/java/org/apache/sysds/runtime/controlprogram/parfor/TaskPartitioner.java
+++ b/src/main/java/org/apache/sysds/runtime/controlprogram/parfor/TaskPartitioner.java
@@ -33,8 +33,8 @@ import org.apache.sysds.runtime.instructions.cp.IntObject;
  */
 public abstract class TaskPartitioner 
 {	
-	protected long           _taskSize     = -1;	
-	protected String  		 _iterVarName  = null;
+	protected long           _taskSize     = -1;
+	protected String         _iterVarName  = null;
 	protected IntObject      _fromVal      = null;
 	protected IntObject      _toVal        = null;
 	protected IntObject      _incrVal      = null;
diff --git a/src/main/java/org/apache/sysds/runtime/controlprogram/parfor/TaskPartitionerFactory.java b/src/main/java/org/apache/sysds/runtime/controlprogram/parfor/TaskPartitionerFactory.java
new file mode 100644
index 0000000000..66d4961104
--- /dev/null
+++ b/src/main/java/org/apache/sysds/runtime/controlprogram/parfor/TaskPartitionerFactory.java
@@ -0,0 +1,50 @@
+/*
+ * 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.controlprogram.parfor;
+
+import org.apache.sysds.runtime.DMLRuntimeException;
+import org.apache.sysds.runtime.controlprogram.ParForProgramBlock.PTaskPartitioner;
+import org.apache.sysds.runtime.instructions.cp.IntObject;
+
+public abstract class TaskPartitionerFactory 
+{
+	public static TaskPartitioner createTaskPartitioner(PTaskPartitioner type, 
+		IntObject from, IntObject to, IntObject incr, long taskSize, int numThreads, String iterPredVar) 
+	{
+		switch( type ) {
+			case FIXED:
+				return new TaskPartitionerFixedsize(taskSize, iterPredVar, from, to, incr);
+			case NAIVE:
+				return new TaskPartitionerNaive(taskSize, iterPredVar, from, to, incr);
+			case STATIC:
+				return new TaskPartitionerStatic(taskSize, numThreads, iterPredVar, from, to, incr);
+			case FACTORING:
+				return new TaskPartitionerFactoring(taskSize, numThreads, iterPredVar, from, to, incr);
+			case FACTORING_CMIN:
+				return new TaskPartitionerFactoringCmin(taskSize,
+					numThreads, taskSize, iterPredVar, from, to, incr);
+			case FACTORING_CMAX:
+				return new TaskPartitionerFactoringCmax(taskSize,
+					numThreads, taskSize, iterPredVar, from, to, incr);
+			default:
+				throw new DMLRuntimeException("Undefined task partitioner: '"+type+"'.");
+		}
+	}
+}
diff --git a/src/main/java/org/apache/sysds/runtime/controlprogram/parfor/TaskPartitionerNaive.java b/src/main/java/org/apache/sysds/runtime/controlprogram/parfor/TaskPartitionerNaive.java
index 7704974548..3894bcff92 100644
--- a/src/main/java/org/apache/sysds/runtime/controlprogram/parfor/TaskPartitionerNaive.java
+++ b/src/main/java/org/apache/sysds/runtime/controlprogram/parfor/TaskPartitionerNaive.java
@@ -29,12 +29,12 @@ import org.apache.sysds.runtime.instructions.cp.IntObject;
  */
 public class TaskPartitionerNaive extends TaskPartitionerFixedsize
 {
-	
-	public TaskPartitionerNaive( long taskSize, String iterVarName, IntObject fromVal, IntObject toVal, IntObject incrVal ) 
+	public TaskPartitionerNaive( long taskSize, String iterVarName,
+		IntObject fromVal, IntObject toVal, IntObject incrVal ) 
 	{
 		super(taskSize, iterVarName, fromVal, toVal, incrVal);
 	
 		//compute the new task size
 		_taskSize = 1;
-	}	
+	}
 }
diff --git a/src/main/java/org/apache/sysds/runtime/controlprogram/parfor/TaskPartitionerStatic.java b/src/main/java/org/apache/sysds/runtime/controlprogram/parfor/TaskPartitionerStatic.java
index 54ea8cbdc1..244cca320d 100644
--- a/src/main/java/org/apache/sysds/runtime/controlprogram/parfor/TaskPartitionerStatic.java
+++ b/src/main/java/org/apache/sysds/runtime/controlprogram/parfor/TaskPartitionerStatic.java
@@ -29,12 +29,12 @@ import org.apache.sysds.runtime.instructions.cp.IntObject;
  */
 public class TaskPartitionerStatic extends TaskPartitionerFixedsize
 {
-	
-	public TaskPartitionerStatic( long taskSize, int numThreads, String iterVarName, IntObject fromVal, IntObject toVal, IntObject incrVal ) 
+	public TaskPartitionerStatic( long taskSize, int numThreads,
+		String iterVarName, IntObject fromVal, IntObject toVal, IntObject incrVal ) 
 	{
 		super(taskSize, iterVarName, fromVal, toVal, incrVal);
 	
 		_taskSize = _numIter / numThreads;
 		_firstnPlus1 = (int)_numIter % numThreads;
-	}	
+	}
 }
diff --git a/src/main/java/org/apache/sysds/runtime/controlprogram/parfor/opt/OptTreePlanChecker.java b/src/main/java/org/apache/sysds/runtime/controlprogram/parfor/opt/OptTreePlanChecker.java
deleted file mode 100644
index 2cbf1228a9..0000000000
--- a/src/main/java/org/apache/sysds/runtime/controlprogram/parfor/opt/OptTreePlanChecker.java
+++ /dev/null
@@ -1,185 +0,0 @@
-/*
- * 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.controlprogram.parfor.opt;
-
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.Set;
-
-import org.apache.sysds.hops.FunctionOp;
-import org.apache.sysds.hops.Hop;
-import org.apache.sysds.parser.DMLProgram;
-import org.apache.sysds.parser.ForStatement;
-import org.apache.sysds.parser.ForStatementBlock;
-import org.apache.sysds.parser.FunctionStatement;
-import org.apache.sysds.parser.FunctionStatementBlock;
-import org.apache.sysds.parser.IfStatement;
-import org.apache.sysds.parser.IfStatementBlock;
-import org.apache.sysds.parser.StatementBlock;
-import org.apache.sysds.parser.WhileStatement;
-import org.apache.sysds.parser.WhileStatementBlock;
-import org.apache.sysds.runtime.DMLRuntimeException;
-import org.apache.sysds.runtime.controlprogram.BasicProgramBlock;
-import org.apache.sysds.runtime.controlprogram.ForProgramBlock;
-import org.apache.sysds.runtime.controlprogram.FunctionProgramBlock;
-import org.apache.sysds.runtime.controlprogram.IfProgramBlock;
-import org.apache.sysds.runtime.controlprogram.Program;
-import org.apache.sysds.runtime.controlprogram.ProgramBlock;
-import org.apache.sysds.runtime.controlprogram.WhileProgramBlock;
-import org.apache.sysds.runtime.instructions.Instruction;
-import org.apache.sysds.runtime.instructions.cp.FunctionCallCPInstruction;
-
-public class OptTreePlanChecker 
-{
-
-	public static void checkProgramCorrectness( ProgramBlock pb, StatementBlock sb, Set<String> fnStack ) 
-	{
-		Program prog = pb.getProgram();
-		DMLProgram dprog = sb.getDMLProg();
-		
-		if (pb instanceof FunctionProgramBlock && sb instanceof FunctionStatementBlock ) {
-			FunctionProgramBlock fpb = (FunctionProgramBlock)pb;
-			FunctionStatementBlock fsb = (FunctionStatementBlock)sb;
-			FunctionStatement fstmt = (FunctionStatement)fsb.getStatement(0);
-			for( int i=0; i<fpb.getChildBlocks().size(); i++ ) {
-				ProgramBlock pbc = fpb.getChildBlocks().get(i);
-				StatementBlock sbc = fstmt.getBody().get(i);
-				checkProgramCorrectness(pbc, sbc, fnStack);
-			}
-		}
-		else if (pb instanceof WhileProgramBlock && sb instanceof WhileStatementBlock) {
-			WhileProgramBlock wpb = (WhileProgramBlock) pb;
-			WhileStatementBlock wsb = (WhileStatementBlock) sb;
-			WhileStatement wstmt = (WhileStatement) wsb.getStatement(0);
-			checkHopDagCorrectness(prog, dprog, wsb.getPredicateHops(), wpb.getPredicate(), fnStack);
-			for( int i=0; i<wpb.getChildBlocks().size(); i++ ) {
-				ProgramBlock pbc = wpb.getChildBlocks().get(i);
-				StatementBlock sbc = wstmt.getBody().get(i);
-				checkProgramCorrectness(pbc, sbc, fnStack);
-			}
-			checkLinksProgramStatementBlock(wpb, wsb);
-		}
-		else if (pb instanceof IfProgramBlock && sb instanceof IfStatementBlock) {
-			IfProgramBlock ipb = (IfProgramBlock) pb;
-			IfStatementBlock isb = (IfStatementBlock) sb;
-			IfStatement istmt = (IfStatement) isb.getStatement(0);
-			checkHopDagCorrectness(prog, dprog, isb.getPredicateHops(), ipb.getPredicate(), fnStack);
-			for( int i=0; i<ipb.getChildBlocksIfBody().size(); i++ ) {
-				ProgramBlock pbc = ipb.getChildBlocksIfBody().get(i);
-				StatementBlock sbc = istmt.getIfBody().get(i);
-				checkProgramCorrectness(pbc, sbc, fnStack);
-			}
-			for( int i=0; i<ipb.getChildBlocksElseBody().size(); i++ ) {
-				ProgramBlock pbc = ipb.getChildBlocksElseBody().get(i);
-				StatementBlock sbc = istmt.getElseBody().get(i);
-				checkProgramCorrectness(pbc, sbc, fnStack);
-			}
-			checkLinksProgramStatementBlock(ipb, isb);
-		}
-		else if (pb instanceof ForProgramBlock && sb instanceof ForStatementBlock) { //incl parfor
-			ForProgramBlock fpb = (ForProgramBlock) pb;
-			ForStatementBlock fsb = (ForStatementBlock) sb;
-			ForStatement fstmt = (ForStatement) sb.getStatement(0);
-			checkHopDagCorrectness(prog, dprog, fsb.getFromHops(), fpb.getFromInstructions(), fnStack);
-			checkHopDagCorrectness(prog, dprog, fsb.getToHops(), fpb.getToInstructions(), fnStack);
-			checkHopDagCorrectness(prog, dprog, fsb.getIncrementHops(), fpb.getIncrementInstructions(), fnStack);
-			for( int i=0; i<fpb.getChildBlocks().size(); i++ ) {
-				ProgramBlock pbc = fpb.getChildBlocks().get(i);
-				StatementBlock sbc = fstmt.getBody().get(i);
-				checkProgramCorrectness(pbc, sbc, fnStack);
-			}
-			checkLinksProgramStatementBlock(fpb, fsb);
-		}
-		else if( pb instanceof BasicProgramBlock ) {
-			BasicProgramBlock bpb = (BasicProgramBlock) pb;
-			checkHopDagCorrectness(prog, dprog, sb.getHops(), bpb.getInstructions(), fnStack);
-		}
-	}
-
-	private static void checkHopDagCorrectness( Program prog, DMLProgram dprog, ArrayList<Hop> roots, ArrayList<Instruction> inst, Set<String> fnStack ) {
-		if( roots != null )
-			for( Hop hop : roots )
-				checkHopDagCorrectness(prog, dprog, hop, inst, fnStack);
-	}
-
-	private static void checkHopDagCorrectness( Program prog, DMLProgram dprog, Hop root, ArrayList<Instruction> inst, Set<String> fnStack ) {
-		//set of checks to perform
-		checkFunctionNames(prog, dprog, root, inst, fnStack);
-	}
-
-	private static void checkLinksProgramStatementBlock( ProgramBlock pb, StatementBlock sb ) {
-		if( pb.getStatementBlock() != sb )
-			throw new DMLRuntimeException("Links between programblocks and statementblocks are incorrect ("+pb+").");
-	}
-
-	private static void checkFunctionNames( Program prog, DMLProgram dprog, Hop root, ArrayList<Instruction> inst, Set<String> fnStack ) {
-		//reset visit status of dag
-		root.resetVisitStatus();
-		
-		//get all function op in this dag
-		HashMap<String, FunctionOp> fops = new HashMap<>();
-		getAllFunctionOps(root, fops);
-		
-		for( Instruction linst : inst )
-			if( linst instanceof FunctionCallCPInstruction )
-			{
-				FunctionCallCPInstruction flinst = (FunctionCallCPInstruction) linst;
-				String fnamespace = flinst.getNamespace();
-				String fname = flinst.getFunctionName();
-				String key = DMLProgram.constructFunctionKey(fnamespace, fname);
-				
-				//check 1: instruction name equal to hop name
-				if( !fops.containsKey(key) )
-					throw new DMLRuntimeException( "Function Check: instruction and hop names differ ("+key+", "+fops.keySet()+")" );
-				
-				//check 2: function exists
-				if( !prog.getFunctionProgramBlocks().containsKey(key) )
-					throw new DMLRuntimeException( "Function Check: function does not exits ("+key+")" );
-	
-				//check 3: recursive program check
-				FunctionProgramBlock fpb = prog.getFunctionProgramBlock(fnamespace, fname);
-				FunctionStatementBlock fsb = dprog.getFunctionStatementBlock(fnamespace, fname);
-				if( !fnStack.contains(key) )
-				{
-					fnStack.add(key);
-					checkProgramCorrectness(fpb, fsb, fnStack);
-					fnStack.remove(key);
-				}
-			}
-	}
-
-	private static void getAllFunctionOps( Hop hop, HashMap<String, FunctionOp> memo )
-	{
-		if( hop.isVisited() )
-			return;
-		
-		//process functionop
-		if( hop instanceof FunctionOp ) {
-			FunctionOp fop = (FunctionOp) hop;
-			memo.put(fop.getFunctionKey(), fop);
-		}
-		
-		//process children
-		for( Hop in : hop.getInput() )
-			getAllFunctionOps(in, memo);
-		
-		hop.setVisited();
-	}
-}
diff --git a/src/main/java/org/apache/sysds/runtime/controlprogram/parfor/opt/OptimizationWrapper.java b/src/main/java/org/apache/sysds/runtime/controlprogram/parfor/opt/OptimizationWrapper.java
index 6dbc952172..4a7cec8cd2 100644
--- a/src/main/java/org/apache/sysds/runtime/controlprogram/parfor/opt/OptimizationWrapper.java
+++ b/src/main/java/org/apache/sysds/runtime/controlprogram/parfor/opt/OptimizationWrapper.java
@@ -20,7 +20,6 @@
 package org.apache.sysds.runtime.controlprogram.parfor.opt;
 
 import java.util.ArrayList;
-import java.util.HashSet;
 import java.util.Set;
 
 import org.apache.commons.logging.Log;
@@ -75,7 +74,6 @@ public class OptimizationWrapper
 	
 	//internal parameters
 	public static final double PAR_FACTOR_INFRASTRUCTURE = 1.0;
-	private static final boolean CHECK_PLAN_CORRECTNESS = false;
 
 
 	/**
@@ -227,17 +225,6 @@ public class OptimizationWrapper
 		//core optimize
 		opt.optimize(sb, pb, tree, est, numRuns, ec);
 		LOG.debug("ParFOR Opt: Optimized plan (after optimization): \n" + tree.explain(false));
-		
-		//assert plan correctness
-		if( CHECK_PLAN_CORRECTNESS && LOG.isDebugEnabled() ) {
-			try{
-				OptTreePlanChecker.checkProgramCorrectness(pb, sb, new HashSet<String>());
-				LOG.debug("ParFOR Opt: Checked plan and program correctness.");
-			}
-			catch(Exception ex) {
-				throw new DMLRuntimeException("Failed to check program correctness.", ex);
-			}
-		}
 
 		long ltime = (long) time.stop();
 		LOG.trace("ParFOR Opt: Optimized plan in "+ltime+"ms.");
diff --git a/src/main/java/org/apache/sysds/runtime/util/BinaryBlockInputFormat.java b/src/main/java/org/apache/sysds/runtime/util/BinaryBlockInputFormat.java
deleted file mode 100644
index d9d8fc080b..0000000000
--- a/src/main/java/org/apache/sysds/runtime/util/BinaryBlockInputFormat.java
+++ /dev/null
@@ -1,48 +0,0 @@
-/*
- * 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.util;
-
-import java.io.IOException;
-
-import org.apache.hadoop.mapred.InputSplit;
-import org.apache.hadoop.mapred.JobConf;
-import org.apache.hadoop.mapred.RecordReader;
-import org.apache.hadoop.mapred.Reporter;
-import org.apache.hadoop.mapred.SequenceFileInputFormat;
-import org.apache.sysds.runtime.matrix.data.MatrixBlock;
-import org.apache.sysds.runtime.matrix.data.MatrixIndexes;
-import org.apache.hadoop.mapred.FileSplit;
-
-/**
- * Custom binary block input format to return the custom record reader.
- * <p>
- * NOTE: Not used by default.
- * <p>
- * NOTE: Used for performance debugging of binary block HDFS reads.
- */
-public class BinaryBlockInputFormat extends SequenceFileInputFormat<MatrixIndexes,MatrixBlock>
-{
-	@Override
-	public RecordReader<MatrixIndexes, MatrixBlock> getRecordReader(InputSplit split, JobConf job, Reporter reporter)
-		throws IOException 
-	{
-		return new BinaryBlockRecordReader(job, (FileSplit)split);
-	}
-}
\ No newline at end of file
diff --git a/src/main/java/org/apache/sysds/runtime/util/BinaryBlockRecordReader.java b/src/main/java/org/apache/sysds/runtime/util/BinaryBlockRecordReader.java
deleted file mode 100644
index ddae00d721..0000000000
--- a/src/main/java/org/apache/sysds/runtime/util/BinaryBlockRecordReader.java
+++ /dev/null
@@ -1,68 +0,0 @@
-/*
- * 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.util;
-
-import java.io.IOException;
-
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.mapred.FileSplit;
-import org.apache.hadoop.mapred.SequenceFileRecordReader;
-import org.apache.sysds.runtime.matrix.data.MatrixBlock;
-import org.apache.sysds.runtime.matrix.data.MatrixIndexes;
-
-/**
- * Custom record reader for binary block. Currently its only purpose is to allow for
- * detailed profiling of overall read time (io, deserialize, decompress).
- * 
- * NOTE: not used by default.
- */
-public class BinaryBlockRecordReader extends SequenceFileRecordReader<MatrixIndexes,MatrixBlock>
-{
-	//private long _time = 0;
-	
-	public BinaryBlockRecordReader(Configuration conf, FileSplit split)
-		throws IOException 
-	{
-		super(conf, split);
-		
-	}
-	
-	@Override
-	public synchronized boolean next(MatrixIndexes key, MatrixBlock value)
-		throws IOException 
-	{
-		//long t0 = System.nanoTime();		
-		boolean ret = super.next(key, value);		
-		//long t1 = System.nanoTime();
-		
-		//_time+=(t1-t0);
-		
-		return ret;
-	}
-
-	@Override
-	public synchronized void close() 
-		throws IOException 
-	{		
-		//in milliseconds.
-		//System.out.println(_time/1000000);
-		super.close();
-	}
-}
diff --git a/src/main/java/org/apache/sysds/utils/SystemDSLoaderUtils.java b/src/main/java/org/apache/sysds/utils/SystemDSLoaderUtils.java
deleted file mode 100644
index 67761b414f..0000000000
--- a/src/main/java/org/apache/sysds/utils/SystemDSLoaderUtils.java
+++ /dev/null
@@ -1,41 +0,0 @@
-/*
- * 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.utils;
-
-import java.lang.reflect.InvocationTargetException;
-import java.lang.reflect.Method;
-import java.net.URL; 
-import java.net.URLClassLoader;
-import java.io.File;
-import java.io.IOException;
-
-public class SystemDSLoaderUtils  {
-	
-	public void loadSystemDS(String filePath) 
-		throws NoSuchMethodException, SecurityException, IllegalAccessException, IllegalArgumentException, InvocationTargetException, IOException {
-		URL url = new File(filePath).toURI().toURL();
-		try( URLClassLoader classLoader = (URLClassLoader)ClassLoader.getSystemClassLoader() ) {
-			Method method = URLClassLoader.class.getDeclaredMethod("addURL", URL.class);
-			method.setAccessible(true);
-			method.invoke(classLoader, url);
-		}
-	}
-
-}
diff --git a/src/test/java/org/apache/sysds/test/component/misc/OpTypeTest.java b/src/test/java/org/apache/sysds/test/component/misc/OpTypeTest.java
index fabff5a459..581ff0d405 100644
--- a/src/test/java/org/apache/sysds/test/component/misc/OpTypeTest.java
+++ b/src/test/java/org/apache/sysds/test/component/misc/OpTypeTest.java
@@ -19,10 +19,6 @@
 
 package org.apache.sysds.test.component.misc;
 
-import java.io.IOException;
-import java.net.URL;
-import java.util.Enumeration;
-
 import org.apache.sysds.common.Types.OpOp1;
 import org.apache.sysds.common.Types.OpOp2;
 import org.apache.sysds.common.Types.OpOp3;
diff --git a/src/test/java/org/apache/sysds/test/component/parfor/TaskPartitionerTest.java b/src/test/java/org/apache/sysds/test/component/parfor/TaskPartitionerTest.java
new file mode 100644
index 0000000000..4e3acbe4c6
--- /dev/null
+++ b/src/test/java/org/apache/sysds/test/component/parfor/TaskPartitionerTest.java
@@ -0,0 +1,85 @@
+/*
+ * 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.parfor;
+
+import org.apache.sysds.runtime.controlprogram.ParForProgramBlock.PTaskPartitioner;
+import org.apache.sysds.runtime.controlprogram.parfor.LocalTaskQueue;
+import org.apache.sysds.runtime.controlprogram.parfor.Task;
+import org.apache.sysds.runtime.controlprogram.parfor.TaskPartitioner;
+import org.apache.sysds.runtime.controlprogram.parfor.TaskPartitionerFactory;
+import org.apache.sysds.runtime.controlprogram.parfor.stat.InfrastructureAnalyzer;
+import org.apache.sysds.runtime.instructions.cp.IntObject;
+import org.apache.sysds.runtime.util.CommonThreadPool;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class TaskPartitionerTest {
+
+	@Test
+	public void testNaive() {
+		testTaskPartitioner(2*LocalTaskQueue.MAX_SIZE, PTaskPartitioner.NAIVE);
+	}
+	
+	@Test
+	public void testStatic() {
+		testTaskPartitioner(2*LocalTaskQueue.MAX_SIZE, PTaskPartitioner.STATIC);
+	}
+	
+	@Test
+	public void testFixed() {
+		testTaskPartitioner(2*LocalTaskQueue.MAX_SIZE, PTaskPartitioner.FIXED);
+	}
+	
+	@Test
+	public void testFactoring() {
+		testTaskPartitioner(2*LocalTaskQueue.MAX_SIZE, PTaskPartitioner.FACTORING);
+	}
+	
+	@Test
+	public void testFactoring2() {
+		testTaskPartitioner(2*LocalTaskQueue.MAX_SIZE, PTaskPartitioner.FACTORING_CMIN);
+	}
+	
+	@Test
+	public void testFactoring3() {
+		testTaskPartitioner(2*LocalTaskQueue.MAX_SIZE, PTaskPartitioner.FACTORING_CMAX);
+	}
+	
+	private void testTaskPartitioner(int numTasks, PTaskPartitioner type) {
+		LocalTaskQueue<Task> queue = new LocalTaskQueue<>();
+		TaskPartitioner partitioner = TaskPartitionerFactory.createTaskPartitioner(
+			type, new IntObject(1), new IntObject(numTasks), new IntObject(1),
+			numTasks, InfrastructureAnalyzer.getLocalParallelism(), "i");
+		//asynchronous task creation
+		CommonThreadPool.get().submit(()->partitioner.createTasks(queue));
+		//consume tasks and check serialization
+		Task t = null;
+		try {
+			while((t = queue.dequeueTask())!=LocalTaskQueue.NO_MORE_TASKS) {
+				Task ts1 = Task.parseCompactString(t.toCompactString());
+				Task ts2 = Task.parseCompactString(t.toCompactString(10));
+				Assert.assertEquals(t.toString(), ts1.toString());
+				Assert.assertEquals(t.toString(), ts2.toString());
+			}
+		} catch (InterruptedException e) {
+			throw new RuntimeException(e);
+		}
+	}
+}
diff --git a/src/test/java/org/apache/sysds/test/functions/builtin/part2/BuiltinRaJoinTest.java b/src/test/java/org/apache/sysds/test/functions/builtin/part2/BuiltinRaJoinTest.java
index 97b820d2b1..ebaf24135c 100644
--- a/src/test/java/org/apache/sysds/test/functions/builtin/part2/BuiltinRaJoinTest.java
+++ b/src/test/java/org/apache/sysds/test/functions/builtin/part2/BuiltinRaJoinTest.java
@@ -26,7 +26,6 @@ import org.apache.sysds.test.TestConfiguration;
 import org.apache.sysds.test.TestUtils;
 import org.junit.Test;
 
-import java.util.Arrays;
 import java.util.HashMap;
 
 public class BuiltinRaJoinTest extends AutomatedTestBase
diff --git a/src/test/java/org/apache/sysds/test/functions/parfor/misc/ParForRecursiveFunctionTest.java b/src/test/java/org/apache/sysds/test/functions/parfor/misc/ParForRecursiveFunctionTest.java
new file mode 100644
index 0000000000..d5dffb8944
--- /dev/null
+++ b/src/test/java/org/apache/sysds/test/functions/parfor/misc/ParForRecursiveFunctionTest.java
@@ -0,0 +1,84 @@
+/*
+ * 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.functions.parfor.misc;
+
+import java.util.HashMap;
+
+import org.junit.Assert;
+import org.junit.Test;
+import org.apache.sysds.common.Types.ExecMode;
+import org.apache.sysds.runtime.matrix.data.MatrixValue.CellIndex;
+import org.apache.sysds.test.AutomatedTestBase;
+import org.apache.sysds.test.TestConfiguration;
+
+public class ParForRecursiveFunctionTest extends AutomatedTestBase
+{
+	private final static String TEST_NAME1 = "parfor_recursive";
+	private final static String TEST_DIR = "functions/parfor/";
+	private final static String TEST_CLASS_DIR = TEST_DIR + ParForRecursiveFunctionTest.class.getSimpleName() + "/";
+	private final static double eps = 1e-10;
+	
+	private final static int rows = 20;
+	private final static int cols = 10;
+	private final static double sparsity = 1.0;
+	
+	@Override
+	public void setUp() {
+		addTestConfiguration(TEST_NAME1, new TestConfiguration(TEST_CLASS_DIR, TEST_NAME1, new String[]{"Rout"}));
+	}
+
+	@Test
+	public void testParForCP() {
+		runParforTest(TEST_NAME1, ExecMode.SINGLE_NODE);
+	}
+	
+	@Test
+	public void testParForHybrid() {
+		runParforTest(TEST_NAME1, ExecMode.HYBRID);
+	}
+	
+	private void runParforTest( String TEST_NAME, ExecMode type )
+	{
+		TestConfiguration config = getTestConfiguration(TEST_NAME);
+		config.addVariable("rows", rows);
+		config.addVariable("cols", cols);
+		loadTestConfiguration(config);
+		ExecMode oldExec = setExecMode(type);
+		
+		try {
+			String HOME = SCRIPT_DIR + TEST_DIR;
+			fullDMLScriptName = HOME + TEST_NAME + ".dml";
+			programArgs = new String[]{"-explain","-stats","-args", input("V"), output("R") };
+			
+			double[][] V = getRandomMatrix(rows, cols, 4, 4, sparsity, 3);
+			writeInputMatrixWithMTD("V", V, true);
+	
+			boolean exceptionExpected = false;
+			runTest(true, exceptionExpected, null, -1);
+			
+			//compare matrices
+			HashMap<CellIndex, Double> dmlfile = readDMLMatrixFromOutputDir("R");
+			Assert.assertEquals(5d*rows*cols, dmlfile.get(new CellIndex(1,1)), eps);
+		}
+		finally {
+			resetExecMode(oldExec);
+		}
+	}
+}
diff --git a/src/test/scripts/functions/parfor/parfor_recursive.dml b/src/test/scripts/functions/parfor/parfor_recursive.dml
new file mode 100644
index 0000000000..42b6c226a2
--- /dev/null
+++ b/src/test/scripts/functions/parfor/parfor_recursive.dml
@@ -0,0 +1,37 @@
+#-------------------------------------------------------------
+#
+# 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.
+#
+#-------------------------------------------------------------
+
+fun = function(Matrix[Double] A)
+  return(Matrix[Double] B)
+{
+    B = matrix(0, nrow(A), ncol(A))
+    parfor(i in 1:nrow(A)) {
+        if( as.scalar(A[1,1]) < 5  )
+           B[i,] = fun(A[i,]+1);
+        else
+           B[i,] = A[i, ]
+    }
+}
+
+A = read($1);
+B = as.matrix(sum(fun(A)));
+write(B, $2);
+