HBASE-28556 Reduce memory copying in Rest server when serializing CellModel to Protobuf (#5870)

Signed-off-by: Duo Zhang <zhangduo@apache.org>
diff --git a/hbase-rest/src/main/java/org/apache/hadoop/hbase/rest/MultiRowResource.java b/hbase-rest/src/main/java/org/apache/hadoop/hbase/rest/MultiRowResource.java
index 99fc0c8..8cce772 100644
--- a/hbase-rest/src/main/java/org/apache/hadoop/hbase/rest/MultiRowResource.java
+++ b/hbase-rest/src/main/java/org/apache/hadoop/hbase/rest/MultiRowResource.java
@@ -22,14 +22,10 @@
 import java.util.Base64;
 import java.util.Base64.Decoder;
 import java.util.List;
-import org.apache.hadoop.hbase.Cell;
-import org.apache.hadoop.hbase.CellUtil;
 import org.apache.hadoop.hbase.client.Result;
 import org.apache.hadoop.hbase.filter.Filter;
 import org.apache.hadoop.hbase.filter.ParseFilter;
-import org.apache.hadoop.hbase.rest.model.CellModel;
 import org.apache.hadoop.hbase.rest.model.CellSetModel;
-import org.apache.hadoop.hbase.rest.model.RowModel;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.yetus.audience.InterfaceAudience;
 import org.slf4j.Logger;
@@ -125,12 +121,7 @@
         if (r.isEmpty()) {
           continue;
         }
-        RowModel rowModel = new RowModel(r.getRow());
-        for (Cell c : r.listCells()) {
-          rowModel.addCell(new CellModel(CellUtil.cloneFamily(c), CellUtil.cloneQualifier(c),
-            c.getTimestamp(), CellUtil.cloneValue(c)));
-        }
-        model.addRow(rowModel);
+        model.addRow(RestUtil.createRowModelFromResult(r));
       }
       if (model.getRows().isEmpty()) {
         // If no rows found.
diff --git a/hbase-rest/src/main/java/org/apache/hadoop/hbase/rest/ProtobufStreamingOutput.java b/hbase-rest/src/main/java/org/apache/hadoop/hbase/rest/ProtobufStreamingOutput.java
index eadd6a9..60c3d36 100644
--- a/hbase-rest/src/main/java/org/apache/hadoop/hbase/rest/ProtobufStreamingOutput.java
+++ b/hbase-rest/src/main/java/org/apache/hadoop/hbase/rest/ProtobufStreamingOutput.java
@@ -19,14 +19,9 @@
 
 import java.io.IOException;
 import java.io.OutputStream;
-import java.util.List;
-import org.apache.hadoop.hbase.Cell;
-import org.apache.hadoop.hbase.CellUtil;
 import org.apache.hadoop.hbase.client.Result;
 import org.apache.hadoop.hbase.client.ResultScanner;
-import org.apache.hadoop.hbase.rest.model.CellModel;
 import org.apache.hadoop.hbase.rest.model.CellSetModel;
-import org.apache.hadoop.hbase.rest.model.RowModel;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.yetus.audience.InterfaceAudience;
 import org.slf4j.Logger;
@@ -91,15 +86,11 @@
 
   private CellSetModel createModelFromResults(Result[] results) {
     CellSetModel cellSetModel = new CellSetModel();
-    for (Result rs : results) {
-      byte[] rowKey = rs.getRow();
-      RowModel rModel = new RowModel(rowKey);
-      List<Cell> kvs = rs.listCells();
-      for (Cell kv : kvs) {
-        rModel.addCell(new CellModel(CellUtil.cloneFamily(kv), CellUtil.cloneQualifier(kv),
-          kv.getTimestamp(), CellUtil.cloneValue(kv)));
+    for (int i = 0; i < results.length; i++) {
+      if (results[i].isEmpty()) {
+        continue;
       }
-      cellSetModel.addRow(rModel);
+      cellSetModel.addRow(RestUtil.createRowModelFromResult(results[i]));
     }
     return cellSetModel;
   }
diff --git a/hbase-rest/src/main/java/org/apache/hadoop/hbase/rest/RestUtil.java b/hbase-rest/src/main/java/org/apache/hadoop/hbase/rest/RestUtil.java
new file mode 100644
index 0000000..5f884c5
--- /dev/null
+++ b/hbase-rest/src/main/java/org/apache/hadoop/hbase/rest/RestUtil.java
@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.rest;
+
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.rest.model.CellModel;
+import org.apache.hadoop.hbase.rest.model.RowModel;
+import org.apache.yetus.audience.InterfaceAudience;
+
+@InterfaceAudience.Private
+public final class RestUtil {
+
+  private RestUtil() {
+    // Do not instantiate
+  }
+
+  /**
+   * Speed-optimized method to convert an HBase result to a RowModel. Avoids iterators and uses the
+   * non-cloning constructors to minimize overhead, especially when using protobuf marshalling.
+   * @param r non-empty Result object
+   */
+  public static RowModel createRowModelFromResult(Result r) {
+    Cell firstCell = r.rawCells()[0];
+    RowModel rowModel =
+      new RowModel(firstCell.getRowArray(), firstCell.getRowOffset(), firstCell.getRowLength());
+    int cellsLength = r.rawCells().length;
+    for (int i = 0; i < cellsLength; i++) {
+      rowModel.addCell(new CellModel(r.rawCells()[i]));
+    }
+    return rowModel;
+  }
+}
diff --git a/hbase-rest/src/main/java/org/apache/hadoop/hbase/rest/RowResource.java b/hbase-rest/src/main/java/org/apache/hadoop/hbase/rest/RowResource.java
index 1f0c75a..20c896a 100644
--- a/hbase-rest/src/main/java/org/apache/hadoop/hbase/rest/RowResource.java
+++ b/hbase-rest/src/main/java/org/apache/hadoop/hbase/rest/RowResource.java
@@ -112,8 +112,7 @@
           rowKey = CellUtil.cloneRow(value);
           rowModel = new RowModel(rowKey);
         }
-        rowModel.addCell(new CellModel(CellUtil.cloneFamily(value), CellUtil.cloneQualifier(value),
-          value.getTimestamp(), CellUtil.cloneValue(value)));
+        rowModel.addCell(new CellModel(value));
         if (++count > rowspec.getMaxValues()) {
           break;
         }
@@ -711,8 +710,7 @@
         CellSetModel rModel = new CellSetModel();
         RowModel rRowModel = new RowModel(result.getRow());
         for (Cell cell : result.listCells()) {
-          rRowModel.addCell(new CellModel(CellUtil.cloneFamily(cell), CellUtil.cloneQualifier(cell),
-            cell.getTimestamp(), CellUtil.cloneValue(cell)));
+          rRowModel.addCell(new CellModel(cell));
         }
         rModel.addRow(rRowModel);
         servlet.getMetrics().incrementSucessfulAppendRequests(1);
@@ -803,8 +801,7 @@
         CellSetModel rModel = new CellSetModel();
         RowModel rRowModel = new RowModel(result.getRow());
         for (Cell cell : result.listCells()) {
-          rRowModel.addCell(new CellModel(CellUtil.cloneFamily(cell), CellUtil.cloneQualifier(cell),
-            cell.getTimestamp(), CellUtil.cloneValue(cell)));
+          rRowModel.addCell(new CellModel(cell));
         }
         rModel.addRow(rowModel);
         servlet.getMetrics().incrementSucessfulIncrementRequests(1);
diff --git a/hbase-rest/src/main/java/org/apache/hadoop/hbase/rest/ScannerInstanceResource.java b/hbase-rest/src/main/java/org/apache/hadoop/hbase/rest/ScannerInstanceResource.java
index 81ab8e2..951cafc 100644
--- a/hbase-rest/src/main/java/org/apache/hadoop/hbase/rest/ScannerInstanceResource.java
+++ b/hbase-rest/src/main/java/org/apache/hadoop/hbase/rest/ScannerInstanceResource.java
@@ -83,7 +83,9 @@
     }
     CellSetModel model = new CellSetModel();
     RowModel rowModel = null;
-    byte[] rowKey = null;
+    byte[] rowKeyArray = null;
+    int rowKeyOffset = 0;
+    int rowKeyLength = 0;
     int limit = batch;
     if (maxValues > 0) {
       limit = maxValues;
@@ -121,11 +123,13 @@
         }
         break;
       }
