[SYSTEMDS-3931] Out-of-core right indexing operations

Closes #2351.

Author: janniklinde

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

@@ -28,6 +28,7 @@
 import org.apache.sysds.runtime.instructions.ooc.CSVReblockOOCInstruction;
 import org.apache.sysds.runtime.instructions.ooc.CentralMomentOOCInstruction;
 import org.apache.sysds.runtime.instructions.ooc.CtableOOCInstruction;
+import org.apache.sysds.runtime.instructions.ooc.IndexingOOCInstruction;
 import org.apache.sysds.runtime.instructions.ooc.OOCInstruction;
 import org.apache.sysds.runtime.instructions.ooc.ParameterizedBuiltinOOCInstruction;
 import org.apache.sysds.runtime.instructions.ooc.ReblockOOCInstruction;
@@ -75,12 +76,14 @@ public static OOCInstruction parseSingleInstruction(InstructionType ooctype, Str
 				return TransposeOOCInstruction.parseInstruction(str);
 			case Tee:
 				return TeeOOCInstruction.parseInstruction(str);
-            case CentralMoment:
-                return  CentralMomentOOCInstruction.parseInstruction(str);
+			case CentralMoment:
+				return  CentralMomentOOCInstruction.parseInstruction(str);
 			case Ctable:
 				return CtableOOCInstruction.parseInstruction(str);
 			case ParameterizedBuiltin:
 				return ParameterizedBuiltinOOCInstruction.parseInstruction(str);
+			case MatrixIndexing:
+				return IndexingOOCInstruction.parseInstruction(str);
 
 			default:
 				throw new DMLRuntimeException("Invalid OOC Instruction Type: " + ooctype);

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

@@ -80,36 +80,37 @@ public CachingStream(OOCStream<IndexedMatrixValue> source, long streamId) {
 		});
 	}
 
-	private boolean fetchFromStream() throws InterruptedException {
-		synchronized (this) {
-			if(!_cacheInProgress)
-				throw new DMLRuntimeException("Stream is closed");
-		}
+	private synchronized boolean fetchFromStream() throws InterruptedException {
+		if(!_cacheInProgress)
+			throw new DMLRuntimeException("Stream is closed");
 
 		IndexedMatrixValue task = _source.dequeue();
 
-		synchronized (this) {
-			if(task != LocalTaskQueue.NO_MORE_TASKS) {
-				OOCEvictionManager.put(_streamId, _numBlocks, task);
-				if (_index != null)
-					_index.put(task.getIndexes(), _numBlocks);
-				_numBlocks++;
-				notifyAll();
-				return false;
-			}
-			else {
-				_cacheInProgress = false; // caching is complete
-				notifyAll();
-				return true;
-			}
+		if(task != LocalTaskQueue.NO_MORE_TASKS) {
+			OOCEvictionManager.put(_streamId, _numBlocks, task);
+			if (_index != null)
+				_index.put(task.getIndexes(), _numBlocks);
+			_numBlocks++;
+			notifyAll();
+			return false;
+		}
+		else {
+			_cacheInProgress = false; // caching is complete
+			notifyAll();
+			return true;
 		}
 	}
 
 	public synchronized IndexedMatrixValue get(int idx) throws InterruptedException {
 		while (true) {
-			if (idx < _numBlocks)
-				return OOCEvictionManager.get(_streamId, idx);
-			else if (!_cacheInProgress)
+			if (idx < _numBlocks) {
+				IndexedMatrixValue out = OOCEvictionManager.get(_streamId, idx);
+
+				if (_index != null) // Ensure index is up to date
+					_index.putIfAbsent(out.getIndexes(), idx);
+
+				return out;
+			} else if (!_cacheInProgress)
 				return (IndexedMatrixValue)LocalTaskQueue.NO_MORE_TASKS;
 
 			wait();