HBASE-8849 - CellCodec should write and read the memstoreTS/mvccVersion (Ram)

git-svn-id: https://svn.apache.org/repos/asf/hbase/trunk@1511043 13f79535-47bb-0310-9956-ffa450edef68

hbase-common/src/main/java/org/apache/hadoop/hbase/CellUtil.java

@@ -129,6 +129,17 @@ public final class CellUtil {
       KeyValue.Type.codeToType(type), value);
   }
   
+  public static Cell createCell(final byte[] row, final byte[] family, final byte[] qualifier,
+      final long timestamp, final byte type, final byte[] value, final long memstoreTS) {
+    // I need a Cell Factory here. Using KeyValue for now. TODO.
+    // TODO: Make a new Cell implementation that just carries these
+    // byte arrays.
+    KeyValue keyValue = new KeyValue(row, family, qualifier, timestamp,
+        KeyValue.Type.codeToType(type), value);
+    keyValue.setMvccVersion(memstoreTS);
+    return keyValue;
+  }
+
   /**
    * @param cellScannerables
    * @return CellScanner interface over <code>cellIterables</code>

hbase-common/src/main/java/org/apache/hadoop/hbase/codec/CellCodec.java

@@ -28,8 +28,7 @@ import org.apache.hadoop.hbase.util.Bytes;
 
 /**
  * Basic Cell codec that just writes out all the individual elements of a Cell.  Uses ints
- * delimiting all lengths. Profligate. Needs tune up.  Does not write the mvcc stamp.
+ * delimiting all lengths. Profligate. Needs tune up. 
- * Use a different codec if you want that in the stream.
  */
 public class CellCodec implements Codec {
   static class CellEncoder extends BaseEncoder {
@@ -52,6 +51,8 @@ public class CellCodec implements Codec {
       this.out.write(cell.getTypeByte());
       // Value
       write(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
+      // MvccVersion
+      this.out.write(Bytes.toBytes(cell.getMvccVersion()));
     }
 
     /**
@@ -82,7 +83,11 @@ public class CellCodec implements Codec {
       long timestamp = Bytes.toLong(longArray);
       byte type = (byte) this.in.read();
       byte [] value = readByteArray(in);
-      return CellUtil.createCell(row, family, qualifier, timestamp, type, value);
+      // Read memstore version
+      byte[] memstoreTSArray = new byte[Bytes.SIZEOF_LONG];
+      IOUtils.readFully(this.in, memstoreTSArray);
+      long memstoreTS = Bytes.toLong(memstoreTSArray);
+      return CellUtil.createCell(row, family, qualifier, timestamp, type, value, memstoreTS);
     }
 
     /**

hbase-common/src/test/java/org/apache/hadoop/hbase/codec/TestCellCodec.java

@@ -70,6 +70,7 @@ public class TestCellCodec {
     Codec.Encoder encoder = codec.getEncoder(dos);
     final KeyValue kv =
       new KeyValue(Bytes.toBytes("r"), Bytes.toBytes("f"), Bytes.toBytes("q"), Bytes.toBytes("v"));
+    kv.setMvccVersion(Long.MAX_VALUE);
     encoder.write(kv);
     encoder.flush();
     dos.close();