-      if (rowKey == null) {
-        rowKey = CellUtil.cloneRow(value);
-        rowModel = new RowModel(rowKey);
+      if (rowKeyArray == null) {
+        rowKeyArray = value.getRowArray();
+        rowKeyOffset = value.getRowOffset();
+        rowKeyLength = value.getRowLength();
+        rowModel = new RowModel(rowKeyArray, rowKeyOffset, rowKeyLength);
       }
-      if (!Bytes.equals(CellUtil.cloneRow(value), rowKey)) {
+      if (!CellUtil.matchingRow(value, rowKeyArray, rowKeyOffset, rowKeyLength)) {
         // if maxRows was given as a query param, stop if we would exceed the
         // specified number of rows
         if (maxRows > 0) {
@@ -135,11 +139,12 @@
           }
         }
         model.addRow(rowModel);
-        rowKey = CellUtil.cloneRow(value);
-        rowModel = new RowModel(rowKey);
+        rowKeyArray = value.getRowArray();
+        rowKeyOffset = value.getRowOffset();
+        rowKeyLength = value.getRowLength();
+        rowModel = new RowModel(rowKeyArray, rowKeyOffset, rowKeyLength);
       }
-      rowModel.addCell(new CellModel(CellUtil.cloneFamily(value), CellUtil.cloneQualifier(value),
-        value.getTimestamp(), CellUtil.cloneValue(value)));
+      rowModel.addCell(new CellModel(value));
     } while (--count > 0);
     model.addRow(rowModel);
     ResponseBuilder response = Response.ok(model);
