protected void fillJoinCollector ( K iterkey ) throws IOException { final PriorityQueue < ComposableRecordReader < K , ? > > q = getRecordReaderQueue ( ) ; if ( ! q . isEmpty ( ) ) { int highpos = - 1 ; ArrayList < ComposableRecordReader < K , ? > > list = new ArrayList < ComposableRecordReader < K , ? > > ( kids . length ) ; q . peek ( ) . key ( iterkey ) ; final WritableComparator cmp = getComparator ( ) ; while ( 0 == cmp . compare ( q . peek ( ) . key ( ) , iterkey ) ) { ComposableRecordReader < K , ? > t = q . poll ( ) ; if ( - 1 == highpos || list . get ( highpos ) . id ( ) < t . id ( ) ) { highpos = list . size ( ) ; } list . add ( t ) ; if ( q . isEmpty ( ) ) break ; } ComposableRecordReader < K , ? > t = list . remove ( highpos ) ; t . accept ( jc , iterkey ) ; for ( ComposableRecordReader < K , ? > rr : list ) { rr . skip ( iterkey ) ; } list . add ( t ) ; for ( ComposableRecordReader < K , ? > rr : list ) { if ( rr . hasNext ( ) ) { q . add ( rr ) ; } } } }
Instead of filling the JoinCollector with iterators from all data sources fill only the rightmost for this key . This not only saves space by discarding the other sources but it also emits the number of key - value pairs in the preferred RecordReader instead of repeating that stream n times where n is the cardinality of the cross product of the discarded streams for the given key .