Add OOC reshape

src/main/java/org/apache/sysds/runtime/data/DenseBlockFP64.java

@@ -181,6 +181,26 @@ public DenseBlock set(int rl, int ru, int ol, int ou, DenseBlock db) {
 		return this;
 	}
 
+	public DenseBlock setPartialRow(DenseBlock row, int rIdx, int srcOffset, int destOffset, int length) {
+		if(destOffset + length > _odims[0])
+			throw new RuntimeException(
+				"Partial row assignment exceeds row length: " + (destOffset + length) + " > " + _odims[0]);
+		System.arraycopy(row.valuesAt(0), srcOffset, _data, this.pos(rIdx, destOffset), length);
+		return this;
+	}
+
+	public DenseBlock setPartialCol(DenseBlock col, int cIdx, int srcOffset, int destOffset, int length) {
+		if(destOffset + length > _rlen)
+			throw new RuntimeException(
+				"Partial column assignment exceeds column length: " + (destOffset + length) + " > " + _rlen);
+		int destPos = this.pos(destOffset, cIdx);
+		double[] src = col.valuesAt(0);
+		for(int i = 0; i < length; i++) {
+			_data[destPos + i * _odims[0]] = src[srcOffset + i];
+		}
+		return this;
+	}
+
 	@Override
 	public DenseBlock set(int r, double[] v) {
 		System.arraycopy(v, 0, _data, pos(r), _odims[0]);

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

@@ -43,6 +43,7 @@
 import org.apache.sysds.runtime.instructions.ooc.ReorgOOCInstruction;
 import org.apache.sysds.runtime.instructions.ooc.TeeOOCInstruction;
 import org.apache.sysds.runtime.instructions.ooc.AppendOOCInstruction;
+import org.apache.sysds.runtime.instructions.ooc.ReshapeOOCInstruction;
 import org.apache.sysds.runtime.instructions.ooc.QuaternaryOOCInstruction;
 
 public class OOCInstructionParser extends InstructionParser {
@@ -97,7 +98,7 @@ else if(parts.length == 4)
 			case Reorg:
 				return ReorgOOCInstruction.parseInstruction(str);
 			case Reshape:
-				return ReorgOOCInstruction.parseInstruction(str);
+				return ReshapeOOCInstruction.parseInstruction(str);
 			case Tee:
 				return TeeOOCInstruction.parseInstruction(str);
 			case CentralMoment:

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

@@ -99,6 +99,7 @@ public CachingStream(OOCStream<IndexedMatrixValue> source, long streamId) {
 			// Capture a short context to help identify origin
 			OOCWatchdog.registerOpen(_watchdogId, toString(), getCtxMsg(), this);
 		}
+		activateIndexing();
 		_downstreamRelays = null;
 		source.setSubscriber(tmp -> {
 			try(tmp) {

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

@@ -80,7 +80,7 @@ public abstract class OOCInstruction extends Instruction {
 
 	public enum OOCType {
 		Reblock, Tee, Binary, Ternary, Unary, AggregateUnary, AggregateBinary, AggregateTernary, MAPMM, MMTSJ,
-		MAPMMCHAIN, Reorg, CM, Ctable, MatrixIndexing, ParameterizedBuiltin, Rand, Append, Quaternary
+		MAPMMCHAIN, Reorg, CM, Ctable, MatrixIndexing, ParameterizedBuiltin, Rand, Append, Quaternary, Reshape
 	}
 
 	protected final OOCInstruction.OOCType _ooctype;

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

@@ -41,31 +41,17 @@ public class ReorgOOCInstruction extends ComputationOOCInstruction {
 	private final CPOperand _col;
 	private final CPOperand _desc;
 	private final CPOperand _ixret;
-	// reshape-specific attributes
-	private final CPOperand _opRows;
-	private final CPOperand _opCols;
-	//private final CPOperand _opDims;
-	private final CPOperand _opByRow;
 
 	protected ReorgOOCInstruction(ReorgOperator op, CPOperand in1, CPOperand out, String opcode, String istr) {
-		this(op, in1, out, null, null, null, null, null, null, null, opcode, istr);
-	}
-
-	private ReorgOOCInstruction(Operator op, CPOperand in, CPOperand out, CPOperand opRows, CPOperand opCols,
-		CPOperand opDims, CPOperand opByRow, String opcode, String istr) {
-		this(op, in, out, null, null, null, opRows, opCols, opDims, opByRow, opcode, istr);
+		this(op, in1, out, null, null, null, opcode, istr);
 	}
 
 	private ReorgOOCInstruction(Operator op, CPOperand in, CPOperand out, CPOperand col, CPOperand desc, CPOperand ixret,
-		CPOperand opRows, CPOperand opCols, CPOperand opDims, CPOperand opByRow, String opcode, String istr) {
+		String opcode, String istr) {
 		super(OOCType.Reorg, op, in, out, opcode, istr);
 		_col = col;
 		_desc = desc;
 		_ixret = ixret;
-		_opRows = opRows;
-		_opCols = opCols;
-		//_opDims = opDims;
-		_opByRow = opByRow;
 	}
 
 	public static ReorgOOCInstruction parseInstruction(String str) {
@@ -92,35 +78,13 @@ else if(opcode.equalsIgnoreCase(Opcodes.SORT.toString())) {
 			CPOperand ixret = new CPOperand(parts[4]);
 			int k = Integer.parseInt(parts[6]);
 			return new ReorgOOCInstruction(new ReorgOperator(new SortIndex(1, false, false), k),
-				in, out, col, desc, ixret, null, null, null, null, opcode, str);
-		}
-		else if(opcode.equalsIgnoreCase(Opcodes.RESHAPE.toString())) {
-			InstructionUtils.checkNumFields(parts, 6);
-			in.split(parts[1]);
-			CPOperand rows = new CPOperand(parts[2]);
-			CPOperand cols = new CPOperand(parts[3]);
-			CPOperand dims = new CPOperand(parts[4]);
-			CPOperand byRow = new CPOperand(parts[5]);
-			out.split(parts[6]);
-			return new ReorgOOCInstruction(new Operator(true), in, out, rows, cols, dims, byRow, opcode, str);
+				in, out, col, desc, ixret, opcode, str);
 		}
 		else
 			throw new NotImplementedException();
 	}
 
 	public void processInstruction( ExecutionContext ec ) {
-		if(getOpcode().equalsIgnoreCase(Opcodes.RESHAPE.toString())) {
-			// TODO Make reshape truly out-of-core
-			int rows = (int) ec.getScalarInput(_opRows).getLongValue();
-			int cols = (int) ec.getScalarInput(_opCols).getLongValue();
-			boolean byRow = ec.getScalarInput(_opByRow).getBooleanValue();
-			MatrixBlock in = ec.getMatrixInput(input1.getName());
-			MatrixBlock out = in.reshape(rows, cols, byRow);
-			ec.releaseMatrixInput(input1.getName());
-			ec.setMatrixOutput(output.getName(), out);
-			return;
-		}
-
 		// Create thread and process the transpose/sort operation
 		MatrixObject min = ec.getMatrixObject(input1);
 		ReorgOperator r_op = (ReorgOperator) _optr;