[MINOR] Updated reader/write tests to force parallel writers

Author: mboehm7

src/main/java/org/apache/sysds/runtime/io/FrameWriter.java

@@ -35,7 +35,13 @@
  */
 public abstract class FrameWriter {
 	protected static final Log LOG = LogFactory.getLog(FrameWriter.class.getName());
+	
+	protected boolean _forcedParallel = false;
+	
 	public abstract void writeFrameToHDFS( FrameBlock src, String fname, long rlen, long clen )
 		throws IOException, DMLRuntimeException;
 
+	public void setForcedParallel(boolean par) {
+		_forcedParallel = par;
+	}
 }

src/main/java/org/apache/sysds/runtime/io/FrameWriterBinaryBlockParallel.java

@@ -44,7 +44,7 @@
  * 
  */
 public class FrameWriterBinaryBlockParallel extends FrameWriterBinaryBlock
-{	
+{
 	@Override
 	protected void writeBinaryBlockFrameToHDFS( Path path, JobConf job, FrameBlock src, long rlen, long clen )
 		throws IOException, DMLRuntimeException
@@ -59,11 +59,11 @@ protected void writeBinaryBlockFrameToHDFS( Path path, JobConf job, FrameBlock s
 		numThreads = Math.min(numThreads, numPartFiles);
 
 		//fall back to sequential write if dop is 1 (e.g., <128MB) in order to create single file
-		if( numThreads <= 1 ) {
+		if( !_forcedParallel && numThreads <= 1 ) {
 			super.writeBinaryBlockFrameToHDFS(path, job, src, rlen, clen);
 			return;
 		}
-		
+	
 		//create directory for concurrent tasks
 		HDFSTool.createDirIfNotExistOnHDFS(path, DMLConfig.DEFAULT_SHARED_DIR_PERMISSION);
 		FileSystem fs = IOUtilFunctions.getFileSystem(path);

src/main/java/org/apache/sysds/runtime/io/FrameWriterTextCSVParallel.java

@@ -60,7 +60,7 @@ protected void writeCSVFrameToHDFS( Path path, JobConf job, FrameBlock src, long
 		numThreads = Math.min(numThreads, numPartFiles);
 
 		//fall back to sequential write if dop is 1 (e.g., <128MB) in order to create single file
-		if( numThreads <= 1 ) {
+		if( !_forcedParallel && numThreads <= 1 ) {
 			super.writeCSVFrameToHDFS(path, job, src, rlen, clen, csvprops);
 			return;
 		}

src/main/java/org/apache/sysds/runtime/io/FrameWriterTextCellParallel.java

@@ -56,7 +56,7 @@ protected void writeTextCellFrameToHDFS( Path path, JobConf job, FrameBlock src,
 		numThreads = Math.min(numThreads, numPartFiles);
 
 		//fall back to sequential write if dop is 1 (e.g., <128MB) in order to create single file
-		if( numThreads <= 1 ) {
+		if( !_forcedParallel && numThreads <= 1 ) {
 			super.writeTextCellFrameToHDFS(path, job, src, rlen, clen);
 			return;
 		}

src/main/java/org/apache/sysds/runtime/io/MatrixWriter.java

@@ -34,13 +34,19 @@
 public abstract class MatrixWriter {
 	protected static final Log LOG = LogFactory.getLog(MatrixWriter.class.getName());
 
+	protected boolean _forcedParallel = false;
+	
 	public void writeMatrixToHDFS( MatrixBlock src, String fname, long rlen, long clen, int blen, long nnz ) throws IOException {
 		writeMatrixToHDFS(src, fname, rlen, clen, blen, nnz, false);
 	}
 
 	public abstract void writeMatrixToHDFS( MatrixBlock src, String fname, long rlen, long clen, int blen, long nnz, boolean diag )
 		throws IOException;
 
+	public void setForcedParallel(boolean par) {
+		_forcedParallel = par;
+	}
+	
 
 	/**
 	 * Writes a minimal entry to represent an empty matrix on hdfs.

src/main/java/org/apache/sysds/runtime/io/WriterHDF5Parallel.java

@@ -59,7 +59,7 @@ public void writeHDF5MatrixToHDFS(Path path, JobConf job, FileSystem fs, MatrixB
 		numThreads = Math.min(numThreads, numPartFiles);
 
 		//fall back to sequential write if dop is 1 (e.g., <128MB) in order to create single file
-		if(numThreads <= 1) {
+		if( !_forcedParallel && numThreads <= 1 ) {
 			super.writeHDF5MatrixToHDFS(path, job, fs, src);
 			return;
 		}

src/main/java/org/apache/sysds/runtime/io/WriterMatrixMarketParallel.java

@@ -55,7 +55,7 @@ protected void writeMatrixMarketMatrixToHDFS( Path path, JobConf job, FileSystem
 		numThreads = Math.min(numThreads, numPartFiles);
 
 		//fall back to sequential write if dop is 1 (e.g., <128MB) in order to create single file
-		if( numThreads <= 1 ) {
+		if( !_forcedParallel && numThreads <= 1 ) {
 			super.writeMatrixMarketMatrixToHDFS(path, job, fs, src);
 			return;
 		}

src/main/java/org/apache/sysds/runtime/io/WriterTextCSVParallel.java

@@ -57,7 +57,7 @@ protected void writeCSVMatrixToHDFS(Path path, JobConf job, FileSystem fs, Matri
 		numThreads = Math.min(numThreads, numPartFiles);
 
 		//fall back to sequential write if dop is 1 (e.g., <128MB) in order to create single file
-		if( numThreads <= 1 ) {
+		if( !_forcedParallel && numThreads <= 1 ) {
 			super.writeCSVMatrixToHDFS(path, job, fs, src, csvprops);
 			return;
 		}

src/main/java/org/apache/sysds/runtime/io/WriterTextCellParallel.java

@@ -54,7 +54,7 @@ protected void writeTextCellMatrixToHDFS( Path path, JobConf job, FileSystem fs,
 		numThreads = Math.min(numThreads, numPartFiles);
 
 		//fall back to sequential write if dop is 1 (e.g., <128MB) in order to create single file
-		if( numThreads <= 1 || src.getNonZeros()==0 ) {
+		if( !_forcedParallel && (numThreads <= 1 || src.getNonZeros()==0) ) {
 			super.writeTextCellMatrixToHDFS(path, job, fs, src, rlen, clen);
 			return;
 		}

src/main/java/org/apache/sysds/runtime/io/WriterTextLIBSVMParallel.java

@@ -57,7 +57,7 @@ protected void writeLIBSVMMatrixToHDFS(Path path, JobConf job, FileSystem fs, Ma
 		numThreads = Math.min(numThreads, numPartFiles);
 
 		//fall back to sequential write if dop is 1 (e.g., <128MB) in order to create single file
-		if( numThreads <= 1 ) {
+		if( !_forcedParallel && numThreads <= 1 ) {
 			super.writeLIBSVMMatrixToHDFS(path, job, fs, src);
 			return;
 		}