[SYSTEMDS-3894] New out-of-core binary scalar-matrix operations

This patch completes the selected example operations for the new
out-of-core backend and related test.

Author: mboehm7

src/main/java/org/apache/sysds/hops/BinaryOp.java

@@ -854,9 +854,6 @@ else if( (op == OpOp2.CBIND && getDataType().isList())
 			_etype = ExecType.CP;
 		}
 
-		if( _etype == ExecType.OOC ) //TODO
-			setExecType(ExecType.CP);
-		
 		//mark for recompile (forever)
 		setRequiresRecompileIfNecessary();
 

src/main/java/org/apache/sysds/runtime/instructions/OOCInstructionParser.java

@@ -24,6 +24,7 @@
 import org.apache.sysds.common.InstructionType;
 import org.apache.sysds.runtime.DMLRuntimeException;
 import org.apache.sysds.runtime.instructions.ooc.AggregateUnaryOOCInstruction;
+import org.apache.sysds.runtime.instructions.ooc.BinaryOOCInstruction;
 import org.apache.sysds.runtime.instructions.ooc.OOCInstruction;
 import org.apache.sysds.runtime.instructions.ooc.ReblockOOCInstruction;
 
@@ -50,10 +51,9 @@ public static OOCInstruction parseSingleInstruction(InstructionType ooctype, Str
 				return ReblockOOCInstruction.parseInstruction(str);
 			case AggregateUnary:
 				return AggregateUnaryOOCInstruction.parseInstruction(str);
-			
-			// TODO:
 			case Binary:
-
+				return BinaryOOCInstruction.parseInstruction(str);
+			
 			default:
 				throw new DMLRuntimeException("Invalid OOC Instruction Type: " + ooctype);
 		}

src/main/java/org/apache/sysds/runtime/instructions/ooc/BinaryOOCInstruction.java

@@ -0,0 +1,95 @@
+/*
+ * 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.instructions.ooc;
+
+import java.util.concurrent.ExecutorService;
+
+import org.apache.sysds.common.Types.DataType;
+import org.apache.sysds.runtime.DMLRuntimeException;
+import org.apache.sysds.runtime.controlprogram.caching.MatrixObject;
+import org.apache.sysds.runtime.controlprogram.context.ExecutionContext;
+import org.apache.sysds.runtime.controlprogram.parfor.LocalTaskQueue;
+import org.apache.sysds.runtime.instructions.InstructionUtils;
+import org.apache.sysds.runtime.instructions.cp.CPOperand;
+import org.apache.sysds.runtime.instructions.cp.ScalarObject;
+import org.apache.sysds.runtime.instructions.spark.data.IndexedMatrixValue;
+import org.apache.sysds.runtime.matrix.data.MatrixBlock;
+import org.apache.sysds.runtime.matrix.operators.Operator;
+import org.apache.sysds.runtime.matrix.operators.ScalarOperator;
+import org.apache.sysds.runtime.util.CommonThreadPool;
+
+public class BinaryOOCInstruction extends ComputationOOCInstruction {
+	
+	protected BinaryOOCInstruction(OOCType type, Operator bop, 
+			CPOperand in1, CPOperand in2, CPOperand out, String opcode, String istr) {
+		super(type, bop, in1, in2, out, opcode, istr);
+	}
+
+	public static BinaryOOCInstruction parseInstruction(String str) {
+		String[] parts = InstructionUtils.getInstructionPartsWithValueType(str);
+		InstructionUtils.checkNumFields(parts, 3);
+		String opcode = parts[0];
+		CPOperand in1 = new CPOperand(parts[1]);
+		CPOperand in2 = new CPOperand(parts[2]);
+		CPOperand out = new CPOperand(parts[3]);
+		Operator bop = InstructionUtils.parseExtendedBinaryOrBuiltinOperator(opcode, in1, in2);
+		
+		return new BinaryOOCInstruction(
+			OOCType.Binary, bop, in1, in2, out, opcode, str);
+	}
+	
+	@Override
+	public void processInstruction( ExecutionContext ec ) {
+		//TODO support all types, currently only binary matrix-scalar
+		
+		//get operator and scalar
+		CPOperand scalar = ( input1.getDataType() == DataType.MATRIX ) ? input2 : input1;
+		ScalarObject constant = ec.getScalarInput(scalar);
+		ScalarOperator sc_op = ((ScalarOperator)_optr).setConstant(constant.getDoubleValue());
+		
+		//create thread and process binary operation
+		MatrixObject min = ec.getMatrixObject(input1);
+		LocalTaskQueue<IndexedMatrixValue> qIn = min.getStreamHandle();
+		LocalTaskQueue<IndexedMatrixValue> qOut = new LocalTaskQueue<>();
+		ec.getMatrixObject(output).setStreamHandle(qOut);
+		
+		ExecutorService pool = CommonThreadPool.get();
+		try {
+			pool.submit(() -> {
+				IndexedMatrixValue tmp = null;
+				try {
+					while((tmp = qIn.dequeueTask()) != LocalTaskQueue.NO_MORE_TASKS) {
+						IndexedMatrixValue tmpOut = new IndexedMatrixValue();
+						tmpOut.set(tmp.getIndexes(),
+							tmp.getValue().scalarOperations(sc_op, new MatrixBlock()));
+						qOut.enqueueTask(tmpOut);
+					}
+					qOut.closeInput();
+				}
+				catch(Exception ex) {
+					throw new DMLRuntimeException(ex);
+				}
+			});
+		}
+		finally {
+			pool.shutdown();
+		}
+	}
+}

src/test/java/org/apache/sysds/test/functions/ooc/SumScalarMultiplicationTest.java

@@ -23,6 +23,7 @@
 import org.apache.sysds.common.Types;
 import org.apache.sysds.common.Types.FileFormat;
 import org.apache.sysds.common.Types.ValueType;
+import org.apache.sysds.hops.OptimizerUtils;
 import org.apache.sysds.runtime.instructions.Instruction;
 import org.apache.sysds.runtime.io.MatrixWriter;
 import org.apache.sysds.runtime.io.MatrixWriterFactory;
@@ -57,11 +58,26 @@ public void setUp() {
 	 * Test the sum of scalar multiplication, "sum(X*7)", with OOC backend.
 	 */
 	@Test
