[SystemDS-3667] Erroneous cleanup of List objects after function return

Currently, after function returns we replace the old variables references
with the returned objects if they share the same variable name. This reference
check logic is not properly done for List objects, leading to incorrect cleanup.

Closes #1985

src/main/java/org/apache/sysds/runtime/instructions/cp/FunctionCallCPInstruction.java

@@ -229,6 +229,7 @@ public void processInstruction(ExecutionContext ec) {
 		// add the updated binding for each return variable to the variables in original symbol table
 		// (with robustness for unbound outputs, i.e., function calls without assignment)
 		int numOutputs = Math.min(_boundOutputNames.size(), fpb.getOutputParams().size());
+		List<Data> toBeCleanedUp = new ArrayList<>();
 		for (int i=0; i< numOutputs; i++) {
 			String boundVarName = _boundOutputNames.get(i);
 			String retVarName = fpb.getOutputParams().get(i).getName();
@@ -237,20 +238,26 @@ public void processInstruction(ExecutionContext ec) {
 				throw new DMLRuntimeException("fcall "+_functionName+": "
 					+boundVarName + " was not assigned a return value");
 
-			//cleanup existing data bound to output variable name
+			// remove existing data bound to output variable name
 			Data exdata = ec.removeVariable(boundVarName);
-			if( exdata != boundValue && !retVars.hasReferences(exdata) )
-				ec.cleanupDataObject(exdata);
+			// save old data for cleanup later
+			if (exdata != boundValue && !retVars.hasReferences(exdata))
+				toBeCleanedUp.add(exdata);
 				//FIXME: interferes with reuse. Removes broadcasts before materialization
 
 			//add/replace data in symbol table
 			ec.setVariable(boundVarName, boundValue);
-			
+
 			//map lineage of function returns back to calling site
 			if( lineage != null ) //unchanged ref
 				ec.getLineage().set(boundVarName, lineage.get(retVarName));
 		}
 
+		// cleanup old data bound to output variable names
+		// needs to be done after return variables are added to ec
+		for (Data dat : toBeCleanedUp)
+			ec.cleanupDataObject(dat);
+
 		//update lineage cache with the functions outputs
 		if ((DMLScript.LINEAGE && LineageCacheConfig.isMultiLevelReuse() && !fpb.isNondeterministic())
 			|| (LineageCacheConfig.isEstimator() && !fpb.isNondeterministic())) {

src/test/java/org/apache/sysds/test/functions/caching/InputVariableOverwriteTest.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.test.functions.caching;
+
+import org.apache.sysds.test.AutomatedTestBase;
+import org.apache.sysds.test.TestConfiguration;
+import org.junit.Test;
+
+public class InputVariableOverwriteTest extends AutomatedTestBase {
+	private final static String TEST_NAME = "InputVariableOverwriteTest";
+	private final static String TEST_DIR = "component/misc/";
+	private final static String TEST_CLASS_DIR = TEST_DIR + InputVariableOverwriteTest.class.getSimpleName() + "/";
+
+	@Override
+	public void setUp() {
+		addTestConfiguration(TEST_NAME, new TestConfiguration(TEST_CLASS_DIR, TEST_NAME));
+	}
+
+	@Test
+	public void testInputVariableOverwrite() {
+		// running the DML script should not raise error
+		runInputVariableOverwriteTest();
+	}
+
+	private void runInputVariableOverwriteTest() {
+		loadTestConfiguration(getTestConfiguration(TEST_NAME));
+
+		String HOME = SCRIPT_DIR + TEST_DIR;
+		fullDMLScriptName = HOME + TEST_NAME + ".dml";
+		programArgs = new String[]{};
+		runTest(true, false, null, -1);
+	}
+}

src/test/scripts/component/misc/InputVariableOverwriteTest.dml

@@ -0,0 +1,51 @@
+#-------------------------------------------------------------
+#
+# 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.
+#
+#-------------------------------------------------------------
+
+/*
+ * This script tests the correct memory management with function
+ * calls that have lists as input and output where the output 
+ * list object overwrites the input list object since they are
+ * bound to the same variable name and contain mutual 
+ * elements.
+ */
+
+# initialize some variables
+some_matrix = matrix(112, rows=2, cols=2)
+some_list = list(some_matrix)
+
+# update the list
+some_list = some_function(some_list, 3)
+
+# try to access matrix of list
+print(toString(as.matrix(some_list[1])))
+
+
+some_function = function(list[unknown] just_a_list, int l)
+  return (list[unknown] also_a_list) {
+  # add an element to list
+  count = 0
+  for (i in 1:l) {
+      count += 1
+  }
+  other_matrix = matrix(count, rows=1, cols=3)
+
+  also_a_list = append(just_a_list, other_matrix)
+}