<commit_before>/**
 * IBM Confidential
 * OCO Source Materials
 * (C) Copyright IBM Corp. 2010, 2013
 * The source code for this program is not published or otherwise divested of its trade secrets, irrespective of what has been deposited with the U.S. Copyright Office.
 */

package com.ibm.bi.dml.runtime.util;

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.EOFException;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.io.OutputStream;
import java.io.InputStream;
import java.util.Arrays;
import java.util.LinkedList;
import java.util.StringTokenizer;
import java.util.regex.Pattern;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.InputFormat;
import org.apache.hadoop.mapred.InputSplit;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.RecordReader;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.TextInputFormat;

import com.ibm.bi.dml.runtime.DMLRuntimeException;
import com.ibm.bi.dml.runtime.DMLUnsupportedOperationException;
import com.ibm.bi.dml.runtime.matrix.MatrixCharacteristics;
import com.ibm.bi.dml.runtime.matrix.io.BinaryBlockToBinaryCellConverter;
import com.ibm.bi.dml.runtime.matrix.io.BinaryBlockToTextCellConverter;
import com.ibm.bi.dml.runtime.matrix.io.CSVFileFormatProperties;
import com.ibm.bi.dml.runtime.matrix.io.Converter;
import com.ibm.bi.dml.runtime.matrix.io.FileFormatProperties;
import com.ibm.bi.dml.runtime.matrix.io.InputInfo;
import com.ibm.bi.dml.runtime.matrix.io.MatrixBlock;
import com.ibm.bi.dml.runtime.matrix.io.MatrixBlockDSM;
import com.ibm.bi.dml.runtime.matrix.io.MatrixCell;
import com.ibm.bi.dml.runtime.matrix.io.MatrixIndexes;
import com.ibm.bi.dml.runtime.matrix.io.OutputInfo;
import com.ibm.bi.dml.runtime.matrix.io.Pair;
import com.ibm.bi.dml.runtime.matrix.io.MatrixBlockDSM.IJV;
import com.ibm.bi.dml.runtime.matrix.io.MatrixBlockDSM.SparseCellIterator;
import com.ibm.bi.dml.runtime.matrix.io.SparseRow;
import com.ibm.bi.dml.runtime.matrix.mapred.MRJobConfiguration;
import com.ibm.bi.dml.runtime.matrix.mapred.MRJobConfiguration.ConvertTarget;


/**
 * This class provides methods to read and write matrix blocks from to HDFS using different data formats.
 * Those functionalities are used especially for CP read/write and exporting in-memory matrices to HDFS
 * (before executing MR jobs).
 * 
 */
public class DataConverter 
{
	@SuppressWarnings("unused")
	private static final String _COPYRIGHT = "Licensed Materials - Property of IBM\n(C) Copyright IBM Corp. 2010, 2013\n" +
	                                         "US Government Users Restricted Rights - Use, duplication  disclosure restricted by GSA ADP Schedule Contract with IBM Corp.";
		
	//////////////
	// READING and WRITING of matrix blocks to/from HDFS
	// (textcell, binarycell, binaryblock)
	///////
	
	private class ReadProperties {
		/* Properties common to all file formats */
		String path;
		long rlen, clen;
		int brlen, bclen;
		double expectedSparsity;
		InputInfo inputInfo;
		boolean localFS;
		
		/* Flag to indicate the input file is in MatrixMarket format */
		//boolean matrixMarket;
		
		/* Properties specific to CSV files */
		FileFormatProperties formatProperties;
		
		public ReadProperties() {
			rlen = clen = -1;
			brlen = bclen = -1;
			expectedSparsity = 0.1d;
			inputInfo = null;
			localFS = false;
			
			//matrixMarket = false;
		}

	}
	
	/**
	 * 
	 * @param mat
	 * @param dir
	 * @param outputinfo
	 * @param rlen
	 * @param clen
	 * @param brlen
	 * @param bclen
	 * @throws IOException
	 */
	public static void writeMatrixToHDFS(MatrixBlock mat, String dir, OutputInfo outputinfo, 
			                             MatrixCharacteristics mc )
		throws IOException
	{
		writeMatrixToHDFS(mat, dir, outputinfo, mc, -1, null);
	}
	
	public static void writeMatrixToHDFS(MatrixBlock mat, String dir, OutputInfo outputinfo, MatrixCharacteristics mc, int replication, FileFormatProperties formatProperties)
	throws IOException
	{
		JobConf job = new JobConf();
		Path path = new Path(dir);

		//System.out.println("write matrix (sparse="+mat.isInSparseFormat()+") to HDFS: "+dir);
		
		//set the replication factor for the output
		if( replication > 0 ) //otherwise: default 3
			job.setInt("dfs.replication", replication);
		
		try
		{
			// If the file already exists on HDFS, remove it.
			MapReduceTool.deleteFileIfExistOnHDFS(dir);

			// core matrix writing
			if ( outputinfo == OutputInfo.TextCellOutputInfo ) 
			{	
				writeTextCellMatrixToHDFS(path, job, mat, mc.get_rows(), mc.get_cols(), mc.get_rows_per_block(), mc.get_cols_per_block());
			}
			else if ( outputinfo == OutputInfo.BinaryCellOutputInfo ) 
			{
				writeBinaryCellMatrixToHDFS(path, job, mat, mc.get_rows(), mc.get_cols(), mc.get_rows_per_block(), mc.get_cols_per_block());
			}
			else if( outputinfo == OutputInfo.BinaryBlockOutputInfo )
			{
				writeBinaryBlockMatrixToHDFS(path, job, mat, mc.get_rows(), mc.get_cols(), mc.get_rows_per_block(), mc.get_cols_per_block());
			}
			else if ( outputinfo == OutputInfo.MatrixMarketOutputInfo ) 
			{
				writeMatrixMarketToHDFS( path, job, mat, mc.get_rows(), mc.get_cols(), mc.getNonZeros());
			}
			else if ( outputinfo == OutputInfo.CSVOutputInfo ) {
				writeCSVToHDFS(path, job, mat, mc.get_rows(), mc.get_cols(), mc.getNonZeros(), formatProperties);
			}
			else {
				throw new RuntimeException("Unknown format (" + OutputInfo.outputInfoToString(outputinfo));
			}
		}
		catch(Exception e)
		{
			throw new IOException(e);
		}
	}
	
	/**
	 * 
	 * @param dir
	 * @param inputinfo
	 * @param rlen
	 * @param clen
	 * @param brlen
	 * @param bclen
	 * @return
	 * @throws IOException
	 */
	public static MatrixBlock readMatrixFromHDFS(String dir, InputInfo inputinfo, long rlen, long clen, int brlen, int bclen, boolean localFS) 
		throws IOException
	{	
		DataConverter dc = new DataConverter();
		ReadProperties prop = dc.new ReadProperties();
		
		prop.path = dir;
		prop.inputInfo = inputinfo;
		prop.rlen = rlen;
		prop.clen = clen;
		prop.brlen = brlen;
		prop.bclen = bclen;
		prop.localFS = localFS;
		
		//expected matrix is sparse (default SystemML usecase)
		return readMatrixFromHDFS(prop);
	}
	
	/**
	 * 
	 * @param dir
	 * @param inputinfo
	 * @param rlen
	 * @param clen
	 * @param brlen
	 * @param bclen
	 * @return
	 * @throws IOException
	 */
	public static MatrixBlock readMatrixFromHDFS(String dir, InputInfo inputinfo, long rlen, long clen, int brlen, int bclen) 
		throws IOException
	{	
		DataConverter dc = new DataConverter();
		ReadProperties prop = dc.new ReadProperties();
		
		prop.path = dir;
		prop.inputInfo = inputinfo;
		prop.rlen = rlen;
		prop.clen = clen;
		prop.brlen = brlen;
		prop.bclen = bclen;
		
		//expected matrix is sparse (default SystemML usecase)
		return readMatrixFromHDFS(prop);
	}

	/**
	 * 
	 * @param dir
	 * @param inputinfo
	 * @param rlen
	 * @param clen
	 * @param brlen
	 * @param bclen
	 * @param expectedSparsity
	 * @return
	 * @throws IOException
	 */
	public static MatrixBlock readMatrixFromHDFS(String dir, InputInfo inputinfo, long rlen, long clen, int brlen, int bclen, double expectedSparsity) 
		throws IOException
	{	
		DataConverter dc = new DataConverter();
		ReadProperties prop = dc.new ReadProperties();
		
		prop.path = dir;
		prop.inputInfo = inputinfo;
		prop.rlen = rlen;
		prop.clen = clen;
		prop.brlen = brlen;
		prop.bclen = bclen;
		prop.expectedSparsity = expectedSparsity;
		
		return readMatrixFromHDFS(prop);
	}