-	public void testSumScalarMult() {
-
+	public void testSumScalarMultNoRewrite() {
+		testSumScalarMult(false);
+	}
+	
+	/**
+	 * Test the sum of scalar multiplication, "sum(X)*7", with OOC backend.
+	 */
+	@Test
+	public void testSumScalarMultRewrite() {
+		testSumScalarMult(true);
+	}
+	
+	
+	public void testSumScalarMult(boolean rewrite)
+	{
 		Types.ExecMode platformOld = rtplatform;
 		rtplatform = Types.ExecMode.SINGLE_NODE;
-
+		boolean oldRewrite = OptimizerUtils.ALLOW_ALGEBRAIC_SIMPLIFICATION;
+		OptimizerUtils.ALLOW_ALGEBRAIC_SIMPLIFICATION = rewrite;
+		
 		try {
 			getAndLoadTestConfiguration(TEST_NAME);
 			String HOME = SCRIPT_DIR + TEST_DIR;
@@ -92,16 +108,17 @@ public void testSumScalarMult() {
 			String prefix = Instruction.OOC_INST_PREFIX;
 			Assert.assertTrue("OOC wasn't used for RBLK",
 				heavyHittersContainsString(prefix + Opcodes.RBLK));
+			if(!rewrite)
+				Assert.assertTrue("OOC wasn't used for SUM",
+					heavyHittersContainsString(prefix + Opcodes.MULT));
 			Assert.assertTrue("OOC wasn't used for SUM",
 				heavyHittersContainsString(prefix + Opcodes.UAKP));
-			
-//			boolean usedOOCMult = Statistics.getCPHeavyHitterOpCodes().contains(prefix + Opcodes.MULT);
-//			Assert.assertTrue("OOC wasn't used for MULT", usedOOCMult);
 		}
 		catch(Exception ex) {
 			Assert.fail(ex.getMessage());
 		}
 		finally {
+			OptimizerUtils.ALLOW_ALGEBRAIC_SIMPLIFICATION = oldRewrite;
 			resetExecMode(platformOld);
 		}
 	}