use file partition to work with multiple blocks in parts

Author: j143

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

@@ -26,9 +26,7 @@
 
 import org.apache.commons.lang3.StringUtils;
 import org.apache.commons.lang3.tuple.Pair;
-import org.apache.hadoop.fs.Options;
 import org.apache.sysds.api.DMLScript;
-import org.apache.sysds.api.mlcontext.MatrixMetadata;
 import org.apache.sysds.common.Opcodes;
 import org.apache.sysds.common.Types.DataType;
 import org.apache.sysds.common.Types.FileFormat;
@@ -50,11 +48,22 @@
 import org.apache.sysds.runtime.instructions.Instruction;
 import org.apache.sysds.runtime.instructions.InstructionUtils;
 import org.apache.sysds.runtime.instructions.spark.data.IndexedMatrixValue;
-import org.apache.sysds.runtime.io.*;
+import org.apache.sysds.runtime.io.FileFormatProperties;
+import org.apache.sysds.runtime.io.FileFormatPropertiesCSV;
+import org.apache.sysds.runtime.io.FileFormatPropertiesHDF5;
+import org.apache.sysds.runtime.io.FileFormatPropertiesLIBSVM;
+import org.apache.sysds.runtime.io.ListReader;
+import org.apache.sysds.runtime.io.ListWriter;
+import org.apache.sysds.runtime.io.MatrixWriter;
+import org.apache.sysds.runtime.io.MatrixWriterFactory;
+import org.apache.sysds.runtime.io.WriterHDF5;
+import org.apache.sysds.runtime.io.WriterMatrixMarket;
+import org.apache.sysds.runtime.io.WriterTextCSV;
 import org.apache.sysds.runtime.lineage.LineageItem;
 import org.apache.sysds.runtime.lineage.LineageItemUtils;
 import org.apache.sysds.runtime.lineage.LineageTraceable;
 import org.apache.sysds.runtime.matrix.data.MatrixBlock;
+import org.apache.sysds.runtime.matrix.data.MatrixIndexes;
 import org.apache.sysds.runtime.meta.DataCharacteristics;
 import org.apache.sysds.runtime.meta.MatrixCharacteristics;
 import org.apache.sysds.runtime.meta.MetaData;
@@ -1061,31 +1070,25 @@ else if( getInput1().getDataType() == DataType.MATRIX ) {
 			LocalTaskQueue<IndexedMatrixValue> stream = mo.getStreamHandle();
 
 			if (stream != null) {
-				System.out.println("Write OOC instruction: " + getInput1().getName() + "to file name: " + fname);
 
 				try {
 					IndexedMatrixValue tmp = null;
 					MatrixWriter writer = MatrixWriterFactory.createMatrixWriter(fmt);
-					System.out.println("mo details: "+ mo.getDataCharacteristics());
-
 
 					while((tmp = stream.dequeueTask()) != LocalTaskQueue.NO_MORE_TASKS) {
 						MatrixBlock mb = (MatrixBlock)tmp.getValue();
+						MatrixIndexes mi = tmp.getIndexes();
 
+						// Construct a unique filename for each part-file inside the output directory
+						String partFilePath = fname + "/part-" + mi.getRowIndex() + "-" + mi.getColumnIndex();
 
-//						writer.writeMatrixToHDFS(tmp.getValue(), fname, FileFormat.HDF5, mo.getMetaData().getDataCharacteristics());
-						writer.writeMatrixToHDFS(mb, fname, mb.getNumRows(), mb.getNumColumns(), mo.getBlocksize(),mb.getNonZeros());
-//						writer.writeMatrixToHDFS((MatrixBlock)tmp.getValue(), fname, mc.getRows(), mc.getCols(),
-//								mc.getBlocksize(), mc.getNonZeros());
-						System.out.println(tmp);
-
+						writer.writeMatrixToHDFS(mb, partFilePath, mb.getNumRows(), mb.getNumColumns(), (int) mb.getLength() , mb.getNonZeros());
 					}
-//					IOUtilFunctions.closeSilently(writer);
-					HDFSTool.writeMetaDataFile(fname + ".mtd", mo.getValueType(),
+					HDFSTool.writeMetaDataFile(fname + "/.mtd", mo.getValueType(),
 							mo.getMetaData().getDataCharacteristics(), FileFormat.HDF5, _formatProperties);
 				}
 				catch(Exception ex) {
-					throw new DMLRuntimeException(ex);
+					throw new DMLRuntimeException("Failed to write OOC stream to " + fname, ex);
 				}
 			}
 			else {

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

@@ -56,7 +56,7 @@ public void setUp() {
 	}
 
 	/**
-	 * Test the sum of scalar multiplication, "sum(X*7)", with OOC backend.
+	 * Test the unary operation, "ceil(X)", with OOC backend.
 	 */
 	@Test
 	public void testUnary() {
@@ -77,7 +77,7 @@ public void testUnaryOperation(boolean rewrite)
 			programArgs = new String[] {"-explain", "-stats", "-ooc", 
 				"-args", input(INPUT_NAME), output(OUTPUT_NAME)};
 
-			int rows = 1000, cols = 1000;
+			int rows = 5000, cols = 5000;
 			MatrixBlock mb = MatrixBlock.randOperations(rows, cols, 1.0, -1, 1, "uniform", 7);
 			MatrixWriter writer = MatrixWriterFactory.createMatrixWriter(FileFormat.BINARY);
 			writer.writeMatrixToHDFS(mb, input(INPUT_NAME), rows, cols, 1000, rows*cols);
@@ -94,8 +94,8 @@ public void testUnaryOperation(boolean rewrite)
 					Double dmlResult = dmlfile.get(new MatrixValue.CellIndex(i+1 , j+1 )); // Note: MM format is 1-based index
 					double actualValue = (dmlResult == null) ? 0.0 : dmlResult;
 					expected = Math.abs(Math.ceil(mb.get(i, j)));
-					Assert.assertEquals(expected, actualValue, 1e-10);
 					System.out.println("("+i+","+j+"): " + actualValue + "actual: " + expected);
+					Assert.assertEquals(expected, actualValue, 1e-10);
 				}
 			}
 
@@ -104,6 +104,8 @@ public void testUnaryOperation(boolean rewrite)
 				heavyHittersContainsString(prefix + Opcodes.RBLK));
 			Assert.assertTrue("OOC wasn't used for CEIL",
 				heavyHittersContainsString(prefix + Opcodes.CEIL));
+			Assert.assertTrue("Stream Aware WRITE wasn't used",
+					heavyHittersContainsString(String.valueOf(Opcodes.WRITE)));
 		}
 		catch(Exception ex) {
 			Assert.fail(ex.getMessage());

src/test/scripts/functions/ooc/Unary.dml

@@ -21,9 +21,7 @@
 
 # Read input matrix and operator from command line args
 X = read($1);
-#print(toString(X))
 Y = ceil(X);
-#print(toString(Y))
-#res = as.matrix(sum(Y));
+
 # Write the final matrix result
-write(Y, $2);
+write(Y, $2, format="binary");