	/**
	 * 
	 * @param dir
	 * @param inputinfo
	 * @param rlen
	 * @param clen
	 * @param brlen
	 * @param bclen
	 * @param expectedSparsity
	 * @param localFS
	 * @return
	 * @throws IOException
	 */
	public static MatrixBlock readMatrixFromHDFS(String dir, InputInfo inputinfo, long rlen, long clen, 
			int brlen, int bclen, double expectedSparsity, boolean localFS) 
	throws IOException
	{
		DataConverter dc = new DataConverter();
		ReadProperties prop = dc.new ReadProperties();
		
		prop.path = dir;
		prop.inputInfo = inputinfo;
		prop.rlen = rlen;
		prop.clen = clen;
		prop.brlen = brlen;
		prop.bclen = bclen;
		prop.expectedSparsity = expectedSparsity;
		prop.localFS = localFS;
		
		return readMatrixFromHDFS(prop);
	}
	/**
	 * 
	 * @param dir
	 * @param inputinfo
	 * @param rlen
	 * @param clen
	 * @param brlen
	 * @param bclen
	 * @param expectedSparsity
	 * @param localFS
	 * @return
	 * @throws IOException
	 */
	public static MatrixBlock readMatrixFromHDFS(String dir, InputInfo inputinfo, long rlen, long clen, 
			int brlen, int bclen, double expectedSparsity, FileFormatProperties formatProperties) 
	throws IOException
	{
		DataConverter dc = new DataConverter();
		ReadProperties prop = dc.new ReadProperties();
		
		prop.path = dir;
		prop.inputInfo = inputinfo;
		prop.rlen = rlen;
		prop.clen = clen;
		prop.brlen = brlen;
		prop.bclen = bclen;
		prop.expectedSparsity = expectedSparsity;
		prop.formatProperties = formatProperties;
		
		return readMatrixFromHDFS(prop);
	}
	
	/**
	 * Core method for reading matrices in format textcell, matrixmarket, binarycell, or binaryblock 
	 * from HDFS into main memory. For expected dense matrices we directly copy value- or block-at-a-time 
	 * into the target matrix. In contrast, for sparse matrices, we append (column-value)-pairs and do a 
	 * final sort if required in order to prevent large reorg overheads and increased memory consumption 
	 * in case of unordered inputs.  
	 * 
	 * DENSE MxN input:
	 *  * best/average/worst: O(M*N)
	 * SPARSE MxN input
	 *  * best (ordered, or binary block w/ clen<=bclen): O(M*N)
	 *  * average (unordered): O(M*N*log(N))
	 *  * worst (descending order per row): O(M * N^2)
	 * 
	 * NOTE: providing an exact estimate of 'expected sparsity' can prevent a full copy of the result
	 * matrix block (required for changing sparse->dense, or vice versa)
	 * 
	 * @param dir
	 * @param inputinfo
	 * @param rlen
	 * @param clen
	 * @param brlen
	 * @param bclen
	 * @param expectedSparsity
	 * @return
	 * @throws IOException
	 */
	public static MatrixBlock readMatrixFromHDFS(ReadProperties prop) 
	throws IOException
	{	
		boolean sparse = prop.expectedSparsity < MatrixBlockDSM.SPARCITY_TURN_POINT;

		//System.out.println("read matrix (sparse="+sparse+") from HDFS: "+prop.dir);
		
		long rlen = prop.rlen;
		long clen = prop.clen;
		InputInfo inputinfo = prop.inputInfo;
		
		//handle vectors specially
		//if result is a column vector, use dense format, otherwise use the normal process to decide
		if ( clen <=MatrixBlock.SKINNY_MATRIX_TURN_POINT )
			sparse = false;
		
		//prepare result matrix block
		MatrixBlock ret = new MatrixBlock((int)rlen, (int)clen, sparse, (int)(prop.expectedSparsity*rlen*clen));
		if( !sparse && inputinfo != InputInfo.BinaryBlockInputInfo )
			ret.spaceAllocForDenseUnsafe((int)rlen, (int)clen);
		
		//prepare file access
		JobConf job = new JobConf();	
		Path path = null; 
		FileSystem fs = null;
		if ( prop.localFS ) {
			path = new Path("file:///" + prop.path);
			//System.out.println("local file system path ..." + path.toUri() + ", " + path.toString());
			fs = FileSystem.getLocal(job);
		}
		else {
			path = new Path(prop.path);
			fs = FileSystem.get(job);
		}
		if( !fs.exists(path) )	
			throw new IOException("File "+prop.path+" does not exist on HDFS/LFS.");
		//System.out.println("dataconverter: reading file " + path + " [" + rlen + "," + clen + "] from localFS=" + localFS);
		
		try 
		{
			//check for empty file
			if( MapReduceTool.isFileEmpty( fs, path.toString() ) )
				throw new EOFException("Empty input file "+ prop.path +".");
			
			boolean isMMFile = (inputinfo == InputInfo.MatrixMarketInputInfo);
			
			//core matrix reading 
			if( inputinfo == InputInfo.TextCellInputInfo )
			{			
				if( fs.getFileStatus(path).isDir() )
					readTextCellMatrixFromHDFS(path, job, ret, rlen, clen, prop.brlen, prop.bclen);
				else
					readRawTextCellMatrixFromHDFS(path, job, fs, ret, rlen, clen, prop.brlen, prop.bclen, isMMFile);
			}
			else if( inputinfo == InputInfo.BinaryCellInputInfo )
			{
				readBinaryCellMatrixFromHDFS( path, job, fs, ret, rlen, clen, prop.brlen, prop.bclen );
			}
			else if( inputinfo == InputInfo.BinaryBlockInputInfo )
			{
				readBinaryBlockMatrixFromHDFS( path, job, fs, ret, rlen, clen, prop.brlen, prop.bclen );
			}
			else if ( inputinfo == InputInfo.MatrixMarketInputInfo ) {
				readRawTextCellMatrixFromHDFS(path, job, fs, ret, rlen, clen, prop.brlen, prop.bclen, isMMFile);
			} 
			else if ( inputinfo == InputInfo.CSVInputInfo ) {
				CSVFileFormatProperties csvprop = (CSVFileFormatProperties) prop.formatProperties;
				readCSVMatrixFromHDFS(path, job, fs, ret, rlen, clen, prop.brlen, prop.bclen, csvprop.isHasHeader(), csvprop.getDelim(), csvprop.isFill(), csvprop.getFillValue());
			}
			else {
				throw new IOException("Can not read files with format = " + InputInfo.inputInfoToString(inputinfo));
			}
			
			//finally check if change of sparse/dense block representation required
			if( !sparse || inputinfo==InputInfo.BinaryBlockInputInfo )
				ret.recomputeNonZeros();
			ret.examSparsity();	
		} 
		catch (Exception e) 
		{
			throw new IOException(e);
		}

		//System.out.println("read matrix (after exec sparse="+ret.isInSparseFormat()+") from HDFS: "+dir);
		//System.out.println("read matrix ("+rlen+","+clen+","+ret.getNonZeros()+") in "+time.stop());
		
		return ret;
	}

	/**
	 * Reads a partition that contains the given block index (rowBlockIndex, colBlockIndex)
	 * from a file on HDFS/LocalFS as specified by <code>dir</code>, which is partitioned using 
	 * <code>pformat</code> and <code>psize</code>.
	 * 
	 * @param dir
	 * @param localFS
	 * @param rows
	 * @param cols
	 * @param pformat
	 * @param psize
	 * @param rowBlockIndex
	 * @param colBlockIndex
	 * @return
	 * @throws DMLRuntimeException
	 */
	public static MatrixBlock readPartitionFromDistCache(String dir, boolean localFS, 
			long rows, long cols, int rowBlockSize, int colBlockSize,
			int partID, int partSize) throws DMLRuntimeException {
		
		dir = dir + "/" + partID;
		
		try {
			MatrixBlock partition = readMatrixFromHDFS(dir, InputInfo.BinaryBlockInputInfo, partSize, 1, rowBlockSize, colBlockSize, localFS);
			//System.out.println("Reading complete...");
			return partition;
		} catch (IOException e) {
			throw new DMLRuntimeException(e);
		}
	}
	