diff --git a/hbase-rest/src/main/java/org/apache/hadoop/hbase/rest/TableScanResource.java b/hbase-rest/src/main/java/org/apache/hadoop/hbase/rest/TableScanResource.java
index e30beaa..4bb30b0 100644
--- a/hbase-rest/src/main/java/org/apache/hadoop/hbase/rest/TableScanResource.java
+++ b/hbase-rest/src/main/java/org/apache/hadoop/hbase/rest/TableScanResource.java
@@ -22,16 +22,12 @@
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Iterator;
-import java.util.List;
 import javax.xml.bind.annotation.XmlAccessType;
 import javax.xml.bind.annotation.XmlAccessorType;
 import javax.xml.bind.annotation.XmlElement;
 import javax.xml.bind.annotation.XmlRootElement;
-import org.apache.hadoop.hbase.Cell;
-import org.apache.hadoop.hbase.CellUtil;
 import org.apache.hadoop.hbase.client.Result;
 import org.apache.hadoop.hbase.client.ResultScanner;
-import org.apache.hadoop.hbase.rest.model.CellModel;
 import org.apache.hadoop.hbase.rest.model.RowModel;
 import org.apache.yetus.audience.InterfaceAudience;
 import org.slf4j.Logger;
@@ -87,13 +83,7 @@
             if ((rs == null) || (count <= 0)) {
               return null;
             }
