public static MatrixChar_N_ReducerGroups computeMatrixCharacteristics(JobConf job,byte[] inputIndexes,String dataGenInstructions,String instructionsInMapper,String reblockInstructions,String aggInstructionsInReducer,String aggBinInstructions,String otherInstructionsInReducer,byte[] resultIndexes,HashSet<Byte> mapOutputIndexes,boolean forMMCJ) throws DMLRuntimeException {
  HashSet<Byte> intermediateMatrixIndexes=new HashSet<Byte>();
  HashMap<Byte,MatrixCharacteristics> dims=new HashMap<Byte,MatrixCharacteristics>();
  for (  byte i : inputIndexes) {
    MatrixCharacteristics dim=new MatrixCharacteristics(getNumRows(job,i),getNumColumns(job,i),getNumRowsPerBlock(job,i),getNumColumnsPerBlock(job,i),getNumNonZero(job,i));
    dims.put(i,dim);
  }
  DataGenMRInstruction[] dataGenIns=null;
  dataGenIns=MRInstructionParser.parseDataGenInstructions(dataGenInstructions);
  if (dataGenIns != null) {
    for (    DataGenMRInstruction ins : dataGenIns) {
      MatrixCharacteristics.computeDimension(dims,ins);
    }
  }
  MRInstruction[] insMapper=MRInstructionParser.parseMixedInstructions(instructionsInMapper);
  if (insMapper != null) {
    for (    MRInstruction ins : insMapper) {
      MatrixCharacteristics.computeDimension(dims,ins);
      if (ins instanceof UnaryMRInstructionBase) {
        UnaryMRInstructionBase tempIns=(UnaryMRInstructionBase)ins;
        setIntermediateMatrixCharactristics(job,tempIns.input,dims.get(tempIns.input));
        intermediateMatrixIndexes.add(tempIns.input);
      }
 else       if (ins instanceof AppendMInstruction) {
        AppendMInstruction tempIns=(AppendMInstruction)ins;
        setIntermediateMatrixCharactristics(job,tempIns.input1,dims.get(tempIns.input1));
        intermediateMatrixIndexes.add(tempIns.input1);
      }
 else       if (ins instanceof AppendGInstruction) {
        AppendGInstruction tempIns=(AppendGInstruction)ins;
        setIntermediateMatrixCharactristics(job,tempIns.input1,dims.get(tempIns.input1));
        intermediateMatrixIndexes.add(tempIns.input1);
      }
 else       if (ins instanceof BinaryMInstruction) {
        BinaryMInstruction tempIns=(BinaryMInstruction)ins;
        setIntermediateMatrixCharactristics(job,tempIns.input1,dims.get(tempIns.input1));
        intermediateMatrixIndexes.add(tempIns.input1);
      }
 else       if (ins instanceof AggregateBinaryInstruction) {
        AggregateBinaryInstruction tempIns=(AggregateBinaryInstruction)ins;
        setIntermediateMatrixCharactristics(job,tempIns.input1,dims.get(tempIns.input1));
        intermediateMatrixIndexes.add(tempIns.input1);
      }
 else       if (ins instanceof MapMultChainInstruction) {
        MapMultChainInstruction tempIns=(MapMultChainInstruction)ins;
        setIntermediateMatrixCharactristics(job,tempIns.getInput1(),dims.get(tempIns.getInput2()));
        intermediateMatrixIndexes.add(tempIns.getInput1());
      }
 else       if (ins instanceof PMMJMRInstruction) {
        PMMJMRInstruction tempIns=(PMMJMRInstruction)ins;
        setIntermediateMatrixCharactristics(job,tempIns.input2,dims.get(tempIns.input2));
        intermediateMatrixIndexes.add(tempIns.input2);
      }
    }
  }
  ReblockInstruction[] reblockIns=MRInstructionParser.parseReblockInstructions(reblockInstructions);
  if (reblockIns != null) {
    for (    ReblockInstruction ins : reblockIns) {
      MatrixCharacteristics.computeDimension(dims,ins);
      setMatrixCharactristicsForReblock(job,ins.output,dims.get(ins.output));
    }
  }
  Instruction[] aggIns=MRInstructionParser.parseAggregateInstructions(aggInstructionsInReducer);
  if (aggIns != null) {
    for (    Instruction ins : aggIns) {
      MatrixCharacteristics.computeDimension(dims,(MRInstruction)ins);
      MRInstruction mrins=(MRInstruction)ins;
      boolean found=false;
      for (      byte b : resultIndexes) {
        if (b == mrins.output) {
          found=true;
          break;
        }
      }
      if (!found) {
        setIntermediateMatrixCharactristics(job,mrins.output,dims.get(mrins.output));
        intermediateMatrixIndexes.add(mrins.output);
      }
    }
  }
  long numReduceGroups=0;
  AggregateBinaryInstruction[] aggBinIns=getAggregateBinaryInstructions(job);
  if (aggBinIns != null) {
    for (    AggregateBinaryInstruction ins : aggBinIns) {
      MatrixCharacteristics dim1=dims.get(ins.input1);
      MatrixCharacteristics dim2=dims.get(ins.input2);
      setMatrixCharactristicsForBinAgg(job,ins.input1,dim1);
      setMatrixCharactristicsForBinAgg(job,ins.input2,dim2);
      MatrixCharacteristics.computeDimension(dims,ins);
      if (forMMCJ)       numReduceGroups=(long)Math.ceil((double)dim1.getCols() / (double)dim1.getColsPerBlock());
    }
  }
  if (!forMMCJ) {
    ArrayList<Long> xs=new ArrayList<Long>(mapOutputIndexes.size());
    ArrayList<Long> ys=new ArrayList<Long>(mapOutputIndexes.size());
    for (    byte idx : mapOutputIndexes) {
      MatrixCharacteristics dim=dims.get(idx);
      long x=(long)Math.ceil((double)dim.getRows() / (double)dim.getRowsPerBlock());
      long y=(long)Math.ceil((double)dim.getCols() / (double)dim.getColsPerBlock());
      int i=0;
      boolean toadd=true;
      while (i < xs.size()) {
        if ((x >= xs.get(i) && y > ys.get(i)) || (x > xs.get(i) && y >= ys.get(i))) {
          xs.remove(i);
          ys.remove(i);
        }
 else         if (x <= xs.get(i) && y <= ys.get(i)) {
          toadd=false;
          break;
        }
 else         i++;
      }
      if (toadd) {
        xs.add(x);
        ys.add(y);
      }
    }
    TreeMap<Long,Long> map=new TreeMap<Long,Long>();
    for (int i=0; i < xs.size(); i++)     map.put(xs.get(i),ys.get(i));
    numReduceGroups=0;
    long prev=0;
    for (    Entry<Long,Long> e : map.entrySet()) {
      numReduceGroups+=(e.getKey() - prev) * e.getValue();
      prev=e.getKey();
    }
  }
  MRInstruction[] insReducer=MRInstructionParser.parseMixedInstructions(otherInstructionsInReducer);
  if (insReducer != null) {
    for (    MRInstruction ins : insReducer) {
      MatrixCharacteristics.computeDimension(dims,ins);
      if (ins instanceof UnaryMRInstructionBase) {
        UnaryMRInstructionBase tempIns=(UnaryMRInstructionBase)ins;
        setIntermediateMatrixCharactristics(job,tempIns.input,dims.get(tempIns.input));
        intermediateMatrixIndexes.add(tempIns.input);
      }
 else       if (ins instanceof RemoveEmptyMRInstruction) {
        RemoveEmptyMRInstruction tempIns=(RemoveEmptyMRInstruction)ins;
        setIntermediateMatrixCharactristics(job,tempIns.input1,dims.get(tempIns.input1));
        intermediateMatrixIndexes.add(tempIns.input1);
      }
      boolean found=false;
      for (      byte b : resultIndexes) {
        if (b == ins.output) {
          found=true;
          break;
        }
      }
      if (!found) {
        setIntermediateMatrixCharactristics(job,ins.output,dims.get(ins.output));
        intermediateMatrixIndexes.add(ins.output);
      }
    }
  }
  setIntermediateMatrixIndexes(job,intermediateMatrixIndexes);
  for (  byte tag : mapOutputIndexes)   setMatrixCharactristicsForMapperOutput(job,tag,dims.get(tag));
  MatrixCharacteristics[] stats=new MatrixCharacteristics[resultIndexes.length];
  MatrixCharacteristics resultDims;
  for (int i=0; i < resultIndexes.length; i++) {
    resultDims=dims.get(resultIndexes[i]);
    stats[i]=resultDims;
    setMatrixCharactristicsForOutput(job,resultIndexes[i],stats[i]);
  }
  return new MatrixChar_N_ReducerGroups(stats,numReduceGroups);
}

