<commit_before>package dml.lops.runtime;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.SequenceFile;

import dml.api.DMLScript;
import dml.api.DMLScript.RUNTIME_PLATFORM;
import dml.lops.Lops;
import dml.lops.compile.JobType;
import dml.meta.PartitionBlockHashMapMR;
import dml.meta.PartitionBlockJoinMR;
import dml.meta.PartitionBlockMR;
import dml.meta.PartitionCellMR;
import dml.meta.PartitionParams;
import dml.meta.PartitionSubMatrixMR;
import dml.parser.DMLTranslator;
import dml.parser.Expression.DataType;
import dml.runtime.controlprogram.ProgramBlock;
import dml.runtime.instructions.MRJobInstruction;
import dml.runtime.instructions.CPInstructions.Data;
import dml.runtime.instructions.CPInstructions.MatrixObject;
import dml.runtime.instructions.CPInstructions.ScalarObject;
import dml.runtime.matrix.CMCOVMR;
import dml.runtime.matrix.CombineMR;
import dml.runtime.matrix.GMR;
import dml.runtime.matrix.GroupedAggMR;
import dml.runtime.matrix.JobReturn;
import dml.runtime.matrix.MMCJMR;
import dml.runtime.matrix.MMRJMR;
import dml.runtime.matrix.MatrixCharacteristics;
import dml.runtime.matrix.MatrixDimensionsMetaData;
import dml.runtime.matrix.RandMR;
import dml.runtime.matrix.ReblockMR;
import dml.runtime.matrix.SortMR;
import dml.runtime.matrix.io.InputInfo;
import dml.runtime.matrix.io.MatrixCell;
import dml.runtime.matrix.io.MatrixIndexes;
import dml.runtime.matrix.io.OutputInfo;
import dml.runtime.util.MapReduceTool;
import dml.utils.DMLRuntimeException;
import dml.utils.configuration.DMLConfig;

public class RunMRJobs {
	public static boolean flagLocalModeOpt = true;
	public enum ExecMode { LOCAL, CLUSTER, INVALID }; 