-            byte[] rowKey = rs.getRow();
-            RowModel rModel = new RowModel(rowKey);
-            List<Cell> kvs = rs.listCells();
-            for (Cell kv : kvs) {
-              rModel.addCell(new CellModel(CellUtil.cloneFamily(kv), CellUtil.cloneQualifier(kv),
-                kv.getTimestamp(), CellUtil.cloneValue(kv)));
-            }
+            RowModel rModel = RestUtil.createRowModelFromResult(rs);
             count--;
             if (count == 0) {
               results.close();
diff --git a/hbase-rest/src/main/java/org/apache/hadoop/hbase/rest/model/CellModel.java b/hbase-rest/src/main/java/org/apache/hadoop/hbase/rest/model/CellModel.java
index eda3267..4284727 100644
--- a/hbase-rest/src/main/java/org/apache/hadoop/hbase/rest/model/CellModel.java
+++ b/hbase-rest/src/main/java/org/apache/hadoop/hbase/rest/model/CellModel.java
@@ -17,6 +17,9 @@
  */
 package org.apache.hadoop.hbase.rest.model;
 
+import static org.apache.hadoop.hbase.KeyValue.COLUMN_FAMILY_DELIMITER;
+
+import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import java.io.IOException;
 import java.io.Serializable;
@@ -58,10 +61,11 @@
  * </pre>
  */
 @XmlRootElement(name = "Cell")
-@XmlAccessorType(XmlAccessType.FIELD)
+@XmlAccessorType(XmlAccessType.NONE)
 @InterfaceAudience.Private
 public class CellModel implements ProtobufMessageHandler, Serializable {
   private static final long serialVersionUID = 1L;
+  public static final int MAGIC_LENGTH = -1;
 
   @JsonProperty("column")
   @XmlAttribute
@@ -71,10 +75,17 @@
   @XmlAttribute
   private long timestamp = HConstants.LATEST_TIMESTAMP;
 
-  @JsonProperty("$")
-  @XmlValue
+  // If valueLength = -1, this represents the cell's value.
+  // If valueLength <> 1, this represents an array containing the cell's value as determined by
+  // offset and length.
   private byte[] value;
 
+  @JsonIgnore
+  private int valueOffset;
+
+  @JsonIgnore
+  private int valueLength = MAGIC_LENGTH;
+
   /**
    * Default constructor
    */
@@ -96,11 +107,16 @@
   }
 
   /**
-   * Constructor from KeyValue
+   * Constructor from KeyValue This avoids copying the value from the cell, and tries to optimize
+   * generating the column value.
    */
   public CellModel(org.apache.hadoop.hbase.Cell cell) {
-    this(CellUtil.cloneFamily(cell), CellUtil.cloneQualifier(cell), cell.getTimestamp(),
-      CellUtil.cloneValue(cell));
+    this.column = makeColumn(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength(),
+      cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
+    this.timestamp = cell.getTimestamp();
+    this.value = cell.getValueArray();
+    this.valueOffset = cell.getValueOffset();
+    this.valueLength = cell.getValueLength();
   }
 
   /**
@@ -109,16 +125,16 @@
   public CellModel(byte[] column, long timestamp, byte[] value) {
     this.column = column;
     this.timestamp = timestamp;
-    this.value = value;
+    setValue(value);
   }
 
   /**
    * Constructor
    */
-  public CellModel(byte[] column, byte[] qualifier, long timestamp, byte[] value) {
-    this.column = CellUtil.makeColumn(column, qualifier);
+  public CellModel(byte[] family, byte[] qualifier, long timestamp, byte[] value) {
+    this.column = CellUtil.makeColumn(family, qualifier);
     this.timestamp = timestamp;
-    this.value = value;
+    setValue(value);
   }
 
   /** Returns the column */
@@ -151,22 +167,49 @@
   }
 
   /** Returns the value */
+  @JsonProperty("$")
+  @XmlValue
   public byte[] getValue() {
+    if (valueLength == MAGIC_LENGTH) {
+      return value;
+    } else {
+      byte[] retValue = new byte[valueLength];
+      System.arraycopy(value, valueOffset, retValue, 0, valueLength);
+      return retValue;
+    }
+  }
+
+  /** Returns the backing array for value (may be the same as value) */
+  public byte[] getValueArray() {
     return value;
   }
 
   /**
    * @param value the value to set
    */
+  @JsonProperty("$")
   public void setValue(byte[] value) {
     this.value = value;
+    this.valueLength = MAGIC_LENGTH;
+  }
+
+  public int getValueOffset() {
+    return valueOffset;
+  }
+
+  public int getValueLength() {
+    return valueLength;
   }
 
   @Override
   public byte[] createProtobufOutput() {
     Cell.Builder builder = Cell.newBuilder();
     builder.setColumn(UnsafeByteOperations.unsafeWrap(getColumn()));
-    builder.setData(UnsafeByteOperations.unsafeWrap(getValue()));
+    if (valueLength == MAGIC_LENGTH) {
+      builder.setData(UnsafeByteOperations.unsafeWrap(getValue()));
+    } else {
+      builder.setData(UnsafeByteOperations.unsafeWrap(value, valueOffset, valueLength));
+    }
     if (hasUserTimestamp()) {
       builder.setTimestamp(getTimestamp());
     }
@@ -185,6 +228,21 @@
     return this;
   }
 
+  /**
+   * Makes a column in family:qualifier form from separate byte arrays with offset and length.
+   * <p>
+   * Not recommended for usage as this is old-style API.
+   * @return family:qualifier
+   */
+  public static byte[] makeColumn(byte[] family, int familyOffset, int familyLength,
+    byte[] qualifier, int qualifierOffset, int qualifierLength) {
+    byte[] column = new byte[familyLength + qualifierLength + 1];
+    System.arraycopy(family, familyOffset, column, 0, familyLength);
+    column[familyLength] = COLUMN_FAMILY_DELIMITER;
+    System.arraycopy(qualifier, qualifierOffset, column, familyLength + 1, qualifierLength);
+    return column;
+  }
+
   @Override
   public boolean equals(Object obj) {
     if (obj == null) {
@@ -198,17 +256,17 @@
     }
     CellModel cellModel = (CellModel) obj;
     return new EqualsBuilder().append(column, cellModel.column)
-      .append(timestamp, cellModel.timestamp).append(value, cellModel.value).isEquals();
+      .append(timestamp, cellModel.timestamp).append(getValue(), cellModel.getValue()).isEquals();
   }
 
   @Override
   public int hashCode() {
-    return new HashCodeBuilder().append(column).append(timestamp).append(value).toHashCode();
+    return new HashCodeBuilder().append(column).append(timestamp).append(getValue()).toHashCode();
   }
 
   @Override
   public String toString() {
     return new ToStringBuilder(this).append("column", column).append("timestamp", timestamp)
-      .append("value", value).toString();
+      .append("value", getValue()).toString();
   }
 }
diff --git a/hbase-rest/src/main/java/org/apache/hadoop/hbase/rest/model/CellSetModel.java b/hbase-rest/src/main/java/org/apache/hadoop/hbase/rest/model/CellSetModel.java
index fff96b3..8908ec7 100644
--- a/hbase-rest/src/main/java/org/apache/hadoop/hbase/rest/model/CellSetModel.java
+++ b/hbase-rest/src/main/java/org/apache/hadoop/hbase/rest/model/CellSetModel.java
@@ -17,6 +17,8 @@
  */
 package org.apache.hadoop.hbase.rest.model;
 
+import static org.apache.hadoop.hbase.rest.model.CellModel.MAGIC_LENGTH;
+
 import java.io.IOException;
 import java.io.Serializable;
 import java.util.ArrayList;
@@ -69,7 +71,7 @@
  * </pre>
  */
 @XmlRootElement(name = "CellSet")
-@XmlAccessorType(XmlAccessType.FIELD)
+@XmlAccessorType(XmlAccessType.NONE)
 @InterfaceAudience.Private
 public class CellSetModel implements Serializable, ProtobufMessageHandler {
   private static final long serialVersionUID = 1L;
@@ -110,11 +112,21 @@
     CellSet.Builder builder = CellSet.newBuilder();
     for (RowModel row : getRows()) {
       CellSet.Row.Builder rowBuilder = CellSet.Row.newBuilder();
-      rowBuilder.setKey(UnsafeByteOperations.unsafeWrap(row.getKey()));
+      if (row.getKeyLength() == MAGIC_LENGTH) {
+        rowBuilder.setKey(UnsafeByteOperations.unsafeWrap(row.getKey()));
+      } else {
+        rowBuilder.setKey(UnsafeByteOperations.unsafeWrap(row.getKeyArray(), row.getKeyOffset(),
+          row.getKeyLength()));
+      }
       for (CellModel cell : row.getCells()) {
         Cell.Builder cellBuilder = Cell.newBuilder();
         cellBuilder.setColumn(UnsafeByteOperations.unsafeWrap(cell.getColumn()));
-        cellBuilder.setData(UnsafeByteOperations.unsafeWrap(cell.getValue()));
+        if (cell.getValueLength() == MAGIC_LENGTH) {
+          cellBuilder.setData(UnsafeByteOperations.unsafeWrap(cell.getValue()));
+        } else {
+          cellBuilder.setData(UnsafeByteOperations.unsafeWrap(cell.getValueArray(),
+            cell.getValueOffset(), cell.getValueLength()));
+        }
         if (cell.hasUserTimestamp()) {
           cellBuilder.setTimestamp(cell.getTimestamp());
         }
diff --git a/hbase-rest/src/main/java/org/apache/hadoop/hbase/rest/model/RowModel.java b/hbase-rest/src/main/java/org/apache/hadoop/hbase/rest/model/RowModel.java
index f3e892a..8b660ac 100644
--- a/hbase-rest/src/main/java/org/apache/hadoop/hbase/rest/model/RowModel.java
+++ b/hbase-rest/src/main/java/org/apache/hadoop/hbase/rest/model/RowModel.java
@@ -17,6 +17,8 @@
  */
 package org.apache.hadoop.hbase.rest.model;
 
+import static org.apache.hadoop.hbase.rest.model.CellModel.MAGIC_LENGTH;
+
 import com.fasterxml.jackson.annotation.JsonProperty;
 import java.io.IOException;
 import java.io.Serializable;
@@ -49,15 +51,18 @@
  * </pre>
  */
 @XmlRootElement(name = "Row")
-@XmlAccessorType(XmlAccessType.FIELD)
+@XmlAccessorType(XmlAccessType.NONE)
 @InterfaceAudience.Private
 public class RowModel implements ProtobufMessageHandler, Serializable {
   private static final long serialVersionUID = 1L;
 
-  @JsonProperty("key")
-  @XmlAttribute
+  // If keyLength = -1, this represents the key
+  // If keyLength <> -1, this represents the base array, and key is determined by offset and length
   private byte[] key;
 
+  private int keyOffset = 0;
+  private int keyLength = MAGIC_LENGTH;
+
   @JsonProperty("Cell")
   @XmlElement(name = "Cell")
   private List<CellModel> cells = new ArrayList<>();
@@ -81,7 +86,18 @@
    * @param key the row key
    */
   public RowModel(final byte[] key) {
+    setKey(key);
+    cells = new ArrayList<>();
+  }
+
+  /**
+   * Constructor
+   * @param key the row key as represented in the Cell
+   */
+  public RowModel(final byte[] key, int keyOffset, int keyLength) {
     this.key = key;
+    this.keyOffset = keyOffset;
+    this.keyLength = keyLength;
     cells = new ArrayList<>();
   }
 
@@ -100,7 +116,17 @@
    * @param cells the cells
    */
   public RowModel(final byte[] key, final List<CellModel> cells) {
-    this.key = key;
+    this(key);
+    this.cells = cells;
+  }
+
+  /**
+   * Constructor
+   * @param key   the row key
+   * @param cells the cells
+   */
+  public RowModel(final byte[] key, int keyOffset, int keyLength, final List<CellModel> cells) {
+    this(key, keyOffset, keyLength);
     this.cells = cells;
   }
 
@@ -113,15 +139,38 @@
   }
 
   /** Returns the row key */
+  @XmlAttribute
+  @JsonProperty("key")
   public byte[] getKey() {
+    if (keyLength == MAGIC_LENGTH) {
+      return key;
+    } else {
+      byte[] retKey = new byte[keyLength];
+      System.arraycopy(key, keyOffset, retKey, 0, keyLength);
+      return retKey;
+    }
+  }
+
+  /** Returns the backing row key array */
+  public byte[] getKeyArray() {
     return key;
   }
 
   /**
    * @param key the row key
    */
+  @JsonProperty("key")
   public void setKey(byte[] key) {
     this.key = key;
+    this.keyLength = MAGIC_LENGTH;
+  }
+
+  public int getKeyOffset() {
+    return keyOffset;
+  }
+
+  public int getKeyLength() {
+    return keyLength;
   }
 
   /** Returns the cells */
@@ -153,16 +202,17 @@
       return false;
     }
     RowModel rowModel = (RowModel) obj;
-    return new EqualsBuilder().append(key, rowModel.key).append(cells, rowModel.cells).isEquals();
+    return new EqualsBuilder().append(getKey(), rowModel.getKey()).append(cells, rowModel.cells)
+      .isEquals();
   }
 
   @Override
   public int hashCode() {
-    return new HashCodeBuilder().append(key).append(cells).toHashCode();
+    return new HashCodeBuilder().append(getKey()).append(cells).toHashCode();
   }
 
   @Override
   public String toString() {
-    return new ToStringBuilder(this).append("key", key).append("cells", cells).toString();
+    return new ToStringBuilder(this).append("key", getKey()).append("cells", cells).toString();
   }
 }