code: 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);
          }
        }
      }
    }
docs: 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.