HBASE-30390 Obtain previous cell value in writer buffer instead of copying in encoding

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

@@ -807,6 +807,20 @@ public static boolean matchingValue(final Cell left, final Cell right, int lvlen
       right.getValueArray(), right.getValueOffset(), rvlength);
   }
 
+  public static boolean matchingValue(final Cell left, final byte[] right, int rightOffset,
+    int rightLength) {
+    if (left.getValueLength() != rightLength) {
+      return false;
+    }
+    if (left instanceof ByteBufferExtendedCell) {
+      return ByteBufferUtils.equals(((ByteBufferExtendedCell) left).getValueByteBuffer(),
+        ((ByteBufferExtendedCell) left).getValuePosition(), rightLength, right, rightOffset,
+        rightLength);
+    }
+    return Bytes.equals(left.getValueArray(), left.getValueOffset(), rightLength, right,
+      rightOffset, rightLength);
+  }
+
   public static boolean matchingType(ExtendedCell a, ExtendedCell b) {
     return a.getTypeByte() == b.getTypeByte();
   }

hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/BufferedDataBlockEncoder.java

@@ -1241,7 +1241,11 @@ public void startBlockEncoding(HFileBlockEncodingContext blkEncodingCtx, DataOut
       }
     }
     StreamUtils.writeInt(out, 0); // DUMMY length. This will be updated in endBlockEncoding()
-    blkEncodingCtx.setEncodingState(new EncodingState());
+    blkEncodingCtx.setEncodingState(createEncodingState());
+  }
+
+  protected EncodingState createEncodingState() {
+    return new EncodingState();
   }
 
   @Override

hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/DiffEncodingState.java