public static MatrixChar_N_ReducerGroups computeMatrixCharacteristics ( JobConf job , byte [ ] inputIndexes , String dataGenInstructions , String instructionsInMapper , String reblockInstructions , String aggInstructionsInReducer , String aggBinInstructions , String otherInstructionsInReducer , byte [ ] resultIndexes , HashSet < Byte > mapOutputIndexes , boolean forMMCJ ) throws DMLRuntimeException { HashSet < Byte > intermediateMatrixIndexes = new HashSet < Byte > ( ) ; HashMap < Byte , MatrixCharacteristics > dims = new HashMap < Byte , MatrixCharacteristics > ( ) ; for ( byte i : inputIndexes ) { MatrixCharacteristics dim = new MatrixCharacteristics ( getNumRows ( job , i ) , getNumColumns ( job , i ) , getNumRowsPerBlock ( job , i ) , getNumColumnsPerBlock ( job , i ) , getNumNonZero ( job , i ) ) ; dims . put ( i , dim ) ; } DataGenMRInstruction [ ] dataGenIns = null ; dataGenIns = MRInstructionParser . parseDataGenInstructions ( dataGenInstructions ) ; if ( dataGenIns != null ) { for ( DataGenMRInstruction ins : dataGenIns ) { MatrixCharacteristics . computeDimension ( dims , ins ) ; } } MRInstruction [ ] insMapper = MRInstructionParser . parseMixedInstructions ( instructionsInMapper ) ; if ( insMapper != null ) { for ( MRInstruction ins : insMapper ) { MatrixCharacteristics . computeDimension ( dims , ins ) ; if ( ins instanceof UnaryMRInstructionBase ) { UnaryMRInstructionBase tempIns = ( UnaryMRInstructionBase ) ins ; setIntermediateMatrixCharactristics ( job , tempIns . input , dims . get ( tempIns . input ) ) ; intermediateMatrixIndexes . add ( tempIns . input ) ; } else if ( ins instanceof AppendMInstruction ) { AppendMInstruction tempIns = ( AppendMInstruction ) ins ; setIntermediateMatrixCharactristics ( job , tempIns . input1 , dims . get ( tempIns . input1 ) ) ; intermediateMatrixIndexes . add ( tempIns . input1 ) ; } else if ( ins instanceof AppendGInstruction ) { AppendGInstruction tempIns = ( AppendGInstruction ) ins ; setIntermediateMatrixCharactristics ( job , tempIns . input1 , dims . get ( tempIns . input1 ) ) ; intermediateMatrixIndexes . add ( tempIns . input1 ) ; } else if ( ins instanceof BinaryMInstruction ) { BinaryMInstruction tempIns = ( BinaryMInstruction ) ins ; setIntermediateMatrixCharactristics ( job , tempIns . input1 , dims . get ( tempIns . input1 ) ) ; intermediateMatrixIndexes . add ( tempIns . input1 ) ; } else if ( ins instanceof AggregateBinaryInstruction ) { AggregateBinaryInstruction tempIns = ( AggregateBinaryInstruction ) ins ; setIntermediateMatrixCharactristics ( job , tempIns . input1 , dims . get ( tempIns . input1 ) ) ; intermediateMatrixIndexes . add ( tempIns . input1 ) ; } else if ( ins instanceof MapMultChainInstruction ) { MapMultChainInstruction tempIns = ( MapMultChainInstruction ) ins ; setIntermediateMatrixCharactristics ( job , tempIns . getInput1 ( ) , dims . get ( tempIns . getInput2 ( ) ) ) ; intermediateMatrixIndexes . add ( tempIns . getInput1 ( ) ) ; } else if ( ins instanceof PMMJMRInstruction ) { PMMJMRInstruction tempIns = ( PMMJMRInstruction ) ins ; setIntermediateMatrixCharactristics ( job , tempIns . input2 , dims . get ( tempIns . input2 ) ) ; intermediateMatrixIndexes . add ( tempIns . input2 ) ; } } } ReblockInstruction [ ] reblockIns = MRInstructionParser . parseReblockInstructions ( reblockInstructions ) ; if ( reblockIns != null ) { for ( ReblockInstruction ins : reblockIns ) { MatrixCharacteristics . computeDimension ( dims , ins ) ; setMatrixCharactristicsForReblock ( job , ins . output , dims . get ( ins . output ) ) ; } } Instruction [ ] aggIns = MRInstructionParser . parseAggregateInstructions ( aggInstructionsInReducer ) ; if ( aggIns != null ) { for ( Instruction ins : aggIns ) { MatrixCharacteristics . computeDimension ( dims , ( MRInstruction ) ins ) ; MRInstruction mrins = ( MRInstruction ) ins ; boolean found = false ; for ( byte b : resultIndexes ) { if ( b == mrins . output ) { found = true ; break ; } } if ( ! found ) { setIntermediateMatrixCharactristics ( job , mrins . output , dims . get ( mrins . output ) ) ; intermediateMatrixIndexes . add ( mrins . output ) ; } } } long numReduceGroups = 0 ; AggregateBinaryInstruction [ ] aggBinIns = getAggregateBinaryInstructions ( job ) ; if ( aggBinIns != null ) { for ( AggregateBinaryInstruction ins : aggBinIns ) { MatrixCharacteristics dim1 = dims . get ( ins . input1 ) ; MatrixCharacteristics dim2 = dims . get ( ins . input2 ) ; setMatrixCharactristicsForBinAgg ( job , ins . input1 , dim1 ) ; setMatrixCharactristicsForBinAgg ( job , ins . input2 , dim2 ) ; MatrixCharacteristics . computeDimension ( dims , ins ) ; if ( forMMCJ ) numReduceGroups = ( long ) Math . ceil ( ( double ) dim1 . getCols ( ) / ( double ) dim1 . getColsPerBlock ( ) ) ; } } if ( ! forMMCJ ) { ArrayList < Long > xs = new ArrayList < Long > ( mapOutputIndexes . size ( ) ) ; ArrayList < Long > ys = new ArrayList < Long > ( mapOutputIndexes . size ( ) ) ; for ( byte idx : mapOutputIndexes ) { MatrixCharacteristics dim = dims . get ( idx ) ; long x = ( long ) Math . ceil ( ( double ) dim . getRows ( ) / ( double ) dim . getRowsPerBlock ( ) ) ; long y = ( long ) Math . ceil ( ( double ) dim . getCols ( ) / ( double ) dim . getColsPerBlock ( ) ) ; int i = 0 ; boolean toadd = true ; while ( i < xs . size ( ) ) { if ( ( x >= xs . get ( i ) && y > ys . get ( i ) ) || ( x > xs . get ( i ) && y >= ys . get ( i ) ) ) { xs . remove ( i ) ; ys . remove ( i ) ; } else if ( x <= xs . get ( i ) && y <= ys . get ( i ) ) { toadd = false ; break ; } else i ++ ; } if ( toadd ) { xs . add ( x ) ; ys . add ( y ) ; } } TreeMap < Long , Long > map = new TreeMap < Long , Long > ( ) ; for ( int i = 0 ; i < xs . size ( ) ; i ++ ) map . put ( xs . get ( i ) , ys . get ( i ) ) ; numReduceGroups = 0 ; long prev = 0 ; for ( Entry < Long , Long > e : map . entrySet ( ) ) { numReduceGroups += ( e . getKey ( ) - prev ) * e . getValue ( ) ; prev = e . getKey ( ) ; } } MRInstruction [ ] insReducer = MRInstructionParser . parseMixedInstructions ( otherInstructionsInReducer ) ; if ( insReducer != null ) { for ( MRInstruction ins : insReducer ) { MatrixCharacteristics . computeDimension ( dims , ins ) ; if ( ins instanceof UnaryMRInstructionBase ) { UnaryMRInstructionBase tempIns = ( UnaryMRInstructionBase ) ins ; setIntermediateMatrixCharactristics ( job , tempIns . input , dims . get ( tempIns . input ) ) ; intermediateMatrixIndexes . add ( tempIns . input ) ; } else if ( ins instanceof RemoveEmptyMRInstruction ) { RemoveEmptyMRInstruction tempIns = ( RemoveEmptyMRInstruction ) ins ; setIntermediateMatrixCharactristics ( job , tempIns . input1 , dims . get ( tempIns . input1 ) ) ; intermediateMatrixIndexes . add ( tempIns . input1 ) ; } boolean found = false ; for ( byte b : resultIndexes ) { if ( b == ins . output ) { found = true ; break ; } } if ( ! found ) { setIntermediateMatrixCharactristics ( job , ins . output , dims . get ( ins . output ) ) ; intermediateMatrixIndexes . add ( ins . output ) ; } } } setIntermediateMatrixIndexes ( job , intermediateMatrixIndexes ) ; for ( byte tag : mapOutputIndexes ) setMatrixCharactristicsForMapperOutput ( job , tag , dims . get ( tag ) ) ; MatrixCharacteristics [ ] stats = new MatrixCharacteristics [ resultIndexes . length ] ; MatrixCharacteristics resultDims ; for ( int i = 0 ; i < resultIndexes . length ; i ++ ) { resultDims = dims . get ( resultIndexes [ i ] ) ; stats [ i ] = resultDims ; setMatrixCharactristicsForOutput ( job , resultIndexes [ i ] , stats [ i ] ) ; } return new MatrixChar_N_ReducerGroups ( stats , numReduceGroups ) ; }
note : this method needs to be in - sync with mrbaseforcommoninstructions . processoneinstruction , otherwise , the latter will potentially fail with missing dimension information .
