question: 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 ) ; }
target:  NOTE: this method needs to be in-sync with MRBaseForCommonInstructions.processOneInstruction, otherwise, the latter will potentially fail with missing dimension information.