[SYSTEMDS-2575] Fix eval function calls (incorrect pinning of inputs)

This patch fixes an issue of indirect eval function calls where wrong
input variable names led to missing pinning of inputs and thus too eager
cleanup of these variables (which causes crashes if the inputs are used
in other operations of the eval call).

The fix is simple. We avoid such inconsistent construction and
invocation of fcall instructions by using a narrower interface and
constructing the materialized names internally in the fcall.

Author: mboehm7

src/main/java/org/apache/sysds/runtime/controlprogram/paramserv/PSWorker.java

@@ -77,12 +77,10 @@ protected void setupUpdateFunction(String updFunc, ExecutionContext ec) {
 		CPOperand[] boundInputs = inputs.stream()
 			.map(input -> new CPOperand(input.getName(), input.getValueType(), input.getDataType()))
 			.toArray(CPOperand[]::new);
-		ArrayList<String> inputNames = inputs.stream().map(DataIdentifier::getName)
-			.collect(Collectors.toCollection(ArrayList::new));
 		ArrayList<String> outputNames = outputs.stream().map(DataIdentifier::getName)
 			.collect(Collectors.toCollection(ArrayList::new));
 		_inst = new FunctionCallCPInstruction(ns, fname, boundInputs,
-			inputNames, func.getInputParamNames(), outputNames, "update function");
+			func.getInputParamNames(), outputNames, "update function");
 
 		// Check the inputs of the update function
 		checkInput(false, inputs, DataType.MATRIX, Statement.PS_FEATURES);

src/main/java/org/apache/sysds/runtime/controlprogram/paramserv/ParamServer.java

@@ -104,12 +104,10 @@ protected void setupAggFunc(ExecutionContext ec, String aggFunc) {
 		CPOperand[] boundInputs = inputs.stream()
 			.map(input -> new CPOperand(input.getName(), input.getValueType(), input.getDataType()))
 			.toArray(CPOperand[]::new);
-		ArrayList<String> inputNames = inputs.stream().map(DataIdentifier::getName)
-			.collect(Collectors.toCollection(ArrayList::new));
 		ArrayList<String> outputNames = outputs.stream().map(DataIdentifier::getName)
 			.collect(Collectors.toCollection(ArrayList::new));
 		_inst = new FunctionCallCPInstruction(ns, fname, boundInputs,
-			inputNames, func.getInputParamNames(), outputNames, "aggregate function");
+			func.getInputParamNames(), outputNames, "aggregate function");
 	}
 
 	public abstract void push(int workerID, ListObject value);

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

@@ -67,10 +67,6 @@ public void processInstruction(ExecutionContext ec) {
 		CPOperand[] boundInputs = Arrays.copyOfRange(inputs, 1, inputs.length);
 		List<String> boundOutputNames = new ArrayList<>();
 		boundOutputNames.add(output.getName());
-		List<String> boundInputNames = new ArrayList<>();
-		for (CPOperand input : boundInputs) {
-			boundInputNames.add(input.getName());
-		}
 
 		//2. copy the created output matrix
 		MatrixObject outputMO = new MatrixObject(ec.getMatrixObject(output.getName()));
@@ -103,32 +99,30 @@ public void processInstruction(ExecutionContext ec) {
 				ec.getVariables().put(varName, in);
 				boundInputs2[i] = new CPOperand(varName, in);
 			}
-			boundInputNames = lo.isNamedList() ? lo.getNames() : fpb.getInputParamNames();
 			boundInputs = boundInputs2;
 		}
 
 		//5. call the function
 		FunctionCallCPInstruction fcpi = new FunctionCallCPInstruction(null, funcName,
-			boundInputs, boundInputNames, fpb.getInputParamNames(), boundOutputNames, "eval func");
+			boundInputs, fpb.getInputParamNames(), boundOutputNames, "eval func");
 		fcpi.processInstruction(ec);
 
 		//6. convert the result to matrix
 		Data newOutput = ec.getVariable(output);
-		if (newOutput instanceof MatrixObject) {
-			return;
-		}
-		MatrixBlock mb = null;
-		if (newOutput instanceof ScalarObject) {
-			//convert scalar to matrix
-			mb = new MatrixBlock(((ScalarObject) newOutput).getDoubleValue());
-		} else if (newOutput instanceof FrameObject) {
-			//convert frame to matrix
-			mb = DataConverter.convertToMatrixBlock(((FrameObject) newOutput).acquireRead());
-			ec.cleanupCacheableData((FrameObject) newOutput);
+		if (!(newOutput instanceof MatrixObject)) {
+			MatrixBlock mb = null;
+			if (newOutput instanceof ScalarObject) {
+				//convert scalar to matrix
+				mb = new MatrixBlock(((ScalarObject) newOutput).getDoubleValue());
+			} else if (newOutput instanceof FrameObject) {
+				//convert frame to matrix
+				mb = DataConverter.convertToMatrixBlock(((FrameObject) newOutput).acquireRead());
+				ec.cleanupCacheableData((FrameObject) newOutput);
+			}
+			outputMO.acquireModify(mb);
+			outputMO.release();
+			ec.setVariable(output.getName(), outputMO);
 		}
-		outputMO.acquireModify(mb);
-		outputMO.release();
-		ec.setVariable(output.getName(), outputMO);
 
 		//7. cleanup of variable expanded from list
 		if( boundInputs2 != null ) {

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

@@ -20,8 +20,10 @@
 package org.apache.sysds.runtime.instructions.cp;
 
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.HashSet;
 import java.util.List;
+import java.util.stream.Collectors;
 
 import org.apache.sysds.api.DMLScript;
 import org.apache.sysds.lops.Lop;
@@ -55,12 +57,13 @@ public class FunctionCallCPInstruction extends CPInstruction {
 	private final List<String> _boundOutputNames;
 
 	public FunctionCallCPInstruction(String namespace, String functName, CPOperand[] boundInputs,
-			List<String> boundInputNames, List<String> funArgNames, List<String> boundOutputNames, String istr) {
+			List<String> funArgNames, List<String> boundOutputNames, String istr) {
 		super(CPType.External, null, functName, istr);
 		_functionName = functName;
 		_namespace = namespace;
 		_boundInputs = boundInputs;
-		_boundInputNames = boundInputNames;
+		_boundInputNames = Arrays.stream(boundInputs).map(i -> i.getName())
+			.collect(Collectors.toCollection(ArrayList::new));
 		_funArgNames = funArgNames;
 		_boundOutputNames = boundOutputNames;
 	}
@@ -81,19 +84,17 @@ public static FunctionCallCPInstruction parseInstruction(String str) {
 		int numInputs = Integer.valueOf(parts[3]);
 		int numOutputs = Integer.valueOf(parts[4]);
 		CPOperand[] boundInputs = new CPOperand[numInputs];
-		List<String> boundInputNames = new ArrayList<>();
 		List<String> funArgNames = new ArrayList<>();
 		List<String> boundOutputNames = new ArrayList<>();
 		for (int i = 0; i < numInputs; i++) {
 			String[] nameValue = IOUtilFunctions.splitByFirst(parts[5 + i], "=");
 			boundInputs[i] = new CPOperand(nameValue[1]);
 			funArgNames.add(nameValue[0]);
-			boundInputNames.add(boundInputs[i].getName());
 		}
 		for (int i = 0; i < numOutputs; i++)
 			boundOutputNames.add(parts[5 + numInputs + i]);
 		return new FunctionCallCPInstruction ( namespace, functionName,
-			boundInputs, boundInputNames, funArgNames, boundOutputNames, str );
+			boundInputs, funArgNames, boundOutputNames, str );
 	}
 
 	@Override