private void uncompressSingleKeyValue(DataInputStream source, ByteBuffer out,
    FastDiffCompressionState state) throws IOException, EncoderBufferTooSmallException {
    byte flag = source.readByte();
    int prevKeyLength = state.keyLength;

    if ((flag & FLAG_SAME_KEY_LENGTH) == 0) {
      state.keyLength = ByteBufferUtils.readCompressedInt(source);
    }
    if ((flag & FLAG_SAME_VALUE_LENGTH) == 0) {
      state.valueLength = ByteBufferUtils.readCompressedInt(source);
    }
    int commonLength = ByteBufferUtils.readCompressedInt(source);

    ensureSpace(out, state.keyLength + state.valueLength + KeyValue.ROW_OFFSET);

    int kvPos = out.position();

    if (!state.isFirst()) {
      // copy the prefix
      int common;
      int prevOffset;

      if ((flag & FLAG_SAME_VALUE_LENGTH) == 0) {
        out.putInt(state.keyLength);
        out.putInt(state.valueLength);
        prevOffset = state.prevOffset + KeyValue.ROW_OFFSET;
        common = commonLength;
      } else {
        if ((flag & FLAG_SAME_KEY_LENGTH) != 0) {
          prevOffset = state.prevOffset;
          common = commonLength + KeyValue.ROW_OFFSET;
        } else {
          out.putInt(state.keyLength);
          prevOffset = state.prevOffset + KeyValue.KEY_LENGTH_SIZE;
          common = commonLength + KeyValue.KEY_LENGTH_SIZE;
        }
      }

      ByteBufferUtils.copyFromBufferToBuffer(out, out, prevOffset, common);

      // copy the rest of the key from the buffer
      int keyRestLength;
      if (commonLength < state.rowLength + KeyValue.ROW_LENGTH_SIZE) {
        // omit the family part of the key, it is always the same
        int rowWithSizeLength;
        int rowRestLength;

        // check length of row
        if (commonLength < KeyValue.ROW_LENGTH_SIZE) {
          // not yet copied, do it now
          ByteBufferUtils.copyFromStreamToBuffer(out, source,
            KeyValue.ROW_LENGTH_SIZE - commonLength);

          rowWithSizeLength =
            out.getShort(out.position() - KeyValue.ROW_LENGTH_SIZE) + KeyValue.ROW_LENGTH_SIZE;
          rowRestLength = rowWithSizeLength - KeyValue.ROW_LENGTH_SIZE;
        } else {
          // already in kvBuffer, just read it
          rowWithSizeLength = out.getShort(kvPos + KeyValue.ROW_OFFSET) + KeyValue.ROW_LENGTH_SIZE;
          rowRestLength = rowWithSizeLength - commonLength;
        }

        // copy the rest of row
        ByteBufferUtils.copyFromStreamToBuffer(out, source, rowRestLength);

        // copy the column family
        ByteBufferUtils.copyFromBufferToBuffer(out, out,
          state.prevOffset + KeyValue.ROW_OFFSET + KeyValue.ROW_LENGTH_SIZE + state.rowLength,
          state.familyLength + KeyValue.FAMILY_LENGTH_SIZE);
        state.rowLength = (short) (rowWithSizeLength - KeyValue.ROW_LENGTH_SIZE);

        keyRestLength = state.keyLength - rowWithSizeLength - state.familyLength
          - (KeyValue.FAMILY_LENGTH_SIZE + KeyValue.TIMESTAMP_TYPE_SIZE);
      } else {
        // prevRowWithSizeLength is the same as on previous row
        keyRestLength = state.keyLength - commonLength - KeyValue.TIMESTAMP_TYPE_SIZE;
      }
      // copy the rest of the key, after column family == column qualifier
      ByteBufferUtils.copyFromStreamToBuffer(out, source, keyRestLength);

      // copy timestamp
      int prefixTimestamp = (flag & MASK_TIMESTAMP_LENGTH) >>> SHIFT_TIMESTAMP_LENGTH;
      ByteBufferUtils.copyFromBufferToBuffer(out, out, state.prevTimestampOffset, prefixTimestamp);
      state.prevTimestampOffset = out.position() - prefixTimestamp;
      ByteBufferUtils.copyFromStreamToBuffer(out, source,
        KeyValue.TIMESTAMP_SIZE - prefixTimestamp);

      // copy the type and value
      if ((flag & FLAG_SAME_TYPE) != 0) {
        out.put(state.type);
        if ((flag & FLAG_SAME_VALUE) != 0) {
          ByteBufferUtils.copyFromBufferToBuffer(out, out,
            state.prevOffset + KeyValue.ROW_OFFSET + prevKeyLength, state.valueLength);
        } else {
          ByteBufferUtils.copyFromStreamToBuffer(out, source, state.valueLength);
        }
      } else {
        if ((flag & FLAG_SAME_VALUE) != 0) {
          ByteBufferUtils.copyFromStreamToBuffer(out, source, KeyValue.TYPE_SIZE);
          ByteBufferUtils.copyFromBufferToBuffer(out, out,
            state.prevOffset + KeyValue.ROW_OFFSET + prevKeyLength, state.valueLength);
        } else {
          ByteBufferUtils.copyFromStreamToBuffer(out, source,
            state.valueLength + KeyValue.TYPE_SIZE);
        }
        state.type = out.get(state.prevTimestampOffset + KeyValue.TIMESTAMP_SIZE);
      }
    } else { // this is the first element
      state.decompressFirstKV(out, source);
    }

    state.prevOffset = kvPos;
  }
Copies the first key/value from the given stream, and initializes decompression state based
on it. Assumes that we have already read key and value lengths. Does not set
{@link #qualifierLength} (not used by decompression) or {@link #prevOffset} (set by the calle
afterwards).
hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/FastDiffDeltaEncoder.java