	/**
	 * 
	 * @param srcfileName
	 * @param fileName
	 * @param rlen
	 * @param clen
	 * @param nnz
	 * @throws IOException
	 */
	public static void mergeTextcellToMatrixMarket( String srcFileName, String fileName, long rlen, long clen, long nnz )
		throws IOException
	{
		  Configuration conf = new Configuration();
		
		  Path src = new Path (srcFileName);
	      Path merge = new Path (fileName);
	      FileSystem hdfs = FileSystem.get(conf);
	    
	      if (hdfs.exists (merge)) {
	    	hdfs.delete(merge, true);
	      }
        
	      OutputStream out = hdfs.create(merge, true);

	      // write out the header first 
	      StringBuilder  sb = new StringBuilder();
	      sb.append ("%%MatrixMarket matrix coordinate real general\n");
	    
	      // output number of rows, number of columns and number of nnz
	 	  sb.append (rlen + " " + clen + " " + nnz + "\n");
	      out.write (sb.toString().getBytes());

	      // if the source is a directory
	      if (hdfs.getFileStatus(src).isDir()) {
	        try {
	          FileStatus contents[] = hdfs.listStatus(src);
	          for (int i = 0; i < contents.length; i++) {
	            if (!contents[i].isDir()) {
	               InputStream in = hdfs.open (contents[i].getPath());
	               try {
	                 IOUtils.copyBytes (in, out, conf, false);
	               }  finally {
	                 in.close();
	               }
	             }
	           }
	         } finally {
	            out.close();
	         }
	      } else if (hdfs.isFile(src))  {
	        InputStream in = null;
	        try {
   	          in = hdfs.open (src);
	          IOUtils.copyBytes (in, out, conf, true);
	        } finally {
	          in.close();
	          out.close();
	        }
	      } else {
	        throw new IOException(src.toString() + ": No such file or directory");
	      }
	}
	
	/**
	 * 
	 * @param fileName
	 * @param src
	 * @param rlen
	 * @param clen
	 * @param nnz
	 * @throws IOException
	 */
	public static void writeMatrixMarketToHDFS( Path path, JobConf job, MatrixBlock src, long rlen, long clen, long nnz )
		throws IOException
	{
		boolean sparse = src.isInSparseFormat();
		boolean entriesWritten = false;
		FileSystem fs = FileSystem.get(job);
        BufferedWriter br=new BufferedWriter(new OutputStreamWriter(fs.create(path,true)));		
        
    	int rows = src.getNumRows();
		int cols = src.getNumColumns();

		//bound check per block
		if( rows > rlen || cols > clen )
		{
			throw new IOException("Matrix block [1:"+rows+",1:"+cols+"] " +
					              "out of overall matrix range [1:"+rlen+",1:"+clen+"].");
		}
		
		try
		{
			//for obj reuse and preventing repeated buffer re-allocations
			StringBuilder sb = new StringBuilder();
			
			// First output MM header
			sb.append ("%%MatrixMarket matrix coordinate real general\n");
		
			// output number of rows, number of columns and number of nnz
			sb.append (rlen + " " + clen + " " + nnz + "\n");
            br.write( sb.toString());
            sb.setLength(0);
            
            // output matrix cell
			if( sparse ) //SPARSE
			{			   
				SparseCellIterator iter = src.getSparseCellIterator();
				while( iter.hasNext() )
				{
					IJV cell = iter.next();

					sb.append(cell.i+1);
					sb.append(' ');
					sb.append(cell.j+1);
					sb.append(' ');
					sb.append(cell.v);
					sb.append('\n');
					br.write( sb.toString() ); //same as append
					sb.setLength(0); 
					entriesWritten = true;					
				}
			}
			else //DENSE
			{
				for( int i=0; i<rows; i++ )
					for( int j=0; j<cols; j++ )
					{
						double lvalue = src.getValueDenseUnsafe(i, j);
						if( lvalue != 0 ) //for nnz
						{
							sb.append(i+1);
							sb.append(' ');
							sb.append(j+1);
							sb.append(' ');
							sb.append(lvalue);
							sb.append('\n');
							br.write( sb.toString() ); //same as append
							sb.setLength(0); 
							entriesWritten = true;
						}
					}
			}
	
			//handle empty result
			if ( !entriesWritten ) {
				br.write("1 1 0\n");
			}
		}
		finally
		{
			if( br != null )
				br.close();
		}
	}
	
	/**
	 * Method to merge multiple CSV part files on HDFS into a single CSV file on HDFS. 
	 * The part files are created by CSV_WRITE MR job. 
	 * 
	 * This method is invoked from CP-write instruction.
	 * 
	 * @param srcFileName
	 * @param destFileName
	 * @param csvprop
	 * @param rlen
	 * @param clen
	 * @throws IOException
	 */
	public static void mergeCSVPartFiles(String srcFileName,
			String destFileName, CSVFileFormatProperties csvprop, long rlen, long clen) 
			throws IOException {
		
		Configuration conf = new Configuration();

		Path srcFilePath = new Path(srcFileName);
		Path mergedFilePath = new Path(destFileName);
		FileSystem hdfs = FileSystem.get(conf);

		if (hdfs.exists(mergedFilePath)) {
			hdfs.delete(mergedFilePath, true);
		}
		OutputStream out = hdfs.create(mergedFilePath, true);

		// write out the header, if needed
		if (csvprop.isHasHeader()) {
			StringBuilder sb = new StringBuilder();
			for (int i = 0; i < clen; i++) {
				sb.append("C" + (i + 1));
				if (i < clen - 1)
					sb.append(csvprop.getDelim());
			}
			sb.append('\n');
			out.write(sb.toString().getBytes());
			sb.setLength(0);
		}

		// if the source is a directory
		if (hdfs.getFileStatus(srcFilePath).isDir()) {
			try {
				FileStatus contents[] = hdfs.listStatus(srcFilePath);
				Path[] partPaths = new Path[contents.length];
				int numPartFiles = 0;
				for (int i = 0; i < contents.length; i++) {
					if (!contents[i].isDir()) {
						partPaths[i] = contents[i].getPath();
						numPartFiles++;
					}
				}
				Arrays.sort(partPaths);

				for (int i = 0; i < numPartFiles; i++) {
					InputStream in = hdfs.open(partPaths[i]);
					try {
						IOUtils.copyBytes(in, out, conf, false);
					} finally {
						in.close();
					}
				}
			} finally {
				out.close();
			}
		} else if (hdfs.isFile(srcFilePath)) {
			InputStream in = null;
			try {
				in = hdfs.open(srcFilePath);
				IOUtils.copyBytes(in, out, conf, true);
			} finally {
				in.close();
				out.close();
			}
		} else {
			throw new IOException(srcFilePath.toString()
					+ ": No such file or directory");
		}
	}
		
	/**
	 * 
	 * @param fileName
	 * @param src
	 * @param rlen
	 * @param clen
	 * @param nnz
	 * @throws IOException
	 */
	public static void writeCSVToHDFS( Path path, JobConf job, MatrixBlock src, long rlen, long clen, long nnz, FileFormatProperties formatProperties )
		throws IOException
	{
		boolean sparse = src.isInSparseFormat();
		FileSystem fs = FileSystem.get(job);
        BufferedWriter br=new BufferedWriter(new OutputStreamWriter(fs.create(path,true)));		
        
    	int rows = src.getNumRows();
		int cols = src.getNumColumns();

		//bound check per block
		if( rows > rlen || cols > clen )
		{
			throw new IOException("Matrix block [1:"+rows+",1:"+cols+"] " +
					              "out of overall matrix range [1:"+rlen+",1:"+clen+"].");
		}
		
		try
		{
			//for obj reuse and preventing repeated buffer re-allocations
			StringBuilder sb = new StringBuilder();
			
			CSVFileFormatProperties csvProperties = (CSVFileFormatProperties)formatProperties;
			String delim = csvProperties.getDelim(); //Pattern.quote(csvProperties.getDelim());
			
			// Write header line, if needed
			if (csvProperties.isHasHeader()) {
				for(int i=0; i < clen; i++) {
					sb.append("C"+ (i+1));
					if ( i < clen-1 )
						sb.append(delim);
				}
				sb.append('\n');
				br.write( sb.toString());
	            sb.setLength(0);
			}
			
			// Write data lines
			if( sparse ) //SPARSE
			{			   
	            int c, prev_c;
	            double v;
				for(int r=0; r < rows; r++) {
					SparseRow spRow = src.getSparseRows()[r];
					prev_c = -1;
					if ( spRow != null) {
						for(int ind=0; ind < spRow.size(); ind++) {
							c = spRow.getIndexContainer()[ind];
							v = spRow.getValueContainer()[ind];
							
							// output empty fields, if needed
							while(prev_c < c-1) {
								if (!csvProperties.isSparse()) {
									sb.append(0.0);
								}
								sb.append(delim);
								prev_c++;
							}
							
							// output the value
							sb.append(v);
							if ( c < clen-1)
								sb.append(delim);
							prev_c = c;
						}
					}
					// Output empty fields at the end of the row.
					// In case of an empty row, output (clen-1) empty fields
					while (prev_c < clen - 1) {
						if (!csvProperties.isSparse()) {
							sb.append(0.0);
						}
						prev_c++;
						if (prev_c < clen-1)
							sb.append(delim);
					}
					sb.append('\n');
					/*if ( sb.toString().split(Pattern.quote(delim)).length != clen) {
						throw new RuntimeException("row = " + r + ", prev_c = " + prev_c + ", filedcount=" + sb.toString().split(Pattern.quote(delim)).length + ": " + sb.toString());
					}*/
					br.write( sb.toString() ); 
					sb.setLength(0); 
				}
			}
			else //DENSE
			{
				for( int i=0; i<rows; i++ ) {
					for( int j=0; j<cols; j++ )
					{
						double lvalue = src.getValueDenseUnsafe(i, j);
						if( lvalue != 0 ) //for nnz
						{
							sb.append(lvalue);
						}
						else {
							if (!csvProperties.isSparse()) {
								// write in dense format
								sb.append(0.0);
							}
						}
						if ( j != cols-1 )
							sb.append(delim);
					}
					sb.append('\n');
					br.write( sb.toString() ); //same as append
					sb.setLength(0); 
				}
			}
		}
		finally
		{
			if( br != null )
				br.close();
		}
	}
	
