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 .