@@ -0,0 +1,36 @@
+/*
+ * 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.io.encoding;
+
+import org.apache.hadoop.hbase.ExtendedCell;
+import org.apache.yetus.audience.InterfaceAudience;
+
+@InterfaceAudience.Private
+class DiffEncodingState extends EncodingState {
+
+  private int previousValueLength;
+
+  int getPreviousValueLength() {
+    return previousValueLength;
+  }
+
+  @Override
+  protected void setPreviousCell(ExtendedCell cell) {
+    super.setPreviousCell(cell);
+    previousValueLength = cell.getValueLength();
+  }
+}

hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/DiffKeyDeltaEncoder.java

@@ -52,6 +52,11 @@ public class DiffKeyDeltaEncoder extends BufferedDataBlockEncoder {
   static final int SHIFT_TIMESTAMP_LENGTH = 4;
   static final int FLAG_TIMESTAMP_SIGN = 1 << 7;
 
+  @Override
+  protected EncodingState createEncodingState() {
+    return new DiffEncodingState();
+  }
+
   protected static class DiffCompressionState extends CompressionState {
     long timestamp;
     byte[] familyNameWithSize;
@@ -183,15 +188,16 @@ private void uncompressSingleKeyValue(DataInputStream source, ByteBuffer buffer,
   @Override
   public int internalEncode(ExtendedCell cell, HFileBlockDefaultEncodingContext encodingContext,
     DataOutputStream out) throws IOException {
-    EncodingState state = encodingContext.getEncodingState();
-    int size = compressSingleKeyValue(out, cell, state.prevCell);
+    DiffEncodingState state = (DiffEncodingState) encodingContext.getEncodingState();
+    int size = compressSingleKeyValue(out, cell, state);
     size += afterEncodingKeyValue(cell, out, encodingContext);
-    state.prevCell = cell;
+    state.setPreviousCell(cell);
     return size;
   }
 
-  private int compressSingleKeyValue(DataOutputStream out, ExtendedCell cell, ExtendedCell prevCell)
-    throws IOException {
+  private int compressSingleKeyValue(DataOutputStream out, ExtendedCell cell,
+    DiffEncodingState state) throws IOException {
+    ExtendedCell prevCell = state.prevCell;
     int flag = 0; // Do not use more bits that can fit into a byte
     int kLength = KeyValueUtil.keyLength(cell);
     int vLength = cell.getValueLength();
@@ -221,7 +227,7 @@ private int compressSingleKeyValue(DataOutputStream out, ExtendedCell cell, Exte
       if (kLength == preKeyLength) {
         flag |= FLAG_SAME_KEY_LENGTH;
       }
-      if (vLength == prevCell.getValueLength()) {
+      if (vLength == state.getPreviousValueLength()) {
         flag |= FLAG_SAME_VALUE_LENGTH;
       }
       if (cell.getTypeByte() == prevCell.getTypeByte()) {

hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/EncodingState.java

@@ -41,13 +41,20 @@ public class EncodingState {
   protected int encodedDataSizeWritten = 0;
 
   public void beforeShipped() {
+    beforeShipped(null, 0, 0);
+  }
+
+  public void beforeShipped(byte[] encodedBlockBuffer, int streamBaseOffset,
+      int encodedBlockLength) {
     if (this.prevCell != null) {
-      // can't use KeyValueUtil#toNewKeyCell, because we need both key and value
-      // from the prevCell in FastDiffDeltaEncoder
-      this.prevCell = KeyValueUtil.copyToNewKeyValue(this.prevCell);
+      this.prevCell = KeyValueUtil.toNewKeyCell(this.prevCell);
     }
   }
 
+  protected void setPreviousCell(ExtendedCell cell) {
+    this.prevCell = cell;
+  }
+
   public void postCellEncode(int unencodedCellSizeWritten, int encodedCellSizeWritten) {
     this.unencodedDataSizeWritten += unencodedCellSizeWritten;
     this.encodedDataSizeWritten += encodedCellSizeWritten;

hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/FastDiffDeltaEncoder.java

@@ -53,6 +53,11 @@ public class FastDiffDeltaEncoder extends BufferedDataBlockEncoder {
   static final int FLAG_SAME_TYPE = 1 << 5;
   static final int FLAG_SAME_VALUE = 1 << 6;
 
+  @Override
+  protected EncodingState createEncodingState() {
+    return new FastDiffEncodingState();
+  }
+
   private static class FastDiffCompressionState extends CompressionState {
     byte[] timestamp = new byte[KeyValue.TIMESTAMP_SIZE];
     int prevTimestampOffset;
@@ -217,15 +222,17 @@ private void uncompressSingleKeyValue(DataInputStream source, ByteBuffer out,
   @Override
   public int internalEncode(ExtendedCell cell, HFileBlockDefaultEncodingContext encodingContext,
     DataOutputStream out) throws IOException {
-    EncodingState state = encodingContext.getEncodingState();
-    int size = compressSingleKeyValue(out, cell, state.prevCell);
+    FastDiffEncodingState state = (FastDiffEncodingState) encodingContext.getEncodingState();
+    state.beginCellEncoding();
+    int size = compressSingleKeyValue(out, cell, state);
     size += afterEncodingKeyValue(cell, out, encodingContext);
-    state.prevCell = cell;
+    state.setPreviousCell(cell);
     return size;
   }
 
-  private int compressSingleKeyValue(DataOutputStream out, ExtendedCell cell, ExtendedCell prevCell)
-    throws IOException {
+  private int compressSingleKeyValue(DataOutputStream out, ExtendedCell cell,
+    FastDiffEncodingState state) throws IOException {
+    ExtendedCell prevCell = state.prevCell;
     int flag = 0; // Do not use more bits than will fit into a byte
     int kLength = KeyValueUtil.keyLength(cell);
     int vLength = cell.getValueLength();
@@ -238,17 +245,19 @@ private int compressSingleKeyValue(DataOutputStream out, ExtendedCell cell, Exte
       ByteBufferUtils.putCompressedInt(out, 0);
       PrivateCellUtil.writeFlatKey(cell, (DataOutput) out);
       // Write the value part
+      int valueStreamOffset = out.size();
       PrivateCellUtil.writeValue(out, cell, cell.getValueLength());
+      state.setCurrentCellValueStreamOffset(valueStreamOffset);
     } else {
       int preKeyLength = KeyValueUtil.keyLength(prevCell);
-      int preValLength = prevCell.getValueLength();
+      int preValLength = state.getPreviousValueLength();
       // 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()) {
+      if (vLength == preValLength) {
         flag |= FLAG_SAME_VALUE_LENGTH;
       }
       if (cell.getTypeByte() == prevCell.getTypeByte()) {
@@ -263,10 +272,7 @@ private int compressSingleKeyValue(DataOutputStream out, ExtendedCell cell, Exte
 
       // 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)
-      ) {
+      if (vLength == preValLength && state.matchingPreviousValue(cell)) {
         flag |= FLAG_SAME_VALUE;
       }
 
@@ -304,7 +310,9 @@ private int compressSingleKeyValue(DataOutputStream out, ExtendedCell cell, Exte
 
       // Write the value if it is not the same as before.
       if ((flag & FLAG_SAME_VALUE) == 0) {
+        int valueStreamOffset = out.size();
         PrivateCellUtil.writeValue(out, cell, vLength);
+        state.setCurrentCellValueStreamOffset(valueStreamOffset);
       }
     }
     return kLength + vLength + KeyValue.KEYVALUE_INFRASTRUCTURE_SIZE;

hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/FastDiffEncodingState.java

@@ -0,0 +1,82 @@
+/*
+ * 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.io.encoding;
+
+import org.apache.hadoop.hbase.ExtendedCell;
+import org.apache.hadoop.hbase.PrivateCellUtil;
+import org.apache.yetus.audience.InterfaceAudience;
+
+@InterfaceAudience.Private
+class FastDiffEncodingState extends DiffEncodingState {
+
+  private int lastMaterializedValueStreamOffset = -1;
+  private int currentCellValueStreamOffset = -1;
+  private byte[] shippedValueBuffer;
+  private int shippedValueAbsoluteOffset;
+
+  boolean matchingPreviousValue(ExtendedCell cell) {
+    if (shippedValueBuffer != null) {
+      return PrivateCellUtil.matchingValue(cell, shippedValueBuffer, shippedValueAbsoluteOffset,
+        getPreviousValueLength());
+    }
+    return prevCell != null && PrivateCellUtil.matchingValue(cell, prevCell,
+        cell.getValueLength(), getPreviousValueLength());
+  }
+
+  void beginCellEncoding() {
+    currentCellValueStreamOffset = -1;
+  }
+
+  void setCurrentCellValueStreamOffset(int valueStreamOffset) {
+    currentCellValueStreamOffset = valueStreamOffset;
+  }
+
+  @Override
+  protected void setPreviousCell(ExtendedCell cell) {
+    super.setPreviousCell(cell);
+    if (currentCellValueStreamOffset >= 0) {
+      lastMaterializedValueStreamOffset = currentCellValueStreamOffset;
+    }
+    currentCellValueStreamOffset = -1;
+    shippedValueBuffer = null;
+  }
+
+  @Override
+  public void beforeShipped(byte[] encodedBlockBuffer, int streamBaseOffset,
+      int encodedBlockLength) {
+    if (shippedValueBuffer != null || prevCell == null
+        || lastMaterializedValueStreamOffset < 0) {
+      return;
+    }
+
+    if (encodedBlockBuffer == null || streamBaseOffset < 0 || streamBaseOffset > encodedBlockLength
+        || encodedBlockLength > encodedBlockBuffer.length) {
+      throw new IllegalArgumentException("Invalid encoded block buffer range");
+    }
+    int valueLength = getPreviousValueLength();
+    int encodedStreamLength = encodedBlockLength - streamBaseOffset;
+    if (lastMaterializedValueStreamOffset > encodedStreamLength
+        || valueLength > encodedStreamLength - lastMaterializedValueStreamOffset) {
+      throw new IllegalArgumentException("Invalid encoded value range");
+    }
+
+    super.beforeShipped(encodedBlockBuffer, streamBaseOffset, encodedBlockLength);
+    shippedValueBuffer = encodedBlockBuffer;
+    shippedValueAbsoluteOffset =
+      streamBaseOffset + lastMaterializedValueStreamOffset;
+  }
+}

hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/PrefixKeyDeltaEncoder.java

@@ -65,7 +65,7 @@ public int internalEncode(ExtendedCell cell, HFileBlockDefaultEncodingContext en
     PrivateCellUtil.writeValue(out, cell, vlength);
     int size = klength + vlength + KeyValue.KEYVALUE_INFRASTRUCTURE_SIZE;
     size += afterEncodingKeyValue(cell, out, encodingContext);
-    state.prevCell = cell;
+    state.setPreviousCell(cell);
     return size;
   }
 

hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/RowIndexCodecV1.java

@@ -46,7 +46,8 @@ private static class RowIndexEncodingState extends EncodingState {
     RowIndexEncoderV1 encoder = null;
 
     @Override
-    public void beforeShipped() {
+    public void beforeShipped(byte[] encodedBlockBuffer, int streamBaseOffset,
+        int encodedBlockLength) {
       if (encoder != null) {
         encoder.beforeShipped();
       }