diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/PrivateCellUtil.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/PrivateCellUtil.java index 33a821d75db4..b53efa886158 100644 --- a/hbase-common/src/main/java/org/apache/hadoop/hbase/PrivateCellUtil.java +++ b/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(); } diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/BufferedDataBlockEncoder.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/BufferedDataBlockEncoder.java index 5ec39fa5803d..3de4247331ea 100644 --- a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/BufferedDataBlockEncoder.java +++ b/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 diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/DiffEncodingState.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/DiffEncodingState.java new file mode 100644 index 000000000000..2b9543b7807d --- /dev/null +++ b/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(); + } +} diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/DiffKeyDeltaEncoder.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/DiffKeyDeltaEncoder.java index d58f5e2c923e..953e6cd8f162 100644 --- a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/DiffKeyDeltaEncoder.java +++ b/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()) { diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/EncodingState.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/EncodingState.java index 8ab4e320552e..ab4b16497c42 100644 --- a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/EncodingState.java +++ b/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; diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/FastDiffDeltaEncoder.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/FastDiffDeltaEncoder.java index 26b695abfca9..f8cba2c6ffed 100644 --- a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/FastDiffDeltaEncoder.java +++ b/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; diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/FastDiffEncodingState.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/FastDiffEncodingState.java new file mode 100644 index 000000000000..f2e4166b4eaa --- /dev/null +++ b/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; + } +} diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/PrefixKeyDeltaEncoder.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/PrefixKeyDeltaEncoder.java index e9858b5ffba1..c34cea52d8ea 100644 --- a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/PrefixKeyDeltaEncoder.java +++ b/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; } diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/RowIndexCodecV1.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/RowIndexCodecV1.java index 9f2014331089..e61916d4c44f 100644 --- a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/RowIndexCodecV1.java +++ b/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(); } diff --git a/hbase-common/src/test/java/org/apache/hadoop/hbase/io/encoding/TestFastDiffEncodingState.java b/hbase-common/src/test/java/org/apache/hadoop/hbase/io/encoding/TestFastDiffEncodingState.java new file mode 100644 index 000000000000..dfac8a20be34 --- /dev/null +++ b/hbase-common/src/test/java/org/apache/hadoop/hbase/io/encoding/TestFastDiffEncodingState.java @@ -0,0 +1,227 @@ +/* + * 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 static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.Assertions.fail; +import java.io.ByteArrayInputStream; +import java.io.DataInputStream; +import java.io.DataOutputStream; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hbase.ByteBufferKeyValue; +import org.apache.hadoop.hbase.HBaseConfiguration; +import org.apache.hadoop.hbase.HConstants; +import org.apache.hadoop.hbase.KeyValue; +import org.apache.hadoop.hbase.KeyValueTestUtil; +import org.apache.hadoop.hbase.KeyValueUtil; +import org.apache.hadoop.hbase.io.ByteArrayOutputStream; +import org.apache.hadoop.hbase.io.hfile.HFileContext; +import org.apache.hadoop.hbase.io.hfile.HFileContextBuilder; +import org.apache.hadoop.hbase.testclassification.IOTests; +import org.apache.hadoop.hbase.testclassification.SmallTests; +import org.apache.hadoop.hbase.util.Bytes; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; + +@Tag(IOTests.TAG) +@Tag(SmallTests.TAG) +public class TestFastDiffEncodingState { + + private static final byte[] FAMILY = Bytes.toBytes("family"); + private static final byte[] QUALIFIER = Bytes.toBytes("qualifier"); + private static final byte[] VALUE_A = Bytes.toBytes("valueA"); + private static final byte[] VALUE_B = Bytes.toBytes("valueB"); + private static final byte[] CORRUPTED_VALUE = Bytes.toBytes("broken"); + private static final byte[] EMPTY_VALUE = new byte[0]; + + private final Configuration configuration = HBaseConfiguration.create(); + + private FastDiffDeltaEncoder encoder; + private HFileContext fileContext; + private HFileBlockDefaultEncodingContext context; + private ByteArrayOutputStream encodedBlock; + private DataOutputStream out; + + @BeforeEach + public void setUp() throws Exception { + initializeContext(new HFileContextBuilder().withIncludesMvcc(false).build()); + } + + @Test + public void testUnencodedCellSize() throws Exception { + KeyValue cell = createCell("row1", VALUE_A); + + encoder.encode(cell, context, out); + + int expectedSize = KeyValueUtil.keyLength(cell) + cell.getValueLength() + + KeyValue.KEYVALUE_INFRASTRUCTURE_SIZE; + assertEquals(expectedSize, getState().getUnencodedDataSizeWritten()); + } + + @Test + public void testFailedCellDoesNotCommitMaterializedValueOffset() throws Exception { + initializeContext(new HFileContextBuilder().withIncludesTags(true).build()); + KeyValue firstCell = createCell("row1", VALUE_A); + KeyValue failedCell = new KeyValue(Bytes.toBytes("row2"), FAMILY, QUALIFIER, VALUE_B) { + @Override + public int getTagsLength() { + throw new IllegalStateException("expected failure"); + } + }; + encoder.encode(firstCell, context, out); + + try { + encoder.encode(failedCell, context, out); + fail("Cell encoding should fail after writing its value"); + } catch (IllegalStateException e) { + assertEquals("expected failure", e.getMessage()); + } + + FastDiffEncodingState state = getState(); + beforeShipped(state); + + assertTrue(state.matchingPreviousValue(createCell("candidate", VALUE_A))); + } + + private void initializeContext(HFileContext newFileContext) throws Exception { + encoder = new FastDiffDeltaEncoder(); + fileContext = newFileContext; + context = new HFileBlockDefaultEncodingContext(configuration, DataBlockEncoding.FAST_DIFF, + HConstants.HFILEBLOCK_DUMMY_HEADER, fileContext); + encodedBlock = new ByteArrayOutputStream(); + encodedBlock.write(HConstants.HFILEBLOCK_DUMMY_HEADER); + out = new DataOutputStream(encodedBlock); + encoder.startBlockEncoding(context, out); + } + + @Test + public void testSameValueAfterShipped() throws Exception { + KeyValue firstCell = createCell("row1", VALUE_A); + KeyValue secondCell = createCell("row2", VALUE_A); + encoder.encode(firstCell, context, out); + encoder.encode(secondCell, context, out); + + FastDiffEncodingState state = shipAndCorrupt(secondCell); + + assertTrue(state.prevCell instanceof KeyValue.KeyOnlyKeyValue); + assertTrue(state.matchingPreviousValue(createCell("candidate", VALUE_A))); + } + + @Test + public void testDifferentValueAfterShipped() throws Exception { + KeyValue firstCell = createCell("row1", VALUE_A); + KeyValue secondCell = createCell("row2", VALUE_B); + encoder.encode(firstCell, context, out); + encoder.encode(secondCell, context, out); + + FastDiffEncodingState state = shipAndCorrupt(secondCell); + + assertTrue(state.prevCell instanceof KeyValue.KeyOnlyKeyValue); + assertTrue(state.matchingPreviousValue(createCell("candidate", VALUE_B))); + assertFalse(state.matchingPreviousValue(createCell("candidate", VALUE_A))); + } + + @Test + public void testRepeatedEncode() throws Exception { + byte[][] values = { VALUE_A, VALUE_A, VALUE_B, VALUE_B, EMPTY_VALUE, EMPTY_VALUE, VALUE_A, + VALUE_B }; + List expectedCells = new ArrayList<>(); + FastDiffEncodingState state = getState(); + + for (int i = 0; i < values.length; i++) { + KeyValue scannerCell = createCell(String.format("row%03d", i), values[i]); + expectedCells.add(KeyValueUtil.copyToNewKeyValue(scannerCell)); + encoder.encode(scannerCell, context, out); + beforeShipped(state); + if ((i & 1) == 0) { + beforeShipped(state); + } + Arrays.fill(scannerCell.getBuffer(), (byte) 0x5a); + } + + ByteBuffer expected = KeyValueTestUtil.toByteBufferAndRewind(expectedCells, false); + assertEquals(expected, finishAndDecode()); + } + + @Test + public void testByteBufferKeyValue() throws Exception { + KeyValue sourceCell = createCell("row1", VALUE_A); + ByteBuffer scannerBuffer = ByteBuffer.allocateDirect(sourceCell.getLength()); + scannerBuffer.put(sourceCell.getBuffer(), sourceCell.getOffset(), sourceCell.getLength()); + scannerBuffer.rewind(); + ByteBufferKeyValue scannerCell = + new ByteBufferKeyValue(scannerBuffer, 0, scannerBuffer.remaining()); + KeyValue expectedFirstCell = KeyValueUtil.copyToNewKeyValue(scannerCell); + KeyValue secondCell = createCell("row2", VALUE_A); + + encoder.encode(scannerCell, context, out); + FastDiffEncodingState state = getState(); + beforeShipped(state); + assertTrue(state.prevCell instanceof KeyValue.KeyOnlyKeyValue); + for (int i = 0; i < scannerBuffer.capacity(); i++) { + scannerBuffer.put(i, (byte) 0x5a); + } + encoder.encode(secondCell, context, out); + + List expectedCells = + Arrays.asList(expectedFirstCell, KeyValueUtil.copyToNewKeyValue(secondCell)); + ByteBuffer expected = KeyValueTestUtil.toByteBufferAndRewind(expectedCells, false); + assertEquals(expected, finishAndDecode()); + } + + private FastDiffEncodingState shipAndCorrupt(KeyValue cell) { + FastDiffEncodingState state = getState(); + beforeShipped(state); + Bytes.putBytes(cell.getValueArray(), cell.getValueOffset(), CORRUPTED_VALUE, 0, + cell.getValueLength()); + return state; + } + + private FastDiffEncodingState getState() { + return (FastDiffEncodingState) context.getEncodingState(); + } + + private void beforeShipped(FastDiffEncodingState state) { + state.beforeShipped(encodedBlock.getBuffer(), getStreamBaseOffset(), encodedBlock.size()); + } + + private int getStreamBaseOffset() { + return encodedBlock.size() - out.size(); + } + + private ByteBuffer finishAndDecode() throws Exception { + encoder.endBlockEncoding(context, out, encodedBlock.getBuffer()); + int encodedDataOffset = HConstants.HFILEBLOCK_HEADER_SIZE + DataBlockEncoding.ID_SIZE; + DataInputStream in = new DataInputStream(new ByteArrayInputStream(encodedBlock.getBuffer(), + encodedDataOffset, encodedBlock.size() - encodedDataOffset)); + ByteBuffer decoded = encoder.decodeKeyValues(in, + encoder.newDataBlockDecodingContext(configuration, fileContext)); + decoded.rewind(); + return decoded; + } + + private static KeyValue createCell(String row, byte[] value) { + return new KeyValue(Bytes.toBytes(row), FAMILY, QUALIFIER, value); + } +} diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileBlock.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileBlock.java index 087c619a7e99..d7a8603e3a6b 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileBlock.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileBlock.java @@ -822,7 +822,9 @@ private enum State { @Override public void beforeShipped() { if (getEncodingState() != null) { - getEncodingState().beforeShipped(); + int streamBaseOffset = baosInMemory.size() - userDataStream.size(); + getEncodingState().beforeShipped(baosInMemory.getBuffer(), streamBaseOffset, + baosInMemory.size()); } }