private int compressSingleKeyValue(DataOutputStream out, ExtendedCell cell, ExtendedCell prevCell)
    throws IOException {
    int flag = 0; // Do not use more bits than will fit into a byte
    int kLength = KeyValueUtil.keyLength(cell);
    int vLength = cell.getValueLength();

    if (prevCell == null) {
      // copy the key, there is no common prefix with none
      out.write(flag);
      ByteBufferUtils.putCompressedInt(out, kLength);
      ByteBufferUtils.putCompressedInt(out, vLength);
      ByteBufferUtils.putCompressedInt(out, 0);
      PrivateCellUtil.writeFlatKey(cell, (DataOutput) out);
      // Write the value part
      PrivateCellUtil.writeValue(out, cell, cell.getValueLength());
    } else {
      int preKeyLength = KeyValueUtil.keyLength(prevCell);
      int preValLength = prevCell.getValueLength();
      // find a common prefix and skip it
      int commonPrefix = PrivateCellUtil.findCommonPrefixInFlatKey(cell, prevCell, true, false);

      if (kLength == preKeyLength) {
        flag |= FLAG_SAME_KEY_LENGTH;
      }
      if (vLength == prevCell.getValueLength()) {
        flag |= FLAG_SAME_VALUE_LENGTH;
      }
      if (cell.getTypeByte() == prevCell.getTypeByte()) {
        flag |= FLAG_SAME_TYPE;
      }

      byte[] curTsBuf = Bytes.toBytes(cell.getTimestamp());
      int commonTimestampPrefix =
        findCommonTimestampPrefix(curTsBuf, Bytes.toBytes(prevCell.getTimestamp()));

      flag |= commonTimestampPrefix << SHIFT_TIMESTAMP_LENGTH;

      // Check if current and previous values are the same. Compare value
      // length first as an optimization.
      if (
        vLength == preValLength
          && PrivateCellUtil.matchingValue(cell, prevCell, vLength, preValLength)
      ) {
        flag |= FLAG_SAME_VALUE;
      }

      out.write(flag);
      if ((flag & FLAG_SAME_KEY_LENGTH) == 0) {
        ByteBufferUtils.putCompressedInt(out, kLength);
      }
      if ((flag & FLAG_SAME_VALUE_LENGTH) == 0) {
        ByteBufferUtils.putCompressedInt(out, vLength);
      }
      ByteBufferUtils.putCompressedInt(out, commonPrefix);
      short rLen = cell.getRowLength();
      if (commonPrefix < rLen + KeyValue.ROW_LENGTH_SIZE) {
        // Previous and current rows are different. Copy the differing part of
        // the row, skip the column family, and copy the qualifier.
        PrivateCellUtil.writeRowKeyExcludingCommon(cell, rLen, commonPrefix, out);
        PrivateCellUtil.writeQualifier(out, cell, cell.getQualifierLength());
      } else {
        // The common part includes the whole row. As the column family is the
        // same across the whole file, it will automatically be included in the
        // common prefix, so we need not special-case it here.
        // What we write here is the non common part of the qualifier
        int commonQualPrefix = commonPrefix - (rLen + KeyValue.ROW_LENGTH_SIZE)
          - (cell.getFamilyLength() + KeyValue.FAMILY_LENGTH_SIZE);
        PrivateCellUtil.writeQualifierSkippingBytes(out, cell, cell.getQualifierLength(),
          commonQualPrefix);
      }
      // Write non common ts part
      out.write(curTsBuf, commonTimestampPrefix, KeyValue.TIMESTAMP_SIZE - commonTimestampPrefix);

      // Write the type if it is not the same as before.
      if ((flag & FLAG_SAME_TYPE) == 0) {
        out.write(cell.getTypeByte());
      }

      // Write the value if it is not the same as before.
      if ((flag & FLAG_SAME_VALUE) == 0) {
        PrivateCellUtil.writeValue(out, cell, vLength);
      }
    }
    return kLength + vLength + KeyValue.KEYVALUE_INFRASTRUCTURE_SIZE;
  }
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