	/**
	 * 
	 * @param path
	 * @param job
	 * @param src
	 * @param rlen
	 * @param clen
	 * @param brlen
	 * @param bclen
	 * @throws IOException
	 */
	private static void writeTextCellMatrixToHDFS( Path path, JobConf job, MatrixBlock src, long rlen, long clen, int brlen, int bclen )
		throws IOException
	{
		boolean sparse = src.isInSparseFormat();
		boolean entriesWritten = false;
		FileSystem fs = FileSystem.get(job);
        BufferedWriter br=new BufferedWriter(new OutputStreamWriter(fs.create(path,true)));		
        
    	int rows = src.getNumRows();
		int cols = src.getNumColumns();

		//bound check per block
		if( rows > rlen || cols > clen )
		{
			throw new IOException("Matrix block [1:"+rows+",1:"+cols+"] " +
					              "out of overall matrix range [1:"+rlen+",1:"+clen+"].");
		}
		
		try
		{
			//for obj reuse and preventing repeated buffer re-allocations
			StringBuilder sb = new StringBuilder();
			
			if( sparse ) //SPARSE
			{			   
				SparseCellIterator iter = src.getSparseCellIterator();
				while( iter.hasNext() )
				{
					IJV cell = iter.next();

					sb.append(cell.i+1);
					sb.append(' ');
					sb.append(cell.j+1);
					sb.append(' ');
					sb.append(cell.v);
					sb.append('\n');
					br.write( sb.toString() ); //same as append
					sb.setLength(0); 
					entriesWritten = true;					
				}
			}
			else //DENSE
			{
				for( int i=0; i<rows; i++ )
					for( int j=0; j<cols; j++ )
					{
						double lvalue = src.getValueDenseUnsafe(i, j);
						if( lvalue != 0 ) //for nnz
						{
							sb.append(i+1);
							sb.append(' ');
							sb.append(j+1);
							sb.append(' ');
							sb.append(lvalue);
							sb.append('\n');
							br.write( sb.toString() ); //same as append
							sb.setLength(0); 
							entriesWritten = true;
						}
					}
			}
	
			//handle empty result
			if ( !entriesWritten ) {
				br.write("1 1 0\n");
			}
		}
		finally
		{
			if( br != null )
				br.close();
		}
	}
	
	/**
	 * 
	 * @param path
	 * @param job
	 * @param src
	 * @param rlen
	 * @param clen
	 * @param brlen
	 * @param bclen
	 * @throws IOException
	 */
	private static void writeBinaryCellMatrixToHDFS( Path path, JobConf job, MatrixBlock src, long rlen, long clen, int brlen, int bclen )
		throws IOException
	{
		boolean sparse = src.isInSparseFormat();
		boolean entriesWritten = false;
		FileSystem fs = FileSystem.get(job);
		SequenceFile.Writer writer = new SequenceFile.Writer(fs, job, path, MatrixIndexes.class, MatrixCell.class);
		
		MatrixIndexes indexes = new MatrixIndexes();
		MatrixCell cell = new MatrixCell();

		int rows = src.getNumRows(); 
		int cols = src.getNumColumns();
        
		//bound check per block
		if( rows > rlen || cols > clen )
		{
			throw new IOException("Matrix block [1:"+rows+",1:"+cols+"] " +
					              "out of overall matrix range [1:"+rlen+",1:"+clen+"].");
		}
		
		try
		{
			if( sparse ) //SPARSE
			{
				SparseCellIterator iter = src.getSparseCellIterator();
				while( iter.hasNext() )
				{
					IJV lcell = iter.next();
					indexes.setIndexes(lcell.i+1, lcell.j+1);
					cell.setValue(lcell.v);
					writer.append(indexes, cell);
					entriesWritten = true;
				}
			}
			else //DENSE
			{
				for( int i=0; i<rows; i++ )
					for( int j=0; j<cols; j++ )
					{
						double lvalue  = src.getValueDenseUnsafe(i, j);
						if( lvalue != 0 ) //for nnz
						{
							indexes.setIndexes(i+1, j+1);
							cell.setValue(lvalue);
							writer.append(indexes, cell);
							entriesWritten = true;
						}
					}
			}
	
			//handle empty result
			if ( !entriesWritten ) {
				writer.append(new MatrixIndexes(1, 1), new MatrixCell(0));
			}
		}
		finally
		{
			if( writer != null )
				writer.close();
		}
	}
	
