private void writeOutData() {
    if (valuesHeld == 0) {
      return;
    }
    int batchSize = batchItemSizePerEmmit == null ? DEFAULT_SORT_BATCH_PER_EMIT_BYTES
        : batchItemSizePerEmmit;
    ByteBuffer currentKey = getKeyValueFromPointer(0).getKey();
    List<ByteBuffer> currentValues = new ArrayList<>();
    int totalSize = 0;

    for (int i = 0; i < valuesHeld; i++) {
      KeyValue<ByteBuffer, ByteBuffer> keyValue = getKeyValueFromPointer(i);
      int compare = LexicographicalComparator.compareBuffers(keyValue.getKey(), currentKey);

      if (compare == 0) {
        if (!currentValues.isEmpty() && totalSize >= batchSize) {
          emitCurrentOrLeftover(currentKey, currentValues);
          totalSize = 0;
        }
        currentValues.add(keyValue.getValue());
        totalSize += Math.max(1, keyValue.getValue().remaining());
      } else if (compare > 0) {
        emitCurrentOrLeftover(currentKey, currentValues);
        currentKey = keyValue.getKey();
        currentValues.add(keyValue.getValue());
        totalSize = Math.max(1, keyValue.getValue().remaining());
      } else {
        throw new IllegalStateException("Sort failed to properly order output");
      }
    }
    emitCurrentOrLeftover(currentKey, currentValues);
    if (leftover != null) {
      emit(leftover.getKey(), leftover.getValue());
    }
  }
private void writeOutData ( ) { if ( valuesHeld == 0 ) { return ; } int batchSize = batchItemSizePerEmmit == null ? DEFAULT_SORT_BATCH_PER_EMIT_BYTES : batchItemSizePerEmmit ; ByteBuffer currentKey = getKeyValueFromPointer ( 0 ) . getKey ( ) ; List < ByteBuffer > currentValues = new ArrayList < > ( ) ; int totalSize = 0 ; for ( int i = 0 ; i < valuesHeld ; i ++ ) { KeyValue < ByteBuffer , ByteBuffer > keyValue = getKeyValueFromPointer ( i ) ; int compare = LexicographicalComparator . compareBuffers ( keyValue . getKey ( ) , currentKey ) ; if ( compare == 0 ) { if ( ! currentValues . isEmpty ( ) && totalSize >= batchSize ) { emitCurrentOrLeftover ( currentKey , currentValues ) ; totalSize = 0 ; } currentValues . add ( keyValue . getValue ( ) ) ; totalSize += Math . max ( 1 , keyValue . getValue ( ) . remaining ( ) ) ; } else if ( compare > 0 ) { emitCurrentOrLeftover ( currentKey , currentValues ) ; currentKey = keyValue . getKey ( ) ; currentValues . add ( keyValue . getValue ( ) ) ; totalSize = Math . max ( 1 , keyValue . getValue ( ) . remaining ( ) ) ; } else { throw new IllegalStateException ( "Sort failed to properly order output" ) ; } } emitCurrentOrLeftover ( currentKey , currentValues ) ; if ( leftover != null ) { emit ( leftover . getKey ( ) , leftover . getValue ( ) ) ; } }
Writes out the key value pairs in order.
If there are multiple consecutive values with the same key, they can be combined to avoid
repeating the key.
In the event the buffer is full, there is one leftover item which did not go into it,
and hence was not sorted. So a merge between this one item and the sorted list is done on the
way out.