	public static JobReturn submitJob(MRJobInstruction inst, ProgramBlock pb ) throws DMLRuntimeException {
		JobReturn ret = new JobReturn();

		if ( DMLScript.DEBUG  )
			System.out.println(inst.toString());

		// Check if any of the input files are empty.. only for those job types
		// for which empty inputs are NOT allowed
		if (!inst.getJobType().areEmptyInputsAllowed()) {
			for (String f : updateLabels(inst.getIv_inputs(), inst.getInputLabelValueMapping())) {
				try {
					if (MapReduceTool.isHDFSFileEmpty(f)) {
						throw new DMLRuntimeException("Can not operate on an empty file: " + f);
					}
				} catch (IOException e) {
					throw new DMLRuntimeException(e);
				}
			}
		}

		try {
			String[] updatedInputLabels = null;
			String[] updatedOutputLabels = null;
			long[] updatedRows = null;
			long[] updatedCols = null;
			int[] updatedRowsPerBlock = null;
			int[] updatedColsPerBlock = null;
			
			/*
			 * Update placeholder labels i.e., the ones of the form ##<label>##
			 * Update rows and columns that are set in piggybacking by 
			 *   looking at the metadata structure (_matrices) in program block 
			 */
			if ( inst.getIv_inputs() != null )
				updatedInputLabels = updateLabels(inst.getIv_inputs(), inst.getInputLabelValueMapping());
			if ( inst.getIv_outputs() != null )
				updatedOutputLabels = updateLabels(inst.getIv_outputs(), inst.getOutputLabelValueMapping());
			if ( inst.getIv_rows() != null && updatedInputLabels != null )
				updatedRows = updateRows(inst.getIv_rows(), updatedInputLabels, inst.getInputLabels(), pb);
			if ( inst.getIv_cols() != null && updatedInputLabels != null )
				updatedCols = updateCols(inst.getIv_cols(), updatedInputLabels, inst.getInputLabels(), pb);
			if ( inst.getIv_num_rows_per_block() != null && updatedInputLabels != null )
				updatedRowsPerBlock = updateRowsPerBlock(inst.getIv_num_rows_per_block(), updatedInputLabels, inst.getInputLabels(), pb);
			if ( inst.getIv_num_cols_per_block() != null && updatedInputLabels != null )
				updatedColsPerBlock = updateColsPerBlock(inst.getIv_num_cols_per_block(), updatedInputLabels, inst.getInputLabels(), pb);
			
			if (inst.getJobType() == JobType.GMR) {
				boolean blocked_rep = true;
				
				// If any input value class is Cell then the whole job should run in "Cell" mode
				// Remaining blocked input is implicitly converted into cell format
				for ( InputInfo ii : inst.getIv_inputInfos()) {
					if ( ii == InputInfo.TextCellInputInfo || ii == InputInfo.BinaryCellInputInfo ) {
						blocked_rep = false;
						break;
					}
				}
				
			/*	//TODO: added by Yuanyuan
				for(int i=0; i<inst.getIv_num_cols_per_block().length; i++)
				{
					if(updatedRowsPerBlock[i]==1 && updatedColsPerBlock[i]==1)
					{
						blocked_rep=false;
						if(inst.getIv_inputInfos()[i]== InputInfo.BinaryBlockInputInfo)
							inst.getIv_inputInfos()[i]=InputInfo.BinaryCellInputInfo;
					}
				}*/
				
				// TODO: statiko -- how to remove this check?
				if ( blocked_rep == false ) {
					// Check if any outputs are of type BinaryBlock!
					// output info must not be blocked
					for ( int oi=0; oi < inst.getIv_outputs().length; oi++ ) {
						if ( inst.getIv_outputInfos()[oi] == OutputInfo.BinaryBlockOutputInfo) {
							inst.getIv_outputInfos()[oi] = OutputInfo.BinaryCellOutputInfo;
						}
					}
					
			/*		//TODO: added by Yuanyuan
					for(int i=0; i<inst.getIv_inputInfos().length; i++)
					{
						updatedRowsPerBlock[i]=1;
						updatedColsPerBlock[i]=1;
					}*/
				}
				
				if ( !inst.getIv_recordReaderInstructions().equals("") ) {
					// if there are record reader instructions, we need to update MetaData
					
					// In the presence of recordReader instructions (valuepick/rangepick), the job must operate in cell mode
					blocked_rep = false;
					
					// output info must not be blocked
					for ( int oi=0; oi < inst.getIv_outputs().length; oi++ ) {
						if ( inst.getIv_outputInfos()[oi] == OutputInfo.BinaryBlockOutputInfo) {<commit_msg>21523: Open this task to track/reflect the TODO's noticed in dml.lops.runtime.RunMRJobs.java - removed unused code<commit_after>package dml.lops.runtime;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.SequenceFile;

import dml.api.DMLScript;
import dml.api.DMLScript.RUNTIME_PLATFORM;
import dml.lops.Lops;
import dml.lops.compile.JobType;
import dml.meta.PartitionBlockHashMapMR;
import dml.meta.PartitionBlockJoinMR;
import dml.meta.PartitionBlockMR;
import dml.meta.PartitionCellMR;
import dml.meta.PartitionParams;
import dml.meta.PartitionSubMatrixMR;
import dml.parser.DMLTranslator;
import dml.parser.Expression.DataType;
import dml.runtime.controlprogram.ProgramBlock;
import dml.runtime.instructions.MRJobInstruction;
import dml.runtime.instructions.CPInstructions.Data;
import dml.runtime.instructions.CPInstructions.MatrixObject;
import dml.runtime.instructions.CPInstructions.ScalarObject;
import dml.runtime.matrix.CMCOVMR;
import dml.runtime.matrix.CombineMR;
import dml.runtime.matrix.GMR;
import dml.runtime.matrix.GroupedAggMR;
import dml.runtime.matrix.JobReturn;
import dml.runtime.matrix.MMCJMR;
import dml.runtime.matrix.MMRJMR;
import dml.runtime.matrix.MatrixCharacteristics;
import dml.runtime.matrix.MatrixDimensionsMetaData;
import dml.runtime.matrix.RandMR;
import dml.runtime.matrix.ReblockMR;
import dml.runtime.matrix.SortMR;
import dml.runtime.matrix.io.InputInfo;
import dml.runtime.matrix.io.MatrixCell;
import dml.runtime.matrix.io.MatrixIndexes;
import dml.runtime.matrix.io.OutputInfo;
import dml.runtime.util.MapReduceTool;
import dml.utils.DMLRuntimeException;
import dml.utils.configuration.DMLConfig;

public class RunMRJobs {
	public static boolean flagLocalModeOpt = true;
	public enum ExecMode { LOCAL, CLUSTER, INVALID }; 

	public static JobReturn submitJob(MRJobInstruction inst, ProgramBlock pb ) throws DMLRuntimeException {
		JobReturn ret = new JobReturn();

		if ( DMLScript.DEBUG  )
			System.out.println(inst.toString());

		// Check if any of the input files are empty.. only for those job types
		// for which empty inputs are NOT allowed
		if (!inst.getJobType().areEmptyInputsAllowed()) {
			for (String f : updateLabels(inst.getIv_inputs(), inst.getInputLabelValueMapping())) {
				try {
					if (MapReduceTool.isHDFSFileEmpty(f)) {
						throw new DMLRuntimeException("Can not operate on an empty file: " + f);
					}
				} catch (IOException e) {
					throw new DMLRuntimeException(e);
				}
			}
		}

		try {
			String[] updatedInputLabels = null;
			String[] updatedOutputLabels = null;
			long[] updatedRows = null;
			long[] updatedCols = null;
			int[] updatedRowsPerBlock = null;
			int[] updatedColsPerBlock = null;
			
			/*
			 * Update placeholder labels i.e., the ones of the form ##<label>##
			 * Update rows and columns that are set in piggybacking by 
			 *   looking at the metadata structure (_matrices) in program block 
			 */
			if ( inst.getIv_inputs() != null )
				updatedInputLabels = updateLabels(inst.getIv_inputs(), inst.getInputLabelValueMapping());
			if ( inst.getIv_outputs() != null )
				updatedOutputLabels = updateLabels(inst.getIv_outputs(), inst.getOutputLabelValueMapping());
			if ( inst.getIv_rows() != null && updatedInputLabels != null )
				updatedRows = updateRows(inst.getIv_rows(), updatedInputLabels, inst.getInputLabels(), pb);
			if ( inst.getIv_cols() != null && updatedInputLabels != null )
				updatedCols = updateCols(inst.getIv_cols(), updatedInputLabels, inst.getInputLabels(), pb);
			if ( inst.getIv_num_rows_per_block() != null && updatedInputLabels != null )
				updatedRowsPerBlock = updateRowsPerBlock(inst.getIv_num_rows_per_block(), updatedInputLabels, inst.getInputLabels(), pb);
			if ( inst.getIv_num_cols_per_block() != null && updatedInputLabels != null )
				updatedColsPerBlock = updateColsPerBlock(inst.getIv_num_cols_per_block(), updatedInputLabels, inst.getInputLabels(), pb);
			
			if (inst.getJobType() == JobType.GMR) {
				boolean blocked_rep = true;
				
				// If any input value class is Cell then the whole job should run in "Cell" mode
				// Remaining blocked input is implicitly converted into cell format
				for ( InputInfo ii : inst.getIv_inputInfos()) {
					if ( ii == InputInfo.TextCellInputInfo || ii == InputInfo.BinaryCellInputInfo ) {
						blocked_rep = false;
						break;
					}
				}
				
				// TODO: statiko -- how to remove this check?
				if ( blocked_rep == false ) {
					// Check if any outputs are of type BinaryBlock!
					// output info must not be blocked
					for ( int oi=0; oi < inst.getIv_outputs().length; oi++ ) {
						if ( inst.getIv_outputInfos()[oi] == OutputInfo.BinaryBlockOutputInfo) {
							inst.getIv_outputInfos()[oi] = OutputInfo.BinaryCellOutputInfo;
						}
					}
				}
				
				if ( !inst.getIv_recordReaderInstructions().equals("") ) {
					// if there are record reader instructions, we need to update MetaData
					
					// In the presence of recordReader instructions (valuepick/rangepick), the job must operate in cell mode
					blocked_rep = false;
					
					// output info must not be blocked
					for ( int oi=0; oi < inst.getIv_outputs().length; oi++ ) {
						if ( inst.getIv_outputInfos()[oi] == OutputInfo.BinaryBlockOutputInfo) {
{'fraction_non_alphanumeric': 0.10967063770147162, 'fraction_numerical': 0.0013139453398738612, 'mean_word_length': 4.122027815163751, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '246849', 'n_tokens_mistral': 4118, 'n_tokens_neox': 3587, 'n_words': 819}