	/**
	 * 
	 * @param path
	 * @param job
	 * @param src
	 * @param rlen
	 * @param clen
	 * @param brlen
	 * @param bclen
	 * @throws IOException
	 * @throws DMLUnsupportedOperationException 
	 * @throws DMLRuntimeException 
	 */
	private static void writeBinaryBlockMatrixToHDFS( Path path, JobConf job, MatrixBlock src, long rlen, long clen, int brlen, int bclen )
		throws IOException, DMLRuntimeException, DMLUnsupportedOperationException
	{
		boolean sparse = src.isInSparseFormat();
		FileSystem fs = FileSystem.get(job);
		SequenceFile.Writer writer = new SequenceFile.Writer(fs, job, path, MatrixIndexes.class, MatrixBlock.class);
		
		//bound check for src block
		if( src.getNumRows() > rlen || src.getNumColumns() > clen )
		{
			throw new IOException("Matrix block [1:"+src.getNumRows()+",1:"+src.getNumColumns()+"] " +
					              "out of overall matrix range [1:"+rlen+",1:"+clen+"].");
		}
		//reblock and write
		try
		{
			MatrixIndexes indexes = new MatrixIndexes();

			if( rlen <= brlen && clen <= bclen ) //opt for single block
			{
				//directly write single block
				indexes.setIndexes(1, 1);
				writer.append(indexes, src);
			}
			else //general case
			{
				//initialize blocks for reuse (at most 4 different blocks required)
				MatrixBlock[] blocks = createMatrixBlocksForReuse(rlen, clen, brlen, bclen, sparse, src.getNonZeros());  
				
				//create and write subblocks of matrix
				for(int blockRow = 0; blockRow < (int)Math.ceil(src.getNumRows()/(double)brlen); blockRow++)
					for(int blockCol = 0; blockCol < (int)Math.ceil(src.getNumColumns()/(double)bclen); blockCol++)
					{
						int maxRow = (blockRow*brlen + brlen < src.getNumRows()) ? brlen : src.getNumRows() - blockRow*brlen;
						int maxCol = (blockCol*bclen + bclen < src.getNumColumns()) ? bclen : src.getNumColumns() - blockCol*bclen;
				<commit_msg>55555: SystemML Engine Enhancements - Fix data converter export replication factor (disregarded in job conf)<commit_after>/**
 * IBM Confidential
 * OCO Source Materials
 * (C) Copyright IBM Corp. 2010, 2014
 * The source code for this program is not published or otherwise divested of its trade secrets, irrespective of what has been deposited with the U.S. Copyright Office.
 */

package com.ibm.bi.dml.runtime.util;

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.EOFException;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.io.OutputStream;
import java.io.InputStream;
import java.util.Arrays;
import java.util.LinkedList;
import java.util.StringTokenizer;
import java.util.regex.Pattern;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.InputFormat;
import org.apache.hadoop.mapred.InputSplit;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.RecordReader;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.TextInputFormat;

import com.ibm.bi.dml.runtime.DMLRuntimeException;
import com.ibm.bi.dml.runtime.DMLUnsupportedOperationException;
import com.ibm.bi.dml.runtime.matrix.MatrixCharacteristics;
import com.ibm.bi.dml.runtime.matrix.io.BinaryBlockToBinaryCellConverter;
import com.ibm.bi.dml.runtime.matrix.io.BinaryBlockToTextCellConverter;
import com.ibm.bi.dml.runtime.matrix.io.CSVFileFormatProperties;
import com.ibm.bi.dml.runtime.matrix.io.Converter;
import com.ibm.bi.dml.runtime.matrix.io.FileFormatProperties;
import com.ibm.bi.dml.runtime.matrix.io.InputInfo;
import com.ibm.bi.dml.runtime.matrix.io.MatrixBlock;
import com.ibm.bi.dml.runtime.matrix.io.MatrixBlockDSM;
import com.ibm.bi.dml.runtime.matrix.io.MatrixCell;
import com.ibm.bi.dml.runtime.matrix.io.MatrixIndexes;
import com.ibm.bi.dml.runtime.matrix.io.OutputInfo;
import com.ibm.bi.dml.runtime.matrix.io.Pair;
import com.ibm.bi.dml.runtime.matrix.io.MatrixBlockDSM.IJV;
import com.ibm.bi.dml.runtime.matrix.io.MatrixBlockDSM.SparseCellIterator;
import com.ibm.bi.dml.runtime.matrix.io.SparseRow;
import com.ibm.bi.dml.runtime.matrix.mapred.MRJobConfiguration;
import com.ibm.bi.dml.runtime.matrix.mapred.MRJobConfiguration.ConvertTarget;


/**
 * This class provides methods to read and write matrix blocks from to HDFS using different data formats.
 * Those functionalities are used especially for CP read/write and exporting in-memory matrices to HDFS
 * (before executing MR jobs).
 * 
 */
public class DataConverter 
{
	@SuppressWarnings("unused")
	private static final String _COPYRIGHT = "Licensed Materials - Property of IBM\n(C) Copyright IBM Corp. 2010, 2013\n" +
	                                         "US Government Users Restricted Rights - Use, duplication  disclosure restricted by GSA ADP Schedule Contract with IBM Corp.";
		
	//////////////
	// READING and WRITING of matrix blocks to/from HDFS
	// (textcell, binarycell, binaryblock)
	///////
	
	private class ReadProperties {
		// Properties common to all file formats 
		String path;
		long rlen, clen;
		int brlen, bclen;
		double expectedSparsity;
		InputInfo inputInfo;
		boolean localFS;
		
		// Properties specific to CSV files
		FileFormatProperties formatProperties;
		
		public ReadProperties() {
			rlen = clen = -1;
			brlen = bclen = -1;
			expectedSparsity = 0.1d;
			inputInfo = null;
			localFS = false;
		}

	}
	
	/**
	 * 
	 * @param mat
	 * @param dir
	 * @param outputinfo
	 * @param rlen
	 * @param clen
	 * @param brlen
	 * @param bclen
	 * @throws IOException
	 */
	public static void writeMatrixToHDFS(MatrixBlock mat, String dir, OutputInfo outputinfo, 
			                             MatrixCharacteristics mc )
		throws IOException
	{
		writeMatrixToHDFS(mat, dir, outputinfo, mc, -1, null);
	}
	
	public static void writeMatrixToHDFS(MatrixBlock mat, String dir, OutputInfo outputinfo, MatrixCharacteristics mc, int replication, FileFormatProperties formatProperties)
			throws IOException
	{
		JobConf job = new JobConf();
		Path path = new Path(dir);

		//System.out.println("write matrix (sparse="+mat.isInSparseFormat()+") to HDFS: "+dir);
		
		try
		{
			// If the file already exists on HDFS, remove it.
			MapReduceTool.deleteFileIfExistOnHDFS(dir);

			// core matrix writing
			if ( outputinfo == OutputInfo.TextCellOutputInfo ) 
			{	
				writeTextCellMatrixToHDFS(path, job, mat, mc.get_rows(), mc.get_cols(), mc.get_rows_per_block(), mc.get_cols_per_block());
			}
			else if ( outputinfo == OutputInfo.BinaryCellOutputInfo ) 
			{
				writeBinaryCellMatrixToHDFS(path, job, mat, mc.get_rows(), mc.get_cols(), mc.get_rows_per_block(), mc.get_cols_per_block());
			}
			else if( outputinfo == OutputInfo.BinaryBlockOutputInfo )
			{
				writeBinaryBlockMatrixToHDFS(path, job, mat, mc.get_rows(), mc.get_cols(), mc.get_rows_per_block(), mc.get_cols_per_block(), replication);
			}
			else if ( outputinfo == OutputInfo.MatrixMarketOutputInfo ) 
			{
				writeMatrixMarketToHDFS( path, job, mat, mc.get_rows(), mc.get_cols(), mc.getNonZeros());
			}
			else if ( outputinfo == OutputInfo.CSVOutputInfo ) {
				writeCSVToHDFS(path, job, mat, mc.get_rows(), mc.get_cols(), mc.getNonZeros(), formatProperties);
			}
			else {
				throw new RuntimeException("Unknown format (" + OutputInfo.outputInfoToString(outputinfo));
			}
		}
		catch(Exception e)
		{
			throw new IOException(e);
		}
	}
	
	/**
	 * 
	 * @param dir
	 * @param inputinfo
	 * @param rlen
	 * @param clen
	 * @param brlen
	 * @param bclen
	 * @return
	 * @throws IOException
	 */
	public static MatrixBlock readMatrixFromHDFS(String dir, InputInfo inputinfo, long rlen, long clen, int brlen, int bclen, boolean localFS) 
		throws IOException
	{	
		DataConverter dc = new DataConverter();
		ReadProperties prop = dc.new ReadProperties();
		
		prop.path = dir;
		prop.inputInfo = inputinfo;
		prop.rlen = rlen;
		prop.clen = clen;
		prop.brlen = brlen;
		prop.bclen = bclen;
		prop.localFS = localFS;
		
		//expected matrix is sparse (default SystemML usecase)
		return readMatrixFromHDFS(prop);
	}
	
	/**
	 * 
	 * @param dir
	 * @param inputinfo
	 * @param rlen
	 * @param clen
	 * @param brlen
	 * @param bclen
	 * @return
	 * @throws IOException
	 */
	public static MatrixBlock readMatrixFromHDFS(String dir, InputInfo inputinfo, long rlen, long clen, int brlen, int bclen) 
		throws IOException
	{	
		DataConverter dc = new DataConverter();
		ReadProperties prop = dc.new ReadProperties();
		
		prop.path = dir;
		prop.inputInfo = inputinfo;
		prop.rlen = rlen;
		prop.clen = clen;
		prop.brlen = brlen;
		prop.bclen = bclen;
		
		//expected matrix is sparse (default SystemML usecase)
		return readMatrixFromHDFS(prop);
	}

	/**
	 * 
	 * @param dir
	 * @param inputinfo
	 * @param rlen
	 * @param clen
	 * @param brlen
	 * @param bclen
	 * @param expectedSparsity
	 * @return
	 * @throws IOException
	 */
	public static MatrixBlock readMatrixFromHDFS(String dir, InputInfo inputinfo, long rlen, long clen, int brlen, int bclen, double expectedSparsity) 
		throws IOException
	{	
		DataConverter dc = new DataConverter();
		ReadProperties prop = dc.new ReadProperties();
		
		prop.path = dir;
		prop.inputInfo = inputinfo;
		prop.rlen = rlen;
		prop.clen = clen;
		prop.brlen = brlen;
		prop.bclen = bclen;
		prop.expectedSparsity = expectedSparsity;
		
		return readMatrixFromHDFS(prop);
	}

	/**
	 * 
	 * @param dir
	 * @param inputinfo
	 * @param rlen
	 * @param clen
	 * @param brlen
	 * @param bclen
	 * @param expectedSparsity
	 * @param localFS
	 * @return
	 * @throws IOException
	 */
	public static MatrixBlock readMatrixFromHDFS(String dir, InputInfo inputinfo, long rlen, long clen, 
			int brlen, int bclen, double expectedSparsity, boolean localFS) 
	throws IOException
	{
		DataConverter dc = new DataConverter();
		ReadProperties prop = dc.new ReadProperties();
		
		prop.path = dir;
		prop.inputInfo = inputinfo;
		prop.rlen = rlen;
		prop.clen = clen;
		prop.brlen = brlen;
		prop.bclen = bclen;
		prop.expectedSparsity = expectedSparsity;
		prop.localFS = localFS;
		
		return readMatrixFromHDFS(prop);
	}
	/**
	 * 
	 * @param dir
	 * @param inputinfo
	 * @param rlen
	 * @param clen
	 * @param brlen
	 * @param bclen
	 * @param expectedSparsity
	 * @param localFS
	 * @return
	 * @throws IOException
	 */
	public static MatrixBlock readMatrixFromHDFS(String dir, InputInfo inputinfo, long rlen, long clen, 
			int brlen, int bclen, double expectedSparsity, FileFormatProperties formatProperties) 
	throws IOException
	{
		DataConverter dc = new DataConverter();
		ReadProperties prop = dc.new ReadProperties();
		
		prop.path = dir;
		prop.inputInfo = inputinfo;
		prop.rlen = rlen;
		prop.clen = clen;
		prop.brlen = brlen;
		prop.bclen = bclen;
		prop.expectedSparsity = expectedSparsity;
		prop.formatProperties = formatProperties;
		
		return readMatrixFromHDFS(prop);
	}
	
	/**
	 * Core method for reading matrices in format textcell, matrixmarket, binarycell, or binaryblock 
	 * from HDFS into main memory. For expected dense matrices we directly copy value- or block-at-a-time 
	 * into the target matrix. In contrast, for sparse matrices, we append (column-value)-pairs and do a 
	 * final sort if required in order to prevent large reorg overheads and increased memory consumption 
	 * in case of unordered inputs.  
	 * 
	 * DENSE MxN input:
	 *  * best/average/worst: O(M*N)
	 * SPARSE MxN input
	 *  * best (ordered, or binary block w/ clen<=bclen): O(M*N)
	 *  * average (unordered): O(M*N*log(N))
	 *  * worst (descending order per row): O(M * N^2)
	 * 
	 * NOTE: providing an exact estimate of 'expected sparsity' can prevent a full copy of the result
	 * matrix block (required for changing sparse->dense, or vice versa)
	 * 
	 * @param dir
	 * @param inputinfo
	 * @param rlen
	 * @param clen
	 * @param brlen
	 * @param bclen
	 * @param expectedSparsity
	 * @return
	 * @throws IOException
	 */
	public static MatrixBlock readMatrixFromHDFS(ReadProperties prop) 
		throws IOException
	{	
		//determine target representation (sparse/dense)
		boolean sparse = (    prop.expectedSparsity < MatrixBlockDSM.SPARCITY_TURN_POINT
				           && prop.clen > MatrixBlock.SKINNY_MATRIX_TURN_POINT ); 

		//System.out.println("read matrix (sparse="+sparse+") from HDFS: "+prop.dir);
		
		long rlen = prop.rlen;
		long clen = prop.clen;
		InputInfo inputinfo = prop.inputInfo;
		
		//prepare result matrix block
		MatrixBlock ret = new MatrixBlock((int)rlen, (int)clen, sparse, (int)(prop.expectedSparsity*rlen*clen));
		if( !sparse && inputinfo != InputInfo.BinaryBlockInputInfo )
			ret.spaceAllocForDenseUnsafe((int)rlen, (int)clen);
		
		//prepare file access
		JobConf job = new JobConf();	
		Path path = null; 
		FileSystem fs = null;
		if ( prop.localFS ) {
			path = new Path("file:///" + prop.path);
			//System.out.println("local file system path ..." + path.toUri() + ", " + path.toString());
			fs = FileSystem.getLocal(job);
		}
		else {
			path = new Path(prop.path);
			fs = FileSystem.get(job);
		}
		
		if( !fs.exists(path) )	
			throw new IOException("File "+prop.path+" does not exist on HDFS/LFS.");
		//System.out.println("dataconverter: reading file " + path + " [" + rlen + "," + clen + "] from localFS=" + localFS);
		
		try 
		{
			//check for empty file
			if( MapReduceTool.isFileEmpty( fs, path.toString() ) )
				throw new EOFException("Empty input file "+ prop.path +".");
			
			boolean isMMFile = (inputinfo == InputInfo.MatrixMarketInputInfo);
			
			//core matrix reading 
			if( inputinfo == InputInfo.TextCellInputInfo )
			{			
				if( fs.getFileStatus(path).isDir() )
					readTextCellMatrixFromHDFS(path, job, ret, rlen, clen, prop.brlen, prop.bclen);
				else
					readRawTextCellMatrixFromHDFS(path, job, fs, ret, rlen, clen, prop.brlen, prop.bclen, isMMFile);
			}
			else if( inputinfo == InputInfo.BinaryCellInputInfo )
			{
				readBinaryCellMatrixFromHDFS( path, job, fs, ret, rlen, clen, prop.brlen, prop.bclen );
			}
			else if( inputinfo == InputInfo.BinaryBlockInputInfo )
			{
				readBinaryBlockMatrixFromHDFS( path, job, fs, ret, rlen, clen, prop.brlen, prop.bclen );
			}
			else if ( inputinfo == InputInfo.MatrixMarketInputInfo ) {
				readRawTextCellMatrixFromHDFS(path, job, fs, ret, rlen, clen, prop.brlen, prop.bclen, isMMFile);
			} 
			else if ( inputinfo == InputInfo.CSVInputInfo ) {
				CSVFileFormatProperties csvprop = (CSVFileFormatProperties) prop.formatProperties;
				readCSVMatrixFromHDFS(path, job, fs, ret, rlen, clen, prop.brlen, prop.bclen, csvprop.isHasHeader(), csvprop.getDelim(), csvprop.isFill(), csvprop.getFillValue());
			}
			else {
				throw new IOException("Can not read files with format = " + InputInfo.inputInfoToString(inputinfo));
			}
			
			//finally check if change of sparse/dense block representation required
			if( !sparse || inputinfo==InputInfo.BinaryBlockInputInfo )
				ret.recomputeNonZeros();
			ret.examSparsity();	
		} 
		catch (Exception e) 
		{
			throw new IOException(e);
		}

		//System.out.println("read matrix (after exec sparse="+ret.isInSparseFormat()+") from HDFS: "+dir);
		//System.out.println("read matrix ("+rlen+","+clen+","+ret.getNonZeros()+") in "+time.stop());
		
		return ret;
	}

	/**
	 * Reads a partition that contains the given block index (rowBlockIndex, colBlockIndex)
	 * from a file on HDFS/LocalFS as specified by <code>dir</code>, which is partitioned using 
	 * <code>pformat</code> and <code>psize</code>.
	 * 
	 * @param dir
	 * @param localFS
	 * @param rows
	 * @param cols
	 * @param pformat
	 * @param psize
	 * @param rowBlockIndex
	 * @param colBlockIndex
	 * @return
	 * @throws DMLRuntimeException
	 */
	public static MatrixBlock readPartitionFromDistCache(String dir, boolean localFS, 
			long rows, long cols, int rowBlockSize, int colBlockSize,
			int partID, int partSize) throws DMLRuntimeException {
		
		dir = dir + "/" + partID;
		
		try {
			MatrixBlock partition = readMatrixFromHDFS(dir, InputInfo.BinaryBlockInputInfo, partSize, 1, rowBlockSize, colBlockSize, localFS);
			//System.out.println("Reading complete...");
			return partition;
		} catch (IOException e) {
			throw new DMLRuntimeException(e);
		}
	}
	
	/**
	 * 
	 * @param srcfileName
	 * @param fileName
	 * @param rlen
	 * @param clen
	 * @param nnz
	 * @throws IOException
	 */
	public static void mergeTextcellToMatrixMarket( String srcFileName, String fileName, long rlen, long clen, long nnz )
		throws IOException
	{
		  Configuration conf = new Configuration();
		
		  Path src = new Path (srcFileName);
	      Path merge = new Path (fileName);
	      FileSystem hdfs = FileSystem.get(conf);
	    
	      if (hdfs.exists (merge)) {
	    	hdfs.delete(merge, true);
	      }
        
	      OutputStream out = hdfs.create(merge, true);

	      // write out the header first 
	      StringBuilder  sb = new StringBuilder();
	      sb.append ("%%MatrixMarket matrix coordinate real general\n");
	    
	      // output number of rows, number of columns and number of nnz
	 	  sb.append (rlen + " " + clen + " " + nnz + "\n");
	      out.write (sb.toString().getBytes());

	      // if the source is a directory
	      if (hdfs.getFileStatus(src).isDir()) {
	        try {
	          FileStatus contents[] = hdfs.listStatus(src);
	          for (int i = 0; i < contents.length; i++) {
	            if (!contents[i].isDir()) {
	               InputStream in = hdfs.open (contents[i].getPath());
	               try {
	                 IOUtils.copyBytes (in, out, conf, false);
	               }  finally {
	                 in.close();
	               }
	             }
	           }
	         } finally {
	            out.close();
	         }
	      } else if (hdfs.isFile(src))  {
	        InputStream in = null;
	        try {
   	          in = hdfs.open (src);
	          IOUtils.copyBytes (in, out, conf, true);
	        } finally {
	          in.close();
	          out.close();
	        }
	      } else {
	        throw new IOException(src.toString() + ": No such file or directory");
	      }
	}
	
	/**
	 * 
	 * @param fileName
	 * @param src
	 * @param rlen
	 * @param clen
	 * @param nnz
	 * @throws IOException
	 */
	public static void writeMatrixMarketToHDFS( Path path, JobConf job, MatrixBlock src, long rlen, long clen, long nnz )
		throws IOException
	{
		boolean sparse = src.isInSparseFormat();
		boolean entriesWritten = false;
		FileSystem fs = FileSystem.get(job);
        BufferedWriter br=new BufferedWriter(new OutputStreamWriter(fs.create(path,true)));		
        
    	int rows = src.getNumRows();
		int cols = src.getNumColumns();

		//bound check per block
		if( rows > rlen || cols > clen )
		{
			throw new IOException("Matrix block [1:"+rows+",1:"+cols+"] " +
					              "out of overall matrix range [1:"+rlen+",1:"+clen+"].");
		}
		
		try
		{
			//for obj reuse and preventing repeated buffer re-allocations
			StringBuilder sb = new StringBuilder();
			
			// First output MM header
			sb.append ("%%MatrixMarket matrix coordinate real general\n");
		
			// output number of rows, number of columns and number of nnz
			sb.append (rlen + " " + clen + " " + nnz + "\n");
            br.write( sb.toString());
            sb.setLength(0);
            
            // output matrix cell
			if( sparse ) //SPARSE
			{			   
				SparseCellIterator iter = src.getSparseCellIterator();
				while( iter.hasNext() )
				{
					IJV cell = iter.next();

					sb.append(cell.i+1);
					sb.append(' ');
					sb.append(cell.j+1);
					sb.append(' ');
					sb.append(cell.v);
					sb.append('\n');
					br.write( sb.toString() ); //same as append
					sb.setLength(0); 
					entriesWritten = true;					
				}
			}
			else //DENSE
			{
				for( int i=0; i<rows; i++ )
					for( int j=0; j<cols; j++ )
					{
						double lvalue = src.getValueDenseUnsafe(i, j);
						if( lvalue != 0 ) //for nnz
						{
							sb.append(i+1);
							sb.append(' ');
							sb.append(j+1);
							sb.append(' ');
							sb.append(lvalue);
							sb.append('\n');
							br.write( sb.toString() ); //same as append
							sb.setLength(0); 
							entriesWritten = true;
						}
					}
			}
	
			//handle empty result
			if ( !entriesWritten ) {
				br.write("1 1 0\n");
			}
		}
		finally
		{
			if( br != null )
				br.close();
		}
	}
	
	/**
	 * Method to merge multiple CSV part files on HDFS into a single CSV file on HDFS. 
	 * The part files are created by CSV_WRITE MR job. 
	 * 
	 * This method is invoked from CP-write instruction.
	 * 
	 * @param srcFileName
	 * @param destFileName
	 * @param csvprop
	 * @param rlen
	 * @param clen
	 * @throws IOException
	 */
	public static void mergeCSVPartFiles(String srcFileName,
			String destFileName, CSVFileFormatProperties csvprop, long rlen, long clen) 
			throws IOException {
		
		Configuration conf = new Configuration();

		Path srcFilePath = new Path(srcFileName);
		Path mergedFilePath = new Path(destFileName);
		FileSystem hdfs = FileSystem.get(conf);

		if (hdfs.exists(mergedFilePath)) {
			hdfs.delete(mergedFilePath, true);
		}
		OutputStream out = hdfs.create(mergedFilePath, true);

		// write out the header, if needed
		if (csvprop.isHasHeader()) {
			StringBuilder sb = new StringBuilder();
			for (int i = 0; i < clen; i++) {
				sb.append("C" + (i + 1));
				if (i < clen - 1)
					sb.append(csvprop.getDelim());
			}
			sb.append('\n');
			out.write(sb.toString().getBytes());
			sb.setLength(0);
		}

		// if the source is a directory
		if (hdfs.getFileStatus(srcFilePath).isDir()) {
			try {
				FileStatus contents[] = hdfs.listStatus(srcFilePath);
				Path[] partPaths = new Path[contents.length];
				int numPartFiles = 0;
				for (int i = 0; i < contents.length; i++) {
					if (!contents[i].isDir()) {
						partPaths[i] = contents[i].getPath();
						numPartFiles++;
					}
				}
				Arrays.sort(partPaths);

				for (int i = 0; i < numPartFiles; i++) {
					InputStream in = hdfs.open(partPaths[i]);
					try {
						IOUtils.copyBytes(in, out, conf, false);
					} finally {
						in.close();
					}
				}
			} finally {
				out.close();
			}
		} else if (hdfs.isFile(srcFilePath)) {
			InputStream in = null;
			try {
				in = hdfs.open(srcFilePath);
				IOUtils.copyBytes(in, out, conf, true);
			} finally {
				in.close();
				out.close();
			}
		} else {
			throw new IOException(srcFilePath.toString()
					+ ": No such file or directory");
		}
	}
		
	/**
	 * 
	 * @param fileName
	 * @param src
	 * @param rlen
	 * @param clen
	 * @param nnz
	 * @throws IOException
	 */
	public static void writeCSVToHDFS( Path path, JobConf job, MatrixBlock src, long rlen, long clen, long nnz, FileFormatProperties formatProperties )
		throws IOException
	{
		boolean sparse = src.isInSparseFormat();
		FileSystem fs = FileSystem.get(job);
        BufferedWriter br=new BufferedWriter(new OutputStreamWriter(fs.create(path,true)));		
        
    	int rows = src.getNumRows();
		int cols = src.getNumColumns();

		//bound check per block
		if( rows > rlen || cols > clen )
		{
			throw new IOException("Matrix block [1:"+rows+",1:"+cols+"] " +
					              "out of overall matrix range [1:"+rlen+",1:"+clen+"].");
		}
		
		try
		{
			//for obj reuse and preventing repeated buffer re-allocations
			StringBuilder sb = new StringBuilder();
			
			CSVFileFormatProperties csvProperties = (CSVFileFormatProperties)formatProperties;
			String delim = csvProperties.getDelim(); //Pattern.quote(csvProperties.getDelim());
			
			// Write header line, if needed
			if (csvProperties.isHasHeader()) {
				for(int i=0; i < clen; i++) {
					sb.append("C"+ (i+1));
					if ( i < clen-1 )
						sb.append(delim);
				}
				sb.append('\n');
				br.write( sb.toString());
	            sb.setLength(0);
			}
			
			// Write data lines
			if( sparse ) //SPARSE
			{			   
	            int c, prev_c;
	            double v;
				for(int r=0; r < rows; r++) {
					SparseRow spRow = src.getSparseRows()[r];
					prev_c = -1;
					if ( spRow != null) {
						for(int ind=0; ind < spRow.size(); ind++) {
							c = spRow.getIndexContainer()[ind];
							v = spRow.getValueContainer()[ind];
							
							// output empty fields, if needed
							while(prev_c < c-1) {
								if (!csvProperties.isSparse()) {
									sb.append(0.0);
								}
								sb.append(delim);
								prev_c++;
							}
							
							// output the value
							sb.append(v);
							if ( c < clen-1)
								sb.append(delim);
							prev_c = c;
						}
					}
					// Output empty fields at the end of the row.
					// In case of an empty row, output (clen-1) empty fields
					while (prev_c < clen - 1) {
						if (!csvProperties.isSparse()) {
							sb.append(0.0);
						}
						prev_c++;
						if (prev_c < clen-1)
							sb.append(delim);
					}
					sb.append('\n');
					/*if ( sb.toString().split(Pattern.quote(delim)).length != clen) {
						throw new RuntimeException("row = " + r + ", prev_c = " + prev_c + ", filedcount=" + sb.toString().split(Pattern.quote(delim)).length + ": " + sb.toString());
					}*/
					br.write( sb.toString() ); 
					sb.setLength(0); 
				}
			}
			else //DENSE
			{
				for( int i=0; i<rows; i++ ) {
					for( int j=0; j<cols; j++ )
					{
						double lvalue = src.getValueDenseUnsafe(i, j);
						if( lvalue != 0 ) //for nnz
						{
							sb.append(lvalue);
						}
						else {
							if (!csvProperties.isSparse()) {
								// write in dense format
								sb.append(0.0);
							}
						}
						if ( j != cols-1 )
							sb.append(delim);
					}
					sb.append('\n');
					br.write( sb.toString() ); //same as append
					sb.setLength(0); 
				}
			}
		}
		finally
		{
			if( br != null )
				br.close();
		}
	}
	
	/**
	 * 
	 * @param path
	 * @param job
	 * @param src
	 * @param rlen
	 * @param clen
	 * @param brlen
	 * @param bclen
	 * @throws IOException
	 */
	private static void writeTextCellMatrixToHDFS( Path path, JobConf job, MatrixBlock src, long rlen, long clen, int brlen, int bclen )
		throws IOException
	{
		boolean sparse = src.isInSparseFormat();
		boolean entriesWritten = false;
		FileSystem fs = FileSystem.get(job);
        BufferedWriter br=new BufferedWriter(new OutputStreamWriter(fs.create(path,true)));		
        
    	int rows = src.getNumRows();
		int cols = src.getNumColumns();

		//bound check per block
		if( rows > rlen || cols > clen )
		{
			throw new IOException("Matrix block [1:"+rows+",1:"+cols+"] " +
					              "out of overall matrix range [1:"+rlen+",1:"+clen+"].");
		}
		
		try
		{
			//for obj reuse and preventing repeated buffer re-allocations
			StringBuilder sb = new StringBuilder();
			
			if( sparse ) //SPARSE
			{			   
				SparseCellIterator iter = src.getSparseCellIterator();
				while( iter.hasNext() )
				{
					IJV cell = iter.next();

					sb.append(cell.i+1);
					sb.append(' ');
					sb.append(cell.j+1);
					sb.append(' ');
					sb.append(cell.v);
					sb.append('\n');
					br.write( sb.toString() ); //same as append
					sb.setLength(0); 
					entriesWritten = true;					
				}
			}
			else //DENSE
			{
				for( int i=0; i<rows; i++ )
					for( int j=0; j<cols; j++ )
					{
						double lvalue = src.getValueDenseUnsafe(i, j);
						if( lvalue != 0 ) //for nnz
						{
							sb.append(i+1);
							sb.append(' ');
							sb.append(j+1);
							sb.append(' ');
							sb.append(lvalue);
							sb.append('\n');
							br.write( sb.toString() ); //same as append
							sb.setLength(0); 
							entriesWritten = true;
						}
					}
			}
	
			//handle empty result
			if ( !entriesWritten ) {
				br.write("1 1 0\n");
			}
		}
		finally
		{
			if( br != null )
				br.close();
		}
	}
	
	/**
	 * 
	 * @param path
	 * @param job
	 * @param src
	 * @param rlen
	 * @param clen
	 * @param brlen
	 * @param bclen
	 * @throws IOException
	 */
	private static void writeBinaryCellMatrixToHDFS( Path path, JobConf job, MatrixBlock src, long rlen, long clen, int brlen, int bclen )
		throws IOException
	{
		boolean sparse = src.isInSparseFormat();
		boolean entriesWritten = false;
		FileSystem fs = FileSystem.get(job);
		SequenceFile.Writer writer = new SequenceFile.Writer(fs, job, path, MatrixIndexes.class, MatrixCell.class);
		
		MatrixIndexes indexes = new MatrixIndexes();
		MatrixCell cell = new MatrixCell();

		int rows = src.getNumRows(); 
		int cols = src.getNumColumns();
        
		//bound check per block
		if( rows > rlen || cols > clen )
		{
			throw new IOException("Matrix block [1:"+rows+",1:"+cols+"] " +
					              "out of overall matrix range [1:"+rlen+",1:"+clen+"].");
		}
		
		try
		{
			if( sparse ) //SPARSE
			{
				SparseCellIterator iter = src.getSparseCellIterator();
				while( iter.hasNext() )
				{
					IJV lcell = iter.next();
					indexes.setIndexes(lcell.i+1, lcell.j+1);
					cell.setValue(lcell.v);
					writer.append(indexes, cell);
					entriesWritten = true;
				}
			}
			else //DENSE
			{
				for( int i=0; i<rows; i++ )
					for( int j=0; j<cols; j++ )
					{
						double lvalue  = src.getValueDenseUnsafe(i, j);
						if( lvalue != 0 ) //for nnz
						{
							indexes.setIndexes(i+1, j+1);
							cell.setValue(lvalue);
							writer.append(indexes, cell);
							entriesWritten = true;
						}
					}
			}
	
			//handle empty result
			if ( !entriesWritten ) {
				writer.append(new MatrixIndexes(1, 1), new MatrixCell(0));
			}
		}
		finally
		{
			if( writer != null )
				writer.close();
		}
	}
	
	/**
	 * 
	 * @param path
	 * @param job
	 * @param src
	 * @param rlen
	 * @param clen
	 * @param brlen
	 * @param bclen
	 * @throws IOException
	 * @throws DMLUnsupportedOperationException 
	 * @throws DMLRuntimeException 
	 */
	private static void writeBinaryBlockMatrixToHDFS( Path path, JobConf job, MatrixBlock src, long rlen, long clen, int brlen, int bclen, int replication )
		throws IOException, DMLRuntimeException, DMLUnsupportedOperationException
	{
		boolean sparse = src.isInSparseFormat();
		FileSystem fs = FileSystem.get(job);
		
		// 1) create sequence file writer, with right replication factor 
		// (config via 'dfs.replication' not possible since sequence file internally calls fs.getDefaultReplication())
		SequenceFile.Writer writer = null;
		if( replication > 0 ) //if replication specified (otherwise default)
		{
			//copy of SequenceFile.Writer(fs, job, path, MatrixIndexes.class, MatrixBlock.class), except for replication
			writer = new SequenceFile.Writer(fs, job, path, MatrixIndexes.class, MatrixBlock.class, job.getInt("io.file.buffer.size", 4096),  
					                         (short)replication, fs.getDefaultBlockSize(), null, new SequenceFile.Metadata());
		}
		else	
		{
			writer = new SequenceFile.Writer(fs, job, path, MatrixIndexes.class, MatrixBlock.class);
		}
		
		// 2) bound check for src block
		if( src.getNumRows() > rlen || src.getNumColumns() > clen )
		{
			throw new IOException("Matrix block [1:"+src.getNumRows()+",1:"+src.getNumColumns()+"] " +
					              "out of overall matrix range [1:"+rlen+",1:"+clen+"].");
		}
		
		//3) reblock and write
		try
		{
			MatrixIndexes indexes = new MatrixIndexes();

			if( rlen <= brlen && clen <= bclen ) //opt for single block
			{
				//directly write single block
				indexes.setIndexes(1, 1);
				writer.append(indexes, src);
			}
			else //general case
			{
				//initialize blocks for reuse (at most 4 different blocks required)
				MatrixBlock[] blocks = createMatrixBlocksForReuse(rlen, clen, brlen, bclen, sparse, src.getNonZeros());  
				
				//create and write subblocks of matrix
				for(int blockRow = 0; blockRow < (int)Math.ceil(src.getNumRows()/(double)brlen); blockRow++)
					for(int blockCol = 0; blockCol < (int)Math.ceil(src.getNumColumns()/(double)bclen); blockCol++)
					{
						int maxRow = (blockRow*brlen + brlen < src.getNumRows()) ? brlen : src.getNumRows() - blockRow*brlen;
						int maxCol = (blockCol*bclen + bclen < src.getNumColumns()) ? bclen : src.getNumColumns() - blockCol*bclen;
				
{'fraction_non_alphanumeric': 0.12797742551249067, 'fraction_numerical': 0.00406672752925554, 'mean_word_length': 2.846880786667518, 'pattern_counts': {'":': 6, '<': 74, '<?xml version=': 0, '>': 40, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7320089', 'n_tokens_mistral': 23769, 'n_tokens_neox': 19579, 'n_words': 5014}